taosdata/TDengine | Build 3800 | source/client/src/clientRawBlockWrite.c | Coveralls

55

static bool  tmqAddJsonObjectItem(cJSON *object, const char *string, cJSON *item){

×

56

  bool ret = cJSON_AddItemToObject(object, string, item);

×

57

  if (!ret){

×

58

    cJSON_Delete(item);

×

60

  return ret;

×

62

static bool  tmqAddJsonArrayItem(cJSON *array, cJSON *item){

×

63

  bool ret = cJSON_AddItemToArray(array, item);

×

64

  if (!ret){

×

65

    cJSON_Delete(item);

×

67

  return ret;

×

72

static tb_uid_t processSuid(tb_uid_t suid, char* db) {

×

73

  if (db == NULL) {

×

74

    return suid;

×

76

  return suid + MurmurHash3_32(db, strlen(db));

×

78

static void buildCreateTableJson(SSchemaWrapper* schemaRow, SSchemaWrapper* schemaTag, char* name, int64_t id, int8_t t,

×

80

  if (schemaRow == NULL || name == NULL || pColCmprRow == NULL || pJson == NULL) {

×

81

    uError("invalid parameter, schemaRow:%p, name:%p, pColCmprRow:%p, pJson:%p", schemaRow, name, pColCmprRow, pJson);

×

82

    return;

×

84

  int32_t code = TSDB_CODE_SUCCESS;

×

85

  int8_t  buildDefaultCompress = 0;

×

86

  if (pColCmprRow->nCols <= 0) {

×

87

    buildDefaultCompress = 1;

×

90

  char*  string = NULL;

×

91

  cJSON* json = cJSON_CreateObject();

×

92

  RAW_NULL_CHECK(json);

×

93

  cJSON* type = cJSON_CreateString("create");

×

94

  RAW_NULL_CHECK(type);

×

96

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));

×

97

  cJSON* tableType = cJSON_CreateString(t == TSDB_NORMAL_TABLE ? "normal" : "super");

×

98

  RAW_NULL_CHECK(tableType);

×

99

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));

×

100

  cJSON* tableName = cJSON_CreateString(name);

×

101

  RAW_NULL_CHECK(tableName);

×

102

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));

×

104

  cJSON* columns = cJSON_CreateArray();

×

105

  RAW_NULL_CHECK(columns);

×

106

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "columns", columns));

×

108

  for (int i = 0; i < schemaRow->nCols; i++) {

×

109

    cJSON* column = cJSON_CreateObject();

×

110

    RAW_NULL_CHECK(column);

×

111

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(columns, column));

×

112

    SSchema* s = schemaRow->pSchema + i;

×

113

    cJSON*   cname = cJSON_CreateString(s->name);

×

114

    RAW_NULL_CHECK(cname);

×

115

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "name", cname));

×

116

    cJSON* ctype = cJSON_CreateNumber(s->type);

×

117

    RAW_NULL_CHECK(ctype);

×

118

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "type", ctype));

×

119

    if (s->type == TSDB_DATA_TYPE_BINARY || s->type == TSDB_DATA_TYPE_VARBINARY || s->type == TSDB_DATA_TYPE_GEOMETRY) {

×

120

      int32_t length = s->bytes - VARSTR_HEADER_SIZE;

×

121

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

122

      RAW_NULL_CHECK(cbytes);

×

123

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "length", cbytes));

×

124

    } else if (s->type == TSDB_DATA_TYPE_NCHAR) {

×

125

      int32_t length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

126

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

127

      RAW_NULL_CHECK(cbytes);

×

128

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "length", cbytes));

×

130

    cJSON* isPk = cJSON_CreateBool(s->flags & COL_IS_KEY);

×

131

    RAW_NULL_CHECK(isPk);

×

132

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "isPrimarykey", isPk));

×

134

    if (pColCmprRow == NULL) {

×

135

      continue;

×

138

    uint32_t alg = 0;

×

139

    if (buildDefaultCompress) {

×

140

      alg = createDefaultColCmprByType(s->type);

×

142

      SColCmpr* pColCmpr = pColCmprRow->pColCmpr + i;

×

143

      alg = pColCmpr->alg;

×

145

    const char* encode = columnEncodeStr(COMPRESS_L1_TYPE_U32(alg));

×

146

    RAW_NULL_CHECK(encode);

×

147

    const char* compress = columnCompressStr(COMPRESS_L2_TYPE_U32(alg));

×

148

    RAW_NULL_CHECK(compress);

×

149

    const char* level = columnLevelStr(COMPRESS_L2_TYPE_LEVEL_U32(alg));

×

150

    RAW_NULL_CHECK(level);

×

152

    cJSON* encodeJson = cJSON_CreateString(encode);

×

153

    RAW_NULL_CHECK(encodeJson);

×

154

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "encode", encodeJson));

×

156

    cJSON* compressJson = cJSON_CreateString(compress);

×

157

    RAW_NULL_CHECK(compressJson);

×

158

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "compress", compressJson));

×

160

    cJSON* levelJson = cJSON_CreateString(level);

×

161

    RAW_NULL_CHECK(levelJson);

×

162

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "level", levelJson));

×

165

  cJSON* tags = cJSON_CreateArray();

×

166

  RAW_NULL_CHECK(tags);

×

167

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tags", tags));

×

169

  for (int i = 0; schemaTag && i < schemaTag->nCols; i++) {

×

170

    cJSON* tag = cJSON_CreateObject();

×

171

    RAW_NULL_CHECK(tag);

×

172

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));

×

173

    SSchema* s = schemaTag->pSchema + i;

×

174

    cJSON*   tname = cJSON_CreateString(s->name);

×

175

    RAW_NULL_CHECK(tname);

×

176

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "name", tname));

×

177

    cJSON* ttype = cJSON_CreateNumber(s->type);

×

178

    RAW_NULL_CHECK(ttype);

×

179

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "type", ttype));

×

180

    if (s->type == TSDB_DATA_TYPE_BINARY || s->type == TSDB_DATA_TYPE_VARBINARY || s->type == TSDB_DATA_TYPE_GEOMETRY) {

×

181

      int32_t length = s->bytes - VARSTR_HEADER_SIZE;

×

182

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

183

      RAW_NULL_CHECK(cbytes);

×

184

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "length", cbytes));

×

185

    } else if (s->type == TSDB_DATA_TYPE_NCHAR) {

×

186

      int32_t length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

187

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

188

      RAW_NULL_CHECK(cbytes);

×

189

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "length", cbytes));

×

193

end:

×

194

  *pJson = json;

×

197

static int32_t setCompressOption(cJSON* json, uint32_t para) {

×

198

  if (json == NULL) {

×

199

    return TSDB_CODE_INVALID_PARA;

×

201

  uint8_t encode = COMPRESS_L1_TYPE_U32(para);

×

202

  int32_t code = 0;

×

203

  if (encode != 0) {

×

204

    const char* encodeStr = columnEncodeStr(encode);

×

205

    RAW_NULL_CHECK(encodeStr);

×

206

    cJSON* encodeJson = cJSON_CreateString(encodeStr);

×

207

    RAW_NULL_CHECK(encodeJson);

×

208

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "encode", encodeJson));

×

209

    return code;

×

211

  uint8_t compress = COMPRESS_L2_TYPE_U32(para);

×

212

  if (compress != 0) {

×

213

    const char* compressStr = columnCompressStr(compress);

×

214

    RAW_NULL_CHECK(compressStr);

×

215

    cJSON* compressJson = cJSON_CreateString(compressStr);

×

216

    RAW_NULL_CHECK(compressJson);

×

217

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "compress", compressJson));

×

218

    return code;

×

220

  uint8_t level = COMPRESS_L2_TYPE_LEVEL_U32(para);

×

221

  if (level != 0) {

×

222

    const char* levelStr = columnLevelStr(level);

×

223

    RAW_NULL_CHECK(levelStr);

×

224

    cJSON* levelJson = cJSON_CreateString(levelStr);

×

225

    RAW_NULL_CHECK(levelJson);

×

226

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "level", levelJson));

×

227

    return code;

×

230

end:

×

231

  return code;

×

233

static void buildAlterSTableJson(void* alterData, int32_t alterDataLen, cJSON** pJson) {

×

234

  if (alterData == NULL || pJson == NULL) {

×

235

    uError("invalid parameter in %s", __func__);

×

236

    return;

×

238

  SMAlterStbReq req = {0};

×

239

  cJSON*        json = NULL;

×

240

  char*         string = NULL;

×

241

  int32_t       code = 0;

×

243

  if (tDeserializeSMAlterStbReq(alterData, alterDataLen, &req) != 0) {

×

244

    goto end;

×

247

  json = cJSON_CreateObject();

×

248

  RAW_NULL_CHECK(json);

×

249

  cJSON* type = cJSON_CreateString("alter");

×

250

  RAW_NULL_CHECK(type);

×

251

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));

×

252

  SName name = {0};

×

253

  RAW_RETURN_CHECK(tNameFromString(&name, req.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));

×

254

  cJSON* tableType = cJSON_CreateString("super");

×

255

  RAW_NULL_CHECK(tableType);

×

256

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));

×

257

  cJSON* tableName = cJSON_CreateString(name.tname);

×

258

  RAW_NULL_CHECK(tableName);

×

259

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));

×

261

  cJSON* alterType = cJSON_CreateNumber(req.alterType);

×

262

  RAW_NULL_CHECK(alterType);

×

263

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "alterType", alterType));

×

264

  switch (req.alterType) {

×

265

    case TSDB_ALTER_TABLE_ADD_TAG:

×

267

      TAOS_FIELD* field = taosArrayGet(req.pFields, 0);

×

268

      RAW_NULL_CHECK(field);

×

269

      cJSON* colName = cJSON_CreateString(field->name);

×

270

      RAW_NULL_CHECK(colName);

×

271

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

272

      cJSON* colType = cJSON_CreateNumber(field->type);

×

273

      RAW_NULL_CHECK(colType);

×

274

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));

×

276

      if (field->type == TSDB_DATA_TYPE_BINARY || field->type == TSDB_DATA_TYPE_VARBINARY ||

×

277

          field->type == TSDB_DATA_TYPE_GEOMETRY) {

×

278

        int32_t length = field->bytes - VARSTR_HEADER_SIZE;

×

279

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

280

        RAW_NULL_CHECK(cbytes);

×

281

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

282

      } else if (field->type == TSDB_DATA_TYPE_NCHAR) {

×

283

        int32_t length = (field->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

284

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

285

        RAW_NULL_CHECK(cbytes);

×

286

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

288

      break;

×

290

    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION: {

×

291

      SFieldWithOptions* field = taosArrayGet(req.pFields, 0);

×

292

      RAW_NULL_CHECK(field);

×

293

      cJSON* colName = cJSON_CreateString(field->name);

×

294

      RAW_NULL_CHECK(colName);

×

295

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

296

      cJSON* colType = cJSON_CreateNumber(field->type);

×

297

      RAW_NULL_CHECK(colType);

×

298

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));

×

300

      if (field->type == TSDB_DATA_TYPE_BINARY || field->type == TSDB_DATA_TYPE_VARBINARY ||

×

301

          field->type == TSDB_DATA_TYPE_GEOMETRY) {

×

302

        int32_t length = field->bytes - VARSTR_HEADER_SIZE;

×

303

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

304

        RAW_NULL_CHECK(cbytes);

×

305

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

306

      } else if (field->type == TSDB_DATA_TYPE_NCHAR) {

×

307

        int32_t length = (field->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

308

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

309

        RAW_NULL_CHECK(cbytes);

×

310

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

312

      RAW_RETURN_CHECK(setCompressOption(json, field->compress));

×

313

      break;

×

315

    case TSDB_ALTER_TABLE_DROP_TAG:

×

317

      TAOS_FIELD* field = taosArrayGet(req.pFields, 0);

×

318

      RAW_NULL_CHECK(field);

×

319

      cJSON* colName = cJSON_CreateString(field->name);

×

320

      RAW_NULL_CHECK(colName);

×

321

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

322

      break;

×

324

    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:

×

326

      TAOS_FIELD* field = taosArrayGet(req.pFields, 0);

×

327

      RAW_NULL_CHECK(field);

×

328

      cJSON* colName = cJSON_CreateString(field->name);

×

329

      RAW_NULL_CHECK(colName);

×

330

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

331

      cJSON* colType = cJSON_CreateNumber(field->type);

×

332

      RAW_NULL_CHECK(colType);

×

333

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));

×

334

      if (field->type == TSDB_DATA_TYPE_BINARY || field->type == TSDB_DATA_TYPE_VARBINARY ||

×

335

          field->type == TSDB_DATA_TYPE_GEOMETRY) {

×

336

        int32_t length = field->bytes - VARSTR_HEADER_SIZE;

×

337

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

338

        RAW_NULL_CHECK(cbytes);

×

339

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

340

      } else if (field->type == TSDB_DATA_TYPE_NCHAR) {

×

341

        int32_t length = (field->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

342

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

343

        RAW_NULL_CHECK(cbytes);

×

344

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

346

      break;

×

348

    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:

×

350

      TAOS_FIELD* oldField = taosArrayGet(req.pFields, 0);

×

351

      RAW_NULL_CHECK(oldField);

×

352

      TAOS_FIELD* newField = taosArrayGet(req.pFields, 1);

×

353

      RAW_NULL_CHECK(newField);

×

354

      cJSON* colName = cJSON_CreateString(oldField->name);

×

355

      RAW_NULL_CHECK(colName);

×

356

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

357

      cJSON* colNewName = cJSON_CreateString(newField->name);

×

358

      RAW_NULL_CHECK(colNewName);

×

359

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colNewName", colNewName));

×

360

      break;

×

362

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {

×

363

      TAOS_FIELD* field = taosArrayGet(req.pFields, 0);

×

364

      RAW_NULL_CHECK(field);

×

365

      cJSON* colName = cJSON_CreateString(field->name);

×

366

      RAW_NULL_CHECK(colName);

×

367

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

368

      RAW_RETURN_CHECK(setCompressOption(json, field->bytes));

×

369

      break;

×

371

    default:

×

372

      break;

×

375

end:

×

376

  tFreeSMAltertbReq(&req);

×

377

  *pJson = json;

×

380

static void processCreateStb(SMqMetaRsp* metaRsp, cJSON** pJson) {

×

381

  if (metaRsp == NULL || pJson == NULL) {

×

382

    uError("invalid parameter in %s", __func__);

×

383

    return;

×

385

  SVCreateStbReq req = {0};

×

386

  SDecoder       coder = {0};

×

388

  uDebug("create stable data:%p", metaRsp);

×

390

  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));

×

391

  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);

×

392

  tDecoderInit(&coder, data, len);

×

394

  if (tDecodeSVCreateStbReq(&coder, &req) < 0) {

×

395

    goto end;

×

397

  buildCreateTableJson(&req.schemaRow, &req.schemaTag, req.name, req.suid, TSDB_SUPER_TABLE, &req.colCmpr, pJson);

×

399

end:

×

400

  uDebug("create stable return");

×

401

  tDecoderClear(&coder);

×

404

static void processAlterStb(SMqMetaRsp* metaRsp, cJSON** pJson) {

×

405

  if (metaRsp == NULL || pJson == NULL) {

×

406

    uError("invalid parameter in %s", __func__);

×

407

    return;

×

409

  SVCreateStbReq req = {0};

×

410

  SDecoder       coder = {0};

×

411

  uDebug("alter stable data:%p", metaRsp);

×

414

  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));

×

415

  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);

×

416

  tDecoderInit(&coder, data, len);

×

418

  if (tDecodeSVCreateStbReq(&coder, &req) < 0) {

×

419

    goto end;

×

421

  buildAlterSTableJson(req.alterOriData, req.alterOriDataLen, pJson);

×

423

end:

×

424

  uDebug("alter stable return");

×

425

  tDecoderClear(&coder);

×

428

static void buildChildElement(cJSON* json, SVCreateTbReq* pCreateReq) {

×

429

  if (json == NULL || pCreateReq == NULL) {

×

430

    uError("invalid parameter in %s", __func__);

×

431

    return;

×

433

  STag*   pTag = (STag*)pCreateReq->ctb.pTag;

×

434

  char*   sname = pCreateReq->ctb.stbName;

×

435

  char*   name = pCreateReq->name;

×

436

  SArray* tagName = pCreateReq->ctb.tagName;

×

437

  int64_t id = pCreateReq->uid;

×

438

  uint8_t tagNum = pCreateReq->ctb.tagNum;

×

439

  int32_t code = 0;

×

440

  SArray* pTagVals = NULL;

×

441

  char*   pJson = NULL;

×

443

  cJSON*  tableName = cJSON_CreateString(name);

×

444

  RAW_NULL_CHECK(tableName);

×

445

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));

×

446

  cJSON* using = cJSON_CreateString(sname);

×

447

  RAW_NULL_CHECK(using);

×

448

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "using", using));

×

449

  cJSON* tagNumJson = cJSON_CreateNumber(tagNum);

×

450

  RAW_NULL_CHECK(tagNumJson);

×

451

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tagNum", tagNumJson));

×

453

  cJSON* tags = cJSON_CreateArray();

×

454

  RAW_NULL_CHECK(tags);

×

455

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tags", tags));

×

456

  RAW_RETURN_CHECK(tTagToValArray(pTag, &pTagVals));

×

457

  if (tTagIsJson(pTag)) {

×

458

    STag* p = (STag*)pTag;

×

459

    if (p->nTag == 0) {

×

460

      uError("p->nTag == 0");

×

461

      goto end;

×

463

    parseTagDatatoJson(pTag, &pJson, NULL);

×

464

    RAW_NULL_CHECK(pJson);

×

465

    cJSON* tag = cJSON_CreateObject();

×

466

    RAW_NULL_CHECK(tag);

×

467

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));

×

468

    STagVal* pTagVal = taosArrayGet(pTagVals, 0);

×

469

    RAW_NULL_CHECK(pTagVal);

×

470

    char* ptname = taosArrayGet(tagName, 0);

×

471

    RAW_NULL_CHECK(ptname);

×

472

    cJSON* tname = cJSON_CreateString(ptname);

×

473

    RAW_NULL_CHECK(tname);

×

474

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "name", tname));

×

475

    cJSON* ttype = cJSON_CreateNumber(TSDB_DATA_TYPE_JSON);

×

476

    RAW_NULL_CHECK(ttype);

×

477

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "type", ttype));

×

478

    cJSON* tvalue = cJSON_CreateString(pJson);

×

479

    RAW_NULL_CHECK(tvalue);

×

480

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "value", tvalue));

×

481

    goto end;

×

484

  for (int i = 0; i < taosArrayGetSize(pTagVals); i++) {

×

485

    STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, i);

×

486

    RAW_NULL_CHECK(pTagVal);

×

487

    cJSON* tag = cJSON_CreateObject();

×

488

    RAW_NULL_CHECK(tag);

×

489

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));

×

490

    char* ptname = taosArrayGet(tagName, i);

×

491

    RAW_NULL_CHECK(ptname);

×

492

    cJSON* tname = cJSON_CreateString(ptname);

×

493

    RAW_NULL_CHECK(tname);

×

494

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "name", tname));

×

495

    cJSON* ttype = cJSON_CreateNumber(pTagVal->type);

×

496

    RAW_NULL_CHECK(ttype);

×

497

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "type", ttype));

×

499

    cJSON* tvalue = NULL;

×

500

    if (IS_VAR_DATA_TYPE(pTagVal->type)) {

×

501

      int64_t bufSize = 0;

×

502

      if (pTagVal->type == TSDB_DATA_TYPE_VARBINARY) {

×

503

        bufSize = pTagVal->nData * 2 + 2 + 3;

×

505

        bufSize = pTagVal->nData + 3;

×

507

      char* buf = taosMemoryCalloc(bufSize, 1);

×

508

      RAW_NULL_CHECK(buf);

×

509

      if (dataConverToStr(buf, bufSize, pTagVal->type, pTagVal->pData, pTagVal->nData, NULL) != TSDB_CODE_SUCCESS) {

×

510

        taosMemoryFree(buf);

×

511

        goto end;

×

514

      tvalue = cJSON_CreateString(buf);

×

515

      taosMemoryFree(buf);

×

516

      RAW_NULL_CHECK(tvalue);

×

518

      double val = 0;

×

519

      GET_TYPED_DATA(val, double, pTagVal->type, &pTagVal->i64, 0); // currently tag type can't be decimal, so pass 0 as typeMod

×

520

      tvalue = cJSON_CreateNumber(val);

×

521

      RAW_NULL_CHECK(tvalue);

×

524

    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "value", tvalue));

×

527

end:

×

528

  taosMemoryFree(pJson);

×

529

  taosArrayDestroy(pTagVals);

×

532

static void buildCreateCTableJson(SVCreateTbReq* pCreateReq, int32_t nReqs, cJSON** pJson) {

×

533

  if (pJson == NULL || pCreateReq == NULL) {

×

534

    uError("invalid parameter in %s", __func__);

×

535

    return;

×

537

  int32_t code = 0;

×

538

  char*   string = NULL;

×

539

  cJSON*  json = cJSON_CreateObject();

×

540

  RAW_NULL_CHECK(json);

×

541

  cJSON* type = cJSON_CreateString("create");

×

542

  RAW_NULL_CHECK(type);

×

543

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));

×

545

  cJSON* tableType = cJSON_CreateString("child");

×

546

  RAW_NULL_CHECK(tableType);

×

547

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));

×

549

  buildChildElement(json, pCreateReq);

×

550

  cJSON* createList = cJSON_CreateArray();

×

551

  RAW_NULL_CHECK(createList);

×

552

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "createList", createList));

×

554

  for (int i = 0; nReqs > 1 && i < nReqs; i++) {

×

555

    cJSON* create = cJSON_CreateObject();

×

556

    RAW_NULL_CHECK(create);

×

557

    buildChildElement(create, pCreateReq + i);

×

558

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(createList, create));

×

561

end:

×

562

  *pJson = json;

×

565

static void processCreateTable(SMqMetaRsp* metaRsp, cJSON** pJson) {

×

566

  if (pJson == NULL || metaRsp == NULL) {

×

567

    uError("invalid parameter in %s", __func__);

×

568

    return;

×

570

  SDecoder           decoder = {0};

×

571

  SVCreateTbBatchReq req = {0};

×

574

  uDebug("create table data:%p", metaRsp);

×

575

  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));

×

576

  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);

×

577

  tDecoderInit(&decoder, data, len);

×

578

  if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {

×

579

    goto end;

×

583

  if (req.nReqs > 0) {

×

584

    pCreateReq = req.pReqs;

×

585

    if (pCreateReq->type == TSDB_CHILD_TABLE) {

×

586

      buildCreateCTableJson(req.pReqs, req.nReqs, pJson);

×

587

    } else if (pCreateReq->type == TSDB_NORMAL_TABLE) {

×

588

      buildCreateTableJson(&pCreateReq->ntb.schemaRow, NULL, pCreateReq->name, pCreateReq->uid, TSDB_NORMAL_TABLE,

×

593

end:

×

594

  uDebug("create table return");

×

595

  tDeleteSVCreateTbBatchReq(&req);

×

596

  tDecoderClear(&decoder);

×

599

static void processAutoCreateTable(SMqDataRsp* rsp, char** string) {

×

600

  if (rsp == NULL || string == NULL) {

×

601

    uError("invalid parameter in %s", __func__);

×

602

    return;

×

604

  SDecoder*      decoder = NULL;

×

605

  SVCreateTbReq* pCreateReq = NULL;

×

606

  int32_t        code = 0;

×

607

  uDebug("auto create table data:%p", rsp);

×

608

  if (rsp->createTableNum <= 0) {

×

609

    uError("processAutoCreateTable rsp->createTableNum <= 0");

×

610

    goto end;

×

613

  decoder = taosMemoryCalloc(rsp->createTableNum, sizeof(SDecoder));

×

614

  RAW_NULL_CHECK(decoder);

×

615

  pCreateReq = taosMemoryCalloc(rsp->createTableNum, sizeof(SVCreateTbReq));

×

616

  RAW_NULL_CHECK(pCreateReq);

×

619

  for (int32_t iReq = 0; iReq < rsp->createTableNum; iReq++) {

×

621

    void** data = taosArrayGet(rsp->createTableReq, iReq);

×

622

    RAW_NULL_CHECK(data);

×

623

    int32_t* len = taosArrayGet(rsp->createTableLen, iReq);

×

624

    RAW_NULL_CHECK(len);

×

625

    tDecoderInit(&decoder[iReq], *data, *len);

×

626

    if (tDecodeSVCreateTbReq(&decoder[iReq], pCreateReq + iReq) < 0) {

×

627

      goto end;

×

630

    if (pCreateReq[iReq].type != TSDB_CHILD_TABLE) {

×

631

      uError("processAutoCreateTable pCreateReq[iReq].type != TSDB_CHILD_TABLE");

×

632

      goto end;

×

635

  cJSON* pJson = NULL;

×

636

  buildCreateCTableJson(pCreateReq, rsp->createTableNum, &pJson);

×

637

  *string = cJSON_PrintUnformatted(pJson);

×

638

  cJSON_Delete(pJson);

×

640

end:

×

641

  uDebug("auto created table return, sql json:%s", *string);

×

642

  for (int i = 0; decoder && pCreateReq && i < rsp->createTableNum; i++) {

×

643

    tDecoderClear(&decoder[i]);

×

644

    taosMemoryFreeClear(pCreateReq[i].comment);

×

645

    if (pCreateReq[i].type == TSDB_CHILD_TABLE) {

×

646

      taosArrayDestroy(pCreateReq[i].ctb.tagName);

×

649

  taosMemoryFree(decoder);

×

650

  taosMemoryFree(pCreateReq);

×

653

static void processAlterTable(SMqMetaRsp* metaRsp, cJSON** pJson) {

×

654

  if (pJson == NULL || metaRsp == NULL) {

×

655

    uError("invalid parameter in %s", __func__);

×

656

    return;

×

658

  SDecoder     decoder = {0};

×

659

  SVAlterTbReq vAlterTbReq = {0};

×

660

  char*        string = NULL;

×

661

  cJSON*       json = NULL;

×

662

  int32_t      code = 0;

×

664

  uDebug("alter table data:%p", metaRsp);

×

666

  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));

×

667

  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);

×

668

  tDecoderInit(&decoder, data, len);

×

669

  if (tDecodeSVAlterTbReq(&decoder, &vAlterTbReq) < 0) {

×

670

    uError("tDecodeSVAlterTbReq error");

×

671

    goto end;

×

674

  json = cJSON_CreateObject();

×

675

  RAW_NULL_CHECK(json);

×

676

  cJSON* type = cJSON_CreateString("alter");

×

677

  RAW_NULL_CHECK(type);

×

678

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));

×

679

  cJSON* tableType = cJSON_CreateString(vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_TAG_VAL ||

×

680

                                                vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL

×

683

  RAW_NULL_CHECK(tableType);

×

684

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));

×

685

  cJSON* tableName = cJSON_CreateString(vAlterTbReq.tbName);

×

686

  RAW_NULL_CHECK(tableName);

×

687

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));

×

688

  cJSON* alterType = cJSON_CreateNumber(vAlterTbReq.action);

×

689

  RAW_NULL_CHECK(alterType);

×

690

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "alterType", alterType));

×

692

  switch (vAlterTbReq.action) {

×

693

    case TSDB_ALTER_TABLE_ADD_COLUMN: {

×

694

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

695

      RAW_NULL_CHECK(colName);

×

696

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

697

      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type);

×

698

      RAW_NULL_CHECK(colType);

×

699

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));

×

701

      if (vAlterTbReq.type == TSDB_DATA_TYPE_BINARY || vAlterTbReq.type == TSDB_DATA_TYPE_VARBINARY ||

×

702

          vAlterTbReq.type == TSDB_DATA_TYPE_GEOMETRY) {

×

703

        int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE;

×

704

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

705

        RAW_NULL_CHECK(cbytes);

×

706

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

707

      } else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR) {

×

708

        int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

709

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

710

        RAW_NULL_CHECK(cbytes);

×

711

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

713

      break;

×

715

    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION: {

×

716

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

717

      RAW_NULL_CHECK(colName);

×

718

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

719

      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type);

×

720

      RAW_NULL_CHECK(colType);

×

721

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));

×

723

      if (vAlterTbReq.type == TSDB_DATA_TYPE_BINARY || vAlterTbReq.type == TSDB_DATA_TYPE_VARBINARY ||

×

724

          vAlterTbReq.type == TSDB_DATA_TYPE_GEOMETRY) {

×

725

        int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE;

×

726

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

727

        RAW_NULL_CHECK(cbytes);

×

728

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

729

      } else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR) {

×

730

        int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

731

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

732

        RAW_NULL_CHECK(cbytes);

×

733

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

735

      RAW_RETURN_CHECK(setCompressOption(json, vAlterTbReq.compress));

×

736

      break;

×

738

    case TSDB_ALTER_TABLE_DROP_COLUMN: {

×

739

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

740

      RAW_NULL_CHECK(colName);

×

741

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

742

      break;

×

744

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: {

×

745

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

746

      RAW_NULL_CHECK(colName);

×

747

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

748

      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.colModType);

×

749

      RAW_NULL_CHECK(colType);

×

750

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));

×

751

      if (vAlterTbReq.colModType == TSDB_DATA_TYPE_BINARY || vAlterTbReq.colModType == TSDB_DATA_TYPE_VARBINARY ||

×

752

          vAlterTbReq.colModType == TSDB_DATA_TYPE_GEOMETRY) {

×

753

        int32_t length = vAlterTbReq.colModBytes - VARSTR_HEADER_SIZE;

×

754

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

755

        RAW_NULL_CHECK(cbytes);

×

756

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

757

      } else if (vAlterTbReq.colModType == TSDB_DATA_TYPE_NCHAR) {

×

758

        int32_t length = (vAlterTbReq.colModBytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

759

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

760

        RAW_NULL_CHECK(cbytes);

×

761

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));

×

763

      break;

×

765

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: {

×

766

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

767

      RAW_NULL_CHECK(colName);

×

768

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

769

      cJSON* colNewName = cJSON_CreateString(vAlterTbReq.colNewName);

×

770

      RAW_NULL_CHECK(colNewName);

×

771

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colNewName", colNewName));

×

772

      break;

×

774

    case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: {

×

775

      cJSON* tagName = cJSON_CreateString(vAlterTbReq.tagName);

×

776

      RAW_NULL_CHECK(tagName);

×

777

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", tagName));

×

779

      bool isNull = vAlterTbReq.isNull;

×

780

      if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) {

×

781

        STag* jsonTag = (STag*)vAlterTbReq.pTagVal;

×

782

        if (jsonTag->nTag == 0) isNull = true;

×

784

      if (!isNull) {

×

785

        char* buf = NULL;

×

787

        if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) {

×

788

          if (!tTagIsJson(vAlterTbReq.pTagVal)) {

×

789

            uError("processAlterTable isJson false");

×

790

            goto end;

×

792

          parseTagDatatoJson(vAlterTbReq.pTagVal, &buf, NULL);

×

793

          if (buf == NULL) {

×

794

            uError("parseTagDatatoJson failed, buf == NULL");

×

795

            goto end;

×

798

          int64_t bufSize = 0;

×

799

          if (vAlterTbReq.tagType == TSDB_DATA_TYPE_VARBINARY) {

×

800

            bufSize = vAlterTbReq.nTagVal * 2 + 2 + 3;

×

802

            bufSize = vAlterTbReq.nTagVal + 32;

×

804

          buf = taosMemoryCalloc(bufSize, 1);

×

805

          RAW_NULL_CHECK(buf);

×

806

          if (dataConverToStr(buf, bufSize, vAlterTbReq.tagType, vAlterTbReq.pTagVal, vAlterTbReq.nTagVal, NULL) !=

×

808

            taosMemoryFree(buf);

×

809

            goto end;

×

813

        cJSON* colValue = cJSON_CreateString(buf);

×

814

        taosMemoryFree(buf);

×

815

        RAW_NULL_CHECK(colValue);

×

816

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colValue", colValue));

×

819

      cJSON* isNullCJson = cJSON_CreateBool(isNull);

×

820

      RAW_NULL_CHECK(isNullCJson);

×

821

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colValueNull", isNullCJson));

×

822

      break;

×

824

    case TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL: {

×

825

      int32_t nTags = taosArrayGetSize(vAlterTbReq.pMultiTag);

×

826

      if (nTags <= 0) {

×

827

        uError("processAlterTable parse multi tags error");

×

828

        goto end;

×

831

      cJSON* tags = cJSON_CreateArray();

×

832

      RAW_NULL_CHECK(tags);

×

833

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tags", tags));

×

835

      for (int32_t i = 0; i < nTags; i++) {

×

836

        cJSON* member = cJSON_CreateObject();

×

837

        RAW_NULL_CHECK(member);

×

838

        RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, member));

×

840

        SMultiTagUpateVal* pTagVal = taosArrayGet(vAlterTbReq.pMultiTag, i);

×

841

        cJSON*             tagName = cJSON_CreateString(pTagVal->tagName);

×

842

        RAW_NULL_CHECK(tagName);

×

843

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(member, "colName", tagName));

×

845

        if (pTagVal->tagType == TSDB_DATA_TYPE_JSON) {

×

846

          uError("processAlterTable isJson false");

×

847

          goto end;

×

849

        bool isNull = pTagVal->isNull;

×

850

        if (!isNull) {

×

851

          int64_t bufSize = 0;

×

852

          if (pTagVal->tagType == TSDB_DATA_TYPE_VARBINARY) {

×

853

            bufSize = pTagVal->nTagVal * 2 + 2 + 3;

×

855

            bufSize = pTagVal->nTagVal + 3;

×

857

          char* buf = taosMemoryCalloc(bufSize, 1);

×

858

          RAW_NULL_CHECK(buf);

×

859

          if (dataConverToStr(buf, bufSize, pTagVal->tagType, pTagVal->pTagVal, pTagVal->nTagVal, NULL) !=

×

861

            taosMemoryFree(buf);

×

862

            goto end;

×

864

          cJSON* colValue = cJSON_CreateString(buf);

×

865

          taosMemoryFree(buf);

×

866

          RAW_NULL_CHECK(colValue);

×

867

          RAW_FALSE_CHECK(tmqAddJsonObjectItem(member, "colValue", colValue));

×

869

        cJSON* isNullCJson = cJSON_CreateBool(isNull);

×

870

        RAW_NULL_CHECK(isNullCJson);

×

871

        RAW_FALSE_CHECK(tmqAddJsonObjectItem(member, "colValueNull", isNullCJson));

×

873

      break;

×

876

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {

×

877

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

878

      RAW_NULL_CHECK(colName);

×

879

      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));

×

880

      RAW_RETURN_CHECK(setCompressOption(json, vAlterTbReq.compress));

×

881

      break;

×

883

    default:

×

884

      break;

×

887

end:

×

888

  uDebug("alter table return");

×

889

  if (vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL) {

×

890

    taosArrayDestroy(vAlterTbReq.pMultiTag);

×

892

  tDecoderClear(&decoder);

×

893

  *pJson = json;

×

896

static void processDropSTable(SMqMetaRsp* metaRsp, cJSON** pJson) {

×

897

  if (pJson == NULL || metaRsp == NULL) {

×

898

    uError("invalid parameter in %s", __func__);

×

899

    return;

×

901

  SDecoder     decoder = {0};

×

902

  SVDropStbReq req = {0};

×

903

  cJSON*       json = NULL;

×

904

  int32_t      code = 0;

×

906

  uDebug("processDropSTable data:%p", metaRsp);

×

909

  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));

×

910

  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);

×

911

  tDecoderInit(&decoder, data, len);

×

912

  if (tDecodeSVDropStbReq(&decoder, &req) < 0) {

×

913

    uError("tDecodeSVDropStbReq failed");

×

914

    goto end;

×

917

  json = cJSON_CreateObject();

×

918

  RAW_NULL_CHECK(json);

×

919

  cJSON* type = cJSON_CreateString("drop");

×

920

  RAW_NULL_CHECK(type);

×

921

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));

×

922

  cJSON* tableType = cJSON_CreateString("super");

×

923

  RAW_NULL_CHECK(tableType);

×

924

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));

×

925

  cJSON* tableName = cJSON_CreateString(req.name);

×

926

  RAW_NULL_CHECK(tableName);

×

927

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));

×

929

end:

×

930

  uDebug("processDropSTable return");

×

931

  tDecoderClear(&decoder);

×

932

  *pJson = json;

×

934

static void processDeleteTable(SMqMetaRsp* metaRsp, cJSON** pJson) {

×

935

  if (pJson == NULL || metaRsp == NULL) {

×

936

    uError("invalid parameter in %s", __func__);

×

937

    return;

×

939

  SDeleteRes req = {0};

×

940

  SDecoder   coder = {0};

×

941

  cJSON*     json = NULL;

×

942

  int32_t    code = 0;

×

944

  uDebug("processDeleteTable data:%p", metaRsp);

×

946

  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));

×

947

  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);

×

949

  tDecoderInit(&coder, data, len);

×

950

  if (tDecodeDeleteRes(&coder, &req) < 0) {

×

951

    uError("tDecodeDeleteRes failed");

×

952

    goto end;

×

956

  char sql[256] = {0};

×

957

  (void)snprintf(sql, sizeof(sql), "delete from `%s` where `%s` >= %" PRId64 " and `%s` <= %" PRId64, req.tableFName,

×

960

  json = cJSON_CreateObject();

×

961

  RAW_NULL_CHECK(json);

×

962

  cJSON* type = cJSON_CreateString("delete");

×

963

  RAW_NULL_CHECK(type);

×

964

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));

×

965

  cJSON* sqlJson = cJSON_CreateString(sql);

×

966

  RAW_NULL_CHECK(sqlJson);

×

967

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "sql", sqlJson));

×

969

end:

×

970

  uDebug("processDeleteTable return");

×

971

  tDecoderClear(&coder);

×

972

  *pJson = json;

×

975

static void processDropTable(SMqMetaRsp* metaRsp, cJSON** pJson) {

×

976

  if (pJson == NULL || metaRsp == NULL) {

×

977

    uError("invalid parameter in %s", __func__);

×

978

    return;

×

980

  SDecoder         decoder = {0};

×

981

  SVDropTbBatchReq req = {0};

×

982

  cJSON*           json = NULL;

×

983

  int32_t          code = 0;

×

985

  uDebug("processDropTable data:%p", metaRsp);

×

987

  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));

×

988

  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);

×

989

  tDecoderInit(&decoder, data, len);

×

990

  if (tDecodeSVDropTbBatchReq(&decoder, &req) < 0) {

×

991

    uError("tDecodeSVDropTbBatchReq failed");

×

992

    goto end;

×

995

  json = cJSON_CreateObject();

×

996

  RAW_NULL_CHECK(json);

×

997

  cJSON* type = cJSON_CreateString("drop");

×

998

  RAW_NULL_CHECK(type);

×

999

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));

×

1000

  cJSON* tableNameList = cJSON_CreateArray();

×

1001

  RAW_NULL_CHECK(tableNameList);

×

1002

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableNameList", tableNameList));

×

1004

  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {

×

1005

    SVDropTbReq* pDropTbReq = req.pReqs + iReq;

×

1006

    cJSON*       tableName = cJSON_CreateString(pDropTbReq->name);

×

1007

    RAW_NULL_CHECK(tableName);

×

1008

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tableNameList, tableName));

×

1011

end:

×

1012

  uDebug("processDropTable return");

×

1013

  tDecoderClear(&decoder);

×

1014

  *pJson = json;

×

1017

static int32_t taosCreateStb(TAOS* taos, void* meta, uint32_t metaLen) {

×

1018

  if (taos == NULL || meta == NULL) {

×

1019

    uError("invalid parameter in %s", __func__);

×

1020

    return TSDB_CODE_INVALID_PARA;

×

1022

  SVCreateStbReq req = {0};

×

1023

  SDecoder       coder = {0};

×

1024

  SMCreateStbReq pReq = {0};

×

1025

  int32_t        code = TSDB_CODE_SUCCESS;

×

1026

  SRequestObj*   pRequest = NULL;

×

1028

  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));

×

1029

  uDebug(LOG_ID_TAG " create stable, meta:%p, metaLen:%d", LOG_ID_VALUE, meta, metaLen);

×

1030

  pRequest->syncQuery = true;

×

1031

  if (!pRequest->pDb) {

×

1032

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1033

    goto end;

×

1036

  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));

×

1037

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1038

  tDecoderInit(&coder, data, len);

×

1039

  if (tDecodeSVCreateStbReq(&coder, &req) < 0) {

×

1040

    code = TSDB_CODE_INVALID_PARA;

×

1041

    goto end;

×

1044

  int8_t           createDefaultCompress = 0;

×

1045

  SColCmprWrapper* p = &req.colCmpr;

×

1046

  if (p->nCols == 0) {

×

1047

    createDefaultCompress = 1;

×

1050

  pReq.pColumns = taosArrayInit(req.schemaRow.nCols, sizeof(SFieldWithOptions));

×

1051

  RAW_NULL_CHECK(pReq.pColumns);

×

1052

  for (int32_t i = 0; i < req.schemaRow.nCols; i++) {

×

1053

    SSchema*          pSchema = req.schemaRow.pSchema + i;

×

1054

    SFieldWithOptions field = {.type = pSchema->type, .flags = pSchema->flags, .bytes = pSchema->bytes};

×

1055

    tstrncpy(field.name, pSchema->name, TSDB_COL_NAME_LEN);

×

1057

    if (createDefaultCompress) {

×

1058

      field.compress = createDefaultColCmprByType(pSchema->type);

×

1060

      SColCmpr* pCmp = &req.colCmpr.pColCmpr[i];

×

1061

      field.compress = pCmp->alg;

×

1063

    if (req.pExtSchemas) field.typeMod = req.pExtSchemas[i].typeMod;

×

1064

    RAW_NULL_CHECK(taosArrayPush(pReq.pColumns, &field));

×

1066

  pReq.pTags = taosArrayInit(req.schemaTag.nCols, sizeof(SField));

×

1067

  RAW_NULL_CHECK(pReq.pTags);

×

1068

  for (int32_t i = 0; i < req.schemaTag.nCols; i++) {

×

1069

    SSchema* pSchema = req.schemaTag.pSchema + i;

×

1070

    SField   field = {.type = pSchema->type, .flags = pSchema->flags, .bytes = pSchema->bytes};

×

1071

    tstrncpy(field.name, pSchema->name, TSDB_COL_NAME_LEN);

×

1072

    RAW_NULL_CHECK(taosArrayPush(pReq.pTags, &field));

×

1075

  pReq.colVer = req.schemaRow.version;

×

1076

  pReq.tagVer = req.schemaTag.version;

×

1077

  pReq.numOfColumns = req.schemaRow.nCols;

×

1078

  pReq.numOfTags = req.schemaTag.nCols;

×

1079

  pReq.commentLen = -1;

×

1080

  pReq.suid = processSuid(req.suid, pRequest->pDb);

×

1081

  pReq.source = TD_REQ_FROM_TAOX;

×

1082

  pReq.igExists = true;

×

1084

  uDebug(LOG_ID_TAG " create stable name:%s suid:%" PRId64 " processSuid:%" PRId64, LOG_ID_VALUE, req.name, req.suid,

×

1086

  STscObj* pTscObj = pRequest->pTscObj;

×

1087

  SName    tableName = {0};

×

1088

  toName(pTscObj->acctId, pRequest->pDb, req.name, &tableName);

×

1089

  RAW_RETURN_CHECK(tNameExtractFullName(&tableName, pReq.name));

×

1090

  SCmdMsgInfo pCmdMsg = {0};

×

1091

  pCmdMsg.epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

×

1092

  pCmdMsg.msgType = TDMT_MND_CREATE_STB;

×

1093

  pCmdMsg.msgLen = tSerializeSMCreateStbReq(NULL, 0, &pReq);

×

1094

  if (pCmdMsg.msgLen <= 0) {

×

1095

    code = TSDB_CODE_INVALID_PARA;

×

1096

    goto end;

×

1098

  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);

×

1099

  RAW_NULL_CHECK(pCmdMsg.pMsg);

×

1100

  if (tSerializeSMCreateStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq) <= 0) {

×

1101

    code = TSDB_CODE_INVALID_PARA;

×

1102

    taosMemoryFree(pCmdMsg.pMsg);

×

1103

    goto end;

×

1106

  SQuery pQuery = {0};

×

1107

  pQuery.execMode = QUERY_EXEC_MODE_RPC;

×

1108

  pQuery.pCmdMsg = &pCmdMsg;

×

1109

  pQuery.msgType = pQuery.pCmdMsg->msgType;

×

1110

  pQuery.stableQuery = true;

×

1112

  launchQueryImpl(pRequest, &pQuery, true, NULL);  // ignore, because return value is pRequest

×

1114

  taosMemoryFree(pCmdMsg.pMsg);

×

1116

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1117

    SCatalog* pCatalog = NULL;

×

1118

    RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));

×

1119

    RAW_RETURN_CHECK(catalogRemoveTableMeta(pCatalog, &tableName));

×

1122

  code = pRequest->code;

×

1124

end:

×

1125

  uDebug(LOG_ID_TAG " create stable return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

1126

  destroyRequest(pRequest);

×

1127

  tFreeSMCreateStbReq(&pReq);

×

1128

  tDecoderClear(&coder);

×

1129

  return code;

×

1132

static int32_t taosDropStb(TAOS* taos, void* meta, uint32_t metaLen) {

×

1133

  if (taos == NULL || meta == NULL) {

×

1134

    uError("invalid parameter in %s", __func__);

×

1135

    return TSDB_CODE_INVALID_PARA;

×

1137

  SVDropStbReq req = {0};

×

1138

  SDecoder     coder = {0};

×

1139

  SMDropStbReq pReq = {0};

×

1140

  int32_t      code = TSDB_CODE_SUCCESS;

×

1141

  SRequestObj* pRequest = NULL;

×

1143

  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));

×

1144

  uDebug(LOG_ID_TAG " drop stable, meta:%p, metaLen:%d", LOG_ID_VALUE, meta, metaLen);

×

1145

  pRequest->syncQuery = true;

×

1146

  if (!pRequest->pDb) {

×

1147

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1148

    goto end;

×

1151

  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));

×

1152

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1153

  tDecoderInit(&coder, data, len);

×

1154

  if (tDecodeSVDropStbReq(&coder, &req) < 0) {

×

1155

    code = TSDB_CODE_INVALID_PARA;

×

1156

    goto end;

×

1159

  SCatalog* pCatalog = NULL;

×

1160

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

×

1161

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

×

1162

                           .requestId = pRequest->requestId,

×

1163

                           .requestObjRefId = pRequest->self,

×

1164

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

×

1165

  SName            pName = {0};

×

1166

  toName(pRequest->pTscObj->acctId, pRequest->pDb, req.name, &pName);

×

1167

  STableMeta* pTableMeta = NULL;

×

1168

  code = catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta);

×

1169

  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1170

    code = TSDB_CODE_SUCCESS;

×

1171

    taosMemoryFreeClear(pTableMeta);

×

1172

    goto end;

×

1174

  if (code != TSDB_CODE_SUCCESS) {

×

1175

    goto end;

×

1177

  pReq.suid = pTableMeta->uid;

×

1178

  taosMemoryFreeClear(pTableMeta);

×

1181

  pReq.igNotExists = true;

×

1182

  pReq.source = TD_REQ_FROM_TAOX;

×

1185

  uDebug(LOG_ID_TAG " drop stable name:%s suid:%" PRId64 " new suid:%" PRId64, LOG_ID_VALUE, req.name, req.suid,

×

1187

  STscObj* pTscObj = pRequest->pTscObj;

×

1188

  SName    tableName = {0};

×

1189

  toName(pTscObj->acctId, pRequest->pDb, req.name, &tableName);

×

1190

  if (tNameExtractFullName(&tableName, pReq.name) != 0) {

×

1191

    code = TSDB_CODE_INVALID_PARA;

×

1192

    goto end;

×

1195

  SCmdMsgInfo pCmdMsg = {0};

×

1196

  pCmdMsg.epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);

×

1197

  pCmdMsg.msgType = TDMT_MND_DROP_STB;

×

1198

  pCmdMsg.msgLen = tSerializeSMDropStbReq(NULL, 0, &pReq);

×

1199

  if (pCmdMsg.msgLen <= 0) {

×

1200

    code = TSDB_CODE_INVALID_PARA;

×

1201

    goto end;

×

1203

  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);

×

1204

  RAW_NULL_CHECK(pCmdMsg.pMsg);

×

1205

  if (tSerializeSMDropStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq) <= 0) {

×

1206

    code = TSDB_CODE_INVALID_PARA;

×

1207

    taosMemoryFree(pCmdMsg.pMsg);

×

1208

    goto end;

×

1211

  SQuery pQuery = {0};

×

1212

  pQuery.execMode = QUERY_EXEC_MODE_RPC;

×

1213

  pQuery.pCmdMsg = &pCmdMsg;

×

1214

  pQuery.msgType = pQuery.pCmdMsg->msgType;

×

1215

  pQuery.stableQuery = true;

×

1217

  launchQueryImpl(pRequest, &pQuery, true, NULL);  // ignore, because return value is pRequest

×

1218

  taosMemoryFree(pCmdMsg.pMsg);

×

1219

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1221

    RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));

×

1222

    RAW_RETURN_CHECK(catalogRemoveTableMeta(pCatalog, &tableName));

×

1225

  code = pRequest->code;

×

1227

end:

×

1228

  uDebug(LOG_ID_TAG " drop stable return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

1229

  destroyRequest(pRequest);

×

1230

  tDecoderClear(&coder);

×

1231

  return code;

×

1240

static void destroyCreateTbReqBatch(void* data) {

×

1241

  if (data == NULL) {

×

1242

    uError("invalid parameter in %s", __func__);

×

1243

    return;

×

1245

  SVgroupCreateTableBatch* pTbBatch = (SVgroupCreateTableBatch*)data;

×

1246

  taosArrayDestroy(pTbBatch->req.pArray);

×

1249

static int32_t taosCreateTable(TAOS* taos, void* meta, uint32_t metaLen) {

×

1250

  if (taos == NULL || meta == NULL) {

×

1251

    uError("invalid parameter in %s", __func__);

×

1252

    return TSDB_CODE_INVALID_PARA;

×

1254

  SVCreateTbBatchReq req = {0};

×

1255

  SDecoder           coder = {0};

×

1256

  int32_t            code = TSDB_CODE_SUCCESS;

×

1257

  SRequestObj*       pRequest = NULL;

×

1258

  SQuery*            pQuery = NULL;

×

1259

  SHashObj*          pVgroupHashmap = NULL;

×

1260

  SArray*            pTagList = taosArrayInit(0, POINTER_BYTES);

×

1261

  RAW_NULL_CHECK(pTagList);

×

1262

  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));

×

1263

  uDebug(LOG_ID_TAG " create table, meta:%p, metaLen:%d", LOG_ID_VALUE, meta, metaLen);

×

1265

  pRequest->syncQuery = true;

×

1266

  if (!pRequest->pDb) {

×

1267

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1268

    goto end;

×

1271

  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));

×

1272

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1273

  tDecoderInit(&coder, data, len);

×

1274

  if (tDecodeSVCreateTbBatchReq(&coder, &req) < 0) {

×

1275

    code = TSDB_CODE_INVALID_PARA;

×

1276

    goto end;

×

1279

  STscObj* pTscObj = pRequest->pTscObj;

×

1281

  SVCreateTbReq* pCreateReq = NULL;

×

1282

  SCatalog*      pCatalog = NULL;

×

1283

  RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));

×

1284

  pVgroupHashmap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);

×

1285

  RAW_NULL_CHECK(pVgroupHashmap);

×

1286

  taosHashSetFreeFp(pVgroupHashmap, destroyCreateTbReqBatch);

×

1288

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

×

1289

                           .requestId = pRequest->requestId,

×

1290

                           .requestObjRefId = pRequest->self,

×

1291

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

×

1293

  pRequest->tableList = taosArrayInit(req.nReqs, sizeof(SName));

×

1294

  RAW_NULL_CHECK(pRequest->tableList);

×

1296

  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {

×

1297

    pCreateReq = req.pReqs + iReq;

×

1299

    SVgroupInfo pInfo = {0};

×

1300

    SName       pName = {0};

×

1301

    toName(pTscObj->acctId, pRequest->pDb, pCreateReq->name, &pName);

×

1302

    code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo);

×

1303

    if (code != TSDB_CODE_SUCCESS) {

×

1304

      goto end;

×

1307

    pCreateReq->flags |= TD_CREATE_IF_NOT_EXISTS;

×

1309

    if (pCreateReq->type == TSDB_CHILD_TABLE) {

×

1310

      STableMeta* pTableMeta = NULL;

×

1311

      SName       sName = {0};

×

1312

      tb_uid_t    oldSuid = pCreateReq->ctb.suid;

×

1314

      toName(pTscObj->acctId, pRequest->pDb, pCreateReq->ctb.stbName, &sName);

×

1315

      code = catalogGetTableMeta(pCatalog, &conn, &sName, &pTableMeta);

×

1316

      if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1317

        code = TSDB_CODE_SUCCESS;

×

1318

        taosMemoryFreeClear(pTableMeta);

×

1319

        continue;

×

1322

      if (code != TSDB_CODE_SUCCESS) {

×

1323

        goto end;

×

1325

      pCreateReq->ctb.suid = pTableMeta->uid;

×

1327

      SArray* pTagVals = NULL;

×

1328

      code = tTagToValArray((STag*)pCreateReq->ctb.pTag, &pTagVals);

×

1329

      if (code != TSDB_CODE_SUCCESS) {

×

1330

        taosMemoryFreeClear(pTableMeta);

×

1331

        goto end;

×

1334

      bool rebuildTag = false;

×

1335

      for (int32_t i = 0; i < taosArrayGetSize(pCreateReq->ctb.tagName); i++) {

×

1336

        char* tName = taosArrayGet(pCreateReq->ctb.tagName, i);

×

1337

        if (tName == NULL) {

×

1338

          continue;

×

1340

        for (int32_t j = pTableMeta->tableInfo.numOfColumns;

×

1341

             j < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; j++) {

×

1342

          SSchema* tag = &pTableMeta->schema[j];

×

1343

          if (strcmp(tag->name, tName) == 0 && tag->type != TSDB_DATA_TYPE_JSON) {

×

1344

            STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, i);

×

1345

            if (pTagVal) {

×

1346

              if (pTagVal->cid != tag->colId) {

×

1347

                pTagVal->cid = tag->colId;

×

1348

                rebuildTag = true;

×

1351

              uError("create tb invalid data %s, size:%d index:%d cid:%d", pCreateReq->name,

×

1357

      taosMemoryFreeClear(pTableMeta);

×

1358

      if (rebuildTag) {

×

1359

        STag* ppTag = NULL;

×

1360

        code = tTagNew(pTagVals, 1, false, &ppTag);

×

1361

        taosArrayDestroy(pTagVals);

×

1362

        pTagVals = NULL;

×

1363

        if (code != TSDB_CODE_SUCCESS) {

×

1364

          goto end;

×

1366

        if (NULL == taosArrayPush(pTagList, &ppTag)) {

×

1367

          tTagFree(ppTag);

×

1368

          goto end;

×

1370

        pCreateReq->ctb.pTag = (uint8_t*)ppTag;

×

1372

      taosArrayDestroy(pTagVals);

×

1374

    RAW_NULL_CHECK(taosArrayPush(pRequest->tableList, &pName));

×

1376

    SVgroupCreateTableBatch* pTableBatch = taosHashGet(pVgroupHashmap, &pInfo.vgId, sizeof(pInfo.vgId));

×

1377

    if (pTableBatch == NULL) {

×

1378

      SVgroupCreateTableBatch tBatch = {0};

×

1379

      tBatch.info = pInfo;

×

1380

      tstrncpy(tBatch.dbName, pRequest->pDb, TSDB_DB_NAME_LEN);

×

1382

      tBatch.req.pArray = taosArrayInit(4, sizeof(struct SVCreateTbReq));

×

1383

      RAW_NULL_CHECK(tBatch.req.pArray);

×

1384

      RAW_NULL_CHECK(taosArrayPush(tBatch.req.pArray, pCreateReq));

×

1385

      tBatch.req.source = TD_REQ_FROM_TAOX;

×

1386

      RAW_RETURN_CHECK(taosHashPut(pVgroupHashmap, &pInfo.vgId, sizeof(pInfo.vgId), &tBatch, sizeof(tBatch)));

×

1388

      RAW_NULL_CHECK(taosArrayPush(pTableBatch->req.pArray, pCreateReq));

×

1392

  if (taosHashGetSize(pVgroupHashmap) == 0) {

×

1393

    goto end;

×

1395

  SArray* pBufArray = NULL;

×

1396

  RAW_RETURN_CHECK(serializeVgroupsCreateTableBatch(pVgroupHashmap, &pBufArray));

×

1397

  pQuery = NULL;

×

1398

  code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);

×

1399

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1400

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1401

  pQuery->msgType = TDMT_VND_CREATE_TABLE;

×

1402

  pQuery->stableQuery = false;

×

1403

  code = nodesMakeNode(QUERY_NODE_CREATE_TABLE_STMT, &pQuery->pRoot);

×

1404

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1405

  RAW_NULL_CHECK(pQuery->pRoot);

×

1407

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pBufArray));

×

1409

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1410

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1411

    RAW_RETURN_CHECK(removeMeta(pTscObj, pRequest->tableList, false));

×

1414

  code = pRequest->code;

×

1416

end:

×

1417

  uDebug(LOG_ID_TAG " create table return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

1418

  tDeleteSVCreateTbBatchReq(&req);

×

1420

  taosHashCleanup(pVgroupHashmap);

×

1421

  destroyRequest(pRequest);

×

1422

  tDecoderClear(&coder);

×

1423

  qDestroyQuery(pQuery);

×

1424

  taosArrayDestroyP(pTagList, NULL);

×

1425

  return code;

×

1434

static void destroyDropTbReqBatch(void* data) {

×

1435

  if (data == NULL) {

×

1436

    uError("invalid parameter in %s", __func__);

×

1437

    return;

×

1439

  SVgroupDropTableBatch* pTbBatch = (SVgroupDropTableBatch*)data;

×

1440

  taosArrayDestroy(pTbBatch->req.pArray);

×

1443

static int32_t taosDropTable(TAOS* taos, void* meta, uint32_t metaLen) {

×

1444

  if (taos == NULL || meta == NULL) {

×

1445

    uError("invalid parameter in %s", __func__);

×

1446

    return TSDB_CODE_INVALID_PARA;

×

1448

  SVDropTbBatchReq req = {0};

×

1449

  SDecoder         coder = {0};

×

1450

  int32_t          code = TSDB_CODE_SUCCESS;

×

1451

  SRequestObj*     pRequest = NULL;

×

1452

  SQuery*          pQuery = NULL;

×

1453

  SHashObj*        pVgroupHashmap = NULL;

×

1455

  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));

×

1456

  uDebug(LOG_ID_TAG " drop table, meta:%p, len:%d", LOG_ID_VALUE, meta, metaLen);

×

1458

  pRequest->syncQuery = true;

×

1459

  if (!pRequest->pDb) {

×

1460

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1461

    goto end;

×

1464

  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));

×

1465

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1466

  tDecoderInit(&coder, data, len);

×

1467

  if (tDecodeSVDropTbBatchReq(&coder, &req) < 0) {

×

1468

    code = TSDB_CODE_INVALID_PARA;

×

1469

    goto end;

×

1472

  STscObj* pTscObj = pRequest->pTscObj;

×

1474

  SVDropTbReq* pDropReq = NULL;

×

1475

  SCatalog*    pCatalog = NULL;

×

1476

  RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));

×

1478

  pVgroupHashmap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);

×

1479

  RAW_NULL_CHECK(pVgroupHashmap);

×

1480

  taosHashSetFreeFp(pVgroupHashmap, destroyDropTbReqBatch);

×

1482

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

×

1483

                           .requestId = pRequest->requestId,

×

1484

                           .requestObjRefId = pRequest->self,

×

1485

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

×

1486

  pRequest->tableList = taosArrayInit(req.nReqs, sizeof(SName));

×

1487

  RAW_NULL_CHECK(pRequest->tableList);

×

1489

  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {

×

1490

    pDropReq = req.pReqs + iReq;

×

1491

    pDropReq->igNotExists = true;

×

1494

    SVgroupInfo pInfo = {0};

×

1495

    SName       pName = {0};

×

1496

    toName(pTscObj->acctId, pRequest->pDb, pDropReq->name, &pName);

×

1497

    RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo));

×

1499

    STableMeta* pTableMeta = NULL;

×

1500

    code = catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta);

×

1501

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1502

      code = TSDB_CODE_SUCCESS;

×

1503

      taosMemoryFreeClear(pTableMeta);

×

1504

      continue;

×

1506

    if (code != TSDB_CODE_SUCCESS) {

×

1507

      goto end;

×

1509

    tb_uid_t oldSuid = pDropReq->suid;

×

1510

    pDropReq->suid = pTableMeta->suid;

×

1511

    taosMemoryFreeClear(pTableMeta);

×

1512

    uDebug(LOG_ID_TAG " drop table name:%s suid:%" PRId64 " new suid:%" PRId64, LOG_ID_VALUE, pDropReq->name, oldSuid,

×

1515

    RAW_NULL_CHECK(taosArrayPush(pRequest->tableList, &pName));

×

1516

    SVgroupDropTableBatch* pTableBatch = taosHashGet(pVgroupHashmap, &pInfo.vgId, sizeof(pInfo.vgId));

×

1517

    if (pTableBatch == NULL) {

×

1518

      SVgroupDropTableBatch tBatch = {0};

×

1519

      tBatch.info = pInfo;

×

1520

      tBatch.req.pArray = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbReq));

×

1521

      RAW_NULL_CHECK(tBatch.req.pArray);

×

1522

      RAW_NULL_CHECK(taosArrayPush(tBatch.req.pArray, pDropReq));

×

1523

      RAW_RETURN_CHECK(taosHashPut(pVgroupHashmap, &pInfo.vgId, sizeof(pInfo.vgId), &tBatch, sizeof(tBatch)));

×

1525

      RAW_NULL_CHECK(taosArrayPush(pTableBatch->req.pArray, pDropReq));

×

1529

  if (taosHashGetSize(pVgroupHashmap) == 0) {

×

1530

    goto end;

×

1532

  SArray* pBufArray = NULL;

×

1533

  RAW_RETURN_CHECK(serializeVgroupsDropTableBatch(pVgroupHashmap, &pBufArray));

×

1534

  code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);

×

1535

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1536

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1537

  pQuery->msgType = TDMT_VND_DROP_TABLE;

×

1538

  pQuery->stableQuery = false;

×

1539

  pQuery->pRoot = NULL;

×

1540

  code = nodesMakeNode(QUERY_NODE_DROP_TABLE_STMT, &pQuery->pRoot);

×

1541

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1542

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pBufArray));

×

1544

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1545

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1546

    RAW_RETURN_CHECK(removeMeta(pTscObj, pRequest->tableList, false));

×

1548

  code = pRequest->code;

×

1550

end:

×

1551

  uDebug(LOG_ID_TAG " drop table return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

1552

  taosHashCleanup(pVgroupHashmap);

×

1553

  destroyRequest(pRequest);

×

1554

  tDecoderClear(&coder);

×

1555

  qDestroyQuery(pQuery);

×

1556

  return code;

×

1559

static int32_t taosDeleteData(TAOS* taos, void* meta, uint32_t metaLen) {

×

1560

  if (taos == NULL || meta == NULL) {

×

1561

    uError("invalid parameter in %s", __func__);

×

1562

    return TSDB_CODE_INVALID_PARA;

×

1564

  SDeleteRes req = {0};

×

1565

  SDecoder   coder = {0};

×

1566

  char       sql[256] = {0};

×

1567

  int32_t    code = TSDB_CODE_SUCCESS;

×

1569

  uDebug("connId:0x%" PRIx64 " delete data, meta:%p, len:%d", *(int64_t*)taos, meta, metaLen);

×

1572

  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));

×

1573

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1574

  tDecoderInit(&coder, data, len);

×

1575

  if (tDecodeDeleteRes(&coder, &req) < 0) {

×

1576

    code = TSDB_CODE_INVALID_PARA;

×

1577

    goto end;

×

1580

  (void)snprintf(sql, sizeof(sql), "delete from `%s` where `%s` >= %" PRId64 " and `%s` <= %" PRId64, req.tableFName,

×

1583

  TAOS_RES* res = taosQueryImpl(taos, sql, false, TD_REQ_FROM_TAOX);

×

1584

  RAW_NULL_CHECK(res);

×

1585

  SRequestObj* pRequest = (SRequestObj*)res;

×

1586

  code = pRequest->code;

×

1587

  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_PAR_GET_META_ERROR) {

×

1588

    code = TSDB_CODE_SUCCESS;

×

1590

  taos_free_result(res);

×

1592

end:

×

1593

  uDebug("connId:0x%" PRIx64 " delete data sql:%s, code:%s", *(int64_t*)taos, sql, tstrerror(code));

×

1594

  tDecoderClear(&coder);

×

1595

  return code;

×

1598

static int32_t taosAlterTable(TAOS* taos, void* meta, uint32_t metaLen) {

×

1599

  if (taos == NULL || meta == NULL) {

×

1600

    uError("invalid parameter in %s", __func__);

×

1601

    return TSDB_CODE_INVALID_PARA;

×

1603

  SVAlterTbReq   req = {0};

×

1604

  SDecoder       dcoder = {0};

×

1605

  int32_t        code = TSDB_CODE_SUCCESS;

×

1606

  SRequestObj*   pRequest = NULL;

×

1607

  SQuery*        pQuery = NULL;

×

1608

  SArray*        pArray = NULL;

×

1609

  SVgDataBlocks* pVgData = NULL;

×

1611

  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));

×

1612

  uDebug(LOG_ID_TAG " alter table, meta:%p, len:%d", LOG_ID_VALUE, meta, metaLen);

×

1613

  pRequest->syncQuery = true;

×

1614

  if (!pRequest->pDb) {

×

1615

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1616

    goto end;

×

1619

  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));

×

1620

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1621

  tDecoderInit(&dcoder, data, len);

×

1622

  if (tDecodeSVAlterTbReq(&dcoder, &req) < 0) {

×

1623

    code = TSDB_CODE_INVALID_PARA;

×

1624

    goto end;

×

1628

  if (req.action == TSDB_ALTER_TABLE_UPDATE_OPTIONS) {

×

1629

    goto end;

×

1632

  STscObj*  pTscObj = pRequest->pTscObj;

×

1633

  SCatalog* pCatalog = NULL;

×

1634

  RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));

×

1635

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

×

1636

                           .requestId = pRequest->requestId,

×

1637

                           .requestObjRefId = pRequest->self,

×

1638

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

×

1640

  SVgroupInfo pInfo = {0};

×

1641

  SName       pName = {0};

×

1642

  toName(pTscObj->acctId, pRequest->pDb, req.tbName, &pName);

×

1643

  RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo));

×

1644

  pArray = taosArrayInit(1, sizeof(void*));

×

1645

  RAW_NULL_CHECK(pArray);

×

1647

  pVgData = taosMemoryCalloc(1, sizeof(SVgDataBlocks));

×

1648

  RAW_NULL_CHECK(pVgData);

×

1649

  pVgData->vg = pInfo;

×

1651

  int tlen = 0;

×

1652

  req.source = TD_REQ_FROM_TAOX;

×

1653

  tEncodeSize(tEncodeSVAlterTbReq, &req, tlen, code);

×

1654

  if (code != 0) {

×

1655

    code = terrno;

×

1656

    goto end;

×

1658

  tlen += sizeof(SMsgHead);

×

1659

  void* pMsg = taosMemoryMalloc(tlen);

×

1660

  RAW_NULL_CHECK(pMsg);

×

1661

  ((SMsgHead*)pMsg)->vgId = htonl(pInfo.vgId);

×

1662

  ((SMsgHead*)pMsg)->contLen = htonl(tlen);

×

1663

  void*    pBuf = POINTER_SHIFT(pMsg, sizeof(SMsgHead));

×

1664

  SEncoder coder = {0};

×

1665

  tEncoderInit(&coder, pBuf, tlen - sizeof(SMsgHead));

×

1666

  code = tEncodeSVAlterTbReq(&coder, &req);

×

1667

  if (code != 0) {

×

1668

    tEncoderClear(&coder);

×

1669

    code = terrno;

×

1670

    goto end;

×

1672

  tEncoderClear(&coder);

×

1674

  pVgData->pData = pMsg;

×

1675

  pVgData->size = tlen;

×

1677

  pVgData->numOfTables = 1;

×

1678

  RAW_NULL_CHECK(taosArrayPush(pArray, &pVgData));

×

1680

  pQuery = NULL;

×

1681

  code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);

×

1682

  if (NULL == pQuery) goto end;

×

1683

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1684

  pQuery->msgType = TDMT_VND_ALTER_TABLE;

×

1685

  pQuery->stableQuery = false;

×

1686

  pQuery->pRoot = NULL;

×

1687

  code = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT, &pQuery->pRoot);

×

1688

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1689

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pArray));

×

1691

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1693

  pVgData = NULL;

×

1694

  pArray = NULL;

×

1695

  code = pRequest->code;

×

1696

  if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

×

1697

    code = TSDB_CODE_SUCCESS;

×

1700

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1701

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

×

1702

    if (pRes->res != NULL) {

×

1703

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

×

1706

end:

×

1707

  uDebug(LOG_ID_TAG " alter table return, meta:%p, len:%d, msg:%s", LOG_ID_VALUE, meta, metaLen, tstrerror(code));

×

1708

  taosArrayDestroy(pArray);

×

1709

  if (pVgData) taosMemoryFreeClear(pVgData->pData);

×

1710

  taosMemoryFreeClear(pVgData);

×

1711

  destroyRequest(pRequest);

×

1712

  tDecoderClear(&dcoder);

×

1713

  qDestroyQuery(pQuery);

×

1714

  return code;

×

1717

int taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const char* tbname, TAOS_FIELD* fields,

×

1719

  return taos_write_raw_block_with_fields_with_reqid(taos, rows, pData, tbname, fields, numFields, 0);

×

1722

int taos_write_raw_block_with_fields_with_reqid(TAOS* taos, int rows, char* pData, const char* tbname,

×

1724

  if (taos == NULL || pData == NULL || tbname == NULL) {

×

1725

    uError("invalid parameter in %s", __func__);

×

1726

    return TSDB_CODE_INVALID_PARA;

×

1728

  int32_t     code = TSDB_CODE_SUCCESS;

×

1729

  STableMeta* pTableMeta = NULL;

×

1730

  SQuery*     pQuery = NULL;

×

1731

  SHashObj*   pVgHash = NULL;

×

1733

  SRequestObj* pRequest = NULL;

×

1734

  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, reqid));

×

1736

  uDebug(LOG_ID_TAG " write raw block with field, rows:%d, pData:%p, tbname:%s, fields:%p, numFields:%d", LOG_ID_VALUE,

×

1739

  pRequest->syncQuery = true;

×

1740

  if (!pRequest->pDb) {

×

1741

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1742

    goto end;

×

1745

  SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};

×

1746

  tstrncpy(pName.dbname, pRequest->pDb, sizeof(pName.dbname));

×

1747

  tstrncpy(pName.tname, tbname, sizeof(pName.tname));

×

1749

  struct SCatalog* pCatalog = NULL;

×

1750

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

×

1752

  SRequestConnInfo conn = {0};

×

1753

  conn.pTrans = pRequest->pTscObj->pAppInfo->pTransporter;

×

1754

  conn.requestId = pRequest->requestId;

×

1755

  conn.requestObjRefId = pRequest->self;

×

1756

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

×

1758

  SVgroupInfo vgData = {0};

×

1759

  RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &vgData));

×

1760

  RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta));

×

1761

  RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

1762

  pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);

×

1763

  RAW_NULL_CHECK(pVgHash);

×

1764

  RAW_RETURN_CHECK(

×

1766

  RAW_RETURN_CHECK(rawBlockBindData(pQuery, pTableMeta, pData, NULL, fields, numFields, false, NULL, 0, false));

×

1767

  RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

1769

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1770

  code = pRequest->code;

×

1772

end:

×

1773

  uDebug(LOG_ID_TAG " write raw block with field return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

1774

  taosMemoryFreeClear(pTableMeta);

×

1775

  qDestroyQuery(pQuery);

×

1776

  destroyRequest(pRequest);

×

1777

  taosHashCleanup(pVgHash);

×

1778

  return code;

×

1781

int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) {

×

1782

  return taos_write_raw_block_with_reqid(taos, rows, pData, tbname, 0);

×

1785

int taos_write_raw_block_with_reqid(TAOS* taos, int rows, char* pData, const char* tbname, int64_t reqid) {

×

1786

  if (taos == NULL || pData == NULL || tbname == NULL) {

×

1787

    return TSDB_CODE_INVALID_PARA;

×

1789

  int32_t     code = TSDB_CODE_SUCCESS;

×

1790

  STableMeta* pTableMeta = NULL;

×

1791

  SQuery*     pQuery = NULL;

×

1792

  SHashObj*   pVgHash = NULL;

×

1794

  SRequestObj* pRequest = NULL;

×

1795

  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, reqid));

×

1797

  uDebug(LOG_ID_TAG " write raw block, rows:%d, pData:%p, tbname:%s", LOG_ID_VALUE, rows, pData, tbname);

×

1799

  pRequest->syncQuery = true;

×

1800

  if (!pRequest->pDb) {

×

1801

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1802

    goto end;

×

1805

  SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};

×

1806

  tstrncpy(pName.dbname, pRequest->pDb, sizeof(pName.dbname));

×

1807

  tstrncpy(pName.tname, tbname, sizeof(pName.tname));

×

1809

  struct SCatalog* pCatalog = NULL;

×

1810

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

×

1812

  SRequestConnInfo conn = {0};

×

1813

  conn.pTrans = pRequest->pTscObj->pAppInfo->pTransporter;

×

1814

  conn.requestId = pRequest->requestId;

×

1815

  conn.requestObjRefId = pRequest->self;

×

1816

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

×

1818

  SVgroupInfo vgData = {0};

×

1819

  RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &vgData));

×

1820

  RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta));

×

1821

  RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

1822

  pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);

×

1823

  RAW_NULL_CHECK(pVgHash);

×

1824

  RAW_RETURN_CHECK(

×

1826

  RAW_RETURN_CHECK(rawBlockBindData(pQuery, pTableMeta, pData, NULL, NULL, 0, false, NULL, 0, false));

×

1827

  RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

1829

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1830

  code = pRequest->code;

×

1832

end:

×

1833

  uDebug(LOG_ID_TAG " write raw block return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

1834

  taosMemoryFreeClear(pTableMeta);

×

1835

  qDestroyQuery(pQuery);

×

1836

  destroyRequest(pRequest);

×

1837

  taosHashCleanup(pVgHash);

×

1838

  return code;

×

1841

static void* getRawDataFromRes(void* pRetrieve) {

×

1842

  if (pRetrieve == NULL) {

×

1843

    uError("invalid parameter in %s", __func__);

×

1844

    return NULL;

×

1846

  void* rawData = NULL;

×

1848

  if (*(int64_t*)pRetrieve == RETRIEVE_TABLE_RSP_VERSION) {

×

1849

    rawData = ((SRetrieveTableRsp*)pRetrieve)->data;

×

1850

  } else if (*(int64_t*)pRetrieve == RETRIEVE_TABLE_RSP_TMQ_VERSION ||

×

1851

             *(int64_t*)pRetrieve == RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION) {

×

1852

    rawData = ((SRetrieveTableRspForTmq*)pRetrieve)->data;

×

1854

  return rawData;

×

1857

static int32_t buildCreateTbMap(SMqDataRsp* rsp, SHashObj* pHashObj) {

×

1858

  if (rsp == NULL || pHashObj == NULL) {

×

1859

    uError("invalid parameter in %s", __func__);

×

1860

    return TSDB_CODE_INVALID_PARA;

×

1863

  int32_t       code = 0;

×

1864

  SVCreateTbReq pCreateReq = {0};

×

1865

  SDecoder      decoderTmp = {0};

×

1867

  for (int j = 0; j < rsp->createTableNum; j++) {

×

1868

    void** dataTmp = taosArrayGet(rsp->createTableReq, j);

×

1869

    RAW_NULL_CHECK(dataTmp);

×

1870

    int32_t* lenTmp = taosArrayGet(rsp->createTableLen, j);

×

1871

    RAW_NULL_CHECK(lenTmp);

×

1873

    tDecoderInit(&decoderTmp, *dataTmp, *lenTmp);

×

1874

    RAW_RETURN_CHECK(tDecodeSVCreateTbReq(&decoderTmp, &pCreateReq));

×

1876

    if (pCreateReq.type != TSDB_CHILD_TABLE) {

×

1877

      code = TSDB_CODE_INVALID_MSG;

×

1878

      goto end;

×

1880

    if (taosHashGet(pHashObj, pCreateReq.name, strlen(pCreateReq.name)) == NULL) {

×

1881

      RAW_RETURN_CHECK(

×

1884

      tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);

×

1885

      pCreateReq = (SVCreateTbReq){0};

×

1888

    tDecoderClear(&decoderTmp);

×

1890

  return 0;

×

1892

end:

×

1893

  tDecoderClear(&decoderTmp);

×

1894

  tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);

×

1895

  return code;

×

1920

static void tmqFreeMeta(void* data) {

×

1921

  if (data == NULL) {

×

1922

    uError("invalid parameter in %s", __func__);

×

1923

    return;

×

1925

  STableMeta* pTableMeta = *(STableMeta**)data;

×

1926

  taosMemoryFree(pTableMeta);

×

1929

static void freeRawCache(void* data) {

×

1930

  if (data == NULL) {

×

1931

    uError("invalid parameter in %s", __func__);

×

1932

    return;

×

1934

  rawCacheInfo* pRawCache = (rawCacheInfo*)data;

×

1935

  taosHashCleanup(pRawCache->pMetaHash);

×

1936

  taosHashCleanup(pRawCache->pNameHash);

×

1937

  taosHashCleanup(pRawCache->pVgHash);

×

1940

static int32_t initRawCacheHash() {

×

1941

  if (writeRawCache == NULL) {

×

1942

    writeRawCache = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);

×

1943

    if (writeRawCache == NULL) {

×

1944

      return terrno;

×

1946

    taosHashSetFreeFp(writeRawCache, freeRawCache);

×

1948

  return 0;

×

1951

static bool needRefreshMeta(void* rawData, STableMeta* pTableMeta, SSchemaWrapper* pSW) {

×

1952

  if (rawData == NULL || pSW == NULL){

×

1953

    return false;

×

1955

  if (pTableMeta == NULL) {

×

1956

    uError("invalid parameter in %s", __func__);

×

1957

    return false;

×

1959

  char* p = (char*)rawData;

×

1962

  p += sizeof(int32_t);

×

1963

  p += sizeof(int32_t);

×

1964

  p += sizeof(int32_t);

×

1965

  p += sizeof(int32_t);

×

1966

  p += sizeof(int32_t);

×

1967

  p += sizeof(uint64_t);

×

1968

  int8_t* fields = p;

×

1970

  if (pSW->nCols != pTableMeta->tableInfo.numOfColumns) {

×

1971

    return true;

×

1974

  for (int i = 0; i < pSW->nCols; i++) {

×

1975

    int j = 0;

×

1976

    for (; j < pTableMeta->tableInfo.numOfColumns; j++) {

×

1977

      SSchema* pColSchema = &pTableMeta->schema[j];

×

1978

      SSchemaExt* pColExtSchema = &pTableMeta->schemaExt[j];

×

1979

      char*    fieldName = pSW->pSchema[i].name;

×

1981

      if (strcmp(pColSchema->name, fieldName) == 0) {

×

1982

        if (checkSchema(pColSchema, pColExtSchema, fields, NULL, 0) != 0){

×

1983

          return true;

×

1985

        break;

×

1988

    fields += sizeof(int8_t) + sizeof(int32_t);

×

1990

    if (j == pTableMeta->tableInfo.numOfColumns) return true;

×

1992

  return false;

×

1995

static int32_t getRawCache(SHashObj** pVgHash, SHashObj** pNameHash, SHashObj** pMetaHash, void* key) {

×

1996

  if (pVgHash == NULL || pNameHash == NULL || pMetaHash == NULL || key == NULL) {

×

1997

    uError("invalid parameter in %s", __func__);

×

1998

    return TSDB_CODE_INVALID_PARA;

×

2000

  int32_t code = 0;

×

2001

  void*   cacheInfo = taosHashGet(writeRawCache, &key, POINTER_BYTES);

×

2002

  if (cacheInfo == NULL) {

×

2003

    *pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);

×

2004

    RAW_NULL_CHECK(*pVgHash);

×

2005

    *pNameHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);

×

2006

    RAW_NULL_CHECK(*pNameHash);

×

2007

    *pMetaHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);

×

2008

    RAW_NULL_CHECK(*pMetaHash);

×

2009

    taosHashSetFreeFp(*pMetaHash, tmqFreeMeta);

×

2010

    rawCacheInfo info = {*pVgHash, *pNameHash, *pMetaHash};

×

2011

    RAW_RETURN_CHECK(taosHashPut(writeRawCache, &key, POINTER_BYTES, &info, sizeof(rawCacheInfo)));

×

2013

    rawCacheInfo* info = (rawCacheInfo*)cacheInfo;

×

2014

    *pVgHash = info->pVgHash;

×

2015

    *pNameHash = info->pNameHash;

×

2016

    *pMetaHash = info->pMetaHash;

×

2019

  return 0;

×

2020

end:

×

2021

  taosHashCleanup(*pMetaHash);

×

2022

  taosHashCleanup(*pNameHash);

×

2023

  taosHashCleanup(*pVgHash);

×

2024

  return code;

×

2027

static int32_t buildRawRequest(TAOS* taos, SRequestObj** pRequest, SCatalog** pCatalog, SRequestConnInfo* conn) {

×

2028

  if (taos == NULL || pRequest == NULL || pCatalog == NULL || conn == NULL) {

×

2029

    uError("invalid parameter in %s", __func__);

×

2030

    return TSDB_CODE_INVALID_PARA;

×

2032

  int32_t code = 0;

×

2033

  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, pRequest, 0));

×

2034

  (*pRequest)->syncQuery = true;

×

2035

  if (!(*pRequest)->pDb) {

×

2036

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

2037

    goto end;

×

2040

  RAW_RETURN_CHECK(catalogGetHandle((*pRequest)->pTscObj->pAppInfo->clusterId, pCatalog));

×

2041

  conn->pTrans = (*pRequest)->pTscObj->pAppInfo->pTransporter;

×

2042

  conn->requestId = (*pRequest)->requestId;

×

2043

  conn->requestObjRefId = (*pRequest)->self;

×

2044

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

×

2046

end:

×

2047

  return code;

×

2051

static int32_t  decodeRawData(SDecoder* decoder, void* data, uint32_t dataLen, _raw_decode_func_ func,

×

2053

  if (decoder == NULL || data == NULL || func == NULL || rspObj == NULL) {

×

2054

    uError("invalid parameter in %s", __func__);

×

2055

    return TSDB_CODE_INVALID_PARA;

×

2057

  int8_t dataVersion = *(int8_t*)data;

×

2058

  if (dataVersion >= MQ_DATA_RSP_VERSION) {

×

2059

    data = POINTER_SHIFT(data, sizeof(int8_t) + sizeof(int32_t));

×

2060

    if (dataLen < sizeof(int8_t) + sizeof(int32_t)) {

×

2061

      return TSDB_CODE_INVALID_PARA;

×

2063

    dataLen -= sizeof(int8_t) + sizeof(int32_t);

×

2066

  rspObj->resIter = -1;

×

2067

  tDecoderInit(decoder, data, dataLen);

×

2068

  int32_t code = func(decoder, &rspObj->dataRsp);

×

2069

  if (code != 0) {

×

2070

    SET_ERROR_MSG("decode mq taosx data rsp failed");

×

2072

  return code;

×

2075

static int32_t processCacheMeta(SHashObj* pVgHash, SHashObj* pNameHash, SHashObj* pMetaHash,

×

2078

  if (pVgHash == NULL || pNameHash == NULL || pMetaHash == NULL || pCatalog == NULL || conn == NULL || pName == NULL ||

×

2080

    uError("invalid parameter in %s", __func__);

×

2081

    return TSDB_CODE_INVALID_PARA;

×

2083

  int32_t     code = 0;

×

2084

  STableMeta* pTableMeta = NULL;

×

2085

  tbInfo*     tmpInfo = (tbInfo*)taosHashGet(pNameHash, pName->tname, strlen(pName->tname));

×

2086

  if (tmpInfo == NULL || retry > 0) {

×

2087

    tbInfo info = {0};

×

2089

    RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, conn, pName, &info.vgInfo));

×

2090

    if (pCreateReqDst && tmpInfo == NULL) {  // change stable name to get meta

×

2091

      tstrncpy(pName->tname, pCreateReqDst->ctb.stbName, TSDB_TABLE_NAME_LEN);

×

2093

    RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, conn, pName, &pTableMeta));

×

2094

    info.uid = pTableMeta->uid;

×

2095

    if (pTableMeta->tableType == TSDB_CHILD_TABLE) {

×

2096

      info.suid = pTableMeta->suid;

×

2098

      info.suid = pTableMeta->uid;

×

2100

    code = taosHashPut(pMetaHash, &info.suid, LONG_BYTES, &pTableMeta, POINTER_BYTES);

×

2101

    if (code != 0) {

×

2102

      taosMemoryFree(pTableMeta);

×

2103

      goto end;

×

2105

    if (pCreateReqDst) {

×

2106

      pTableMeta->vgId = info.vgInfo.vgId;

×

2107

      pTableMeta->uid = pCreateReqDst->uid;

×

2108

      pCreateReqDst->ctb.suid = pTableMeta->suid;

×

2111

    RAW_RETURN_CHECK(taosHashPut(pNameHash, pName->tname, strlen(pName->tname), &info, sizeof(tbInfo)));

×

2112

    tmpInfo = (tbInfo*)taosHashGet(pNameHash, pName->tname, strlen(pName->tname));

×

2113

    RAW_RETURN_CHECK(

×

2117

  if (pTableMeta == NULL || retry > 0) {

×

2118

    STableMeta** pTableMetaTmp = (STableMeta**)taosHashGet(pMetaHash, &tmpInfo->suid, LONG_BYTES);

×

2119

    if (pTableMetaTmp == NULL || retry > 0 || needRefreshMeta(rawData, *pTableMetaTmp, pSW)) {

×

2120

      RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, conn, pName, &pTableMeta));

×

2121

      code = taosHashPut(pMetaHash, &tmpInfo->suid, LONG_BYTES, &pTableMeta, POINTER_BYTES);

×

2122

      if (code != 0) {

×

2123

        taosMemoryFree(pTableMeta);

×

2124

        goto end;

×

2128

      pTableMeta = *pTableMetaTmp;

×

2129

      pTableMeta->uid = tmpInfo->uid;

×

2130

      pTableMeta->vgId = tmpInfo->vgInfo.vgId;

×

2133

  *pMeta = pTableMeta;

×

2135

end:

×

2136

  return code;

×

2139

static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, uint32_t dataLen) {

×

2140

  if (taos == NULL || data == NULL) {

×

2141

    uError("invalid parameter in %s", __func__);

×

2142

    return TSDB_CODE_INVALID_PARA;

×

2144

  int32_t   code = TSDB_CODE_SUCCESS;

×

2145

  SQuery*   pQuery = NULL;

×

2146

  SMqRspObj rspObj = {0};

×

2147

  SDecoder  decoder = {0};

×

2149

  SRequestObj*     pRequest = NULL;

×

2150

  SCatalog*        pCatalog = NULL;

×

2151

  SRequestConnInfo conn = {0};

×

2152

  RAW_RETURN_CHECK(buildRawRequest(taos, &pRequest, &pCatalog, &conn));

×

2153

  uDebug(LOG_ID_TAG " write raw data, data:%p, dataLen:%d", LOG_ID_VALUE, data, dataLen);

×

2154

  RAW_RETURN_CHECK(decodeRawData(&decoder, data, dataLen, tDecodeMqDataRsp, &rspObj));

×

2156

  SHashObj* pVgHash = NULL;

×

2157

  SHashObj* pNameHash = NULL;

×

2158

  SHashObj* pMetaHash = NULL;

×

2159

  RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos));

×

2160

  int retry = 0;

×

2162

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2163

    uDebug(LOG_ID_TAG " write raw meta data block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum);

×

2164

    while (++rspObj.resIter < rspObj.dataRsp.blockNum) {

×

2165

      if (!rspObj.dataRsp.withSchema) {

×

2166

        goto end;

×

2169

      const char* tbName = (const char*)taosArrayGetP(rspObj.dataRsp.blockTbName, rspObj.resIter);

×

2170

      RAW_NULL_CHECK(tbName);

×

2171

      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(rspObj.dataRsp.blockSchema, rspObj.resIter);

×

2172

      RAW_NULL_CHECK(pSW);

×

2173

      void* pRetrieve = taosArrayGetP(rspObj.dataRsp.blockData, rspObj.resIter);

×

2174

      RAW_NULL_CHECK(pRetrieve);

×

2175

      void* rawData = getRawDataFromRes(pRetrieve);

×

2176

      RAW_NULL_CHECK(rawData);

×

2178

      uTrace(LOG_ID_TAG " write raw data block tbname:%s", LOG_ID_VALUE, tbName);

×

2179

      SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};

×

2180

      tstrncpy(pName.dbname, pRequest->pDb, TSDB_DB_NAME_LEN);

×

2181

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2183

      STableMeta* pTableMeta = NULL;

×

2184

      RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, NULL, pCatalog, &conn, &pName, &pTableMeta, pSW,

×

2186

      char err[ERR_MSG_LEN] = {0};

×

2187

      code = rawBlockBindData(pQuery, pTableMeta, rawData, NULL, pSW, pSW->nCols, true, err, ERR_MSG_LEN, true);

×

2188

      if (code != TSDB_CODE_SUCCESS) {

×

2189

        SET_ERROR_MSG("table:%s, err:%s", pName.tname, err);

×

2190

        goto end;

×

2193

    RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

2194

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2195

    code = pRequest->code;

×

2197

    if (NEED_CLIENT_HANDLE_ERROR(code) && retry++ < 3) {

×

2198

      uInfo("write raw retry:%d/3 end code:%d, msg:%s", retry, code, tstrerror(code));

×

2199

      qDestroyQuery(pQuery);

×

2200

      pQuery = NULL;

×

2201

      rspObj.resIter = -1;

×

2202

      continue;

×

2204

    break;

×

2207

end:

×

2208

  uDebug(LOG_ID_TAG " write raw data return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

2209

  tDeleteMqDataRsp(&rspObj.dataRsp);

×

2210

  tDecoderClear(&decoder);

×

2211

  qDestroyQuery(pQuery);

×

2212

  destroyRequest(pRequest);

×

2213

  return code;

×

2216

static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, uint32_t dataLen) {

×

2217

  if (taos == NULL || data == NULL) {

×

2218

    uError("invalid parameter in %s", __func__);

×

2219

    return TSDB_CODE_INVALID_PARA;

×

2221

  int32_t   code = TSDB_CODE_SUCCESS;

×

2222

  SQuery*   pQuery = NULL;

×

2223

  SMqRspObj rspObj = {0};

×

2224

  SDecoder  decoder = {0};

×

2225

  SHashObj* pCreateTbHash = NULL;

×

2227

  SRequestObj*     pRequest = NULL;

×

2228

  SCatalog*        pCatalog = NULL;

×

2229

  SRequestConnInfo conn = {0};

×

2231

  RAW_RETURN_CHECK(buildRawRequest(taos, &pRequest, &pCatalog, &conn));

×

2232

  uDebug(LOG_ID_TAG " write raw metadata, data:%p, dataLen:%d", LOG_ID_VALUE, data, dataLen);

×

2233

  RAW_RETURN_CHECK(decodeRawData(&decoder, data, dataLen, tDecodeSTaosxRsp, &rspObj));

×

2235

  pCreateTbHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);

×

2236

  RAW_NULL_CHECK(pCreateTbHash);

×

2237

  RAW_RETURN_CHECK(buildCreateTbMap(&rspObj.dataRsp, pCreateTbHash));

×

2239

  SHashObj* pVgHash = NULL;

×

2240

  SHashObj* pNameHash = NULL;

×

2241

  SHashObj* pMetaHash = NULL;

×

2242

  RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos));

×

2243

  int retry = 0;

×

2245

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2246

    uDebug(LOG_ID_TAG " write raw meta data block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum);

×

2247

    while (++rspObj.resIter < rspObj.dataRsp.blockNum) {

×

2248

      if (!rspObj.dataRsp.withSchema) {

×

2249

        goto end;

×

2252

      const char* tbName = (const char*)taosArrayGetP(rspObj.dataRsp.blockTbName, rspObj.resIter);

×

2253

      RAW_NULL_CHECK(tbName);

×

2254

      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(rspObj.dataRsp.blockSchema, rspObj.resIter);

×

2255

      RAW_NULL_CHECK(pSW);

×

2256

      void* pRetrieve = taosArrayGetP(rspObj.dataRsp.blockData, rspObj.resIter);

×

2257

      RAW_NULL_CHECK(pRetrieve);

×

2258

      void* rawData = getRawDataFromRes(pRetrieve);

×

2259

      RAW_NULL_CHECK(rawData);

×

2261

      uTrace(LOG_ID_TAG " write raw data block tbname:%s", LOG_ID_VALUE, tbName);

×

2262

      SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};

×

2263

      tstrncpy(pName.dbname, pRequest->pDb, TSDB_DB_NAME_LEN);

×

2264

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2267

      SVCreateTbReq* pCreateReqDst = (SVCreateTbReq*)taosHashGet(pCreateTbHash, pName.tname, strlen(pName.tname));

×

2268

      STableMeta*    pTableMeta = NULL;

×

2269

      RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, pCreateReqDst, pCatalog, &conn, &pName,

×

2271

      char err[ERR_MSG_LEN] = {0};

×

2273

          rawBlockBindData(pQuery, pTableMeta, rawData, pCreateReqDst, pSW, pSW->nCols, true, err, ERR_MSG_LEN, true);

×

2274

      if (code != TSDB_CODE_SUCCESS) {

×

2275

        SET_ERROR_MSG("table:%s, err:%s", pName.tname, err);

×

2276

        goto end;

×

2279

    RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

2280

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2281

    code = pRequest->code;

×

2283

    if (NEED_CLIENT_HANDLE_ERROR(code) && retry++ < 3) {

×

2284

      uInfo("write raw retry:%d/3 end code:%d, msg:%s", retry, code, tstrerror(code));

×

2285

      qDestroyQuery(pQuery);

×

2286

      pQuery = NULL;

×

2287

      rspObj.resIter = -1;

×

2288

      continue;

×

2290

    break;

×

2293

end:

×

2294

  uDebug(LOG_ID_TAG " write raw metadata return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

2295

  tDeleteSTaosxRsp(&rspObj.dataRsp);

×

2296

  void* pIter = taosHashIterate(pCreateTbHash, NULL);

×

2297

  while (pIter) {

×

2298

    tDestroySVCreateTbReq(pIter, TSDB_MSG_FLG_DECODE);

×

2299

    pIter = taosHashIterate(pCreateTbHash, pIter);

×

2301

  taosHashCleanup(pCreateTbHash);

×

2302

  tDecoderClear(&decoder);

×

2303

  qDestroyQuery(pQuery);

×

2304

  destroyRequest(pRequest);

×

2305

  return code;

×

2308

static int32_t tmqWriteRawRawDataImpl(TAOS* taos, void* data, uint32_t dataLen) {

×

2309

  if (taos == NULL || data == NULL) {

×

2310

    uError("invalid parameter in %s", __func__);

×

2311

    return TSDB_CODE_INVALID_PARA;

×

2313

  int32_t   code = TSDB_CODE_SUCCESS;

×

2314

  SQuery*   pQuery = NULL;

×

2315

  SHashObj* pVgroupHash = NULL;

×

2316

  SMqRspObj rspObj = {0};

×

2317

  SDecoder  decoder = {0};

×

2319

  SRequestObj*     pRequest = NULL;

×

2320

  SCatalog*        pCatalog = NULL;

×

2321

  SRequestConnInfo conn = {0};

×

2323

  RAW_RETURN_CHECK(buildRawRequest(taos, &pRequest, &pCatalog, &conn));

×

2324

  uDebug(LOG_ID_TAG " write raw rawdata, data:%p, dataLen:%d", LOG_ID_VALUE, data, dataLen);

×

2325

  RAW_RETURN_CHECK(decodeRawData(&decoder, data, dataLen, tDecodeMqDataRsp, &rspObj));

×

2327

  SHashObj* pVgHash = NULL;

×

2328

  SHashObj* pNameHash = NULL;

×

2329

  SHashObj* pMetaHash = NULL;

×

2330

  RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos));

×

2331

  int retry = 0;

×

2332

  while (1) {

×

2333

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2334

    uDebug(LOG_ID_TAG " write raw rawdata block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum);

×

2335

    SVnodeModifyOpStmt* pStmt = (SVnodeModifyOpStmt*)(pQuery)->pRoot;

×

2336

    pVgroupHash = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);

×

2337

    RAW_NULL_CHECK(pVgroupHash);

×

2338

    pStmt->pVgDataBlocks = taosArrayInit(8, POINTER_BYTES);

×

2339

    RAW_NULL_CHECK(pStmt->pVgDataBlocks);

×

2341

    while (++rspObj.resIter < rspObj.dataRsp.blockNum) {

×

2342

      const char* tbName = (const char*)taosArrayGetP(rspObj.dataRsp.blockTbName, rspObj.resIter);

×

2343

      RAW_NULL_CHECK(tbName);

×

2344

      void* pRetrieve = taosArrayGetP(rspObj.dataRsp.blockData, rspObj.resIter);

×

2345

      RAW_NULL_CHECK(pRetrieve);

×

2346

      void* rawData = getRawDataFromRes(pRetrieve);

×

2347

      RAW_NULL_CHECK(rawData);

×

2349

      uTrace(LOG_ID_TAG " write raw data block tbname:%s", LOG_ID_VALUE, tbName);

×

2350

      SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};

×

2351

      tstrncpy(pName.dbname, pRequest->pDb, TSDB_DB_NAME_LEN);

×

2352

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2355

      STableMeta*    pTableMeta = NULL;

×

2356

      RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, NULL, pCatalog, &conn, &pName,

×

2358

      char err[ERR_MSG_LEN] = {0};

×

2359

      code = rawBlockBindRawData(pVgroupHash, pStmt->pVgDataBlocks, pTableMeta, rawData);

×

2360

      if (code != TSDB_CODE_SUCCESS) {

×

2361

        SET_ERROR_MSG("table:%s, err:%s", pName.tname, err);

×

2362

        goto end;

×

2365

    taosHashCleanup(pVgroupHash);

×

2366

    pVgroupHash = NULL;

×

2368

    RAW_RETURN_CHECK(smlBuildOutputRaw(pQuery, pVgHash));

×

2369

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2370

    code = pRequest->code;

×

2372

    if (NEED_CLIENT_HANDLE_ERROR(code) && retry++ < 3) {

×

2373

      uInfo("write raw retry:%d/3 end code:%d, msg:%s", retry, code, tstrerror(code));

×

2374

      qDestroyQuery(pQuery);

×

2375

      pQuery = NULL;

×

2376

      rspObj.resIter = -1;

×

2377

      continue;

×

2379

    break;

×

2382

  end:

×

2383

  uDebug(LOG_ID_TAG " write raw rawdata return, msg:%s", LOG_ID_VALUE, tstrerror(code));

×

2384

  tDeleteMqDataRsp(&rspObj.dataRsp);

×

2385

  tDecoderClear(&decoder);

×

2386

  qDestroyQuery(pQuery);

×

2387

  taosHashCleanup(pVgroupHash);

×

2388

  destroyRequest(pRequest);

×

2389

  return code;

×

2392

static void processSimpleMeta(SMqMetaRsp* pMetaRsp, cJSON** meta) {

×

2393

  if (pMetaRsp == NULL || meta == NULL) {

×

2394

    uError("invalid parameter in %s", __func__);

×

2395

    return;

×

2397

  if (pMetaRsp->resMsgType == TDMT_VND_CREATE_STB) {

×

2398

    processCreateStb(pMetaRsp, meta);

×

2399

  } else if (pMetaRsp->resMsgType == TDMT_VND_ALTER_STB) {

×

2400

    processAlterStb(pMetaRsp, meta);

×

2401

  } else if (pMetaRsp->resMsgType == TDMT_VND_DROP_STB) {

×

2402

    processDropSTable(pMetaRsp, meta);

×

2403

  } else if (pMetaRsp->resMsgType == TDMT_VND_CREATE_TABLE) {

×

2404

    processCreateTable(pMetaRsp, meta);

×

2405

  } else if (pMetaRsp->resMsgType == TDMT_VND_ALTER_TABLE) {

×

2406

    processAlterTable(pMetaRsp, meta);

×

2407

  } else if (pMetaRsp->resMsgType == TDMT_VND_DROP_TABLE) {

×

2408

    processDropTable(pMetaRsp, meta);

×

2409

  } else if (pMetaRsp->resMsgType == TDMT_VND_DELETE) {

×

2410

    processDeleteTable(pMetaRsp, meta);

×

2414

static void processBatchMetaToJson(SMqBatchMetaRsp* pMsgRsp, char** string) {

×

2415

  if (pMsgRsp == NULL || string == NULL) {

×

2416

    uError("invalid parameter in %s", __func__);

×

2417

    return;

×

2419

  SDecoder        coder = {0};

×

2420

  SMqBatchMetaRsp rsp = {0};

×

2421

  int32_t         code = 0;

×

2422

  cJSON*          pJson = NULL;

×

2423

  tDecoderInit(&coder, pMsgRsp->pMetaBuff, pMsgRsp->metaBuffLen);

×

2424

  if (tDecodeMqBatchMetaRsp(&coder, &rsp) < 0) {

×

2425

    goto end;

×

2428

  pJson = cJSON_CreateObject();

×

2429

  RAW_NULL_CHECK(pJson);

×

2430

  RAW_FALSE_CHECK(cJSON_AddStringToObject(pJson, "tmq_meta_version", TMQ_META_VERSION));

×

2431

  cJSON* pMetaArr = cJSON_CreateArray();

×

2432

  RAW_NULL_CHECK(pMetaArr);

×

2433

  RAW_FALSE_CHECK(tmqAddJsonObjectItem(pJson, "metas", pMetaArr));

×

2435

  int32_t num = taosArrayGetSize(rsp.batchMetaReq);

×

2436

  for (int32_t i = 0; i < num; i++) {

×

2437

    int32_t* len = taosArrayGet(rsp.batchMetaLen, i);

×

2438

    RAW_NULL_CHECK(len);

×

2439

    void* tmpBuf = taosArrayGetP(rsp.batchMetaReq, i);

×

2440

    RAW_NULL_CHECK(tmpBuf);

×

2441

    SDecoder   metaCoder = {0};

×

2442

    SMqMetaRsp metaRsp = {0};

×

2443

    tDecoderInit(&metaCoder, POINTER_SHIFT(tmpBuf, sizeof(SMqRspHead)), *len - sizeof(SMqRspHead));

×

2444

    if (tDecodeMqMetaRsp(&metaCoder, &metaRsp) < 0) {

×

2445

      goto end;

×

2447

    cJSON* pItem = NULL;

×

2448

    processSimpleMeta(&metaRsp, &pItem);

×

2449

    tDeleteMqMetaRsp(&metaRsp);

×

2450

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(pMetaArr, pItem));

×

2453

  tDeleteMqBatchMetaRsp(&rsp);

×

2454

  char* fullStr = cJSON_PrintUnformatted(pJson);

×

2455

  cJSON_Delete(pJson);

×

2456

  *string = fullStr;

×

2457

  return;

×

2459

end:

×

2460

  cJSON_Delete(pJson);

×

2461

  tDeleteMqBatchMetaRsp(&rsp);

×

2464

char* tmq_get_json_meta(TAOS_RES* res) {

×

2465

  if (res == NULL) {

×

2466

    uError("invalid parameter in %s", __func__);

×

2467

    return NULL;

×

2469

  uDebug("tmq_get_json_meta res:%p", res);

×

2470

  if (!TD_RES_TMQ_META(res) && !TD_RES_TMQ_METADATA(res) && !TD_RES_TMQ_BATCH_META(res)) {

×

2471

    return NULL;

×

2474

  char*      string = NULL;

×

2475

  SMqRspObj* rspObj = (SMqRspObj*)res;

×

2476

  if (TD_RES_TMQ_METADATA(res)) {

×

2477

    processAutoCreateTable(&rspObj->dataRsp, &string);

×

2478

  } else if (TD_RES_TMQ_BATCH_META(res)) {

×

2479

    processBatchMetaToJson(&rspObj->batchMetaRsp, &string);

×

2480

  } else if (TD_RES_TMQ_META(res)) {

×

2481

    cJSON* pJson = NULL;

×

2482

    processSimpleMeta(&rspObj->metaRsp, &pJson);

×

2483

    string = cJSON_PrintUnformatted(pJson);

×

2484

    cJSON_Delete(pJson);

×

2486

    uError("tmq_get_json_meta res:%d, invalid type", *(int8_t*)res);

×

2489

  uDebug("tmq_get_json_meta string:%s", string);

×

2490

  return string;

×

2493

void tmq_free_json_meta(char* jsonMeta) { taosMemoryFreeClear(jsonMeta); }

×

2495

static int32_t getOffSetLen(const SMqDataRsp* pRsp) {

×

2496

  if (pRsp == NULL) {

×

2497

    uError("invalid parameter in %s", __func__);

×

2498

    return TSDB_CODE_INVALID_PARA;

×

2500

  SEncoder coder = {0};

×

2501

  tEncoderInit(&coder, NULL, 0);

×

2502

  if (tEncodeSTqOffsetVal(&coder, &pRsp->reqOffset) < 0) return -1;

×

2503

  if (tEncodeSTqOffsetVal(&coder, &pRsp->rspOffset) < 0) return -1;

×

2504

  int32_t pos = coder.pos;

×

2505

  tEncoderClear(&coder);

×

2506

  return pos;

×

2510

static int32_t  encodeMqDataRsp(__encode_func__* encodeFunc, SMqDataRsp* rspObj, tmq_raw_data* raw) {

×

2511

  if (raw == NULL || encodeFunc == NULL || rspObj == NULL) {

×

2512

    uError("invalid parameter in %s", __func__);

×

2513

    return TSDB_CODE_INVALID_PARA;

×

2515

  uint32_t len = 0;

×

2516

  int32_t  code = 0;

×

2517

  SEncoder encoder = {0};

×

2518

  void*    buf = NULL;

×

2519

  tEncodeSize(encodeFunc, rspObj, len, code);

×

2520

  if (code < 0) {

×

2521

    code = TSDB_CODE_INVALID_MSG;

×

2522

    goto FAILED;

×

2524

  len += sizeof(int8_t) + sizeof(int32_t);

×

2525

  buf = taosMemoryCalloc(1, len);

×

2526

  if (buf == NULL) {

×

2527

    code = terrno;

×

2528

    goto FAILED;

×

2530

  tEncoderInit(&encoder, buf, len);

×

2531

  if (tEncodeI8(&encoder, MQ_DATA_RSP_VERSION) < 0) {

×

2532

    code = TSDB_CODE_INVALID_MSG;

×

2533

    goto FAILED;

×

2535

  int32_t offsetLen = getOffSetLen(rspObj);

×

2536

  if (offsetLen <= 0) {

×

2537

    code = TSDB_CODE_INVALID_MSG;

×

2538

    goto FAILED;

×

2540

  if (tEncodeI32(&encoder, offsetLen) < 0) {

×

2541

    code = TSDB_CODE_INVALID_MSG;

×

2542

    goto FAILED;

×

2544

  if (encodeFunc(&encoder, rspObj) < 0) {

×

2545

    code = TSDB_CODE_INVALID_MSG;

×

2546

    goto FAILED;

×

2548

  tEncoderClear(&encoder);

×

2550

  raw->raw = buf;

×

2551

  raw->raw_len = len;

×

2552

  return code;

×

2553

FAILED:

×

2554

  tEncoderClear(&encoder);

×

2555

  taosMemoryFree(buf);

×

2556

  return code;

×

2559

int32_t tmq_get_raw(TAOS_RES* res, tmq_raw_data* raw) {

×

2560

  if (raw == NULL || res == NULL) {

×

2561

    uError("invalid parameter in %s", __func__);

×

2562

    return TSDB_CODE_INVALID_PARA;

×

2564

  *raw = (tmq_raw_data){0};

×

2565

  SMqRspObj* rspObj = ((SMqRspObj*)res);

×

2566

  if (TD_RES_TMQ_META(res)) {

×

2567

    raw->raw = rspObj->metaRsp.metaRsp;

×

2568

    raw->raw_len = rspObj->metaRsp.metaRspLen >= 0 ? rspObj->metaRsp.metaRspLen : 0;

×

2569

    raw->raw_type = rspObj->metaRsp.resMsgType;

×

2570

    uDebug("tmq get raw type meta:%p", raw);

×

2571

  } else if (TD_RES_TMQ(res)) {

×

2572

    int32_t code = encodeMqDataRsp(tEncodeMqDataRsp, &rspObj->dataRsp, raw);

×

2573

    if (code != 0) {

×

2574

      uError("tmq get raw type error:%d", terrno);

×

2575

      return code;

×

2577

    raw->raw_type = RES_TYPE__TMQ;

×

2578

    uDebug("tmq get raw type data:%p", raw);

×

2579

  } else if (TD_RES_TMQ_METADATA(res)) {

×

2580

    int32_t code = encodeMqDataRsp(tEncodeSTaosxRsp, &rspObj->dataRsp, raw);

×

2581

    if (code != 0) {

×

2582

      uError("tmq get raw type error:%d", terrno);

×

2583

      return code;

×

2585

    raw->raw_type = RES_TYPE__TMQ_METADATA;

×

2586

    uDebug("tmq get raw type metadata:%p", raw);

×

2587

  } else if (TD_RES_TMQ_BATCH_META(res)) {

×

2588

    raw->raw = rspObj->batchMetaRsp.pMetaBuff;

×

2589

    raw->raw_len = rspObj->batchMetaRsp.metaBuffLen;

×

2590

    raw->raw_type = rspObj->resType;

×

2591

    uDebug("tmq get raw batch meta:%p", raw);

×

2592

  } else if (TD_RES_TMQ_RAW(res)) {

×

2593

    raw->raw = rspObj->dataRsp.rawData;

×

2594

    rspObj->dataRsp.rawData = NULL;

×

2595

    raw->raw_len = rspObj->dataRsp.len;

×

2596

    raw->raw_type = rspObj->resType;

×

2597

    uDebug("tmq get raw raw:%p", raw);

×

2599

    uError("tmq get raw error type:%d", *(int8_t*)res);

×

2600

    return TSDB_CODE_TMQ_INVALID_MSG;

×

2602

  return TSDB_CODE_SUCCESS;

×

2605

void tmq_free_raw(tmq_raw_data raw) {

×

2606

  uDebug("tmq free raw data type:%d", raw.raw_type);

×

2607

  if (raw.raw_type == RES_TYPE__TMQ ||

×

2608

      raw.raw_type == RES_TYPE__TMQ_METADATA) {

×

2609

    taosMemoryFree(raw.raw);

×

2610

  } else if(raw.raw_type == RES_TYPE__TMQ_RAWDATA && raw.raw != NULL){

×

2611

    taosMemoryFree(POINTER_SHIFT(raw.raw, - sizeof(SMqRspHead)));

×

2613

  (void)memset(terrMsg, 0, ERR_MSG_LEN);

×

2614

×

2616

static int32_t writeRawInit() {

×

2617

  while (atomic_load_8(&initedFlag) == WRITE_RAW_INIT_START) {

×

2618

    int8_t old = atomic_val_compare_exchange_8(&initFlag, 0, 1);

×

2619

    if (old == 0) {

×

2620

      int32_t code = initRawCacheHash();

×

2621

      if (code != 0) {

×

2622

        uError("tmq writeRawImpl init error:%d", code);

×

2623

        atomic_store_8(&initedFlag, WRITE_RAW_INIT_FAIL);

×

2624

        return code;

×

2626

      atomic_store_8(&initedFlag, WRITE_RAW_INIT_OK);

×

2630

  if (atomic_load_8(&initedFlag) == WRITE_RAW_INIT_FAIL) {

×

2631

    return TSDB_CODE_INTERNAL_ERROR;

×

2633

  return 0;

×

2636

static int32_t writeRawImpl(TAOS* taos, void* buf, uint32_t len, uint16_t type) {

×

2637

  if (taos == NULL || buf == NULL) {

×

2638

    uError("invalid parameter in %s", __func__);

×

2639

    return TSDB_CODE_INVALID_PARA;

×

2641

  if (writeRawInit() != 0) {

×

2642

    return TSDB_CODE_INTERNAL_ERROR;

×

2645

  if (type == TDMT_VND_CREATE_STB) {

×

2646

    return taosCreateStb(taos, buf, len);

×

2647

  } else if (type == TDMT_VND_ALTER_STB) {

×

2648

    return taosCreateStb(taos, buf, len);

×

2649

  } else if (type == TDMT_VND_DROP_STB) {

×

2650

    return taosDropStb(taos, buf, len);

×

2651

  } else if (type == TDMT_VND_CREATE_TABLE) {

×

2652

    return taosCreateTable(taos, buf, len);

×

2653

  } else if (type == TDMT_VND_ALTER_TABLE) {

×

2654

    return taosAlterTable(taos, buf, len);

×

2655

  } else if (type == TDMT_VND_DROP_TABLE) {

×

2656

    return taosDropTable(taos, buf, len);

×

2657

  } else if (type == TDMT_VND_DELETE) {

×

2658

    return taosDeleteData(taos, buf, len);

×

2659

  } else if (type == RES_TYPE__TMQ_METADATA) {

×

2660

    return tmqWriteRawMetaDataImpl(taos, buf, len);

×

2661

  } else if (type == RES_TYPE__TMQ_RAWDATA) {

×

2662

    return tmqWriteRawRawDataImpl(taos, buf, len);

×

2663

  } else if (type == RES_TYPE__TMQ) {

×

2664

    return tmqWriteRawDataImpl(taos, buf, len);

×

2665

  } else if (type == RES_TYPE__TMQ_BATCH_META) {

×

2666

    return tmqWriteBatchMetaDataImpl(taos, buf, len);

×

2668

  return TSDB_CODE_INVALID_PARA;

×

2671

int32_t tmq_write_raw(TAOS* taos, tmq_raw_data raw) {

×

2672

  if (taos == NULL || raw.raw == NULL || raw.raw_len <= 0) {

×

2673

    SET_ERROR_MSG("taos:%p or data:%p is NULL or raw_len <= 0", taos, raw.raw);

×

2674

    return TSDB_CODE_INVALID_PARA;

×

2676

  taosClearErrMsg(); // clear global error message

×

2678

  return writeRawImpl(taos, raw.raw, raw.raw_len, raw.raw_type);

×

2681

static int32_t tmqWriteBatchMetaDataImpl(TAOS* taos, void* meta, uint32_t metaLen) {

×

2682

  if (taos == NULL || meta == NULL) {

×

2683

    uError("invalid parameter in %s", __func__);

×

2684

    return TSDB_CODE_INVALID_PARA;

×

2686

  SMqBatchMetaRsp rsp = {0};

×

2687

  SDecoder        coder = {0};

×

2688

  int32_t         code = TSDB_CODE_SUCCESS;

×

2691

  tDecoderInit(&coder, meta, metaLen);

×

2692

  if (tDecodeMqBatchMetaRsp(&coder, &rsp) < 0) {

×

2693

    code = TSDB_CODE_INVALID_PARA;

×

2694

    goto end;

×

2696

  int32_t num = taosArrayGetSize(rsp.batchMetaReq);

×

2697

  for (int32_t i = 0; i < num; i++) {

×

2698

    int32_t* len = taosArrayGet(rsp.batchMetaLen, i);

×

2699

    RAW_NULL_CHECK(len);

×

2700

    void* tmpBuf = taosArrayGetP(rsp.batchMetaReq, i);

×

2701

    RAW_NULL_CHECK(tmpBuf);

×

2702

    SDecoder   metaCoder = {0};

×

2703

    SMqMetaRsp metaRsp = {0};

×

2704

    tDecoderInit(&metaCoder, POINTER_SHIFT(tmpBuf, sizeof(SMqRspHead)), *len - sizeof(SMqRspHead));

×

2705

    if (tDecodeMqMetaRsp(&metaCoder, &metaRsp) < 0) {

×

2706

      code = TSDB_CODE_INVALID_PARA;

×

2707

      goto end;

×

2709

    code = writeRawImpl(taos, metaRsp.metaRsp, metaRsp.metaRspLen, metaRsp.resMsgType);

×

2710

    tDeleteMqMetaRsp(&metaRsp);

×

2711

    if (code != TSDB_CODE_SUCCESS) {

×

2712

      goto end;

×

2716

end:

×

2717

  tDeleteMqBatchMetaRsp(&rsp);

×

2718

  return code;

×

taosdata / TDengine / #3800

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

taosdata / TDengine / #3800

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