#5053

Committed 13 May 2026 12:00PM UTC coverage: 73.397% (+0.06%) from 73.338%

Build # #5053

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat: taosdump support stream backup/restore (#35326)

Coverage Stats

139 of 170 new or added lines in 3 files covered. (81.76%)

627 existing lines in 131 files now uncovered.

281694 of 383795 relevant lines covered (73.4%)

132505311.38 hits per line

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

55.88

/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 "clientSession.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 "tmisce.h"
#include "tmsg.h"
#include "tref.h"
#include "trpc.h"
#include "ttime.h"
#include "tversion.h"
#include "version.h"

#define CLIENT_CLEANUP_WAIT_TIMEOUT_MS 10000

#ifdef TAOSD_INTEGRATED
extern void shellStopDaemon();
#endif

static void instanceRpcGlobalCleanup(void);

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

static int32_t waitRefSetToBaseCount(int32_t rsetId, const char *name, int64_t startMs, int64_t timeoutMs) {
  if (rsetId < 0) {
    return TSDB_CODE_SUCCESS;
  }

  while (true) {
    int32_t count = 0;
    int32_t code = taosGetRefSetCount(rsetId, &count);
    if (code != TSDB_CODE_SUCCESS) {
      tscWarn("failed to inspect %s ref pool:%d before cleanup, code:%s", name, rsetId, tstrerror(code));
      return code;
    }

    if (count <= 1) {
      return TSDB_CODE_SUCCESS;
    }

    if (timeoutMs >= 0 && taosGetTimestampMs() - startMs >= timeoutMs) {
      tscWarn("timeout waiting for %s ref pool:%d to drain, count:%d", name, rsetId, count);
      return TSDB_CODE_TIMEOUT_ERROR;
    }

    taosMsleep(1);
  }
}

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);
}
#endif

int32_t tzInit() {
#if !defined(WINDOWS) && !defined(TD_ASTRA)
  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;
  }
#endif
  return 0;
}

void tzCleanup() {
#if !defined(WINDOWS) && !defined(TD_ASTRA)
  taosHashCleanup(pTimezoneMap);
  taosHashCleanup(pTimezoneNameMap);
#endif
}

#if !defined(WINDOWS) && !defined(TD_ASTRA)
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_CONNECTOR_INFO || option == TSDB_OPTION_CONNECTION_CLEAR) {
    if (val != NULL) {
      tstrncpy(pObj->optionInfo.cInfo, val, sizeof(pObj->optionInfo.cInfo));
    } else {
      pObj->optionInfo.cInfo[0] = 0;
    }
  }

  if (option == TSDB_OPTION_CONNECTION_USER_IP || option == TSDB_OPTION_CONNECTION_CLEAR) {
    SIpRange dualIp = {0};
    if (val != NULL) {
      pObj->optionInfo.userIp = taosInetAddr(val);
      SIpAddr addr = {0};
      code = taosGetIpFromFqdn(tsEnableIpv6, val, &addr);
      if (code == 0) {
        code = tIpStrToUint(&addr, &pObj->optionInfo.userDualIp);
      }
      if (code != 0) {
        tscError("ipv6 flag %d failed to convert user ip %s to dual ip since %s", tsEnableIpv6 ? 1 : 0, val,
                 tstrerror(code));
        pObj->optionInfo.userIp = INADDR_NONE;
        pObj->optionInfo.userDualIp = dualIp;
        code = 0;
      }
    } else {
      pObj->optionInfo.userIp = INADDR_NONE;
      pObj->optionInfo.userDualIp = dualIp;
    }
  }

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) {
  tscInfo("start to cleanup client environment");
  if (!beginAsyncWorkShutdown()) {
    return;
  }

  int64_t cleanupStartMs = taosGetTimestampMs();

  if (TSDB_CODE_SUCCESS !=
      waitRefSetToBaseCount(clientReqRefPool, "request", cleanupStartMs, CLIENT_CLEANUP_WAIT_TIMEOUT_MS)) {
    tscWarn("request ref pool did not drain cleanly before cleanup continues");
  }

  monitorClose();
  tscStopCrashReport();

  hbMgrCleanUp();

  catalogDestroy();
  schedulerDestroy();

  fmFuncMgtDestroy();
  qCleanupKeywordsTable();

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

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

  if (TSDB_CODE_SUCCESS !=
      waitRefSetToBaseCount(clientConnRefPool, "connection", cleanupStartMs, CLIENT_CLEANUP_WAIT_TIMEOUT_MS)) {
    tscWarn("connection ref pool did not drain cleanly before cleanup continues");
  }

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

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

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

  sessMgtDestroy();

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

void taos_set_option(OPTIONS *options, const char *key, const char *value) {
  if (options == NULL || key == NULL || value == NULL) {
    terrno = TSDB_CODE_INVALID_PARA;
    tscError("taos_set_option invalid parameter, options: %p, key: %p, value: %p", options, key, value);
    return;
  }

  size_t count = (size_t)options->count;
  size_t len = sizeof(options->keys) / sizeof(options->keys[0]);
  if (count >= len) {
    terrno = TSDB_CODE_INVALID_PARA;
    tscError("taos_set_option overflow, count: %zu, reached capacity: %zu", count, len);
    return;
  }

  options->keys[count] = key;
  options->values[count] = value;
  options->count = (uint16_t)(count + 1);
}

static int set_connection_option_or_close(TAOS *taos, TSDB_OPTION_CONNECTION option, const char *value) {
  if (value == NULL) return TSDB_CODE_SUCCESS;
  int code = taos_options_connection(taos, option, value);
  if (code != TSDB_CODE_SUCCESS) {
    tscError("failed to set option(%d): %s", (int)option, value);
    taos_close(taos);
    return code;
  }
  return TSDB_CODE_SUCCESS;
}

TAOS *taos_connect_with(const OPTIONS *options) {
  const char *ip = NULL;
  const char *user = NULL;
  const char *pass = NULL;
  const char *db = NULL;
  uint16_t    port = 0;

  const char *charset = NULL;
  const char *timezone = NULL;
  const char *userIp = NULL;
  const char *userApp = NULL;
  const char *connectorInfo = NULL;

  if (options && options->count > 0) {
    size_t count = (size_t)options->count;
    for (size_t i = 0; i < count; ++i) {
      const char *key = options->keys[i];
      const char *value = options->values[i];
      if (key == NULL || value == NULL) {
        tscWarn("taos_connect_with option key or value is NULL, index: %zu", i);
        continue;
      }

      if (strcmp(key, "ip") == 0) {
        ip = value;
      } else if (strcmp(key, "user") == 0) {
        user = value;
      } else if (strcmp(key, "pass") == 0) {
        pass = value;
      } else if (strcmp(key, "db") == 0) {
        db = value;
      } else if (strcmp(key, "port") == 0) {
        port = (uint16_t)taosStr2Int32(value, NULL, 10);
      } else if (strcmp(key, "charset") == 0) {
        charset = value;
      } else if (strcmp(key, "timezone") == 0) {
        timezone = value;
      } else if (strcmp(key, "userIp") == 0) {
        userIp = value;
      } else if (strcmp(key, "userApp") == 0) {
        userApp = value;
      } else if (strcmp(key, "connectorInfo") == 0) {
        connectorInfo = value;
      } else {
        tscWarn("taos_connect_with unknown option key: %s", key);
      }
    }
  }

  TAOS *taos = taos_connect(ip, user, pass, db, port);
  if (taos == NULL) return NULL;

  if (set_connection_option_or_close(taos, TSDB_OPTION_CONNECTION_CHARSET, charset) != TSDB_CODE_SUCCESS) return NULL;
  if (set_connection_option_or_close(taos, TSDB_OPTION_CONNECTION_TIMEZONE, timezone) != TSDB_CODE_SUCCESS) return NULL;
  if (set_connection_option_or_close(taos, TSDB_OPTION_CONNECTION_USER_IP, userIp) != TSDB_CODE_SUCCESS) return NULL;
  if (set_connection_option_or_close(taos, TSDB_OPTION_CONNECTION_USER_APP, userApp) != TSDB_CODE_SUCCESS) return NULL;
  if (set_connection_option_or_close(taos, TSDB_OPTION_CONNECTION_CONNECTOR_INFO, connectorInfo) != TSDB_CODE_SUCCESS)
    return NULL;

  return taos;
}

TAOS *taos_connect_with_dsn(const char *dsn) {
  terrno = TSDB_CODE_OPS_NOT_SUPPORT;
  tscError("taos_connect_with_dsn not supported");
  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;
    }
    case TAOS_NOTIFY_DATETIME_WHITELIST_VER: {
      TSC_ERR_RET(taosThreadMutexLock(&pObj->mutex));
      pObj->dateTimeWhiteListInfo.fp = fp;
      pObj->dateTimeWhiteListInfo.param = param;
      TSC_ERR_RET(taosThreadMutexUnlock(&pObj->mutex));
      break;
    }
    case TAOS_NOTIFY_TOKEN: {
      TSC_ERR_RET(taosThreadMutexLock(&pObj->mutex));
      pObj->tokenNotifyInfo.fp = fp;
      pObj->tokenNotifyInfo.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;
  }

  SGetUserIpWhiteListRsp wlRsp;
  if (TSDB_CODE_SUCCESS != tDeserializeSGetUserIpWhiteListRsp(pMsg->pData, pMsg->len, &wlRsp)) {
    taosMemoryFree(pMsg->pData);
    taosMemoryFree(pMsg->pEpSet);
    taosMemoryFree(pInfo);
    tFreeSGetUserIpWhiteListRsp(&wlRsp);
    return terrno;
  }

  uint64_t *pWhiteLists = taosMemoryMalloc(wlRsp.numWhiteLists * sizeof(uint64_t));
  if (pWhiteLists == NULL) {
    taosMemoryFree(pMsg->pData);
    taosMemoryFree(pMsg->pEpSet);
    taosMemoryFree(pInfo);
    tFreeSGetUserIpWhiteListRsp(&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);
  tFreeSGetUserIpWhiteListRsp(&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_IP_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;
}

typedef struct SFetchIpWhiteListInfo {
  int64_t connId;
  bool    supportNeg;
  void   *userParam;

  __taos_async_ip_whitelist_fn_t userCbFn;
} SFetchIpWhiteListInfo;

int32_t fetchIpWhiteListCallbackFn(void *param, SDataBuf *pMsg, int32_t code) {
  int32_t lino = 0;
  char  **pWhiteLists = NULL;

  SGetUserIpWhiteListRsp wlRsp = {0};

  SFetchIpWhiteListInfo *pInfo = (SFetchIpWhiteListInfo *)param;
  TAOS                  *taos = &pInfo->connId;

  if (code != TSDB_CODE_SUCCESS) {
    pInfo->userCbFn(pInfo->userParam, code, taos, 0, NULL);
    TAOS_CHECK_GOTO(code, &lino, _error);
  }

  if ((code = tDeserializeSGetUserIpWhiteListDualRsp(pMsg->pData, pMsg->len, &wlRsp)) != TSDB_CODE_SUCCESS) {
    TAOS_CHECK_GOTO(code, &lino, _error);
  }

  pWhiteLists = taosMemoryMalloc(wlRsp.numWhiteLists * sizeof(char *));
  if (pWhiteLists == NULL) {
    code = terrno;
    TAOS_CHECK_GOTO(code, &lino, _error);
  }

  int32_t numWhiteLists = 0;
  for (int32_t i = 0; i < wlRsp.numWhiteLists; i++) {
    SIpRange *pIpRange = &wlRsp.pWhiteListsDual[i];
    if (!pInfo->supportNeg && pIpRange->neg) {
      continue;
    }
    SIpAddr ipAddr = {0};

    code = tIpUintToStr(pIpRange, &ipAddr);
    TAOS_CHECK_GOTO(code, &lino, _error);

    char *ip = taosMemCalloc(1, IP_RESERVE_CAP);
    if (ip == NULL) {
      code = terrno;
      TAOS_CHECK_GOTO(code, &lino, _error);
    }
    if (ipAddr.type == 0) {
      if (pInfo->supportNeg) {
        snprintf(ip, IP_RESERVE_CAP, "%c %s/%d", pIpRange->neg ? '-' : '+', ipAddr.ipv4, ipAddr.mask);
      } else {
        snprintf(ip, IP_RESERVE_CAP, "%s/%d", ipAddr.ipv4, ipAddr.mask);
      }
    } else {
      if (ipAddr.ipv6[0] == 0) {
        memcpy(ipAddr.ipv6, "::", 2);
      }
      if (pInfo->supportNeg) {
        snprintf(ip, IP_RESERVE_CAP, "%c %s/%d", pIpRange->neg ? '-' : '+', ipAddr.ipv6, ipAddr.mask);
      } else {
        snprintf(ip, IP_RESERVE_CAP, "%s/%d", ipAddr.ipv6, ipAddr.mask);
      }
    }
    pWhiteLists[numWhiteLists++] = ip;
  }

  pInfo->userCbFn(pInfo->userParam, code, taos, numWhiteLists, pWhiteLists);
_error:
  if (pWhiteLists != NULL) {
    for (int32_t i = 0; i < numWhiteLists; i++) {
      taosMemFree(pWhiteLists[i]);
    }
    taosMemoryFree(pWhiteLists);
  }
  taosMemoryFree(pMsg->pData);
  taosMemoryFree(pMsg->pEpSet);
  taosMemoryFree(pInfo);
  tFreeSGetUserIpWhiteListDualRsp(&wlRsp);
  return code;
}

static void taosFetchIpWhiteList(TAOS *taos, __taos_async_whitelist_dual_stack_fn_t fp, void *param, bool supportNeg) {
  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;
  }

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

  pParam->connId = connId;
  pParam->supportNeg = supportNeg;
  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 = fetchIpWhiteListCallbackFn;
  pSendInfo->msgType = TDMT_MND_GET_USER_IP_WHITELIST_DUAL;

  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_fetch_whitelist_dual_stack_a(TAOS *taos, __taos_async_whitelist_dual_stack_fn_t fp, void *param) {
  taosFetchIpWhiteList(taos, fp, param, false);
}

void taos_fetch_ip_whitelist_a(TAOS *taos, __taos_async_ip_whitelist_fn_t fp, void *param) {
  taosFetchIpWhiteList(taos, fp, param, true);
}

typedef struct SFetchDateTimeWhiteListInfo {
  int64_t                              connId;
  void                                *userParam;
  __taos_async_datetime_whitelist_fn_t userCbFn;
} SFetchDateTimeWhiteListInfo;

static const char *weekdays[] = {"SUN", "MON", "TUE", "WED", "THU", "FRI", "SAT"};
int32_t            fetchDateTimeWhiteListCallbackFn(void *param, SDataBuf *pMsg, int32_t code) {
  int32_t lino = 0;
  char  **pWhiteLists = NULL;

  SUserDateTimeWhiteList wlRsp = {0};

  SFetchDateTimeWhiteListInfo *pInfo = (SFetchDateTimeWhiteListInfo *)param;
  TAOS                        *taos = &pInfo->connId;

  if (code != TSDB_CODE_SUCCESS) {
    pInfo->userCbFn(pInfo->userParam, code, taos, 0, NULL);
    TAOS_CHECK_GOTO(code, &lino, _error);
  }

  if ((code = tDeserializeSUserDateTimeWhiteList(pMsg->pData, pMsg->len, &wlRsp)) != TSDB_CODE_SUCCESS) {
    TAOS_CHECK_GOTO(code, &lino, _error);
  }

  pWhiteLists = taosMemoryMalloc(wlRsp.numWhiteLists * sizeof(char *));
  if (pWhiteLists == NULL) {
    code = terrno;
    TAOS_CHECK_GOTO(code, &lino, _error);
  }

  int32_t numWhiteLists = 0;
  for (int32_t i = 0; i < wlRsp.numWhiteLists; i++) {
    SDateTimeWhiteListItem *item = &wlRsp.pWhiteLists[i];

    char *p = taosMemCalloc(1, 128);
    if (p == NULL) {
      code = terrno;
      TAOS_CHECK_GOTO(code, &lino, _error);
    }

    int duration = item->duration / 60;

    if (item->absolute) {
      struct STm tm;
      (void)taosTs2Tm(item->start, TSDB_TIME_PRECISION_SECONDS, &tm, NULL);
      snprintf(p, 128, "%c %04d-%02d-%02d %02d:%02d %d", item->neg ? '-' : '+', tm.tm.tm_year + 1900, tm.tm.tm_mon + 1,
                          tm.tm.tm_mday, tm.tm.tm_hour, tm.tm.tm_min, duration);
    } else {
      int day = item->start / 86400;
      int hour = (item->start % 86400) / 3600;
      int minute = (item->start % 3600) / 60;
      snprintf(p, 128, "%c %s %02d:%02d %d", item->neg ? '-' : '+', weekdays[day], hour, minute, duration);
    }
    pWhiteLists[numWhiteLists++] = p;
  }

  pInfo->userCbFn(pInfo->userParam, code, taos, numWhiteLists, pWhiteLists);
_error:
  if (pWhiteLists != NULL) {
    for (int32_t i = 0; i < numWhiteLists; i++) {
      taosMemFree(pWhiteLists[i]);
    }
    taosMemoryFree(pWhiteLists);
  }
  taosMemoryFree(pMsg->pData);
  taosMemoryFree(pMsg->pEpSet);
  taosMemoryFree(pInfo);
  tFreeSUserDateTimeWhiteList(&wlRsp);
  return code;
}

void taos_fetch_datetime_whitelist_a(TAOS *taos, __taos_async_datetime_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;
  }

  SFetchDateTimeWhiteListInfo *pParam = taosMemoryMalloc(sizeof(SFetchDateTimeWhiteListInfo));
  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 = fetchDateTimeWhiteListCallbackFn;
  pSendInfo->msgType = TDMT_MND_GET_USER_DATETIME_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;
  }
  int32_t code = 0;

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

  SSessParam para = {.type = SESSION_PER_USER, .value = -1, .noCheck = 1};

  code = tscUpdateSessMetric(pTscObj, &para);
  if (code != TSDB_CODE_SUCCESS) {
    tscWarn("conn:0x%" PRIx64 ", failed to update user:%s metric when close connection, code:%d", pTscObj->id,
            pTscObj->user, code);
  }

  code = tscUnrefSessMetric(pTscObj);
  if (TSDB_CODE_SUCCESS != taosRemoveRef(clientConnRefPool, pTscObj->id)) {
    tscError("conn: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)) {
    if (*(taosGetErrMsg()) == 0) {
      return (const char *)tstrerror(terrno);
    } else {
      (void)snprintf(taosGetErrMsgReturn(), ERR_MSG_LEN, "%s", taosGetErrMsg());
      return (const char *)taosGetErrMsgReturn();
    }
  }

  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, res:%p", pRequest->requestId, res);
    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->killed) {
      tscInfo("query has been killed, can not fetch more row.");
      pRequest->code = TSDB_CODE_TSC_QUERY_KILLED;
      CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_FETCH_RETURNED);
      return NULL;
    }

    if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT || pRequest->type == TSDB_SQL_INSERT ||
        pRequest->code != TSDB_CODE_SUCCESS || taos_num_fields(res) == 0) {
      if (pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT) {
        CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_FETCH_RETURNED);
      }
      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);
      pResultInfo->current += 1;
      return pResultInfo->row;
    } else {
      if (tmqGetNextResInfo(res, true, &pResultInfo) != 0) {
        return NULL;
      }

      doSetOneRowPtr(pResultInfo);
      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 += snprintf(str + len, size - len, "%s", TSDB_DATA_NULL_STR);
      continue;
    }

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

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

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

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

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

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

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

      case TSDB_DATA_TYPE_UBIGINT:
        len += snprintf(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_BLOB:
      case TSDB_DATA_TYPE_MEDIUMBLOB: {
        void    *data = NULL;
        uint32_t tmp = 0;
        int32_t  charLen = blobDataLen((char *)row[i] - BLOBSTR_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_TIMESTAMP:
        len += snprintf(str + len, size - len, "%" PRId64, *((int64_t *)row[i]));
        break;

      case TSDB_DATA_TYPE_BOOL:
        len += snprintf(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, 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) {
      CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_FETCH_RETURNED);
      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) {
    if (pRequest->code == TSDB_SQL_RETRIEVE_EMPTY_RESULT) {
      CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_FETCH_RETURNED);
    }
    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 NULL;
  }

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

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

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

// buffer is allocated by caller, len is in/out parameter, input is buffer length, output is actual length.
// because this is a general purpose api, buffer is not null-terminated string even for string info, and
// the return length is the actual length of the info, not including null-terminator.
int taos_get_connection_info(TAOS *taos, TSDB_CONNECTION_INFO info, char *buffer, int *len) {
  if (len == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }

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

  int code = TSDB_CODE_SUCCESS;
  (void)taosThreadMutexLock(&pTscObj->mutex);

  switch (info) {
    case TSDB_CONNECTION_INFO_USER: {
      int userLen = strlen(pTscObj->user);
      if (buffer == NULL || *len < userLen) {
        *len = userLen;
        TSC_ERR_JRET(TSDB_CODE_INVALID_PARA);
      } else {
        *len = userLen;
        (void)memcpy(buffer, pTscObj->user, userLen);
      }
      break;
    }

    case TSDB_CONNECTION_INFO_TOKEN: {
      int tokenLen = strlen(pTscObj->tokenName);
      if (tokenLen == 0) {
        *len = 0;
      } else if (buffer == NULL || *len < tokenLen) {
        *len = tokenLen;
        TSC_ERR_JRET(TSDB_CODE_INVALID_PARA);
      } else {
        *len = tokenLen;
        (void)memcpy(buffer, pTscObj->tokenName, tokenLen);
      }
      break;
    }

    default:
      TSC_ERR_JRET(TSDB_CODE_INVALID_PARA);
  }

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

  if (TSDB_CODE_SUCCESS == code) {
    code = sqlSecurityCheckASTLevel(pRequest, 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->pVStbRefDbs = taosArrayDup(pSrc->pVStbRefDbs, 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 (nodeType(pQuery->pRoot) == QUERY_NODE_DELETE_STMT) {
        pRequest->secureDelete = ((SDeleteStmt *)pQuery->pRoot)->secureDelete;
      }
    }

    if (pQuery->haveResultSet) {
      code = setResSchemaInfo(&pRequest->body.resInfo, pQuery->pResSchema, pQuery->numOfResCols, pQuery->pResExtSchema,
                              pRequest->stmtBindVersion > 0);
      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) && pRequest->stmtBindVersion == 0) {
      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;
  SRequestObj *pRequest = pWrapper->pRequest;

  switch (pRequest->pQuery->execStage) {
    case QUERY_EXEC_STAGE_PARSE: {
      CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_CATALOG);
      code = getAllMetaAsync(pWrapper, doAsyncQueryFromParse);
      break;
    }
    case QUERY_EXEC_STAGE_ANALYSE: {
      CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_CATALOG);
      code = getAllMetaAsync(pWrapper, doAsyncQueryFromAnalyse);
      break;
    }
    case QUERY_EXEC_STAGE_SCHEDULE: {
      CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_SCHEDULE);
      launchAsyncQuery(pRequest, 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 ", continue parse query, 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;
  taosAsyncQueryImpl(connId, sql, fp, param, false, TD_REQ_FROM_APP);
}

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,
                           .userId = pTscObj->userId,
                           .pEffectiveUser = pRequest->effectiveUser,
                           .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
                           .enableSysInfo = pTscObj->sysInfo,
                           .privInfo = pWrapper->pParseCtx ? pWrapper->pParseCtx->privInfo : 0,
                           .sodInitial = pTscObj->pAppInfo->serverCfg.sodInitial,
                           .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};
  int8_t biMode = atomic_load_8(&((STscObj *)pTscObj)->biMode);
  (*pCxt)->biMode = biMode;
  (*pCxt)->minSecLevel = pTscObj->minSecLevel;
  (*pCxt)->maxSecLevel = pTscObj->maxSecLevel;
  (*pCxt)->macMode = pTscObj->pAppInfo->serverCfg.macActive;
  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;

  CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_PARSE);

  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) {
    if (NULL != pRequest->msgBuf && strlen(pRequest->msgBuf) > 0) {
      tscError("req:0x%" PRIx64 ", error happens, code:%d - %s, QID:0x%" PRIx64, pRequest->self, code, pRequest->msgBuf,
               pRequest->requestId);
    } else {
      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) && pRequest->stmtBindVersion == 0) {
      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;

  // Each fetch call sets phase to IN_PROGRESS

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

  SAsyncFetchParam *pParam = taosMemoryCalloc(1, sizeof(SAsyncFetchParam));
  if (!pParam) {
    CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_FETCH_RETURNED);
    fp(param, res, terrno);
    return;
  }

  CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_FETCH_IN_PROGRESS);
  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;

  CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_FETCH_IN_PROGRESS);
  // 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;
  int32_t    code = TSDB_CODE_SUCCESS;
  STMT2_DLOG_E("start to bind param");

  // check query bind number
  bool isQuery = (STMT_TYPE_QUERY == pStmt->sql.type || (pStmt->sql.type == 0 && stmt2IsSelect(stmt)));
  if (isQuery) {
    if (bindv->count != 1 || bindv->bind_cols[0]->num != 1) {
      terrno = TSDB_CODE_TSC_STMT_BIND_NUMBER_ERROR;
      STMT2_ELOG_E("query only support one table and one row bind");
      return terrno;
    }
  }

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

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

  for (int i = 0; i < bindv->count; ++i) {
    SVCreateTbReq *pCreateTbReq = NULL;
    if (!isQuery) {
      STMT2_TLOG("start to bind %dth table", i);
      if (bindv->tbnames && bindv->tbnames[i]) {
        code = stmtSetTbName2(stmt, bindv->tbnames[i]);
        if (code) {
          terrno = code;
          STMT2_ELOG("set tbname failed, code:%s", stmt2Errstr(stmt));
          return terrno;
        }
      }

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

      if (code) {
        terrno = code;
        STMT2_ELOG("set tags failed, code:%s", stmt2Errstr(stmt));
        if (pCreateTbReq) {
          tdDestroySVCreateTbReq(pCreateTbReq);
          taosMemoryFreeClear(pCreateTbReq);
        }
        return terrno;
      }
    }

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

      if (bind->num <= 0 || bind->num > INT16_MAX) {
        STMT2_ELOG("bind num:%d must > 0 and < INT16_MAX", bind->num);
        code = terrno = TSDB_CODE_TSC_STMT_BIND_NUMBER_ERROR;
        if (pCreateTbReq) {
          tdDestroySVCreateTbReq(pCreateTbReq);
          taosMemoryFreeClear(pCreateTbReq);
        }
        return terrno;
      }

      code = stmtBindBatch2(stmt, bind, col_idx, pCreateTbReq);
      if (TSDB_CODE_SUCCESS != code) {
        terrno = code;
        STMT2_ELOG("bind batch failed, code:%s", stmt2Errstr(stmt));
        if (pCreateTbReq) {
          tdDestroySVCreateTbReq(pCreateTbReq);
          taosMemoryFreeClear(pCreateTbReq);
        }
        return terrno;
      }
    }
  }

  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");
    terrno = TSDB_CODE_TSC_STMT_API_ERROR;
    return terrno;
  }
  (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) {
    terrno = code_s;
    (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;
  }
  *insert = stmt2IsInsert(stmt);
  return TSDB_CODE_SUCCESS;
}

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;
  STMT2_DLOG_E("start to get fields");

  if (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type ||
      (pStmt->sql.type == 0 && stmt2IsInsert(stmt))) {
    return stmtGetStbColFields2(stmt, count, fields);
  }
  if (STMT_TYPE_QUERY == pStmt->sql.type || (pStmt->sql.type == 0 && stmt2IsSelect(stmt))) {
    return stmtGetParamNum2(stmt, count, fields);
  }

  tscError("Invalid sql for stmt %s", pStmt->sql.sqlStr);
  return TSDB_CODE_PAR_SYNTAX_ERROR;
}

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 *)stmt2Errstr(stmt); }

int taos_set_conn_mode(TAOS *taos, int mode, int value) {
  int32_t code = 0;
  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.");
      code = TSDB_CODE_INVALID_PARA;
  }
  releaseTscObj(*(int64_t *)taos);
  return code;
}

char *getBuildInfo() { return td_buildinfo; }

int32_t taos_connect_is_alive(TAOS *taos) {
  int32_t code = 0, lino = 0;
  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;
  }

  code = tscCheckConnSessionMetric(pObj);
  TAOS_CHECK_GOTO(code, &lino, _error);

_error:
  releaseTscObj(*(int64_t *)taos);

  if (code != 0) {
    tscError("taos conn failed to check alive, code:%d - %s", code, tstrerror(code));
  }

  return code != 0 ? 0 : 1;
}
static int32_t buildInstanceRegisterSql(const SInstanceRegisterReq *req, char **ppSql, uint32_t *pLen) {
  const char *action = (req->expire < 0) ? "UNREGISTER" : "REGISTER";
  int32_t     len = 0;

  len += snprintf(NULL, 0, "%s INSTANCE '%s'", action, req->id);
  if (req->type[0] != 0) {
    len += snprintf(NULL, 0, " TYPE '%s'", req->type);
  }
  if (req->desc[0] != 0) {
    len += snprintf(NULL, 0, " DESC '%s'", req->desc);
  }
  if (req->expire >= 0) {
    len += snprintf(NULL, 0, " EXPIRE %d", req->expire);
  }

  char *sql = taosMemoryMalloc((size_t)len + 1);
  if (sql == NULL) {
    return terrno;
  }

  int32_t offset = snprintf(sql, (size_t)len + 1, "%s INSTANCE '%s'", action, req->id);
  if (req->type[0] != 0) {
    offset += snprintf(sql + offset, (size_t)len + 1 - (size_t)offset, " TYPE '%s'", req->type);
  }
  if (req->desc[0] != 0) {
    offset += snprintf(sql + offset, (size_t)len + 1 - (size_t)offset, " DESC '%s'", req->desc);
  }
  if (req->expire >= 0) {
    (void)snprintf(sql + offset, (size_t)len + 1 - (size_t)offset, " EXPIRE %d", req->expire);
  }

  *ppSql = sql;
  if (pLen != NULL) {
    *pLen = (uint32_t)len;
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t sendInstanceRegisterReq(STscObj *pObj, const SInstanceRegisterReq *req) {
  SRequestObj *pRequest = NULL;
  int32_t      code = createRequest(pObj->id, TDMT_MND_REGISTER_INSTANCE, 0, &pRequest);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return code;
  }

  code = buildInstanceRegisterSql(req, &pRequest->sqlstr, (uint32_t *)&pRequest->sqlLen);
  if (code != TSDB_CODE_SUCCESS) {
    goto _cleanup;
  }

  int32_t msgLen = tSerializeSInstanceRegisterReq(NULL, 0, (SInstanceRegisterReq *)req);
  if (msgLen <= 0) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    goto _cleanup;
  }

  void *pMsg = taosMemoryMalloc(msgLen);
  if (pMsg == NULL) {
    code = terrno != 0 ? terrno : TSDB_CODE_OUT_OF_MEMORY;
    goto _cleanup;
  }

  if (tSerializeSInstanceRegisterReq(pMsg, msgLen, (SInstanceRegisterReq *)req) < 0) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    taosMemoryFree(pMsg);
    goto _cleanup;
  }

  pRequest->type = TDMT_MND_REGISTER_INSTANCE;
  pRequest->body.requestMsg = (SDataBuf){.pData = pMsg, .len = msgLen, .handle = NULL};

  SMsgSendInfo *pSend = buildMsgInfoImpl(pRequest);
  if (pSend == NULL) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    taosMemoryFree(pMsg);
    pRequest->body.requestMsg.pData = NULL;
    goto _cleanup;
  }

  SEpSet epSet = getEpSet_s(&pObj->pAppInfo->mgmtEp);
  code = asyncSendMsgToServer(pObj->pAppInfo->pTransporter, &epSet, NULL, pSend);
  if (code != TSDB_CODE_SUCCESS) {
    destroySendMsgInfo(pSend);
    pRequest->body.requestMsg = (SDataBuf){0};
    goto _cleanup;
  }

  code = tsem_wait(&pRequest->body.rspSem);
  if (code != TSDB_CODE_SUCCESS) {
    code = terrno != 0 ? terrno : code;
    goto _cleanup;
  }

  code = pRequest->code;
  terrno = code;

_cleanup:
  destroyRequest(pRequest);
  return code;
}

static bool instanceRegisterRpcRfp(int32_t code, tmsg_t msgType) {
  if (NEED_REDIRECT_ERROR(code)) {
    return true;
  } else if (code == TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY || code == TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE ||
             code == TSDB_CODE_SYN_WRITE_STALL || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
             code == TSDB_CODE_SYN_RESTORING) {
    tscDebug("client msg type %s should retry since %s", TMSG_INFO(msgType), tstrerror(code));
    return true;
  } else {
    return false;
  }
}

/** Build epSet from firstEp and secondEp in config. pCfg must have valid firstEp. */
static int32_t instanceBuildEpSetFromCfg(SConfig *pCfg, SEpSet *pEpSet) {
  SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp");
  if (pFirstEpItem == NULL || pFirstEpItem->str == NULL || pFirstEpItem->str[0] == 0) {
    return TSDB_CODE_CFG_NOT_FOUND;
  }
  SEp     firstEp = {0};
  int32_t code = taosGetFqdnPortFromEp(pFirstEpItem->str, &firstEp);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
  pEpSet->inUse = 0;
  pEpSet->numOfEps = 1;
  tstrncpy(pEpSet->eps[0].fqdn, firstEp.fqdn, TSDB_FQDN_LEN);
  pEpSet->eps[0].port = firstEp.port;

  SConfigItem *pSecondEpItem = cfgGetItem(pCfg, "secondEp");
  if (pSecondEpItem != NULL && pSecondEpItem->str != NULL && pSecondEpItem->str[0] != 0) {
    SEp secondEp = {0};
    if (taosGetFqdnPortFromEp(pSecondEpItem->str, &secondEp) == TSDB_CODE_SUCCESS) {
      tstrncpy(pEpSet->eps[1].fqdn, secondEp.fqdn, TSDB_FQDN_LEN);
      pEpSet->eps[1].port = secondEp.port;
      pEpSet->numOfEps = 2;
    }
  }
  return TSDB_CODE_SUCCESS;
}

/** Init and open instance RPC client. label e.g. "INST" or "LIST". Returns handle or NULL. */
static void *instanceOpenRpcClient(const char *label) {
  SRpcInit rpcInit = {0};
  rpcInit.label = (char *)label;
  rpcInit.numOfThreads = 1;
  rpcInit.cfp = NULL;
  rpcInit.sessions = 16;
  rpcInit.connType = TAOS_CONN_CLIENT;
  rpcInit.idleTime = tsShellActivityTimer * 1000;
  rpcInit.compressSize = tsCompressMsgSize;
  rpcInit.user = TSDB_DEFAULT_USER;
  rpcInit.rfp = instanceRegisterRpcRfp;
  rpcInit.retryMinInterval = tsRedirectPeriod;
  rpcInit.retryStepFactor = tsRedirectFactor;
  rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
  rpcInit.retryMaxTimeout = tsMaxRetryWaitTime;

  int32_t connLimitNum = tsNumOfRpcSessions / (tsNumOfRpcThreads * 3);
  connLimitNum = TMAX(connLimitNum, 10);
  connLimitNum = TMIN(connLimitNum, 500);
  rpcInit.connLimitNum = connLimitNum;
  rpcInit.timeToGetConn = tsTimeToGetAvailableConn;
  rpcInit.readTimeout = tsReadTimeout;
  rpcInit.ipv6 = tsEnableIpv6;
  rpcInit.enableSSL = tsEnableTLS;

  memcpy(rpcInit.caPath, tsTLSCaPath, strlen(tsTLSCaPath));
  memcpy(rpcInit.certPath, tsTLSSvrCertPath, strlen(tsTLSSvrCertPath));
  memcpy(rpcInit.keyPath, tsTLSSvrKeyPath, strlen(tsTLSSvrKeyPath));
  memcpy(rpcInit.cliCertPath, tsTLSCliCertPath, strlen(tsTLSCliCertPath));
  memcpy(rpcInit.cliKeyPath, tsTLSCliKeyPath, strlen(tsTLSCliKeyPath));

  int32_t code = taosVersionStrToInt(td_version, &rpcInit.compatibilityVer);
  if (code != TSDB_CODE_SUCCESS) {
    tscError("failed to convert taos version from str to int, errcode:%s", terrstr(code));
    terrno = code;
    return NULL;
  }

  void *clientRpc = rpcOpen(&rpcInit);
  if (clientRpc == NULL) {
    tscError("failed to init instance rpc client since %s", tstrerror(terrno));
  }
  return clientRpc;
}

/** Client-side rate limit: fixed 100 calls per 1s (register + list combined, protect mnode). */
#define TSC_INSTANCE_API_RL_WINDOW_MS   1000
#define TSC_INSTANCE_API_RL_MAX_PER_SEC 100

static TdThreadOnce  gInstRlOnce = PTHREAD_ONCE_INIT;
static TdThreadMutex gInstRlMutex;
static int32_t       gInstRlMutexInited = 0;
static int64_t       gInstRlWindowStartMs = 0;
static int32_t       gInstRlCountInWindow = 0;

static void instRlMutexInit(void) {
  if (taosThreadMutexInit(&gInstRlMutex, NULL) == TSDB_CODE_SUCCESS) {
    gInstRlMutexInited = 1;
  }
}

/** Call before instance RPC; shared by register and list. */
static int32_t instanceApiRateLimitTry(void) {
  int32_t c = taosThreadOnce(&gInstRlOnce, instRlMutexInit);
  if (c != TSDB_CODE_SUCCESS) {
    terrno = c;
    return c;
  }
  if (!gInstRlMutexInited) {
    tscError("instance API rate limiter init failed, block request");
    terrno = TSDB_CODE_TSC_INTERNAL_ERROR;
    return TSDB_CODE_TSC_INTERNAL_ERROR;
  }

  int64_t now = taosGetTimestampMs();
  (void)taosThreadMutexLock(&gInstRlMutex);
  /* Reset window if first use, window elapsed, or clock moved backwards (NTP). */
  if (gInstRlWindowStartMs == 0 || now < gInstRlWindowStartMs ||
      (now - gInstRlWindowStartMs) >= (int64_t)TSC_INSTANCE_API_RL_WINDOW_MS) {
    gInstRlWindowStartMs = now;
    gInstRlCountInWindow = 0;
  }
  if (gInstRlCountInWindow >= TSC_INSTANCE_API_RL_MAX_PER_SEC) {
    (void)taosThreadMutexUnlock(&gInstRlMutex);
    tscWarn("instance API rate limit exceeded (max %d calls per %d ms, register and list combined)",
            TSC_INSTANCE_API_RL_MAX_PER_SEC, TSC_INSTANCE_API_RL_WINDOW_MS);
    terrno = TSDB_CODE_TSC_INSTANCE_API_RATE_LIMIT;
    return TSDB_CODE_TSC_INSTANCE_API_RATE_LIMIT;
  }
  gInstRlCountInWindow++;
  (void)taosThreadMutexUnlock(&gInstRlMutex);
  return TSDB_CODE_SUCCESS;
}

/** Process-wide singleton: connectionless instance APIs share one rpcOpen; closed in taos_cleanup. */
static TdThreadOnce     gInstRpcOnce = PTHREAD_ONCE_INIT;
static TdThreadMutex    gInstRpcMutex;
static TdThreadCond     gInstRpcCond;
static volatile int32_t gInstRpcMutexReady = 0;
static volatile int32_t gInstRpcCondReady = 0;
static void            *gInstRpc = NULL;
static int32_t          gInstRpcRef = 0;
static int32_t          gInstRpcClosing = 0;

static void instRpcMutexInit(void) {
  if (taosThreadMutexInit(&gInstRpcMutex, NULL) == TSDB_CODE_SUCCESS) {
    if (taosThreadCondInit(&gInstRpcCond, NULL) == TSDB_CODE_SUCCESS) {
      gInstRpcCondReady = 1;
      gInstRpcMutexReady = 1;
      return;
    }
    (void)taosThreadMutexDestroy(&gInstRpcMutex);
  }
}

static int32_t instanceRpcAcquire(void **ppRpc) {
  int32_t code = taosThreadOnce(&gInstRpcOnce, instRpcMutexInit);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return code;
  }
  if (!gInstRpcMutexReady || !gInstRpcCondReady) {
    tscError("instance RPC singleton not ready, block request");
    terrno = TSDB_CODE_TSC_INTERNAL_ERROR;
    return TSDB_CODE_TSC_INTERNAL_ERROR;
  }
  code = taosThreadMutexLock(&gInstRpcMutex);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
  if (gInstRpcClosing) {
    (void)taosThreadMutexUnlock(&gInstRpcMutex);
    return TSDB_CODE_TSC_INTERNAL_ERROR;
  }
  if (gInstRpc == NULL) {
    gInstRpc = instanceOpenRpcClient("INST");
    if (gInstRpc == NULL) {
      code = terrno;
      (void)taosThreadMutexUnlock(&gInstRpcMutex);
      return code;
    }
    tscInfo("instance RPC singleton opened, handle:%p (search this line to count rpcOpen)", gInstRpc);
  }
  gInstRpcRef++;
  *ppRpc = gInstRpc;
  (void)taosThreadMutexUnlock(&gInstRpcMutex);
  return TSDB_CODE_SUCCESS;
}

static void instanceRpcRelease(void) {
  if (!gInstRpcMutexReady || !gInstRpcCondReady) {
    return;
  }
  (void)taosThreadMutexLock(&gInstRpcMutex);
  if (gInstRpcRef > 0) {
    gInstRpcRef--;
    if (gInstRpcClosing && gInstRpcRef == 0) {
      (void)taosThreadCondSignal(&gInstRpcCond);
    }
  }
  (void)taosThreadMutexUnlock(&gInstRpcMutex);
}

static void instanceRpcGlobalCleanup(void) {
  if (!gInstRpcMutexReady || !gInstRpcCondReady) {
    return;
  }
  (void)taosThreadMutexLock(&gInstRpcMutex);
  gInstRpcClosing = 1;
  while (gInstRpcRef > 0) {
    (void)taosThreadCondWait(&gInstRpcCond, &gInstRpcMutex);
  }
  if (gInstRpc != NULL) {
    tscInfo("instance RPC singleton closing, handle:%p (search this line to count rpcClose)", gInstRpc);
    rpcClose(gInstRpc);
    gInstRpc = NULL;
  }
  gInstRpcCondReady = 0;
  gInstRpcMutexReady = 0;
  (void)taosThreadMutexUnlock(&gInstRpcMutex);
}

int32_t taos_register_instance(const char *id, const char *type, const char *desc, int32_t expire) {
  if (id == NULL || id[0] == 0) {
    return terrno = TSDB_CODE_INVALID_PARA;
  }

  // Validate string lengths
  size_t idLen = strlen(id);
  if (idLen >= TSDB_INSTANCE_ID_LEN) {
    tscError("instance id length %zu exceeds limit %d", idLen, TSDB_INSTANCE_ID_LEN - 1);
    return terrno = TSDB_CODE_INVALID_PARA;
  }

  if (type != NULL && type[0] != 0) {
    size_t typeLen = strlen(type);
    if (typeLen >= TSDB_INSTANCE_TYPE_LEN) {
      tscError("instance type length %zu exceeds limit %d", typeLen, TSDB_INSTANCE_TYPE_LEN - 1);
      return terrno = TSDB_CODE_INVALID_PARA;
    }
  }

  if (desc != NULL && desc[0] != 0) {
    size_t descLen = strlen(desc);
    if (descLen >= TSDB_INSTANCE_DESC_LEN) {
      tscError("instance desc length %zu exceeds limit %d", descLen, TSDB_INSTANCE_DESC_LEN - 1);
      return terrno = TSDB_CODE_INVALID_PARA;
    }
  }

  int32_t code = taos_init();
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  SConfig *pCfg = taosGetCfg();
  if (pCfg == NULL) {
    return terrno = TSDB_CODE_CFG_NOT_FOUND;
  }

  SEpSet epSet = {0};
  code = instanceBuildEpSetFromCfg(pCfg, &epSet);
  if (code != TSDB_CODE_SUCCESS) {
    return terrno = code;
  }

  code = instanceApiRateLimitTry();
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  void *clientRpc = NULL;
  code = instanceRpcAcquire(&clientRpc);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return code;
  }

  SRpcMsg rpcMsg = {0};
  SRpcMsg rpcRsp = {0};

  // Prepare request
  SInstanceRegisterReq req = {0};
  tstrncpy(req.id, id, sizeof(req.id));
  if (type != NULL && type[0] != 0) {
    tstrncpy(req.type, type, sizeof(req.type));
  }
  if (desc != NULL && desc[0] != 0) {
    tstrncpy(req.desc, desc, sizeof(req.desc));
  }
  req.expire = expire;

  int32_t contLen = tSerializeSInstanceRegisterReq(NULL, 0, &req);
  if (contLen <= 0) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    goto _register_inst_end;
  }

  void *pCont = rpcMallocCont(contLen);
  if (pCont == NULL) {
    code = terrno != 0 ? terrno : TSDB_CODE_OUT_OF_MEMORY;
    goto _register_inst_end;
  }

  if (tSerializeSInstanceRegisterReq(pCont, contLen, &req) < 0) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    rpcFreeCont(pCont);
    goto _register_inst_end;
  }

  rpcMsg.pCont = pCont;
  rpcMsg.contLen = contLen;
  rpcMsg.msgType = TDMT_MND_REGISTER_INSTANCE;
  rpcMsg.info.ahandle = (void *)0x9528;  // Different magic number from server status
  rpcMsg.info.notFreeAhandle = 1;

  code = rpcSendRecv(clientRpc, &epSet, &rpcMsg, &rpcRsp);
  if (TSDB_CODE_SUCCESS != code) {
    tscError("failed to send instance register req since %s", tstrerror(code));
    goto _register_inst_end;
  }

  if (rpcRsp.code != 0) {
    code = rpcRsp.code;
    tscError("instance register failed, code:%s", tstrerror(code));
  } else {
    code = TSDB_CODE_SUCCESS;
  }

  if (rpcRsp.pCont != NULL) {
    rpcFreeCont(rpcRsp.pCont);
  }

_register_inst_end:
  instanceRpcRelease();
  terrno = code;
  return code;
}

int32_t taos_list_instances(const char *filter_type, char ***pList, int32_t *pCount) {
  if (pList == NULL || pCount == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }

  int32_t code = taos_init();
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return code;
  }

  SConfig *pCfg = taosGetCfg();
  if (pCfg == NULL) {
    terrno = TSDB_CODE_CFG_NOT_FOUND;
    return TSDB_CODE_CFG_NOT_FOUND;
  }

  SEpSet epSet = {0};
  code = instanceBuildEpSetFromCfg(pCfg, &epSet);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return code;
  }

  code = instanceApiRateLimitTry();
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return code;
  }

  void *clientRpc = NULL;
  code = instanceRpcAcquire(&clientRpc);
  if (code != TSDB_CODE_SUCCESS) {
    terrno = code;
    return code;
  }

  SRpcMsg rpcMsg = {0};
  SRpcMsg rpcRsp = {0};

  SInstanceListReq req = {0};
  if (filter_type != NULL && filter_type[0] != 0) {
    tstrncpy(req.filter_type, filter_type, sizeof(req.filter_type));
  }

  int32_t contLen = tSerializeSInstanceListReq(NULL, 0, &req);
  if (contLen <= 0) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    goto _list_inst_end;
  }

  void *pCont = rpcMallocCont(contLen);
  if (pCont == NULL) {
    code = terrno != 0 ? terrno : TSDB_CODE_OUT_OF_MEMORY;
    goto _list_inst_end;
  }

  if (tSerializeSInstanceListReq(pCont, contLen, &req) < 0) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    rpcFreeCont(pCont);
    goto _list_inst_end;
  }

  rpcMsg.pCont = pCont;
  rpcMsg.contLen = contLen;
  rpcMsg.msgType = TDMT_MND_LIST_INSTANCES;
  rpcMsg.info.ahandle = (void *)0x9529;  // Different magic number from register
  rpcMsg.info.notFreeAhandle = 1;

  code = rpcSendRecv(clientRpc, &epSet, &rpcMsg, &rpcRsp);
  if (TSDB_CODE_SUCCESS != code) {
    tscError("failed to send instance list req since %s", tstrerror(code));
    goto _list_inst_end;
  }

  if (rpcRsp.code != 0) {
    code = rpcRsp.code;
    tscError("instance list failed, code:%s", tstrerror(code));
    if (rpcRsp.pCont != NULL) {
      rpcFreeCont(rpcRsp.pCont);
    }
    goto _list_inst_end;
  }

  if (rpcRsp.pCont != NULL && rpcRsp.contLen > 0) {
    SInstanceListRsp rsp = {0};
    code = tDeserializeSInstanceListRsp(rpcRsp.pCont, rpcRsp.contLen, &rsp);
    if (code != TSDB_CODE_SUCCESS) {
      tscError("failed to deserialize instance list rsp, code:%s", tstrerror(code));
      if (rsp.ids != NULL) {
        for (int32_t i = 0; i < rsp.count; i++) {
          if (rsp.ids[i] != NULL) {
            taosMemoryFree(rsp.ids[i]);
          }
        }
        taosMemoryFree(rsp.ids);
        rsp.ids = NULL;
      }
      rsp.count = 0;
      rpcFreeCont(rpcRsp.pCont);
      goto _list_inst_end;
    }
    *pList = rsp.ids;
    *pCount = rsp.count;
  } else {
    *pList = NULL;
    *pCount = 0;
  }

  if (rpcRsp.pCont != NULL) {
    rpcFreeCont(rpcRsp.pCont);
  }
  code = TSDB_CODE_SUCCESS;

_list_inst_end:
  instanceRpcRelease();
  terrno = code;
  return code;
}

void taos_free_instances(char ***pList, int32_t count) {
  if (pList == NULL || *pList == NULL || count <= 0) {
    return;
  }

  // Free each string in the array
  for (int32_t i = 0; i < count; i++) {
    if ((*pList)[i] != NULL) {
      taosMemoryFree((*pList)[i]);
      (*pList)[i] = NULL;
    }
  }

  // Free the array itself
  taosMemoryFree(*pList);
  *pList = NULL;
}

taosdata / TDengine / #5053

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