taosdata/TDengine | Build 4884 | 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;

×

71

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

×

72

  if (db == NULL) {

×

73

    return suid;

×

75

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

×

77

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

×

79

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

×

80

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

×

81

    return;

×

83

  int32_t code = TSDB_CODE_SUCCESS;

×

84

  int8_t  buildDefaultCompress = 0;

×

85

  if (pColCmprRow->nCols <= 0) {

×

86

    buildDefaultCompress = 1;

×

89

  char*  string = NULL;

×

90

  cJSON* json = cJSON_CreateObject();

×

91

  RAW_NULL_CHECK(json);

×

92

  cJSON* type = cJSON_CreateString("create");

×

93

  RAW_NULL_CHECK(type);

×

95

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

×

96

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

×

97

  RAW_NULL_CHECK(tableType);

×

98

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

×

99

  cJSON* tableName = cJSON_CreateString(name);

×

100

  RAW_NULL_CHECK(tableName);

×

101

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

×

103

  cJSON* columns = cJSON_CreateArray();

×

104

  RAW_NULL_CHECK(columns);

×

105

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

×

107

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

×

108

    cJSON* column = cJSON_CreateObject();

×

109

    RAW_NULL_CHECK(column);

×

110

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(columns, column));

×

111

    SSchema* s = schemaRow->pSchema + i;

×

112

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

×

113

    RAW_NULL_CHECK(cname);

×

114

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

×

115

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

×

116

    RAW_NULL_CHECK(ctype);

×

117

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

×

118

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

×

119

      int32_t length = s->bytes - VARSTR_HEADER_SIZE;

×

120

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

121

      RAW_NULL_CHECK(cbytes);

×

122

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

×

123

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

×

124

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

×

125

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

126

      RAW_NULL_CHECK(cbytes);

×

127

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

×

128

    } else if (IS_STR_DATA_BLOB(s->type)) {

×

129

      int32_t length = s->bytes - BLOBSTR_HEADER_SIZE;

×

130

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

131

      RAW_NULL_CHECK(cbytes);

×

132

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

×

133

    } else if (s->type == TSDB_DATA_TYPE_DECIMAL || s->type == TSDB_DATA_TYPE_DECIMAL64) {

×

134

      int32_t length = pExtSchemas[i].typeMod;

×

135

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

136

      RAW_NULL_CHECK(cbytes);

×

137

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

×

139

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

×

140

    RAW_NULL_CHECK(isPk);

×

141

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

×

143

    if (pColCmprRow == NULL) {

×

144

      continue;

×

147

    uint32_t alg = 0;

×

148

    if (buildDefaultCompress) {

×

149

      alg = createDefaultColCmprByType(s->type);

×

151

      SColCmpr* pColCmpr = pColCmprRow->pColCmpr + i;

×

152

      alg = pColCmpr->alg;

×

154

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

×

155

    RAW_NULL_CHECK(encode);

×

156

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

×

157

    RAW_NULL_CHECK(compress);

×

158

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

×

159

    RAW_NULL_CHECK(level);

×

161

    cJSON* encodeJson = cJSON_CreateString(encode);

×

162

    RAW_NULL_CHECK(encodeJson);

×

163

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

×

165

    cJSON* compressJson = cJSON_CreateString(compress);

×

166

    RAW_NULL_CHECK(compressJson);

×

167

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

×

169

    cJSON* levelJson = cJSON_CreateString(level);

×

170

    RAW_NULL_CHECK(levelJson);

×

171

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

×

174

  cJSON* tags = cJSON_CreateArray();

×

175

  RAW_NULL_CHECK(tags);

×

176

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

×

178

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

×

179

    cJSON* tag = cJSON_CreateObject();

×

180

    RAW_NULL_CHECK(tag);

×

181

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));

×

182

    SSchema* s = schemaTag->pSchema + i;

×

183

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

×

184

    RAW_NULL_CHECK(tname);

×

185

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

×

186

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

×

187

    RAW_NULL_CHECK(ttype);

×

188

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

×

189

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

×

190

      int32_t length = s->bytes - VARSTR_HEADER_SIZE;

×

191

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

192

      RAW_NULL_CHECK(cbytes);

×

193

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

×

194

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

×

195

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

×

196

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

197

      RAW_NULL_CHECK(cbytes);

×

198

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

×

199

    } else if (IS_STR_DATA_BLOB(s->type)) {

×

200

      int32_t length = s->bytes - BLOBSTR_HEADER_SIZE;

×

201

      cJSON*  cbytes = cJSON_CreateNumber(length);

×

202

      RAW_NULL_CHECK(cbytes);

×

203

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

×

207

end:

×

208

  *pJson = json;

×

211

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

×

212

  if (json == NULL) {

×

213

    return TSDB_CODE_INVALID_PARA;

×

215

  uint8_t encode = COMPRESS_L1_TYPE_U32(para);

×

216

  int32_t code = 0;

×

217

  if (encode != 0) {

×

218

    const char* encodeStr = columnEncodeStr(encode);

×

219

    RAW_NULL_CHECK(encodeStr);

×

220

    cJSON* encodeJson = cJSON_CreateString(encodeStr);

×

221

    RAW_NULL_CHECK(encodeJson);

×

222

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

×

223

    return code;

×

225

  uint8_t compress = COMPRESS_L2_TYPE_U32(para);

×

226

  if (compress != 0) {

×

227

    const char* compressStr = columnCompressStr(compress);

×

228

    RAW_NULL_CHECK(compressStr);

×

229

    cJSON* compressJson = cJSON_CreateString(compressStr);

×

230

    RAW_NULL_CHECK(compressJson);

×

231

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

×

232

    return code;

×

234

  uint8_t level = COMPRESS_L2_TYPE_LEVEL_U32(para);

×

235

  if (level != 0) {

×

236

    const char* levelStr = columnLevelStr(level);

×

237

    RAW_NULL_CHECK(levelStr);

×

238

    cJSON* levelJson = cJSON_CreateString(levelStr);

×

239

    RAW_NULL_CHECK(levelJson);

×

240

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

×

241

    return code;

×

244

end:

×

245

  return code;

×

247

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

×

248

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

×

249

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

×

250

    return;

×

252

  SMAlterStbReq req = {0};

×

253

  cJSON*        json = NULL;

×

254

  char*         string = NULL;

×

255

  int32_t       code = 0;

×

257

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

×

258

    goto end;

×

261

  json = cJSON_CreateObject();

×

262

  RAW_NULL_CHECK(json);

×

263

  cJSON* type = cJSON_CreateString("alter");

×

264

  RAW_NULL_CHECK(type);

×

265

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

×

266

  SName name = {0};

×

267

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

×

268

  cJSON* tableType = cJSON_CreateString("super");

×

269

  RAW_NULL_CHECK(tableType);

×

270

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

×

271

  cJSON* tableName = cJSON_CreateString(name.tname);

×

272

  RAW_NULL_CHECK(tableName);

×

273

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

×

275

  cJSON* alterType = cJSON_CreateNumber(req.alterType);

×

276

  RAW_NULL_CHECK(alterType);

×

277

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

×

278

  switch (req.alterType) {

×

279

    case TSDB_ALTER_TABLE_ADD_TAG:

×

281

      if (taosArrayGetSize(req.pFields) != 1) {

×

282

        uError("invalid field num %" PRIzu " for alter type %d", taosArrayGetSize(req.pFields), req.alterType);

×

283

        cJSON_Delete(json);

×

284

        json = NULL;

×

285

        goto end;

×

287

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

×

288

      RAW_NULL_CHECK(field);

×

289

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

×

290

      RAW_NULL_CHECK(colName);

×

291

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

×

292

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

×

293

      RAW_NULL_CHECK(colType);

×

294

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

×

296

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

×

297

          field->type == TSDB_DATA_TYPE_GEOMETRY) {

×

298

        int32_t length = field->bytes - VARSTR_HEADER_SIZE;

×

299

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

300

        RAW_NULL_CHECK(cbytes);

×

301

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

×

302

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

×

303

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

×

304

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

305

        RAW_NULL_CHECK(cbytes);

×

306

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

×

307

      } else if (IS_STR_DATA_BLOB(field->type)) {

×

308

        int32_t length = field->bytes - BLOBSTR_HEADER_SIZE;

×

309

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

310

        RAW_NULL_CHECK(cbytes);

×

311

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

×

313

      break;

×

315

    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION: {

×

316

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

×

317

      RAW_NULL_CHECK(field);

×

318

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

×

319

      RAW_NULL_CHECK(colName);

×

320

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

×

321

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

×

322

      RAW_NULL_CHECK(colType);

×

323

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

×

325

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

×

326

          field->type == TSDB_DATA_TYPE_GEOMETRY) {

×

327

        int32_t length = field->bytes - VARSTR_HEADER_SIZE;

×

328

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

329

        RAW_NULL_CHECK(cbytes);

×

330

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

×

331

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

×

332

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

×

333

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

334

        RAW_NULL_CHECK(cbytes);

×

335

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

×

336

      } else if (IS_STR_DATA_BLOB(field->type)) {

×

337

        int32_t length = field->bytes - BLOBSTR_HEADER_SIZE;

×

338

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

339

        RAW_NULL_CHECK(cbytes);

×

340

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

×

343

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

×

344

      break;

×

346

    case TSDB_ALTER_TABLE_DROP_TAG:

×

348

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

×

349

      RAW_NULL_CHECK(field);

×

350

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

×

351

      RAW_NULL_CHECK(colName);

×

352

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

×

353

      break;

×

355

    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:

×

357

      if (taosArrayGetSize(req.pFields) != 1) {

×

358

        uError("invalid field num %" PRIzu " for alter type %d", taosArrayGetSize(req.pFields), req.alterType);

×

359

        cJSON_Delete(json);

×

360

        json = NULL;

×

361

        goto end;

×

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

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

×

369

      RAW_NULL_CHECK(colType);

×

370

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

×

371

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

×

372

          field->type == TSDB_DATA_TYPE_GEOMETRY) {

×

373

        int32_t length = field->bytes - VARSTR_HEADER_SIZE;

×

374

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

375

        RAW_NULL_CHECK(cbytes);

×

376

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

×

377

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

×

378

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

×

379

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

380

        RAW_NULL_CHECK(cbytes);

×

381

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

×

383

      break;

×

385

    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:

×

387

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

×

388

      RAW_NULL_CHECK(oldField);

×

389

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

×

390

      RAW_NULL_CHECK(newField);

×

391

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

×

392

      RAW_NULL_CHECK(colName);

×

393

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

×

394

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

×

395

      RAW_NULL_CHECK(colNewName);

×

396

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

×

397

      break;

×

399

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {

×

400

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

×

401

      RAW_NULL_CHECK(field);

×

402

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

×

403

      RAW_NULL_CHECK(colName);

×

404

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

×

405

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

×

406

      break;

×

408

    default:

×

409

      break;

×

412

end:

×

413

  tFreeSMAltertbReq(&req);

×

414

  *pJson = json;

×

417

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

×

418

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

×

419

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

×

420

    return;

×

422

  SVCreateStbReq req = {0};

×

423

  SDecoder       coder = {0};

×

425

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

×

427

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

×

428

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

×

429

  tDecoderInit(&coder, data, len);

×

431

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

×

432

    goto end;

×

434

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

×

436

end:

×

437

  uDebug("create stable return");

×

438

  tDecoderClear(&coder);

×

441

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

×

442

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

×

443

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

×

444

    return;

×

446

  SVCreateStbReq req = {0};

×

447

  SDecoder       coder = {0};

×

448

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

×

451

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

×

452

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

×

453

  tDecoderInit(&coder, data, len);

×

455

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

×

456

    goto end;

×

458

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

×

460

end:

×

461

  uDebug("alter stable return");

×

462

  tDecoderClear(&coder);

×

465

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

×

466

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

×

467

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

×

468

    return;

×

470

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

×

471

  char*   sname = pCreateReq->ctb.stbName;

×

472

  char*   name = pCreateReq->name;

×

473

  SArray* tagName = pCreateReq->ctb.tagName;

×

474

  int64_t id = pCreateReq->uid;

×

475

  uint8_t tagNum = pCreateReq->ctb.tagNum;

×

476

  int32_t code = 0;

×

477

  SArray* pTagVals = NULL;

×

478

  char*   pJson = NULL;

×

480

  cJSON* tableName = cJSON_CreateString(name);

×

481

  RAW_NULL_CHECK(tableName);

×

482

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

×

483

  cJSON* using = cJSON_CreateString(sname);

×

484

  RAW_NULL_CHECK(using);

×

485

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

×

486

  cJSON* tagNumJson = cJSON_CreateNumber(tagNum);

×

487

  RAW_NULL_CHECK(tagNumJson);

×

488

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

×

490

  cJSON* tags = cJSON_CreateArray();

×

491

  RAW_NULL_CHECK(tags);

×

492

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

×

493

  RAW_RETURN_CHECK(tTagToValArray(pTag, &pTagVals));

×

494

  if (tTagIsJson(pTag)) {

×

495

    STag* p = (STag*)pTag;

×

496

    if (p->nTag == 0) {

×

497

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

×

498

      goto end;

×

500

    parseTagDatatoJson(pTag, &pJson, NULL);

×

501

    RAW_NULL_CHECK(pJson);

×

502

    cJSON* tag = cJSON_CreateObject();

×

503

    RAW_NULL_CHECK(tag);

×

504

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));

×

505

    STagVal* pTagVal = taosArrayGet(pTagVals, 0);

×

506

    RAW_NULL_CHECK(pTagVal);

×

507

    char* ptname = taosArrayGet(tagName, 0);

×

508

    RAW_NULL_CHECK(ptname);

×

509

    cJSON* tname = cJSON_CreateString(ptname);

×

510

    RAW_NULL_CHECK(tname);

×

511

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

×

512

    cJSON* ttype = cJSON_CreateNumber(TSDB_DATA_TYPE_JSON);

×

513

    RAW_NULL_CHECK(ttype);

×

514

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

×

515

    cJSON* tvalue = cJSON_CreateString(pJson);

×

516

    RAW_NULL_CHECK(tvalue);

×

517

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

×

518

    goto end;

×

521

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

×

522

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

×

523

    RAW_NULL_CHECK(pTagVal);

×

524

    cJSON* tag = cJSON_CreateObject();

×

525

    RAW_NULL_CHECK(tag);

×

526

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));

×

527

    char* ptname = taosArrayGet(tagName, i);

×

528

    RAW_NULL_CHECK(ptname);

×

529

    cJSON* tname = cJSON_CreateString(ptname);

×

530

    RAW_NULL_CHECK(tname);

×

531

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

×

532

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

×

533

    RAW_NULL_CHECK(ttype);

×

534

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

×

536

    cJSON* tvalue = NULL;

×

537

    if (IS_VAR_DATA_TYPE(pTagVal->type)) {

×

538

      if (IS_STR_DATA_BLOB(pTagVal->type)) {

×

539

        goto end;

×

541

      int64_t bufSize = 0;

×

542

      if (pTagVal->type == TSDB_DATA_TYPE_VARBINARY) {

×

543

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

×

545

        bufSize = pTagVal->nData + 3;

×

547

      char* buf = taosMemoryCalloc(bufSize, 1);

×

548

      RAW_NULL_CHECK(buf);

×

549

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

×

550

        taosMemoryFree(buf);

×

551

        goto end;

×

554

      tvalue = cJSON_CreateString(buf);

×

555

      taosMemoryFree(buf);

×

556

      RAW_NULL_CHECK(tvalue);

×

558

      double val = 0;

×

559

      GET_TYPED_DATA(val, double, pTagVal->type, &pTagVal->i64,

×

561

      tvalue = cJSON_CreateNumber(val);

×

562

      RAW_NULL_CHECK(tvalue);

×

565

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

×

568

end:

×

569

  taosMemoryFree(pJson);

×

570

  taosArrayDestroy(pTagVals);

×

573

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

×

574

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

×

575

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

×

576

    return;

×

578

  int32_t code = 0;

×

579

  char*   string = NULL;

×

580

  cJSON*  json = cJSON_CreateObject();

×

581

  RAW_NULL_CHECK(json);

×

582

  cJSON* type = cJSON_CreateString("create");

×

583

  RAW_NULL_CHECK(type);

×

584

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

×

586

  cJSON* tableType = cJSON_CreateString("child");

×

587

  RAW_NULL_CHECK(tableType);

×

588

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

×

590

  buildChildElement(json, pCreateReq);

×

591

  cJSON* createList = cJSON_CreateArray();

×

592

  RAW_NULL_CHECK(createList);

×

593

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

×

595

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

×

596

    cJSON* create = cJSON_CreateObject();

×

597

    RAW_NULL_CHECK(create);

×

598

    buildChildElement(create, pCreateReq + i);

×

599

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(createList, create));

×

602

end:

×

603

  *pJson = json;

×

606

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

×

607

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

×

608

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

×

609

    return;

×

611

  SDecoder           decoder = {0};

×

612

  SVCreateTbBatchReq req = {0};

×

615

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

×

616

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

×

617

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

×

618

  tDecoderInit(&decoder, data, len);

×

619

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

×

620

    goto end;

×

624

  if (req.nReqs > 0) {

×

625

    pCreateReq = req.pReqs;

×

626

    if (pCreateReq->type == TSDB_CHILD_TABLE) {

×

627

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

×

628

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

×

629

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

×

634

end:

×

635

  uDebug("create table return");

×

636

  tDeleteSVCreateTbBatchReq(&req);

×

637

  tDecoderClear(&decoder);

×

640

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

×

641

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

×

642

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

×

643

    return;

×

645

  SDecoder*      decoder = NULL;

×

646

  SVCreateTbReq* pCreateReq = NULL;

×

647

  int32_t        code = 0;

×

648

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

×

649

  if (rsp->createTableNum <= 0) {

×

650

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

×

651

    goto end;

×

654

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

×

655

  RAW_NULL_CHECK(decoder);

×

656

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

×

657

  RAW_NULL_CHECK(pCreateReq);

×

660

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

×

662

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

×

663

    RAW_NULL_CHECK(data);

×

664

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

×

665

    RAW_NULL_CHECK(len);

×

666

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

×

667

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

×

668

      goto end;

×

671

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

×

672

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

×

673

      goto end;

×

676

  cJSON* pJson = NULL;

×

677

  if (pCreateReq->type == TSDB_NORMAL_TABLE) {

×

678

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

×

680

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

×

681

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

×

684

  *string = cJSON_PrintUnformatted(pJson);

×

685

  cJSON_Delete(pJson);

×

687

end:

×

688

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

×

689

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

×

690

    tDecoderClear(&decoder[i]);

×

691

    taosMemoryFreeClear(pCreateReq[i].comment);

×

692

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

×

693

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

×

696

  taosMemoryFree(decoder);

×

697

  taosMemoryFree(pCreateReq);

×

700

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

×

701

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

×

702

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

×

703

    return;

×

705

  SDecoder     decoder = {0};

×

706

  SVAlterTbReq vAlterTbReq = {0};

×

707

  char*        string = NULL;

×

708

  cJSON*       json = NULL;

×

709

  int32_t      code = 0;

×

711

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

×

713

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

×

714

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

×

715

  tDecoderInit(&decoder, data, len);

×

716

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

×

717

    uError("tDecodeSVAlterTbReq error");

×

718

    goto end;

×

721

  json = cJSON_CreateObject();

×

722

  RAW_NULL_CHECK(json);

×

723

  cJSON* type = cJSON_CreateString("alter");

×

724

  RAW_NULL_CHECK(type);

×

725

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

×

726

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

×

727

                                                vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL

×

730

  RAW_NULL_CHECK(tableType);

×

731

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

×

732

  cJSON* tableName = cJSON_CreateString(vAlterTbReq.tbName);

×

733

  RAW_NULL_CHECK(tableName);

×

734

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

×

735

  cJSON* alterType = cJSON_CreateNumber(vAlterTbReq.action);

×

736

  RAW_NULL_CHECK(alterType);

×

737

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

×

739

  switch (vAlterTbReq.action) {

×

740

    case TSDB_ALTER_TABLE_ADD_COLUMN: {

×

741

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

742

      RAW_NULL_CHECK(colName);

×

743

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

×

744

      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type);

×

745

      RAW_NULL_CHECK(colType);

×

746

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

×

748

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

×

749

          vAlterTbReq.type == TSDB_DATA_TYPE_GEOMETRY) {

×

750

        int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE;

×

751

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

752

        RAW_NULL_CHECK(cbytes);

×

753

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

×

754

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

×

755

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

×

756

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

757

        RAW_NULL_CHECK(cbytes);

×

758

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

×

760

      break;

×

762

    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION: {

×

763

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

764

      RAW_NULL_CHECK(colName);

×

765

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

×

766

      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type);

×

767

      RAW_NULL_CHECK(colType);

×

768

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

×

770

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

×

771

          vAlterTbReq.type == TSDB_DATA_TYPE_GEOMETRY) {

×

772

        int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE;

×

773

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

774

        RAW_NULL_CHECK(cbytes);

×

775

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

×

776

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

×

777

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

×

778

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

779

        RAW_NULL_CHECK(cbytes);

×

780

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

×

782

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

×

783

      break;

×

785

    case TSDB_ALTER_TABLE_DROP_COLUMN: {

×

786

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

787

      RAW_NULL_CHECK(colName);

×

788

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

×

789

      break;

×

791

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: {

×

792

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

793

      RAW_NULL_CHECK(colName);

×

794

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

×

795

      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.colModType);

×

796

      RAW_NULL_CHECK(colType);

×

797

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

×

798

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

×

799

          vAlterTbReq.colModType == TSDB_DATA_TYPE_GEOMETRY) {

×

800

        int32_t length = vAlterTbReq.colModBytes - VARSTR_HEADER_SIZE;

×

801

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

802

        RAW_NULL_CHECK(cbytes);

×

803

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

×

804

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

×

805

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

×

806

        cJSON*  cbytes = cJSON_CreateNumber(length);

×

807

        RAW_NULL_CHECK(cbytes);

×

808

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

×

810

      break;

×

812

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: {

×

813

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

814

      RAW_NULL_CHECK(colName);

×

815

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

×

816

      cJSON* colNewName = cJSON_CreateString(vAlterTbReq.colNewName);

×

817

      RAW_NULL_CHECK(colNewName);

×

818

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

×

819

      break;

×

821

    case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: {

×

822

      cJSON* tagName = cJSON_CreateString(vAlterTbReq.tagName);

×

823

      RAW_NULL_CHECK(tagName);

×

824

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

×

826

      bool isNull = vAlterTbReq.isNull;

×

827

      if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) {

×

828

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

×

829

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

×

831

      if (!isNull) {

×

832

        char* buf = NULL;

×

834

        if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) {

×

835

          if (!tTagIsJson(vAlterTbReq.pTagVal)) {

×

836

            uError("processAlterTable isJson false");

×

837

            goto end;

×

839

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

×

840

          if (buf == NULL) {

×

841

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

×

842

            goto end;

×

845

          int64_t bufSize = 0;

×

846

          if (vAlterTbReq.tagType == TSDB_DATA_TYPE_VARBINARY) {

×

847

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

×

849

            bufSize = vAlterTbReq.nTagVal + 32;

×

851

          buf = taosMemoryCalloc(bufSize, 1);

×

852

          RAW_NULL_CHECK(buf);

×

853

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

×

855

            taosMemoryFree(buf);

×

856

            goto end;

×

860

        cJSON* colValue = cJSON_CreateString(buf);

×

861

        taosMemoryFree(buf);

×

862

        RAW_NULL_CHECK(colValue);

×

863

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

×

866

      cJSON* isNullCJson = cJSON_CreateBool(isNull);

×

867

      RAW_NULL_CHECK(isNullCJson);

×

868

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

×

869

      break;

×

871

    case TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL: {

×

872

      int32_t nTags = taosArrayGetSize(vAlterTbReq.pMultiTag);

×

873

      if (nTags <= 0) {

×

874

        uError("processAlterTable parse multi tags error");

×

875

        goto end;

×

878

      cJSON* tags = cJSON_CreateArray();

×

879

      RAW_NULL_CHECK(tags);

×

880

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

×

882

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

×

883

        cJSON* member = cJSON_CreateObject();

×

884

        RAW_NULL_CHECK(member);

×

885

        RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, member));

×

887

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

×

888

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

×

889

        RAW_NULL_CHECK(tagName);

×

890

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

×

892

        if (pTagVal->tagType == TSDB_DATA_TYPE_JSON) {

×

893

          uError("processAlterTable isJson false");

×

894

          goto end;

×

896

        bool isNull = pTagVal->isNull;

×

897

        if (!isNull) {

×

898

          int64_t bufSize = 0;

×

899

          if (pTagVal->tagType == TSDB_DATA_TYPE_VARBINARY) {

×

900

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

×

902

            bufSize = pTagVal->nTagVal + 3;

×

904

          char* buf = taosMemoryCalloc(bufSize, 1);

×

905

          RAW_NULL_CHECK(buf);

×

906

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

×

908

            taosMemoryFree(buf);

×

909

            goto end;

×

911

          cJSON* colValue = cJSON_CreateString(buf);

×

912

          taosMemoryFree(buf);

×

913

          RAW_NULL_CHECK(colValue);

×

914

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

×

916

        cJSON* isNullCJson = cJSON_CreateBool(isNull);

×

917

        RAW_NULL_CHECK(isNullCJson);

×

918

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

×

920

      break;

×

923

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {

×

924

      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);

×

925

      RAW_NULL_CHECK(colName);

×

926

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

×

927

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

×

928

      break;

×

930

    default:

×

931

      break;

×

934

end:

×

935

  uDebug("alter table return");

×

936

  if (vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL) {

×

937

    taosArrayDestroy(vAlterTbReq.pMultiTag);

×

939

  tDecoderClear(&decoder);

×

940

  *pJson = json;

×

943

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

×

944

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

×

945

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

×

946

    return;

×

948

  SDecoder     decoder = {0};

×

949

  SVDropStbReq req = {0};

×

950

  cJSON*       json = NULL;

×

951

  int32_t      code = 0;

×

953

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

×

956

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

×

957

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

×

958

  tDecoderInit(&decoder, data, len);

×

959

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

×

960

    uError("tDecodeSVDropStbReq failed");

×

961

    goto end;

×

964

  json = cJSON_CreateObject();

×

965

  RAW_NULL_CHECK(json);

×

966

  cJSON* type = cJSON_CreateString("drop");

×

967

  RAW_NULL_CHECK(type);

×

968

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

×

969

  cJSON* tableType = cJSON_CreateString("super");

×

970

  RAW_NULL_CHECK(tableType);

×

971

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

×

972

  cJSON* tableName = cJSON_CreateString(req.name);

×

973

  RAW_NULL_CHECK(tableName);

×

974

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

×

976

end:

×

977

  uDebug("processDropSTable return");

×

978

  tDecoderClear(&decoder);

×

979

  *pJson = json;

×

981

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

×

982

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

×

983

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

×

984

    return;

×

986

  SDeleteRes req = {0};

×

987

  SDecoder   coder = {0};

×

988

  cJSON*     json = NULL;

×

989

  int32_t    code = 0;

×

991

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

×

993

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

×

994

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

×

996

  tDecoderInit(&coder, data, len);

×

997

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

×

998

    uError("tDecodeDeleteRes failed");

×

999

    goto end;

×

1003

  char sql[256] = {0};

×

1004

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

×

1007

  json = cJSON_CreateObject();

×

1008

  RAW_NULL_CHECK(json);

×

1009

  cJSON* type = cJSON_CreateString("delete");

×

1010

  RAW_NULL_CHECK(type);

×

1011

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

×

1012

  cJSON* sqlJson = cJSON_CreateString(sql);

×

1013

  RAW_NULL_CHECK(sqlJson);

×

1014

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

×

1016

end:

×

1017

  uDebug("processDeleteTable return");

×

1018

  tDecoderClear(&coder);

×

1019

  *pJson = json;

×

1022

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

×

1023

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

×

1024

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

×

1025

    return;

×

1027

  SDecoder         decoder = {0};

×

1028

  SVDropTbBatchReq req = {0};

×

1029

  cJSON*           json = NULL;

×

1030

  int32_t          code = 0;

×

1032

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

×

1034

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

×

1035

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

×

1036

  tDecoderInit(&decoder, data, len);

×

1037

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

×

1038

    uError("tDecodeSVDropTbBatchReq failed");

×

1039

    goto end;

×

1042

  json = cJSON_CreateObject();

×

1043

  RAW_NULL_CHECK(json);

×

1044

  cJSON* type = cJSON_CreateString("drop");

×

1045

  RAW_NULL_CHECK(type);

×

1046

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

×

1047

  cJSON* tableNameList = cJSON_CreateArray();

×

1048

  RAW_NULL_CHECK(tableNameList);

×

1049

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

×

1051

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

×

1052

    SVDropTbReq* pDropTbReq = req.pReqs + iReq;

×

1053

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

×

1054

    RAW_NULL_CHECK(tableName);

×

1055

    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tableNameList, tableName));

×

1058

end:

×

1059

  uDebug("processDropTable return");

×

1060

  tDecoderClear(&decoder);

×

1061

  *pJson = json;

×

1064

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

×

1065

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

×

1066

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

×

1067

    return TSDB_CODE_INVALID_PARA;

×

1069

  SVCreateStbReq req = {0};

×

1070

  SDecoder       coder = {0};

×

1071

  SMCreateStbReq pReq = {0};

×

1072

  int32_t        code = TSDB_CODE_SUCCESS;

×

1073

  SRequestObj*   pRequest = NULL;

×

1075

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

×

1076

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

×

1077

  pRequest->syncQuery = true;

×

1078

  if (!pRequest->pDb) {

×

1079

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1080

    goto end;

×

1083

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

×

1084

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1085

  tDecoderInit(&coder, data, len);

×

1086

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

×

1087

    code = TSDB_CODE_INVALID_PARA;

×

1088

    goto end;

×

1091

  int8_t           createDefaultCompress = 0;

×

1092

  SColCmprWrapper* p = &req.colCmpr;

×

1093

  if (p->nCols == 0) {

×

1094

    createDefaultCompress = 1;

×

1097

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

×

1098

  RAW_NULL_CHECK(pReq.pColumns);

×

1099

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

×

1100

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

×

1101

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

×

1102

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

×

1104

    if (createDefaultCompress) {

×

1105

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

×

1107

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

×

1108

      field.compress = pCmp->alg;

×

1110

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

×

1111

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

×

1113

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

×

1114

  RAW_NULL_CHECK(pReq.pTags);

×

1115

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

×

1116

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

×

1117

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

×

1118

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

×

1119

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

×

1122

  pReq.colVer = req.schemaRow.version;

×

1123

  pReq.tagVer = req.schemaTag.version;

×

1124

  pReq.numOfColumns = req.schemaRow.nCols;

×

1125

  pReq.numOfTags = req.schemaTag.nCols;

×

1126

  pReq.commentLen = -1;

×

1127

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

×

1128

  pReq.source = TD_REQ_FROM_TAOX;

×

1129

  pReq.igExists = true;

×

1131

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

×

1133

  STscObj* pTscObj = pRequest->pTscObj;

×

1134

  SName    tableName = {0};

×

1135

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

×

1136

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

×

1137

  SCmdMsgInfo pCmdMsg = {0};

×

1138

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

×

1139

  pCmdMsg.msgType = TDMT_MND_CREATE_STB;

×

1140

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

×

1141

  if (pCmdMsg.msgLen <= 0) {

×

1142

    code = TSDB_CODE_INVALID_PARA;

×

1143

    goto end;

×

1145

  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);

×

1146

  RAW_NULL_CHECK(pCmdMsg.pMsg);

×

1147

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

×

1148

    code = TSDB_CODE_INVALID_PARA;

×

1149

    taosMemoryFree(pCmdMsg.pMsg);

×

1150

    goto end;

×

1153

  SQuery pQuery = {0};

×

1154

  pQuery.execMode = QUERY_EXEC_MODE_RPC;

×

1155

  pQuery.pCmdMsg = &pCmdMsg;

×

1156

  pQuery.msgType = pQuery.pCmdMsg->msgType;

×

1157

  pQuery.stableQuery = true;

×

1159

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

×

1161

  taosMemoryFree(pCmdMsg.pMsg);

×

1163

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1164

    SCatalog* pCatalog = NULL;

×

1165

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

×

1166

    RAW_RETURN_CHECK(catalogRemoveTableMeta(pCatalog, &tableName));

×

1169

  code = pRequest->code;

×

1171

end:

×

1172

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

×

1173

  destroyRequest(pRequest);

×

1174

  tFreeSMCreateStbReq(&pReq);

×

1175

  tDecoderClear(&coder);

×

1176

  return code;

×

1179

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

×

1180

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

×

1181

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

×

1182

    return TSDB_CODE_INVALID_PARA;

×

1184

  SVDropStbReq req = {0};

×

1185

  SDecoder     coder = {0};

×

1186

  SMDropStbReq pReq = {0};

×

1187

  int32_t      code = TSDB_CODE_SUCCESS;

×

1188

  SRequestObj* pRequest = NULL;

×

1190

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

×

1191

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

×

1192

  pRequest->syncQuery = true;

×

1193

  if (!pRequest->pDb) {

×

1194

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1195

    goto end;

×

1198

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

×

1199

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1200

  tDecoderInit(&coder, data, len);

×

1201

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

×

1202

    code = TSDB_CODE_INVALID_PARA;

×

1203

    goto end;

×

1206

  SCatalog* pCatalog = NULL;

×

1207

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

×

1208

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

×

1209

                           .requestId = pRequest->requestId,

×

1210

                           .requestObjRefId = pRequest->self,

×

1211

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

×

1212

  SName            pName = {0};

×

1213

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

×

1214

  STableMeta* pTableMeta = NULL;

×

1215

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

×

1216

  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1217

    code = TSDB_CODE_SUCCESS;

×

1218

    taosMemoryFreeClear(pTableMeta);

×

1219

    goto end;

×

1221

  if (code != TSDB_CODE_SUCCESS) {

×

1222

    goto end;

×

1224

  pReq.suid = pTableMeta->uid;

×

1225

  taosMemoryFreeClear(pTableMeta);

×

1228

  pReq.igNotExists = true;

×

1229

  pReq.source = TD_REQ_FROM_TAOX;

×

1232

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

×

1234

  STscObj* pTscObj = pRequest->pTscObj;

×

1235

  SName    tableName = {0};

×

1236

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

×

1237

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

×

1238

    code = TSDB_CODE_INVALID_PARA;

×

1239

    goto end;

×

1242

  SCmdMsgInfo pCmdMsg = {0};

×

1243

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

×

1244

  pCmdMsg.msgType = TDMT_MND_DROP_STB;

×

1245

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

×

1246

  if (pCmdMsg.msgLen <= 0) {

×

1247

    code = TSDB_CODE_INVALID_PARA;

×

1248

    goto end;

×

1250

  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);

×

1251

  RAW_NULL_CHECK(pCmdMsg.pMsg);

×

1252

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

×

1253

    code = TSDB_CODE_INVALID_PARA;

×

1254

    taosMemoryFree(pCmdMsg.pMsg);

×

1255

    goto end;

×

1258

  SQuery pQuery = {0};

×

1259

  pQuery.execMode = QUERY_EXEC_MODE_RPC;

×

1260

  pQuery.pCmdMsg = &pCmdMsg;

×

1261

  pQuery.msgType = pQuery.pCmdMsg->msgType;

×

1262

  pQuery.stableQuery = true;

×

1264

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

×

1265

  taosMemoryFree(pCmdMsg.pMsg);

×

1266

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1268

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

×

1269

    RAW_RETURN_CHECK(catalogRemoveTableMeta(pCatalog, &tableName));

×

1272

  code = pRequest->code;

×

1274

end:

×

1275

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

×

1276

  destroyRequest(pRequest);

×

1277

  tDecoderClear(&coder);

×

1278

  return code;

×

1287

static void destroyCreateTbReqBatch(void* data) {

×

1288

  if (data == NULL) {

×

1289

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

×

1290

    return;

×

1292

  SVgroupCreateTableBatch* pTbBatch = (SVgroupCreateTableBatch*)data;

×

1293

  taosArrayDestroy(pTbBatch->req.pArray);

×

1296

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

×

1297

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

×

1298

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

×

1299

    return TSDB_CODE_INVALID_PARA;

×

1301

  SVCreateTbBatchReq req = {0};

×

1302

  SDecoder           coder = {0};

×

1303

  int32_t            code = TSDB_CODE_SUCCESS;

×

1304

  SRequestObj*       pRequest = NULL;

×

1305

  SQuery*            pQuery = NULL;

×

1306

  SHashObj*          pVgroupHashmap = NULL;

×

1307

  SArray*            pTagList = taosArrayInit(0, POINTER_BYTES);

×

1308

  RAW_NULL_CHECK(pTagList);

×

1309

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

×

1310

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

×

1312

  pRequest->syncQuery = true;

×

1313

  if (!pRequest->pDb) {

×

1314

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1315

    goto end;

×

1318

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

×

1319

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1320

  tDecoderInit(&coder, data, len);

×

1321

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

×

1322

    code = TSDB_CODE_INVALID_PARA;

×

1323

    goto end;

×

1326

  STscObj* pTscObj = pRequest->pTscObj;

×

1328

  SVCreateTbReq* pCreateReq = NULL;

×

1329

  SCatalog*      pCatalog = NULL;

×

1330

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

×

1331

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

×

1332

  RAW_NULL_CHECK(pVgroupHashmap);

×

1333

  taosHashSetFreeFp(pVgroupHashmap, destroyCreateTbReqBatch);

×

1335

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

×

1336

                           .requestId = pRequest->requestId,

×

1337

                           .requestObjRefId = pRequest->self,

×

1338

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

×

1340

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

×

1341

  RAW_NULL_CHECK(pRequest->tableList);

×

1343

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

×

1344

    pCreateReq = req.pReqs + iReq;

×

1346

    SVgroupInfo pInfo = {0};

×

1347

    SName       pName = {0};

×

1348

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

×

1349

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

×

1350

    if (code != TSDB_CODE_SUCCESS) {

×

1351

      goto end;

×

1354

    pCreateReq->flags |= TD_CREATE_IF_NOT_EXISTS;

×

1356

    if (pCreateReq->type == TSDB_CHILD_TABLE) {

×

1357

      STableMeta* pTableMeta = NULL;

×

1358

      SName       sName = {0};

×

1359

      tb_uid_t    oldSuid = pCreateReq->ctb.suid;

×

1361

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

×

1362

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

×

1363

      if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1364

        code = TSDB_CODE_SUCCESS;

×

1365

        taosMemoryFreeClear(pTableMeta);

×

1366

        continue;

×

1369

      if (code != TSDB_CODE_SUCCESS) {

×

1370

        goto end;

×

1372

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

×

1374

      SArray* pTagVals = NULL;

×

1375

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

×

1376

      if (code != TSDB_CODE_SUCCESS) {

×

1377

        taosMemoryFreeClear(pTableMeta);

×

1378

        goto end;

×

1381

      bool rebuildTag = false;

×

1382

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

×

1383

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

×

1384

        if (tName == NULL) {

×

1385

          continue;

×

1387

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

×

1388

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

×

1389

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

×

1390

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

×

1391

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

×

1392

            if (pTagVal) {

×

1393

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

×

1394

                pTagVal->cid = tag->colId;

×

1395

                rebuildTag = true;

×

1398

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

×

1404

      taosMemoryFreeClear(pTableMeta);

×

1405

      if (rebuildTag) {

×

1406

        STag* ppTag = NULL;

×

1407

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

×

1408

        taosArrayDestroy(pTagVals);

×

1409

        pTagVals = NULL;

×

1410

        if (code != TSDB_CODE_SUCCESS) {

×

1411

          goto end;

×

1413

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

×

1414

          tTagFree(ppTag);

×

1415

          goto end;

×

1417

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

×

1419

      taosArrayDestroy(pTagVals);

×

1421

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

×

1423

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

×

1424

    if (pTableBatch == NULL) {

×

1425

      SVgroupCreateTableBatch tBatch = {0};

×

1426

      tBatch.info = pInfo;

×

1427

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

×

1429

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

×

1430

      RAW_NULL_CHECK(tBatch.req.pArray);

×

1431

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

×

1432

      tBatch.req.source = TD_REQ_FROM_TAOX;

×

1433

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

×

1435

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

×

1439

  if (taosHashGetSize(pVgroupHashmap) == 0) {

×

1440

    goto end;

×

1442

  SArray* pBufArray = NULL;

×

1443

  RAW_RETURN_CHECK(serializeVgroupsCreateTableBatch(pVgroupHashmap, &pBufArray));

×

1444

  pQuery = NULL;

×

1445

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

×

1446

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1447

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1448

  pQuery->msgType = TDMT_VND_CREATE_TABLE;

×

1449

  pQuery->stableQuery = false;

×

1450

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

×

1451

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1452

  RAW_NULL_CHECK(pQuery->pRoot);

×

1454

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pBufArray));

×

1456

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1457

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1458

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

×

1461

  code = pRequest->code;

×

1463

end:

×

1464

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

×

1465

  tDeleteSVCreateTbBatchReq(&req);

×

1467

  taosHashCleanup(pVgroupHashmap);

×

1468

  destroyRequest(pRequest);

×

1469

  tDecoderClear(&coder);

×

1470

  qDestroyQuery(pQuery);

×

1471

  taosArrayDestroyP(pTagList, NULL);

×

1472

  return code;

×

1481

static void destroyDropTbReqBatch(void* data) {

×

1482

  if (data == NULL) {

×

1483

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

×

1484

    return;

×

1486

  SVgroupDropTableBatch* pTbBatch = (SVgroupDropTableBatch*)data;

×

1487

  taosArrayDestroy(pTbBatch->req.pArray);

×

1490

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

×

1491

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

×

1492

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

×

1493

    return TSDB_CODE_INVALID_PARA;

×

1495

  SVDropTbBatchReq req = {0};

×

1496

  SDecoder         coder = {0};

×

1497

  int32_t          code = TSDB_CODE_SUCCESS;

×

1498

  SRequestObj*     pRequest = NULL;

×

1499

  SQuery*          pQuery = NULL;

×

1500

  SHashObj*        pVgroupHashmap = NULL;

×

1502

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

×

1503

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

×

1505

  pRequest->syncQuery = true;

×

1506

  if (!pRequest->pDb) {

×

1507

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1508

    goto end;

×

1511

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

×

1512

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1513

  tDecoderInit(&coder, data, len);

×

1514

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

×

1515

    code = TSDB_CODE_INVALID_PARA;

×

1516

    goto end;

×

1519

  STscObj* pTscObj = pRequest->pTscObj;

×

1521

  SVDropTbReq* pDropReq = NULL;

×

1522

  SCatalog*    pCatalog = NULL;

×

1523

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

×

1525

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

×

1526

  RAW_NULL_CHECK(pVgroupHashmap);

×

1527

  taosHashSetFreeFp(pVgroupHashmap, destroyDropTbReqBatch);

×

1529

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

×

1530

                           .requestId = pRequest->requestId,

×

1531

                           .requestObjRefId = pRequest->self,

×

1532

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

×

1533

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

×

1534

  RAW_NULL_CHECK(pRequest->tableList);

×

1536

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

×

1537

    pDropReq = req.pReqs + iReq;

×

1538

    pDropReq->igNotExists = true;

×

1541

    SVgroupInfo pInfo = {0};

×

1542

    SName       pName = {0};

×

1543

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

×

1544

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

×

1546

    STableMeta* pTableMeta = NULL;

×

1547

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

×

1548

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1549

      code = TSDB_CODE_SUCCESS;

×

1550

      taosMemoryFreeClear(pTableMeta);

×

1551

      continue;

×

1553

    if (code != TSDB_CODE_SUCCESS) {

×

1554

      goto end;

×

1556

    tb_uid_t oldSuid = pDropReq->suid;

×

1557

    pDropReq->suid = pTableMeta->suid;

×

1558

    taosMemoryFreeClear(pTableMeta);

×

1559

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

×

1562

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

×

1563

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

×

1564

    if (pTableBatch == NULL) {

×

1565

      SVgroupDropTableBatch tBatch = {0};

×

1566

      tBatch.info = pInfo;

×

1567

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

×

1568

      RAW_NULL_CHECK(tBatch.req.pArray);

×

1569

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

×

1570

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

×

1572

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

×

1576

  if (taosHashGetSize(pVgroupHashmap) == 0) {

×

1577

    goto end;

×

1579

  SArray* pBufArray = NULL;

×

1580

  RAW_RETURN_CHECK(serializeVgroupsDropTableBatch(pVgroupHashmap, &pBufArray));

×

1581

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

×

1582

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1583

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1584

  pQuery->msgType = TDMT_VND_DROP_TABLE;

×

1585

  pQuery->stableQuery = false;

×

1586

  pQuery->pRoot = NULL;

×

1587

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

×

1588

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1589

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pBufArray));

×

1591

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1592

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1593

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

×

1595

  code = pRequest->code;

×

1597

end:

×

1598

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

×

1599

  taosHashCleanup(pVgroupHashmap);

×

1600

  destroyRequest(pRequest);

×

1601

  tDecoderClear(&coder);

×

1602

  qDestroyQuery(pQuery);

×

1603

  return code;

×

1606

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

×

1607

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

×

1608

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

×

1609

    return TSDB_CODE_INVALID_PARA;

×

1611

  SDeleteRes req = {0};

×

1612

  SDecoder   coder = {0};

×

1613

  char       sql[256] = {0};

×

1614

  int32_t    code = TSDB_CODE_SUCCESS;

×

1616

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

×

1619

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

×

1620

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1621

  tDecoderInit(&coder, data, len);

×

1622

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

×

1623

    code = TSDB_CODE_INVALID_PARA;

×

1624

    goto end;

×

1627

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

×

1630

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

×

1631

  RAW_NULL_CHECK(res);

×

1632

  SRequestObj* pRequest = (SRequestObj*)res;

×

1633

  code = pRequest->code;

×

1634

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

×

1635

    code = TSDB_CODE_SUCCESS;

×

1637

  taos_free_result(res);

×

1639

end:

×

1640

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

×

1641

  tDecoderClear(&coder);

×

1642

  return code;

×

1645

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

×

1646

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

×

1647

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

×

1648

    return TSDB_CODE_INVALID_PARA;

×

1650

  SVAlterTbReq   req = {0};

×

1651

  SDecoder       dcoder = {0};

×

1652

  int32_t        code = TSDB_CODE_SUCCESS;

×

1653

  SRequestObj*   pRequest = NULL;

×

1654

  SQuery*        pQuery = NULL;

×

1655

  SArray*        pArray = NULL;

×

1656

  SVgDataBlocks* pVgData = NULL;

×

1658

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

×

1659

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

×

1660

  pRequest->syncQuery = true;

×

1661

  if (!pRequest->pDb) {

×

1662

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1663

    goto end;

×

1666

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

×

1667

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1668

  tDecoderInit(&dcoder, data, len);

×

1669

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

×

1670

    code = TSDB_CODE_INVALID_PARA;

×

1671

    goto end;

×

1675

  if (req.action == TSDB_ALTER_TABLE_UPDATE_OPTIONS) {

×

1676

    goto end;

×

1679

  STscObj*  pTscObj = pRequest->pTscObj;

×

1680

  SCatalog* pCatalog = NULL;

×

1681

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

×

1682

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

×

1683

                           .requestId = pRequest->requestId,

×

1684

                           .requestObjRefId = pRequest->self,

×

1685

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

×

1687

  SVgroupInfo pInfo = {0};

×

1688

  SName       pName = {0};

×

1689

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

×

1690

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

×

1691

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

×

1692

  RAW_NULL_CHECK(pArray);

×

1694

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

×

1695

  RAW_NULL_CHECK(pVgData);

×

1696

  pVgData->vg = pInfo;

×

1698

  int tlen = 0;

×

1699

  req.source = TD_REQ_FROM_TAOX;

×

1700

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

×

1701

  if (code != 0) {

×

1702

    code = terrno;

×

1703

    goto end;

×

1705

  tlen += sizeof(SMsgHead);

×

1706

  void* pMsg = taosMemoryMalloc(tlen);

×

1707

  RAW_NULL_CHECK(pMsg);

×

1708

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

×

1709

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

×

1710

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

×

1711

  SEncoder coder = {0};

×

1712

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

×

1713

  code = tEncodeSVAlterTbReq(&coder, &req);

×

1714

  if (code != 0) {

×

1715

    tEncoderClear(&coder);

×

1716

    code = terrno;

×

1717

    goto end;

×

1719

  tEncoderClear(&coder);

×

1721

  pVgData->pData = pMsg;

×

1722

  pVgData->size = tlen;

×

1724

  pVgData->numOfTables = 1;

×

1725

  RAW_NULL_CHECK(taosArrayPush(pArray, &pVgData));

×

1727

  pQuery = NULL;

×

1728

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

×

1729

  if (NULL == pQuery) goto end;

×

1730

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1731

  pQuery->msgType = TDMT_VND_ALTER_TABLE;

×

1732

  pQuery->stableQuery = false;

×

1733

  pQuery->pRoot = NULL;

×

1734

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

×

1735

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1736

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pArray));

×

1738

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1740

  pVgData = NULL;

×

1741

  pArray = NULL;

×

1742

  code = pRequest->code;

×

1743

  if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

×

1744

    code = TSDB_CODE_SUCCESS;

×

1747

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1748

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

×

1749

    if (pRes->res != NULL) {

×

1750

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

×

1753

end:

×

1754

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

×

1755

  taosArrayDestroy(pArray);

×

1756

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

×

1757

  taosMemoryFreeClear(pVgData);

×

1758

  destroyRequest(pRequest);

×

1759

  tDecoderClear(&dcoder);

×

1760

  qDestroyQuery(pQuery);

×

1761

  taosArrayDestroy(req.pMultiTag);

×

1762

  return code;

×

1765

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

×

1767

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

×

1770

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

×

1772

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

×

1773

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

×

1774

    return TSDB_CODE_INVALID_PARA;

×

1776

  int32_t     code = TSDB_CODE_SUCCESS;

×

1777

  STableMeta* pTableMeta = NULL;

×

1778

  SQuery*     pQuery = NULL;

×

1779

  SHashObj*   pVgHash = NULL;

×

1781

  SRequestObj* pRequest = NULL;

×

1782

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

×

1784

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

×

1787

  pRequest->syncQuery = true;

×

1788

  if (!pRequest->pDb) {

×

1789

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1790

    goto end;

×

1793

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

×

1794

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

×

1795

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

×

1797

  struct SCatalog* pCatalog = NULL;

×

1798

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

×

1800

  SRequestConnInfo conn = {0};

×

1801

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

×

1802

  conn.requestId = pRequest->requestId;

×

1803

  conn.requestObjRefId = pRequest->self;

×

1804

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

×

1806

  SVgroupInfo vgData = {0};

×

1807

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

×

1808

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

×

1809

  RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

1810

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

×

1811

  RAW_NULL_CHECK(pVgHash);

×

1812

  RAW_RETURN_CHECK(

×

1814

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

×

1815

  RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

1817

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1818

  code = pRequest->code;

×

1820

end:

×

1821

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

×

1822

  taosMemoryFreeClear(pTableMeta);

×

1823

  qDestroyQuery(pQuery);

×

1824

  destroyRequest(pRequest);

×

1825

  taosHashCleanup(pVgHash);

×

1826

  return code;

×

1829

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

×

1830

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

×

1833

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

×

1834

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

×

1835

    return TSDB_CODE_INVALID_PARA;

×

1837

  int32_t     code = TSDB_CODE_SUCCESS;

×

1838

  STableMeta* pTableMeta = NULL;

×

1839

  SQuery*     pQuery = NULL;

×

1840

  SHashObj*   pVgHash = NULL;

×

1842

  SRequestObj* pRequest = NULL;

×

1843

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

×

1845

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

×

1847

  pRequest->syncQuery = true;

×

1848

  if (!pRequest->pDb) {

×

1849

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1850

    goto end;

×

1853

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

×

1854

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

×

1855

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

×

1857

  struct SCatalog* pCatalog = NULL;

×

1858

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

×

1860

  SRequestConnInfo conn = {0};

×

1861

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

×

1862

  conn.requestId = pRequest->requestId;

×

1863

  conn.requestObjRefId = pRequest->self;

×

1864

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

×

1866

  SVgroupInfo vgData = {0};

×

1867

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

×

1868

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

×

1869

  RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

1870

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

×

1871

  RAW_NULL_CHECK(pVgHash);

×

1872

  RAW_RETURN_CHECK(

×

1874

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

×

1875

  RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

1877

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1878

  code = pRequest->code;

×

1880

end:

×

1881

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

×

1882

  taosMemoryFreeClear(pTableMeta);

×

1883

  qDestroyQuery(pQuery);

×

1884

  destroyRequest(pRequest);

×

1885

  taosHashCleanup(pVgHash);

×

1886

  return code;

×

1889

static void* getRawDataFromRes(void* pRetrieve) {

×

1890

  if (pRetrieve == NULL) {

×

1891

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

×

1892

    return NULL;

×

1894

  void* rawData = NULL;

×

1896

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

×

1897

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

×

1898

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

×

1899

             *(int64_t*)pRetrieve == RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION) {

×

1900

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

×

1902

  return rawData;

×

1905

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

×

1906

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

×

1907

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

×

1908

    return TSDB_CODE_INVALID_PARA;

×

1911

  int32_t       code = 0;

×

1912

  SVCreateTbReq pCreateReq = {0};

×

1913

  SDecoder      decoderTmp = {0};

×

1915

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

×

1916

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

×

1917

    RAW_NULL_CHECK(dataTmp);

×

1918

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

×

1919

    RAW_NULL_CHECK(lenTmp);

×

1921

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

×

1922

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

×

1924

    if (pCreateReq.type != TSDB_CHILD_TABLE) {

×

1925

      code = TSDB_CODE_INVALID_MSG;

×

1926

      goto end;

×

1928

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

×

1929

      RAW_RETURN_CHECK(

×

1932

      tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);

×

1933

      pCreateReq = (SVCreateTbReq){0};

×

1936

    tDecoderClear(&decoderTmp);

×

1938

  return 0;

×

1940

end:

×

1941

  tDecoderClear(&decoderTmp);

×

1942

  tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);

×

1943

  return code;

×

1968

static void tmqFreeMeta(void* data) {

×

1969

  if (data == NULL) {

×

1970

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

×

1971

    return;

×

1973

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

×

1974

  taosMemoryFree(pTableMeta);

×

1977

static void freeRawCache(void* data) {

×

1978

  if (data == NULL) {

×

1979

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

×

1980

    return;

×

1982

  rawCacheInfo* pRawCache = (rawCacheInfo*)data;

×

1983

  taosHashCleanup(pRawCache->pMetaHash);

×

1984

  taosHashCleanup(pRawCache->pNameHash);

×

1985

  taosHashCleanup(pRawCache->pVgHash);

×

1988

static int32_t initRawCacheHash() {

×

1989

  if (writeRawCache == NULL) {

×

1990

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

×

1991

    if (writeRawCache == NULL) {

×

1992

      return terrno;

×

1994

    taosHashSetFreeFp(writeRawCache, freeRawCache);

×

1996

  return 0;

×

1999

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

×

2000

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

×

2001

    return false;

×

2003

  if (pTableMeta == NULL) {

×

2004

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

×

2005

    return false;

×

2007

  char* p = (char*)rawData;

×

2010

  p += sizeof(int32_t);

×

2011

  p += sizeof(int32_t);

×

2012

  p += sizeof(int32_t);

×

2013

  p += sizeof(int32_t);

×

2014

  p += sizeof(int32_t);

×

2015

  p += sizeof(uint64_t);

×

2016

  int8_t* fields = p;

×

2018

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

×

2019

    return true;

×

2022

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

×

2023

    int j = 0;

×

2024

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

×

2025

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

×

2026

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

×

2027

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

×

2029

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

×

2030

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

×

2031

          return true;

×

2033

        break;

×

2036

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

×

2038

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

×

2040

  return false;

×

2043

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

×

2044

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

×

2045

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

×

2046

    return TSDB_CODE_INVALID_PARA;

×

2048

  int32_t code = 0;

×

2049

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

×

2050

  if (cacheInfo == NULL) {

×

2051

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

×

2052

    RAW_NULL_CHECK(*pVgHash);

×

2053

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

×

2054

    RAW_NULL_CHECK(*pNameHash);

×

2055

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

×

2056

    RAW_NULL_CHECK(*pMetaHash);

×

2057

    taosHashSetFreeFp(*pMetaHash, tmqFreeMeta);

×

2058

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

×

2059

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

×

2061

    rawCacheInfo* info = (rawCacheInfo*)cacheInfo;

×

2062

    *pVgHash = info->pVgHash;

×

2063

    *pNameHash = info->pNameHash;

×

2064

    *pMetaHash = info->pMetaHash;

×

2067

  return 0;

×

2068

end:

×

2069

  taosHashCleanup(*pMetaHash);

×

2070

  taosHashCleanup(*pNameHash);

×

2071

  taosHashCleanup(*pVgHash);

×

2072

  return code;

×

2075

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

×

2076

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

×

2077

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

×

2078

    return TSDB_CODE_INVALID_PARA;

×

2080

  int32_t code = 0;

×

2081

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

×

2082

  (*pRequest)->syncQuery = true;

×

2083

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

×

2084

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

2085

    goto end;

×

2088

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

×

2089

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

×

2090

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

×

2091

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

×

2092

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

×

2094

end:

×

2095

  return code;

×

2099

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

×

2101

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

×

2102

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

×

2103

    return TSDB_CODE_INVALID_PARA;

×

2105

  int8_t dataVersion = *(int8_t*)data;

×

2106

  if (dataVersion >= MQ_DATA_RSP_VERSION) {

×

2107

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

×

2108

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

×

2109

      return TSDB_CODE_INVALID_PARA;

×

2111

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

×

2114

  rspObj->resIter = -1;

×

2115

  tDecoderInit(decoder, data, dataLen);

×

2116

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

×

2117

  if (code != 0) {

×

2118

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

×

2120

  return code;

×

2123

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

×

2126

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

×

2128

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

×

2129

    return TSDB_CODE_INVALID_PARA;

×

2131

  int32_t     code = 0;

×

2132

  STableMeta* pTableMeta = NULL;

×

2133

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

×

2134

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

×

2135

    tbInfo info = {0};

×

2137

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

×

2138

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

×

2139

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

×

2141

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

×

2142

    info.uid = pTableMeta->uid;

×

2143

    if (pTableMeta->tableType == TSDB_CHILD_TABLE) {

×

2144

      info.suid = pTableMeta->suid;

×

2146

      info.suid = pTableMeta->uid;

×

2148

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

×

2149

    if (code != 0) {

×

2150

      taosMemoryFree(pTableMeta);

×

2151

      goto end;

×

2153

    uDebug("put table meta to hash1, suid:%" PRId64 ", metaHashSIze:%d, nameHashSize:%d, vgHashSize:%d", info.suid, taosHashGetSize(pMetaHash),

×

2155

    if (pCreateReqDst) {

×

2156

      pTableMeta->vgId = info.vgInfo.vgId;

×

2157

      pTableMeta->uid = pCreateReqDst->uid;

×

2158

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

×

2161

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

×

2162

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

×

2163

    RAW_RETURN_CHECK(

×

2167

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

×

2168

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

×

2169

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

×

2170

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

×

2171

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

×

2172

      if (code != 0) {

×

2173

        taosMemoryFree(pTableMeta);

×

2174

        goto end;

×

2176

      uDebug("put table meta to hash2, suid:%" PRId64 ", metaHashSIze:%d, nameHashSize:%d, vgHashSize:%d", tmpInfo->suid, taosHashGetSize(pMetaHash),

×

2179

      pTableMeta = *pTableMetaTmp;

×

2180

      pTableMeta->uid = tmpInfo->uid;

×

2181

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

×

2184

  *pMeta = pTableMeta;

×

2186

end:

×

2187

  return code;

×

2190

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

×

2191

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

×

2192

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

×

2193

    return TSDB_CODE_INVALID_PARA;

×

2195

  int32_t   code = TSDB_CODE_SUCCESS;

×

2196

  SQuery*   pQuery = NULL;

×

2197

  SMqRspObj rspObj = {0};

×

2198

  SDecoder  decoder = {0};

×

2200

  SRequestObj*     pRequest = NULL;

×

2201

  SCatalog*        pCatalog = NULL;

×

2202

  SRequestConnInfo conn = {0};

×

2203

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

×

2204

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

×

2205

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

×

2207

  SHashObj* pVgHash = NULL;

×

2208

  SHashObj* pNameHash = NULL;

×

2209

  SHashObj* pMetaHash = NULL;

×

2210

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

×

2211

  int retry = 0;

×

2213

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2214

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

×

2215

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

×

2216

      if (!rspObj.dataRsp.withSchema) {

×

2217

        goto end;

×

2220

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

×

2221

      RAW_NULL_CHECK(tbName);

×

2222

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

×

2223

      RAW_NULL_CHECK(pSW);

×

2224

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

×

2225

      RAW_NULL_CHECK(pRetrieve);

×

2226

      void* rawData = getRawDataFromRes(pRetrieve);

×

2227

      RAW_NULL_CHECK(rawData);

×

2229

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

×

2230

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

×

2231

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

×

2232

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2234

      STableMeta* pTableMeta = NULL;

×

2235

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

×

2237

      char err[ERR_MSG_LEN] = {0};

×

2238

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

×

2239

      if (code != TSDB_CODE_SUCCESS) {

×

2240

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

×

2241

        goto end;

×

2244

    RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

2245

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2246

    code = pRequest->code;

×

2248

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

×

2249

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

×

2250

      qDestroyQuery(pQuery);

×

2251

      pQuery = NULL;

×

2252

      rspObj.resIter = -1;

×

2253

      continue;

×

2255

    break;

×

2258

end:

×

2259

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

×

2260

  tDeleteMqDataRsp(&rspObj.dataRsp);

×

2261

  tDecoderClear(&decoder);

×

2262

  qDestroyQuery(pQuery);

×

2263

  destroyRequest(pRequest);

×

2264

  return code;

×

2267

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

×

2268

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

×

2269

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

×

2270

    return TSDB_CODE_INVALID_PARA;

×

2272

  int32_t   code = TSDB_CODE_SUCCESS;

×

2273

  SQuery*   pQuery = NULL;

×

2274

  SMqRspObj rspObj = {0};

×

2275

  SDecoder  decoder = {0};

×

2276

  SHashObj* pCreateTbHash = NULL;

×

2278

  SRequestObj*     pRequest = NULL;

×

2279

  SCatalog*        pCatalog = NULL;

×

2280

  SRequestConnInfo conn = {0};

×

2282

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

×

2283

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

×

2284

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

×

2286

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

×

2287

  RAW_NULL_CHECK(pCreateTbHash);

×

2288

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

×

2290

  SHashObj* pVgHash = NULL;

×

2291

  SHashObj* pNameHash = NULL;

×

2292

  SHashObj* pMetaHash = NULL;

×

2293

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

×

2294

  int retry = 0;

×

2296

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2297

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

×

2298

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

×

2299

      if (!rspObj.dataRsp.withSchema) {

×

2300

        goto end;

×

2303

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

×

2304

      RAW_NULL_CHECK(tbName);

×

2305

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

×

2306

      RAW_NULL_CHECK(pSW);

×

2307

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

×

2308

      RAW_NULL_CHECK(pRetrieve);

×

2309

      void* rawData = getRawDataFromRes(pRetrieve);

×

2310

      RAW_NULL_CHECK(rawData);

×

2312

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

×

2313

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

×

2314

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

×

2315

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2318

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

×

2319

      STableMeta*    pTableMeta = NULL;

×

2320

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

×

2322

      char err[ERR_MSG_LEN] = {0};

×

2324

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

×

2325

      if (code != TSDB_CODE_SUCCESS) {

×

2326

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

×

2327

        goto end;

×

2330

    RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

2331

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2332

    code = pRequest->code;

×

2334

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

×

2335

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

×

2336

      qDestroyQuery(pQuery);

×

2337

      pQuery = NULL;

×

2338

      rspObj.resIter = -1;

×

2339

      continue;

×

2341

    break;

×

2344

end:

×

2345

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

×

2346

  tDeleteSTaosxRsp(&rspObj.dataRsp);

×

2347

  void* pIter = taosHashIterate(pCreateTbHash, NULL);

×

2348

  while (pIter) {

×

2349

    tDestroySVCreateTbReq(pIter, TSDB_MSG_FLG_DECODE);

×

2350

    pIter = taosHashIterate(pCreateTbHash, pIter);

×

2352

  taosHashCleanup(pCreateTbHash);

×

2353

  tDecoderClear(&decoder);

×

2354

  qDestroyQuery(pQuery);

×

2355

  destroyRequest(pRequest);

×

2356

  return code;

×

2359

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

×

2360

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

×

2361

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

×

2362

    return TSDB_CODE_INVALID_PARA;

×

2364

  int32_t   code = TSDB_CODE_SUCCESS;

×

2365

  SQuery*   pQuery = NULL;

×

2366

  SHashObj* pVgroupHash = NULL;

×

2367

  SMqRspObj rspObj = {0};

×

2368

  SDecoder  decoder = {0};

×

2370

  SRequestObj*     pRequest = NULL;

×

2371

  SCatalog*        pCatalog = NULL;

×

2372

  SRequestConnInfo conn = {0};

×

2374

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

×

2375

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

×

2376

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

×

2378

  SHashObj* pVgHash = NULL;

×

2379

  SHashObj* pNameHash = NULL;

×

2380

  SHashObj* pMetaHash = NULL;

×

2381

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

×

2382

  int retry = 0;

×

2383

  while (1) {

×

2384

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2385

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

×

2386

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

×

2387

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

×

2388

    RAW_NULL_CHECK(pVgroupHash);

×

2389

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

×

2390

    RAW_NULL_CHECK(pStmt->pVgDataBlocks);

×

2392

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

×

2393

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

×

2394

      RAW_NULL_CHECK(tbName);

×

2395

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

×

2396

      RAW_NULL_CHECK(pRetrieve);

×

2397

      void* rawData = getRawDataFromRes(pRetrieve);

×

2398

      RAW_NULL_CHECK(rawData);

×

2400

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

×

2401

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

×

2402

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

×

2403

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2406

      STableMeta* pTableMeta = NULL;

×

2407

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

×

2409

      char err[ERR_MSG_LEN] = {0};

×

2410

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

×

2411

      if (code != TSDB_CODE_SUCCESS) {

×

2412

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

×

2413

        goto end;

×

2416

    taosHashCleanup(pVgroupHash);

×

2417

    pVgroupHash = NULL;

×

2419

    RAW_RETURN_CHECK(smlBuildOutputRaw(pQuery, pVgHash));

×

2420

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2421

    code = pRequest->code;

×

2423

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

×

2424

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

×

2425

      qDestroyQuery(pQuery);

×

2426

      pQuery = NULL;

×

2427

      rspObj.resIter = -1;

×

2428

      continue;

×

2430

    break;

×

2433

end:

×

2434

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

×

2435

  tDeleteMqDataRsp(&rspObj.dataRsp);

×

2436

  tDecoderClear(&decoder);

×

2437

  qDestroyQuery(pQuery);

×

2438

  taosHashCleanup(pVgroupHash);

×

2439

  destroyRequest(pRequest);

×

2440

  return code;

×

2443

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

×

2444

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

×

2445

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

×

2446

    return;

×

2448

  if (pMetaRsp->resMsgType == TDMT_VND_CREATE_STB) {

×

2449

    processCreateStb(pMetaRsp, meta);

×

2450

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

×

2451

    processAlterStb(pMetaRsp, meta);

×

2452

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

×

2453

    processDropSTable(pMetaRsp, meta);

×

2454

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

×

2455

    processCreateTable(pMetaRsp, meta);

×

2456

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

×

2457

    processAlterTable(pMetaRsp, meta);

×

2458

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

×

2459

    processDropTable(pMetaRsp, meta);

×

2460

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

×

2461

    processDeleteTable(pMetaRsp, meta);

×

2465

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

×

2466

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

×

2467

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

×

2468

    return;

×

2470

  SDecoder        coder = {0};

×

2471

  SMqBatchMetaRsp rsp = {0};

×

2472

  int32_t         code = 0;

×

2473

  cJSON*          pJson = NULL;

×

2474

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

×

2475

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

×

2476

    goto end;

×

2479

  pJson = cJSON_CreateObject();

×

2480

  RAW_NULL_CHECK(pJson);

×

2481

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

×

2482

  cJSON* pMetaArr = cJSON_CreateArray();

×

2483

  RAW_NULL_CHECK(pMetaArr);

×

2484

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

×

2486

  int32_t num = taosArrayGetSize(rsp.batchMetaReq);

×

2487

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

×

2488

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

×

2489

    RAW_NULL_CHECK(len);

×

2490

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

×

2491

    RAW_NULL_CHECK(tmpBuf);

×

2492

    SDecoder   metaCoder = {0};

×

2493

    SMqMetaRsp metaRsp = {0};

×

2494

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

×

2495

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

×

2496

      goto end;

×

2498

    cJSON* pItem = NULL;

×

2499

    processSimpleMeta(&metaRsp, &pItem);

×

2500

    tDeleteMqMetaRsp(&metaRsp);

×

2501

    if (pItem != NULL) RAW_FALSE_CHECK(tmqAddJsonArrayItem(pMetaArr, pItem));

×

2504

  tDeleteMqBatchMetaRsp(&rsp);

×

2505

  char* fullStr = cJSON_PrintUnformatted(pJson);

×

2506

  cJSON_Delete(pJson);

×

2507

  *string = fullStr;

×

2508

  return;

×

2510

end:

×

2511

  cJSON_Delete(pJson);

×

2512

  tDeleteMqBatchMetaRsp(&rsp);

×

2515

char* tmq_get_json_meta(TAOS_RES* res) {

×

2516

  if (res == NULL) {

×

2517

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

×

2518

    return NULL;

×

2520

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

×

2521

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

×

2522

    return NULL;

×

2525

  char*      string = NULL;

×

2526

  SMqRspObj* rspObj = (SMqRspObj*)res;

×

2527

  if (TD_RES_TMQ_METADATA(res)) {

×

2528

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

×

2529

  } else if (TD_RES_TMQ_BATCH_META(res)) {

×

2530

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

×

2531

  } else if (TD_RES_TMQ_META(res)) {

×

2532

    cJSON* pJson = NULL;

×

2533

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

×

2534

    string = cJSON_PrintUnformatted(pJson);

×

2535

    cJSON_Delete(pJson);

×

2537

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

×

2540

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

×

2541

  return string;

×

2544

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

×

2548

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

×

2549

    return TSDB_CODE_INVALID_PARA;

×

2563

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

×

2564

    return TSDB_CODE_INVALID_PARA;

×

2572

    code = TSDB_CODE_INVALID_MSG;

×

2573

    goto FAILED;

×

2578

    code = terrno;

×

2579

    goto FAILED;

×

2583

    code = TSDB_CODE_INVALID_MSG;

×

2584

    goto FAILED;

×

2588

    code = TSDB_CODE_INVALID_MSG;

×

2589

    goto FAILED;

×

2592

    code = TSDB_CODE_INVALID_MSG;

×

2593

    goto FAILED;

×

2596

    code = TSDB_CODE_INVALID_MSG;

×

2597

    goto FAILED;

×

2604

FAILED:

×

2605

  tEncoderClear(&encoder);

×

2606

  taosMemoryFree(buf);

×

2607

  return code;

×

2612

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

×

2613

    return TSDB_CODE_INVALID_PARA;

×

2618

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

×

2619

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

×

2620

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

×

2621

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

×

2625

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

×

2626

      return code;

×

2630

  } else if (TD_RES_TMQ_METADATA(res)) {

×

2631

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

×

2632

    if (code != 0) {

×

2633

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

×

2634

      return code;

×

2636

    raw->raw_type = RES_TYPE__TMQ_METADATA;

×

2637

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

×

2638

  } else if (TD_RES_TMQ_BATCH_META(res)) {

×

2639

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

×

2640

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

×

2641

    raw->raw_type = rspObj->resType;

×

2642

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

×

2643

  } else if (TD_RES_TMQ_RAW(res)) {

×

2644

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

×

2645

    rspObj->dataRsp.rawData = NULL;

×

2646

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

×

2647

    raw->raw_type = rspObj->resType;

×

2648

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

×

2650

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

×

2651

    return TSDB_CODE_TMQ_INVALID_MSG;

×

2656

void tmq_free_raw(tmq_raw_data raw) {

×

2657

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

×

2658

  if (raw.raw_type == RES_TYPE__TMQ || raw.raw_type == RES_TYPE__TMQ_METADATA) {

×

2659

    taosMemoryFree(raw.raw);

×

2660

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

×

2661

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

×

2663

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

×

2664

×

2666

static int32_t writeRawInit() {

×

2667

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

×

2668

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

×

2669

    if (old == 0) {

×

2670

      int32_t code = initRawCacheHash();

×

2671

      if (code != 0) {

×

2672

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

×

2673

        atomic_store_8(&initedFlag, WRITE_RAW_INIT_FAIL);

×

2674

        return code;

×

2676

      atomic_store_8(&initedFlag, WRITE_RAW_INIT_OK);

×

2680

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

×

2681

    return TSDB_CODE_INTERNAL_ERROR;

×

2683

  return 0;

×

2686

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

×

2687

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

×

2688

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

×

2689

    return TSDB_CODE_INVALID_PARA;

×

2691

  if (writeRawInit() != 0) {

×

2692

    return TSDB_CODE_INTERNAL_ERROR;

×

2695

  if (type == TDMT_VND_CREATE_STB) {

×

2696

    return taosCreateStb(taos, buf, len);

×

2697

  } else if (type == TDMT_VND_ALTER_STB) {

×

2698

    return taosCreateStb(taos, buf, len);

×

2699

  } else if (type == TDMT_VND_DROP_STB) {

×

2700

    return taosDropStb(taos, buf, len);

×

2701

  } else if (type == TDMT_VND_CREATE_TABLE) {

×

2702

    return taosCreateTable(taos, buf, len);

×

2703

  } else if (type == TDMT_VND_ALTER_TABLE) {

×

2704

    return taosAlterTable(taos, buf, len);

×

2705

  } else if (type == TDMT_VND_DROP_TABLE) {

×

2706

    return taosDropTable(taos, buf, len);

×

2707

  } else if (type == TDMT_VND_DELETE) {

×

2708

    return taosDeleteData(taos, buf, len);

×

2709

  } else if (type == RES_TYPE__TMQ_METADATA) {

×

2710

    return tmqWriteRawMetaDataImpl(taos, buf, len);

×

2711

  } else if (type == RES_TYPE__TMQ_RAWDATA) {

×

2712

    return tmqWriteRawRawDataImpl(taos, buf, len);

×

2713

  } else if (type == RES_TYPE__TMQ) {

×

2714

    return tmqWriteRawDataImpl(taos, buf, len);

×

2715

  } else if (type == RES_TYPE__TMQ_BATCH_META) {

×

2716

    return tmqWriteBatchMetaDataImpl(taos, buf, len);

×

2718

  return TSDB_CODE_INVALID_PARA;

×

2721

int32_t tmq_write_raw(TAOS* taos, tmq_raw_data raw) {

×

2722

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

×

2723

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

×

2724

    return TSDB_CODE_INVALID_PARA;

×

2726

  taosClearErrMsg();  // clear global error message

×

2728

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

×

2731

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

×

2732

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

×

2733

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

×

2734

    return TSDB_CODE_INVALID_PARA;

×

2736

  SMqBatchMetaRsp rsp = {0};

×

2737

  SDecoder        coder = {0};

×

2738

  int32_t         code = TSDB_CODE_SUCCESS;

×

2741

  tDecoderInit(&coder, meta, metaLen);

×

2742

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

×

2743

    code = TSDB_CODE_INVALID_PARA;

×

2744

    goto end;

×

2746

  int32_t num = taosArrayGetSize(rsp.batchMetaReq);

×

2747

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

×

2748

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

×

2749

    RAW_NULL_CHECK(len);

×

2750

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

×

2751

    RAW_NULL_CHECK(tmpBuf);

×

2752

    SDecoder   metaCoder = {0};

×

2753

    SMqMetaRsp metaRsp = {0};

×

2754

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

×

2755

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

×

2756

      code = TSDB_CODE_INVALID_PARA;

×

2757

      goto end;

×

2759

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

×

2760

    tDeleteMqMetaRsp(&metaRsp);

×

2761

    if (code != TSDB_CODE_SUCCESS) {

×

2762

      goto end;

×

2766

end:

×

2767

  tDeleteMqBatchMetaRsp(&rsp);

×

2768

  return code;

×

taosdata / TDengine / #4884

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

taosdata / TDengine / #4884

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