taosdata/TDengine | Build 3633 | source/client/src/clientMsgHandler.c | Coveralls

33

static void setErrno(SRequestObj* pRequest, int32_t code) {

×

34

  pRequest->code = code;

×

35

  terrno = code;

×

36

×

38

int32_t genericRspCallback(void* param, SDataBuf* pMsg, int32_t code) {

×

39

  SRequestObj* pRequest = param;

×

40

  setErrno(pRequest, code);

×

42

  if (NEED_CLIENT_RM_TBLMETA_REQ(pRequest->type)) {

×

43

    if (removeMeta(pRequest->pTscObj, pRequest->targetTableList, IS_VIEW_REQUEST(pRequest->type)) != 0) {

×

44

      tscError("failed to remove meta data for table");

×

48

  taosMemoryFree(pMsg->pEpSet);

×

49

  taosMemoryFree(pMsg->pData);

×

50

  if (pRequest->body.queryFp != NULL) {

×

51

    doRequestCallback(pRequest, code);

×

53

    if (tsem_post(&pRequest->body.rspSem) != 0) {

×

54

      tscError("failed to post semaphore");

×

57

  return code;

×

60

int32_t processConnectRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

61

  SRequestObj* pRequest = acquireRequest(*(int64_t*)param);

×

62

  if (NULL == pRequest) {

×

63

    goto EXIT;

×

66

  if (code != TSDB_CODE_SUCCESS) {

×

67

    goto End;

×

70

  STscObj* pTscObj = pRequest->pTscObj;

×

72

  if (NULL == pTscObj->pAppInfo) {

×

73

    code = TSDB_CODE_TSC_DISCONNECTED;

×

74

    goto End;

×

77

  SConnectRsp connectRsp = {0};

×

78

  if (tDeserializeSConnectRsp(pMsg->pData, pMsg->len, &connectRsp) != 0) {

×

79

    code = TSDB_CODE_TSC_INVALID_VERSION;

×

80

    goto End;

×

83

  if ((code = taosCheckVersionCompatibleFromStr(td_version, connectRsp.sVer, 3)) != 0) {

×

84

    tscError("version not compatible. client version:%s, server version:%s", td_version, connectRsp.sVer);

×

85

    goto End;

×

88

  int32_t now = taosGetTimestampSec();

×

89

  int32_t delta = abs(now - connectRsp.svrTimestamp);

×

90

  if (delta > timestampDeltaLimit) {

×

91

    code = TSDB_CODE_TIME_UNSYNCED;

×

92

    tscError("time diff:%ds is too big", delta);

×

93

    goto End;

×

96

  if (connectRsp.epSet.numOfEps == 0) {

×

97

    code = TSDB_CODE_APP_ERROR;

×

98

    goto End;

×

101

  int updateEpSet = 1;

×

102

  if (connectRsp.dnodeNum == 1) {

×

103

    SEpSet srcEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

×

104

    SEpSet dstEpSet = connectRsp.epSet;

×

105

    if (srcEpSet.numOfEps == 1) {

×

106

      if (rpcSetDefaultAddr(pTscObj->pAppInfo->pTransporter, srcEpSet.eps[srcEpSet.inUse].fqdn,

×

107

                            dstEpSet.eps[dstEpSet.inUse].fqdn) != 0) {

×

108

        tscError("failed to set default addr for rpc");

×

110

      updateEpSet = 0;

×

113

  if (updateEpSet == 1 && !isEpsetEqual(&pTscObj->pAppInfo->mgmtEp.epSet, &connectRsp.epSet)) {

×

114

    SEpSet corEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

×

116

    SEpSet* pOrig = &corEpSet;

×

117

    SEp*    pOrigEp = &pOrig->eps[pOrig->inUse];

×

118

    SEp*    pNewEp = &connectRsp.epSet.eps[connectRsp.epSet.inUse];

×

119

    tscDebug("mnode epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d in connRsp", pOrig->inUse, pOrig->numOfEps,

×

122

    updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &connectRsp.epSet);

×

125

  for (int32_t i = 0; i < connectRsp.epSet.numOfEps; ++i) {

×

126

    tscDebug("QID:0x%" PRIx64 ", epSet.fqdn[%d]:%s port:%d, connObj:0x%" PRIx64, pRequest->requestId, i,

×

130

  pTscObj->sysInfo = connectRsp.sysInfo;

×

131

  pTscObj->connId = connectRsp.connId;

×

132

  pTscObj->acctId = connectRsp.acctId;

×

133

  tstrncpy(pTscObj->sVer, connectRsp.sVer, tListLen(pTscObj->sVer));

×

134

  tstrncpy(pTscObj->sDetailVer, connectRsp.sDetailVer, tListLen(pTscObj->sDetailVer));

×

137

  pTscObj->pAppInfo->clusterId = connectRsp.clusterId;

×

138

  pTscObj->pAppInfo->serverCfg.monitorParas = connectRsp.monitorParas;

×

139

  pTscObj->pAppInfo->serverCfg.enableAuditDelete = connectRsp.enableAuditDelete;

×

140

  tscDebug("monitor paras from connect rsp, clusterId:0x%" PRIx64 ", threshold:%d scope:%d",

×

142

  lastClusterId = connectRsp.clusterId;

×

144

  pTscObj->connType = connectRsp.connType;

×

145

  pTscObj->passInfo.ver = connectRsp.passVer;

×

146

  pTscObj->authVer = connectRsp.authVer;

×

147

  pTscObj->whiteListInfo.ver = connectRsp.whiteListVer;

×

149

  if (taosHashGet(appInfo.pInstMapByClusterId, &connectRsp.clusterId, LONG_BYTES) == NULL) {

×

150

    if (taosHashPut(appInfo.pInstMapByClusterId, &connectRsp.clusterId, LONG_BYTES, &pTscObj->pAppInfo,

×

152

      tscError("failed to put appInfo into appInfo.pInstMapByClusterId");

×

154

      MonitorSlowLogData data = {0};

×

155

      data.clusterId = pTscObj->pAppInfo->clusterId;

×

156

      data.type = SLOW_LOG_READ_BEGINNIG;

×

157

      (void)monitorPutData2MonitorQueue(data);  // ignore

×

158

      monitorClientSlowQueryInit(connectRsp.clusterId);

×

159

      monitorClientSQLReqInit(connectRsp.clusterId);

×

163

  (void)taosThreadMutexLock(&clientHbMgr.lock);

×

164

  SAppHbMgr* pAppHbMgr = taosArrayGetP(clientHbMgr.appHbMgrs, pTscObj->appHbMgrIdx);

×

165

  if (pAppHbMgr) {

×

166

    if (hbRegisterConn(pAppHbMgr, pTscObj->id, connectRsp.clusterId, connectRsp.connType) != 0) {

×

167

      tscError("QID:0x%" PRIx64 ", failed to register conn to hbMgr", pRequest->requestId);

×

170

    (void)taosThreadMutexUnlock(&clientHbMgr.lock);

×

171

    code = TSDB_CODE_TSC_DISCONNECTED;

×

172

    goto End;

×

174

  (void)taosThreadMutexUnlock(&clientHbMgr.lock);

×

176

  tscDebug("QID:0x%" PRIx64 ", clusterId:0x%" PRIx64 ", totalConn:%" PRId64, pRequest->requestId, connectRsp.clusterId,

×

179

End:

×

180

  if (code != 0) {

×

181

    setErrno(pRequest, code);

×

183

  if (tsem_post(&pRequest->body.rspSem) != 0) {

×

184

    tscError("failed to post semaphore");

×

187

  if (pRequest) {

×

188

    (void)releaseRequest(pRequest->self);

×

191

EXIT:

×

192

  taosMemoryFree(param);

×

193

  taosMemoryFree(pMsg->pEpSet);

×

194

  taosMemoryFree(pMsg->pData);

×

195

  return code;

×

198

SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pRequest) {

×

199

  SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));

×

200

  if (pMsgSendInfo == NULL) return pMsgSendInfo;

×

201

  pMsgSendInfo->requestObjRefId = pRequest->self;

×

202

  pMsgSendInfo->requestId = pRequest->requestId;

×

203

  pMsgSendInfo->param = pRequest;

×

204

  pMsgSendInfo->msgType = pRequest->type;

×

205

  pMsgSendInfo->target.type = TARGET_TYPE_MNODE;

×

207

  pMsgSendInfo->msgInfo = pRequest->body.requestMsg;

×

208

  pMsgSendInfo->fp = getMsgRspHandle(pRequest->type);

×

209

  return pMsgSendInfo;

×

212

int32_t processCreateDbRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

214

  SRequestObj* pRequest = param;

×

215

  taosMemoryFree(pMsg->pData);

×

216

  taosMemoryFree(pMsg->pEpSet);

×

217

  if (code != TSDB_CODE_SUCCESS) {

×

218

    setErrno(pRequest, code);

×

220

    struct SCatalog* pCatalog = NULL;

×

221

    code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);

×

222

    if (TSDB_CODE_SUCCESS == code) {

×

223

      STscObj* pTscObj = pRequest->pTscObj;

×

225

      SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,

×

226

                               .requestId = pRequest->requestId,

×

227

                               .requestObjRefId = pRequest->self,

×

228

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

×

230

      (void)snprintf(dbFName, sizeof(dbFName) - 1, "%d.%s", pTscObj->acctId, TSDB_INFORMATION_SCHEMA_DB);

×

231

      if (catalogRefreshDBVgInfo(pCatalog, &conn, dbFName) != 0) {

×

232

        tscError("QID:0x%" PRIx64 ", failed to refresh db vg info", pRequest->requestId);

×

234

      (void)snprintf(dbFName, sizeof(dbFName) - 1, "%d.%s", pTscObj->acctId, TSDB_PERFORMANCE_SCHEMA_DB);

×

235

      if (catalogRefreshDBVgInfo(pCatalog, &conn, dbFName) != 0) {

×

236

        tscError("QID:0x%" PRIx64 ", failed to refresh db vg info", pRequest->requestId);

×

241

  if (pRequest->body.queryFp) {

×

242

    doRequestCallback(pRequest, code);

×

244

    if (tsem_post(&pRequest->body.rspSem) != 0) {

×

245

      tscError("failed to post semaphore");

×

248

  return code;

×

251

int32_t processUseDbRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

252

  SRequestObj* pRequest = param;

×

253

  if (TSDB_CODE_MND_DB_NOT_EXIST == code || TSDB_CODE_MND_DB_IN_CREATING == code ||

×

255

    SUseDbRsp usedbRsp = {0};

×

256

    if (tDeserializeSUseDbRsp(pMsg->pData, pMsg->len, &usedbRsp) != 0) {

×

257

      tscError("QID:0x%" PRIx64 ", deserialize SUseDbRsp failed", pRequest->requestId);

×

259

    struct SCatalog* pCatalog = NULL;

×

261

    if (usedbRsp.vgVersion >= 0) {  // cached in local

×

262

      int64_t clusterId = pRequest->pTscObj->pAppInfo->clusterId;

×

263

      int32_t code1 = catalogGetHandle(clusterId, &pCatalog);

×

264

      if (code1 != TSDB_CODE_SUCCESS) {

×

265

        tscWarn("QID:0x%" PRIx64 ", catalogGetHandle failed, clusterId:0x%" PRIx64 ", error:%s", pRequest->requestId, clusterId,

×

268

        if (catalogRemoveDB(pCatalog, usedbRsp.db, usedbRsp.uid) != 0) {

×

269

          tscError("QID:0x%" PRIx64 ", catalogRemoveDB failed, db:%s, uid:%" PRId64, pRequest->requestId, usedbRsp.db,

×

274

    tFreeSUsedbRsp(&usedbRsp);

×

277

  if (code != TSDB_CODE_SUCCESS) {

×

278

    taosMemoryFree(pMsg->pData);

×

279

    taosMemoryFree(pMsg->pEpSet);

×

280

    setErrno(pRequest, code);

×

282

    if (pRequest->body.queryFp != NULL) {

×

283

      doRequestCallback(pRequest, pRequest->code);

×

286

      if (tsem_post(&pRequest->body.rspSem) != 0) {

×

287

        tscError("failed to post semaphore");

×

291

    return code;

×

294

  SUseDbRsp usedbRsp = {0};

×

295

  if (tDeserializeSUseDbRsp(pMsg->pData, pMsg->len, &usedbRsp) != 0) {

×

296

    tscError("QID:0x%" PRIx64 ", deserialize SUseDbRsp failed", pRequest->requestId);

×

299

  if (strlen(usedbRsp.db) == 0) {

×

300

    taosMemoryFree(pMsg->pData);

×

301

    taosMemoryFree(pMsg->pEpSet);

×

303

    if (usedbRsp.errCode != 0) {

×

304

      return usedbRsp.errCode;

×

306

      return TSDB_CODE_APP_ERROR;

×

310

  tscTrace("db:%s, usedbRsp received, numOfVgroups:%d", usedbRsp.db, usedbRsp.vgNum);

×

311

  for (int32_t i = 0; i < usedbRsp.vgNum; ++i) {

×

312

    SVgroupInfo* pInfo = taosArrayGet(usedbRsp.pVgroupInfos, i);

×

313

    if (pInfo == NULL) {

×

314

      continue;

×

316

    tscTrace("vgId:%d, numOfEps:%d inUse:%d ", pInfo->vgId, pInfo->epSet.numOfEps, pInfo->epSet.inUse);

×

317

    for (int32_t j = 0; j < pInfo->epSet.numOfEps; ++j) {

×

318

      tscTrace("vgId:%d, index:%d epset:%s:%u", pInfo->vgId, j, pInfo->epSet.eps[j].fqdn, pInfo->epSet.eps[j].port);

×

322

  SName name = {0};

×

323

  if (tNameFromString(&name, usedbRsp.db, T_NAME_ACCT | T_NAME_DB) != TSDB_CODE_SUCCESS) {

×

324

    tscError("QID:0x%" PRIx64 ", failed to parse db name:%s", pRequest->requestId, usedbRsp.db);

×

327

  SUseDbOutput output = {0};

×

328

  code = queryBuildUseDbOutput(&output, &usedbRsp);

×

329

  if (code != 0) {

×

330

    terrno = code;

×

331

    if (output.dbVgroup) taosHashCleanup(output.dbVgroup->vgHash);

×

333

    tscError("QID:0x%" PRIx64 ", failed to build use db output since %s", pRequest->requestId, terrstr());

×

334

  } else if (output.dbVgroup && output.dbVgroup->vgHash) {

×

335

    struct SCatalog* pCatalog = NULL;

×

337

    int32_t code1 = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);

×

338

    if (code1 != TSDB_CODE_SUCCESS) {

×

339

      tscWarn("catalogGetHandle failed, clusterId:0x%" PRIx64 ", error:%s", pRequest->pTscObj->pAppInfo->clusterId,

×

342

      if (catalogUpdateDBVgInfo(pCatalog, output.db, output.dbId, output.dbVgroup) != 0) {

×

343

        tscError("QID:0x%" PRIx64 ", failed to update db vg info, db:%s, dbId:%" PRId64, pRequest->requestId, output.db,

×

346

      output.dbVgroup = NULL;

×

350

  taosMemoryFreeClear(output.dbVgroup);

×

351

  tFreeSUsedbRsp(&usedbRsp);

×

353

  char db[TSDB_DB_NAME_LEN] = {0};

×

354

  if (tNameGetDbName(&name, db) != TSDB_CODE_SUCCESS) {

×

355

    tscError("QID:0x%" PRIx64 ", failed to get db name since %s", pRequest->requestId, tstrerror(code));

×

358

  setConnectionDB(pRequest->pTscObj, db);

×

360

  taosMemoryFree(pMsg->pData);

×

361

  taosMemoryFree(pMsg->pEpSet);

×

363

  if (pRequest->body.queryFp != NULL) {

×

364

    doRequestCallback(pRequest, pRequest->code);

×

366

    if (tsem_post(&pRequest->body.rspSem) != 0) {

×

367

      tscError("failed to post semaphore");

×

370

  return 0;

×

373

int32_t processCreateSTableRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

374

  if (pMsg == NULL) {

×

375

    tscError("processCreateSTableRsp: invalid input param, pMsg is NULL");

×

376

    return TSDB_CODE_TSC_INVALID_INPUT;

×

378

  if (param == NULL) {

×

379

    taosMemoryFree(pMsg->pEpSet);

×

380

    taosMemoryFree(pMsg->pData);

×

381

    tscError("processCreateSTableRsp: invalid input param, param is NULL");

×

382

    return TSDB_CODE_TSC_INVALID_INPUT;

×

385

  SRequestObj* pRequest = param;

×

387

  if (code != TSDB_CODE_SUCCESS) {

×

388

    setErrno(pRequest, code);

×

390

    SMCreateStbRsp createRsp = {0};

×

391

    SDecoder       coder = {0};

×

392

    tDecoderInit(&coder, pMsg->pData, pMsg->len);

×

393

    if (pMsg->len > 0) {

×

394

      code = tDecodeSMCreateStbRsp(&coder, &createRsp);  // pMsg->len == 0

×

395

      if (code != TSDB_CODE_SUCCESS) {

×

396

        setErrno(pRequest, code);

×

399

    tDecoderClear(&coder);

×

401

    pRequest->body.resInfo.execRes.msgType = TDMT_MND_CREATE_STB;

×

402

    pRequest->body.resInfo.execRes.res = createRsp.pMeta;

×

405

  taosMemoryFree(pMsg->pEpSet);

×

406

  taosMemoryFree(pMsg->pData);

×

408

  if (pRequest->body.queryFp != NULL) {

×

409

    SExecResult* pRes = &pRequest->body.resInfo.execRes;

×

411

    if (code == TSDB_CODE_SUCCESS) {

×

412

      SCatalog* pCatalog = NULL;

×

413

      int32_t   ret = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);

×

414

      if (pRes->res != NULL) {

×

415

        ret = handleCreateTbExecRes(pRes->res, pCatalog);

×

418

      if (ret != TSDB_CODE_SUCCESS) {

×

419

        code = ret;

×

423

    doRequestCallback(pRequest, code);

×

425

    if (tsem_post(&pRequest->body.rspSem) != 0) {

×

426

      tscError("failed to post semaphore");

×

429

  return code;

×

432

int32_t processDropDbRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

433

  SRequestObj* pRequest = param;

×

434

  if (code != TSDB_CODE_SUCCESS) {

×

435

    setErrno(pRequest, code);

×

437

    SDropDbRsp dropdbRsp = {0};

×

438

    if (tDeserializeSDropDbRsp(pMsg->pData, pMsg->len, &dropdbRsp) != 0) {

×

439

      tscError("QID:0x%" PRIx64 ", deserialize SDropDbRsp failed", pRequest->requestId);

×

441

    struct SCatalog* pCatalog = NULL;

×

442

    code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);

×

443

    if (TSDB_CODE_SUCCESS == code) {

×

444

      if (catalogRemoveDB(pCatalog, dropdbRsp.db, dropdbRsp.uid) != 0) {

×

445

        tscError("QID:0x%" PRIx64 ", failed to remove db:%s", pRequest->requestId, dropdbRsp.db);

×

447

      STscObj* pTscObj = pRequest->pTscObj;

×

449

      SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,

×

450

                               .requestId = pRequest->requestId,

×

451

                               .requestObjRefId = pRequest->self,

×

452

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

×

453

      char             dbFName[TSDB_DB_FNAME_LEN] = {0};

×

454

      (void)snprintf(dbFName, sizeof(dbFName) - 1, "%d.%s", pTscObj->acctId, TSDB_INFORMATION_SCHEMA_DB);

×

455

      if (catalogRefreshDBVgInfo(pCatalog, &conn, dbFName) != TSDB_CODE_SUCCESS) {

×

456

        tscError("QID:0x%" PRIx64 ", failed to refresh db vg info, db:%s", pRequest->requestId, dbFName);

×

458

      (void)snprintf(dbFName, sizeof(dbFName) - 1, "%d.%s", pTscObj->acctId, TSDB_PERFORMANCE_SCHEMA_DB);

×

459

      if (catalogRefreshDBVgInfo(pCatalog, &conn, dbFName) != 0) {

×

460

        tscError("QID:0x%" PRIx64 ", failed to refresh db vg info, db:%s", pRequest->requestId, dbFName);

×

465

  taosMemoryFree(pMsg->pData);

×

466

  taosMemoryFree(pMsg->pEpSet);

×

468

  if (pRequest->body.queryFp != NULL) {

×

469

    doRequestCallback(pRequest, code);

×

471

    if (tsem_post(&pRequest->body.rspSem) != 0) {

×

472

      tscError("failed to post semaphore");

×

475

  return code;

×

478

int32_t processAlterStbRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

479

  SRequestObj* pRequest = param;

×

480

  if (code != TSDB_CODE_SUCCESS) {

×

481

    setErrno(pRequest, code);

×

483

    SMAlterStbRsp alterRsp = {0};

×

484

    SDecoder      coder = {0};

×

485

    tDecoderInit(&coder, pMsg->pData, pMsg->len);

×

486

    if (pMsg->len > 0) {

×

487

      code = tDecodeSMAlterStbRsp(&coder, &alterRsp);  // pMsg->len == 0

×

488

      if (code != TSDB_CODE_SUCCESS) {

×

489

        setErrno(pRequest, code);

×

492

    tDecoderClear(&coder);

×

494

    pRequest->body.resInfo.execRes.msgType = TDMT_MND_ALTER_STB;

×

495

    pRequest->body.resInfo.execRes.res = alterRsp.pMeta;

×

498

  taosMemoryFree(pMsg->pData);

×

499

  taosMemoryFree(pMsg->pEpSet);

×

501

  if (pRequest->body.queryFp != NULL) {

×

502

    SExecResult* pRes = &pRequest->body.resInfo.execRes;

×

504

    if (code == TSDB_CODE_SUCCESS) {

×

505

      SCatalog* pCatalog = NULL;

×

506

      int32_t   ret = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);

×

507

      if (pRes->res != NULL) {

×

508

        ret = handleAlterTbExecRes(pRes->res, pCatalog);

×

511

      if (ret != TSDB_CODE_SUCCESS) {

×

512

        code = ret;

×

516

    doRequestCallback(pRequest, code);

×

518

    if (tsem_post(&pRequest->body.rspSem) != 0) {

×

519

      tscError("failed to post semaphore");

×

522

  return code;

×

525

static int32_t buildShowVariablesBlock(SArray* pVars, SSDataBlock** block) {

×

526

  int32_t      code = 0;

×

527

  int32_t      line = 0;

×

528

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

×

529

  TSDB_CHECK_NULL(pBlock, code, line, END, terrno);

×

530

  pBlock->info.hasVarCol = true;

×

532

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

×

533

  TSDB_CHECK_NULL(pBlock->pDataBlock, code, line, END, terrno);

×

534

  SColumnInfoData infoData = {0};

×

535

  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;

×

536

  infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD1_LEN;

×

537

  TSDB_CHECK_NULL(taosArrayPush(pBlock->pDataBlock, &infoData), code, line, END, terrno);

×

539

  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;

×

540

  infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD2_LEN;

×

541

  TSDB_CHECK_NULL(taosArrayPush(pBlock->pDataBlock, &infoData), code, line, END, terrno);

×

543

  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;

×

544

  infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD3_LEN;

×

545

  TSDB_CHECK_NULL(taosArrayPush(pBlock->pDataBlock, &infoData), code, line, END, terrno);

×

547

  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;

×

548

  infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD4_LEN;

×

549

  TSDB_CHECK_NULL(taosArrayPush(pBlock->pDataBlock, &infoData), code, line, END, terrno);

×

551

  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;

×

552

  infoData.info.bytes = SHOW_VARIABLES_RESULT_FIELD5_LEN;

×

553

  TSDB_CHECK_NULL(taosArrayPush(pBlock->pDataBlock, &infoData), code, line, END, terrno);

×

555

  int32_t numOfCfg = taosArrayGetSize(pVars);

×

556

  code = blockDataEnsureCapacity(pBlock, numOfCfg);

×

557

  TSDB_CHECK_CODE(code, line, END);

×

559

  for (int32_t i = 0, c = 0; i < numOfCfg; ++i, c = 0) {

×

560

    SVariablesInfo* pInfo = taosArrayGet(pVars, i);

×

561

    TSDB_CHECK_NULL(pInfo, code, line, END, terrno);

×

563

    char name[TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE] = {0};

×

564

    STR_WITH_MAXSIZE_TO_VARSTR(name, pInfo->name, TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE);

×

565

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, c++);

×

566

    TSDB_CHECK_NULL(pColInfo, code, line, END, terrno);

×

567

    code = colDataSetVal(pColInfo, i, name, false);

×

568

    TSDB_CHECK_CODE(code, line, END);

×

570

    char value[TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE] = {0};

×

571

    STR_WITH_MAXSIZE_TO_VARSTR(value, pInfo->value, TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE);

×

572

    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);

×

573

    TSDB_CHECK_NULL(pColInfo, code, line, END, terrno);

×

574

    code = colDataSetVal(pColInfo, i, value, false);

×

575

    TSDB_CHECK_CODE(code, line, END);

×

577

    char scope[TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE] = {0};

×

578

    STR_WITH_MAXSIZE_TO_VARSTR(scope, pInfo->scope, TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE);

×

579

    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);

×

580

    TSDB_CHECK_NULL(pColInfo, code, line, END, terrno);

×

581

    code = colDataSetVal(pColInfo, i, scope, false);

×

582

    TSDB_CHECK_CODE(code, line, END);

×

584

    char category[TSDB_CONFIG_CATEGORY_LEN + VARSTR_HEADER_SIZE] = {0};

×

585

    STR_WITH_MAXSIZE_TO_VARSTR(category, pInfo->category, TSDB_CONFIG_CATEGORY_LEN + VARSTR_HEADER_SIZE);

×

586

    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);

×

587

    TSDB_CHECK_NULL(pColInfo, code, line, END, terrno);

×

588

    code = colDataSetVal(pColInfo, i, category, false);

×

589

    TSDB_CHECK_CODE(code, line, END);

×

591

    char info[TSDB_CONFIG_INFO_LEN + VARSTR_HEADER_SIZE] = {0};

×

592

    STR_WITH_MAXSIZE_TO_VARSTR(info, pInfo->info, TSDB_CONFIG_INFO_LEN + VARSTR_HEADER_SIZE);

×

593

    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);

×

594

    TSDB_CHECK_NULL(pColInfo, code, line, END, terrno);

×

595

    code = colDataSetVal(pColInfo, i, info, false);

×

596

    TSDB_CHECK_CODE(code, line, END);

×

599

  pBlock->info.rows = numOfCfg;

×

601

  *block = pBlock;

×

602

  return code;

×

604

END:

×

605

  taosArrayDestroy(pBlock->pDataBlock);

×

606

  taosMemoryFree(pBlock);

×

607

  return code;

×

610

static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) {

×

611

  SSDataBlock* pBlock = NULL;

×

612

  int32_t      code = buildShowVariablesBlock(pVars, &pBlock);

×

613

  if (code) {

×

614

    return code;

×

617

  size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);

×

618

  size_t rspSize = sizeof(SRetrieveTableRsp) + dataEncodeBufSize + PAYLOAD_PREFIX_LEN;

×

619

  *pRsp = taosMemoryCalloc(1, rspSize);

×

620

  if (NULL == *pRsp) {

×

621

    code = terrno;

×

622

    goto _exit;

×

625

  (*pRsp)->useconds = 0;

×

626

  (*pRsp)->completed = 1;

×

627

  (*pRsp)->precision = 0;

×

628

  (*pRsp)->compressed = 0;

×

630

  (*pRsp)->numOfRows = htobe64((int64_t)pBlock->info.rows);

×

631

  (*pRsp)->numOfCols = htonl(SHOW_VARIABLES_RESULT_COLS);

×

633

  int32_t len = 0;

×

634

  if ((*pRsp)->numOfRows > 0) {

×

635

    len = blockEncode(pBlock, (*pRsp)->data + PAYLOAD_PREFIX_LEN, dataEncodeBufSize, SHOW_VARIABLES_RESULT_COLS);

×

636

    if (len < 0) {

×

637

      uError("buildShowVariablesRsp error, len:%d", len);

×

638

      code = terrno;

×

639

      goto _exit;

×

641

    SET_PAYLOAD_LEN((*pRsp)->data, len, len);

×

643

    int32_t payloadLen = len + PAYLOAD_PREFIX_LEN;

×

644

    (*pRsp)->payloadLen = htonl(payloadLen);

×

645

    (*pRsp)->compLen = htonl(payloadLen);

×

647

    if (payloadLen != rspSize - sizeof(SRetrieveTableRsp)) {

×

648

      uError("buildShowVariablesRsp error, len:%d != rspSize - sizeof(SRetrieveTableRsp):%" PRIu64, len,

×

650

      code = TSDB_CODE_TSC_INVALID_INPUT;

×

651

      goto _exit;

×

655

  blockDataDestroy(pBlock);

×

656

  pBlock = NULL;

×

658

  return TSDB_CODE_SUCCESS;

×

659

_exit:

×

660

  if (*pRsp) {

×

661

    taosMemoryFree(*pRsp);

×

662

    *pRsp = NULL;

×

664

  if (pBlock) {

×

665

    blockDataDestroy(pBlock);

×

666

    pBlock = NULL;

×

668

  return code;

×

671

int32_t processShowVariablesRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

672

  SRequestObj* pRequest = param;

×

673

  if (code != TSDB_CODE_SUCCESS) {

×

674

    setErrno(pRequest, code);

×

676

    SShowVariablesRsp  rsp = {0};

×

677

    SRetrieveTableRsp* pRes = NULL;

×

678

    code = tDeserializeSShowVariablesRsp(pMsg->pData, pMsg->len, &rsp);

×

679

    if (TSDB_CODE_SUCCESS == code) {

×

680

      code = buildShowVariablesRsp(rsp.variables, &pRes);

×

682

    if (TSDB_CODE_SUCCESS == code) {

×

683

      code = setQueryResultFromRsp(&pRequest->body.resInfo, pRes, false);

×

686

    if (code != 0) {

×

687

      taosMemoryFree(pRes);

×

689

    tFreeSShowVariablesRsp(&rsp);

×

692

  taosMemoryFree(pMsg->pData);

×

693

  taosMemoryFree(pMsg->pEpSet);

×

695

  if (pRequest->body.queryFp != NULL) {

×

696

    doRequestCallback(pRequest, code);

×

698

    if (tsem_post(&pRequest->body.rspSem) != 0) {

×

699

      tscError("failed to post semaphore");

×

702

  return code;

×

705

static int32_t buildCompactDbBlock(SCompactDbRsp* pRsp, SSDataBlock** block) {

×

706

  int32_t      code = 0;

×

707

  int32_t      line = 0;

×

708

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

×

709

  TSDB_CHECK_NULL(pBlock, code, line, END, terrno);

×

710

  pBlock->info.hasVarCol = true;

×

712

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

×

713

  TSDB_CHECK_NULL(pBlock->pDataBlock, code, line, END, terrno);

×

714

  SColumnInfoData infoData = {0};

×

715

  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;

×

716

  infoData.info.bytes = COMPACT_DB_RESULT_FIELD1_LEN;

×

717

  TSDB_CHECK_NULL(taosArrayPush(pBlock->pDataBlock, &infoData), code, line, END, terrno);

×

719

  infoData.info.type = TSDB_DATA_TYPE_INT;

×

720

  infoData.info.bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes;

×

721

  TSDB_CHECK_NULL(taosArrayPush(pBlock->pDataBlock, &infoData), code, line, END, terrno);

×

723

  infoData.info.type = TSDB_DATA_TYPE_VARCHAR;

×

724

  infoData.info.bytes = COMPACT_DB_RESULT_FIELD3_LEN;

×

725

  TSDB_CHECK_NULL(taosArrayPush(pBlock->pDataBlock, &infoData), code, line, END, terrno);

×

727

  code = blockDataEnsureCapacity(pBlock, 1);

×

728

  TSDB_CHECK_CODE(code, line, END);

×

730

  SColumnInfoData* pResultCol = taosArrayGet(pBlock->pDataBlock, 0);

×

731

  TSDB_CHECK_NULL(pResultCol, code, line, END, terrno);

×

732

  SColumnInfoData* pIdCol = taosArrayGet(pBlock->pDataBlock, 1);

×

733

  TSDB_CHECK_NULL(pIdCol, code, line, END, terrno);

×

734

  SColumnInfoData* pReasonCol = taosArrayGet(pBlock->pDataBlock, 2);

×

735

  TSDB_CHECK_NULL(pReasonCol, code, line, END, terrno);

×

737

  char result[COMPACT_DB_RESULT_FIELD1_LEN] = {0};

×

738

  char reason[COMPACT_DB_RESULT_FIELD3_LEN] = {0};

×

739

  if (pRsp->bAccepted) {

×

740

    STR_TO_VARSTR(result, "accepted");

×

741

    code = colDataSetVal(pResultCol, 0, result, false);

×

742

    TSDB_CHECK_CODE(code, line, END);

×

743

    code = colDataSetVal(pIdCol, 0, (void*)&pRsp->compactId, false);

×

744

    TSDB_CHECK_CODE(code, line, END);

×

745

    STR_TO_VARSTR(reason, "success");

×

746

    code = colDataSetVal(pReasonCol, 0, reason, false);

×

747

    TSDB_CHECK_CODE(code, line, END);

×

749

    STR_TO_VARSTR(result, "rejected");

×

750

    code = colDataSetVal(pResultCol, 0, result, false);

×

751

    TSDB_CHECK_CODE(code, line, END);

×

753

    STR_TO_VARSTR(reason, "compaction is ongoing");

×

754

    code = colDataSetVal(pReasonCol, 0, reason, false);

×

755

    TSDB_CHECK_CODE(code, line, END);

×

757

  pBlock->info.rows = 1;

×

759

  *block = pBlock;

×

761

  return TSDB_CODE_SUCCESS;

×

762

END:

×

763

  taosMemoryFree(pBlock);

×

764

  taosArrayDestroy(pBlock->pDataBlock);

×

765

  return code;

×

768

static int32_t buildRetriveTableRspForCompactDb(SCompactDbRsp* pCompactDb, SRetrieveTableRsp** pRsp) {

×

769

  SSDataBlock* pBlock = NULL;

×

770

  int32_t      code = buildCompactDbBlock(pCompactDb, &pBlock);

×

771

  if (code) {

×

772

    return code;

×

775

  size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);

×

776

  size_t rspSize = sizeof(SRetrieveTableRsp) + dataEncodeBufSize + PAYLOAD_PREFIX_LEN;

×

777

  *pRsp = taosMemoryCalloc(1, rspSize);

×

778

  if (NULL == *pRsp) {

×

779

    code = terrno;

×

780

    goto _exit;

×

783

  (*pRsp)->useconds = 0;

×

784

  (*pRsp)->completed = 1;

×

785

  (*pRsp)->precision = 0;

×

786

  (*pRsp)->compressed = 0;

×

787

  (*pRsp)->compLen = 0;

×

788

  (*pRsp)->payloadLen = 0;

×

789

  (*pRsp)->numOfRows = htobe64((int64_t)pBlock->info.rows);

×

790

  (*pRsp)->numOfCols = htonl(COMPACT_DB_RESULT_COLS);

×

792

  int32_t len = blockEncode(pBlock, (*pRsp)->data + PAYLOAD_PREFIX_LEN, dataEncodeBufSize, COMPACT_DB_RESULT_COLS);

×

793

  if (len < 0) {

×

794

    uError("buildRetriveTableRspForCompactDb error, len:%d", len);

×

795

    code = terrno;

×

796

    goto _exit;

×

798

  blockDataDestroy(pBlock);

×

800

  SET_PAYLOAD_LEN((*pRsp)->data, len, len);

×

802

  int32_t payloadLen = len + PAYLOAD_PREFIX_LEN;

×

803

  (*pRsp)->payloadLen = htonl(payloadLen);

×

804

  (*pRsp)->compLen = htonl(payloadLen);

×

806

  if (payloadLen != rspSize - sizeof(SRetrieveTableRsp)) {

×

807

    uError("buildRetriveTableRspForCompactDb error, len:%d != rspSize - sizeof(SRetrieveTableRsp):%" PRIu64, len,

×

809

    code = TSDB_CODE_TSC_INVALID_INPUT;

×

810

    goto _exit;

×

813

  return TSDB_CODE_SUCCESS;

×

814

_exit:

×

815

  if (*pRsp) {

×

816

    taosMemoryFree(*pRsp);

×

817

    *pRsp = NULL;

×

819

  if (pBlock) {

×

820

    blockDataDestroy(pBlock);

×

821

    pBlock = NULL;

×

823

  return code;

×

826

int32_t processCompactDbRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

827

  SRequestObj* pRequest = param;

×

828

  if (code != TSDB_CODE_SUCCESS) {

×

829

    setErrno(pRequest, code);

×

831

    SCompactDbRsp      rsp = {0};

×

832

    SRetrieveTableRsp* pRes = NULL;

×

833

    code = tDeserializeSCompactDbRsp(pMsg->pData, pMsg->len, &rsp);

×

834

    if (TSDB_CODE_SUCCESS == code) {

×

835

      code = buildRetriveTableRspForCompactDb(&rsp, &pRes);

×

837

    if (TSDB_CODE_SUCCESS == code) {

×

838

      code = setQueryResultFromRsp(&pRequest->body.resInfo, pRes, false);

×

841

    if (code != 0) {

×

842

      taosMemoryFree(pRes);

×

846

  taosMemoryFree(pMsg->pData);

×

847

  taosMemoryFree(pMsg->pEpSet);

×

849

  if (pRequest->body.queryFp != NULL) {

×

850

    pRequest->body.queryFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, code);

×

852

    if (tsem_post(&pRequest->body.rspSem) != 0) {

×

853

      tscError("failed to post semaphore");

×

856

  return code;

×

859

__async_send_cb_fn_t getMsgRspHandle(int32_t msgType) {

×

860

  switch (msgType) {

×

861

    case TDMT_MND_CONNECT:

×

862

      return processConnectRsp;

×

863

    case TDMT_MND_CREATE_DB:

×

864

      return processCreateDbRsp;

×

865

    case TDMT_MND_USE_DB:

×

866

      return processUseDbRsp;

×

867

    case TDMT_MND_CREATE_STB:

×

868

      return processCreateSTableRsp;

×

869

    case TDMT_MND_DROP_DB:

×

870

      return processDropDbRsp;

×

871

    case TDMT_MND_ALTER_STB:

×

872

      return processAlterStbRsp;

×

873

    case TDMT_MND_SHOW_VARIABLES:

×

874

      return processShowVariablesRsp;

×

875

    case TDMT_MND_COMPACT_DB:

×

876

      return processCompactDbRsp;

×

877

    default:

×

878

      return genericRspCallback;

×

taosdata / TDengine / #3633

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

taosdata / TDengine / #3633

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