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

72

static cJSON* tmqAddObjectToArray(cJSON* array) {

×

73

  cJSON* item = cJSON_CreateObject();

×

74

  if (cJSON_AddItemToArray(array, item)) {

×

75

    return item;

×

77

  cJSON_Delete(item);

×

78

  return NULL;

×

81

static cJSON* tmqAddStringToArray(cJSON* array, const char* str) {

×

82

  cJSON* item = cJSON_CreateString(str);

×

83

  if (cJSON_AddItemToArray(array, item)) {

×

84

    return item;

×

86

  cJSON_Delete(item);

×

87

  return NULL;

×

100

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

×

101

  if (db == NULL) {

×

102

    return suid;

×

104

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

×

107

static int32_t getLength(int8_t type, int32_t bytes, int32_t typeMod) {

×

108

  int32_t length = 0;

×

109

  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_VARBINARY || type == TSDB_DATA_TYPE_GEOMETRY) {

×

110

    length = bytes - VARSTR_HEADER_SIZE;

×

111

  } else if (type == TSDB_DATA_TYPE_NCHAR) {

×

112

    length = (bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;

×

113

  } else if (IS_STR_DATA_BLOB(type)) {

×

114

    length = bytes - BLOBSTR_HEADER_SIZE;

×

115

  } else if (type == TSDB_DATA_TYPE_DECIMAL || type == TSDB_DATA_TYPE_DECIMAL64) {

×

116

    length = typeMod;

×

118

  return length;

×

121

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

×

123

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

×

124

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

×

125

    return TSDB_CODE_INVALID_PARA;

×

127

  int32_t code = TSDB_CODE_SUCCESS;

×

128

  int32_t lino = 0;

×

129

  int8_t  buildDefaultCompress = 0;

×

130

  if (pColCmprRow->nCols <= 0) {

×

131

    buildDefaultCompress = 1;

×

133

  RAW_LOG_START

×

134

  char*  string = NULL;

×

135

  cJSON* json = cJSON_CreateObject();

×

136

  RAW_NULL_CHECK(json);

×

138

  ADD_TO_JSON_STRING(json, "type", "create");

×

139

  ADD_TO_JSON_STRING(json, "tableType", (t == TSDB_SUPER_TABLE ? "super" : "normal"));

×

140

  if (isVirtual){

×

141

    ADD_TO_JSON_BOOL(json, "isVirtual", isVirtual);

×

143

  ADD_TO_JSON_STRING(json, "tableName", name);

×

145

  cJSON* columns = cJSON_AddArrayToObject(json, "columns");

×

146

  RAW_NULL_CHECK(columns);

×

148

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

×

149

    cJSON* column = tmqAddObjectToArray(columns);

×

150

    RAW_NULL_CHECK(column);

×

151

    SSchema* s = schemaRow->pSchema + i;

×

152

    ADD_TO_JSON_STRING(column, "name", s->name);

×

153

    ADD_TO_JSON_NUMBER(column, "type", s->type);

×

154

    int32_t typeMod = 0;

×

155

    if (pExtSchemas != NULL) {

×

156

      typeMod = pExtSchemas[i].typeMod;

×

158

    int32_t length = getLength(s->type, s->bytes, typeMod);

×

159

    if (length > 0) {

×

160

      ADD_TO_JSON_NUMBER(column, "length", length);

×

163

    if (isVirtual && colRef != NULL && i < colRef->nCols){

×

164

      SColRef* pColRef = colRef->pColRef + i;

×

165

      if (pColRef->hasRef) {

×

166

        cJSON* ref = cJSON_AddObjectToObject(column, "ref");

×

167

        RAW_NULL_CHECK(ref);

×

168

        ADD_TO_JSON_STRING(ref, "refDbName", pColRef->refDbName);

×

169

        ADD_TO_JSON_STRING(ref, "refTableName", pColRef->refTableName);

×

170

        ADD_TO_JSON_STRING(ref, "refColName", pColRef->refColName);

×

173

    ADD_TO_JSON_BOOL(column, "isPrimarykey", (s->flags & COL_IS_KEY));

×

174

    if (pColCmprRow == NULL) {

×

175

      continue;

×

178

    uint32_t alg = 0;

×

179

    if (buildDefaultCompress) {

×

180

      alg = createDefaultColCmprByType(s->type);

×

182

      SColCmpr* pColCmpr = pColCmprRow->pColCmpr + i;

×

183

      alg = pColCmpr->alg;

×

185

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

×

186

    RAW_NULL_CHECK(encode);

×

187

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

×

188

    RAW_NULL_CHECK(compress);

×

189

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

×

190

    RAW_NULL_CHECK(level);

×

192

    ADD_TO_JSON_STRING(column, "encode", encode);

×

193

    ADD_TO_JSON_STRING(column, "compress", compress);

×

194

    ADD_TO_JSON_STRING(column, "level", level);

×

197

  cJSON* tags = cJSON_AddArrayToObject(json, "tags");

×

198

  RAW_NULL_CHECK(tags);

×

200

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

×

201

    cJSON* tag = tmqAddObjectToArray(tags);

×

202

    RAW_NULL_CHECK(tag);

×

203

    SSchema* s = schemaTag->pSchema + i;

×

204

    ADD_TO_JSON_STRING(tag, "name", s->name);

×

205

    ADD_TO_JSON_NUMBER(tag, "type", s->type);

×

206

    int32_t length = getLength(s->type, s->bytes, 0);

×

207

    if (length > 0) {

×

208

      ADD_TO_JSON_NUMBER(tag, "length", length);

×

212

end:

×

213

  *pJson = json;

×

214

  RAW_LOG_END

×

215

  return code;

×

218

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

×

219

  if (json == NULL) {

×

220

    return TSDB_CODE_INVALID_PARA;

×

222

  uint8_t encode = COMPRESS_L1_TYPE_U32(para);

×

223

  int32_t code = 0;

×

224

  int32_t lino = 0;

×

225

  RAW_LOG_START

×

226

  if (encode != 0) {

×

227

    const char* encodeStr = columnEncodeStr(encode);

×

228

    RAW_NULL_CHECK(encodeStr);

×

229

    ADD_TO_JSON_STRING(json, "encode", encodeStr);

×

230

    goto end;

×

232

  uint8_t compress = COMPRESS_L2_TYPE_U32(para);

×

233

  if (compress != 0) {

×

234

    const char* compressStr = columnCompressStr(compress);

×

235

    RAW_NULL_CHECK(compressStr);

×

236

    ADD_TO_JSON_STRING(json, "compress", compressStr);

×

237

    goto end;

×

239

  uint8_t level = COMPRESS_L2_TYPE_LEVEL_U32(para);

×

240

  if (level != 0) {

×

241

    const char* levelStr = columnLevelStr(level);

×

242

    RAW_NULL_CHECK(levelStr);

×

243

    ADD_TO_JSON_STRING(json, "level", levelStr);

×

244

    goto end;

×

247

end:

×

248

  RAW_LOG_END

×

249

  return code;

×

251

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

×

252

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

×

253

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

×

254

    return TSDB_CODE_INVALID_PARA;

×

256

  SMAlterStbReq req = {0};

×

257

  cJSON*        json = NULL;

×

258

  char*         string = NULL;

×

259

  int32_t       code = 0;

×

260

  int32_t       lino = 0;

×

261

  RAW_LOG_START

×

262

  RAW_RETURN_CHECK(tDeserializeSMAlterStbReq(alterData, alterDataLen, &req));

×

263

  json = cJSON_CreateObject();

×

264

  RAW_NULL_CHECK(json);

×

265

  ADD_TO_JSON_STRING(json, "type", "alter");

×

266

  SName name = {0};

×

267

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

×

268

  ADD_TO_JSON_STRING(json, "tableType", "super");

×

269

  ADD_TO_JSON_STRING(json, "tableName", name.tname);

×

270

  ADD_TO_JSON_NUMBER(json, "alterType", req.alterType);

×

272

  switch (req.alterType) {

×

273

    case TSDB_ALTER_TABLE_ADD_TAG:

×

275

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

×

276

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

×

277

        cJSON_Delete(json);

×

278

        json = NULL;

×

279

        code = TSDB_CODE_INVALID_PARA;

×

280

        goto end;

×

282

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

×

283

      RAW_NULL_CHECK(field);

×

284

      ADD_TO_JSON_STRING(json, "colName", field->name);

×

285

      ADD_TO_JSON_NUMBER(json, "colType", field->type);

×

286

      int32_t typeMode = 0;

×

287

      if (taosArrayGetSize(req.pTypeMods) > 0) {

×

288

        typeMode = *(STypeMod*)taosArrayGet(req.pTypeMods, 0);

×

290

      int32_t length = getLength(field->type, field->bytes, typeMode);

×

291

      if (length > 0) {

×

292

        ADD_TO_JSON_NUMBER(json, "colLength", length);

×

295

      break;

×

297

    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION: {

×

298

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

×

299

      RAW_NULL_CHECK(field);

×

300

      ADD_TO_JSON_STRING(json, "colName", field->name);

×

301

      ADD_TO_JSON_NUMBER(json, "colType", field->type);

×

302

      int32_t typeMode = 0;

×

303

      if (taosArrayGetSize(req.pTypeMods) > 0) {

×

304

        typeMode = *(STypeMod*)taosArrayGet(req.pTypeMods, 0);

×

306

      int32_t length = getLength(field->type, field->bytes, typeMode);

×

307

      if (length > 0) {

×

308

        ADD_TO_JSON_NUMBER(json, "colLength", length);

×

311

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

×

312

      break;

×

314

    case TSDB_ALTER_TABLE_DROP_TAG:

×

316

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

×

317

      RAW_NULL_CHECK(field);

×

318

      ADD_TO_JSON_STRING(json, "colName", field->name);

×

319

      break;

×

321

    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:

×

323

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

×

324

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

×

325

        cJSON_Delete(json);

×

326

        json = NULL;

×

327

        code = TSDB_CODE_INVALID_PARA;

×

328

        goto end;

×

330

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

×

331

      RAW_NULL_CHECK(field);

×

332

      ADD_TO_JSON_STRING(json, "colName", field->name);

×

333

      ADD_TO_JSON_NUMBER(json, "colType", field->type);

×

334

      int32_t length = getLength(field->type, field->bytes, 0);

×

335

      if (length > 0) {

×

336

        ADD_TO_JSON_NUMBER(json, "colLength", length);

×

339

      break;

×

341

    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:

×

343

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

×

344

      RAW_NULL_CHECK(oldField);

×

345

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

×

346

      RAW_NULL_CHECK(newField);

×

347

      ADD_TO_JSON_STRING(json, "colName", oldField->name);

×

348

      ADD_TO_JSON_STRING(json, "colNewName", newField->name);

×

349

      break;

×

351

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {

×

352

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

×

353

      RAW_NULL_CHECK(field);

×

354

      ADD_TO_JSON_STRING(json, "colName", field->name);

×

355

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

×

356

      break;

×

358

    default:

×

359

      break;

×

362

end:

×

363

  tFreeSMAltertbReq(&req);

×

364

  *pJson = json;

×

365

  RAW_LOG_END

×

366

  return code;

×

369

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

×

370

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

×

371

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

×

372

    return TSDB_CODE_INVALID_PARA;

×

374

  int32_t        code = TSDB_CODE_SUCCESS;

×

375

  int32_t        lino = 0;

×

376

  SVCreateStbReq req = {0};

×

377

  SDecoder       coder = {0};

×

379

  RAW_LOG_START

×

381

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

×

382

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

×

383

  tDecoderInit(&coder, data, len);

×

385

  RAW_RETURN_CHECK(tDecodeSVCreateStbReq(&coder, &req));

×

386

  RAW_RETURN_CHECK(buildCreateTableJson(&req.schemaRow, &req.schemaTag, req.pExtSchemas, req.name, req.suid,

×

389

end:

×

390

  tDecoderClear(&coder);

×

391

  RAW_LOG_END

×

392

  return code;

×

395

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

×

396

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

×

397

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

×

398

    return TSDB_CODE_INVALID_PARA;

×

400

  SVCreateStbReq req = {0};

×

401

  SDecoder       coder = {0};

×

402

  int32_t        code = TSDB_CODE_SUCCESS;

×

403

  int32_t        lino = 0;

×

404

  RAW_LOG_START

×

407

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

×

408

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

×

409

  tDecoderInit(&coder, data, len);

×

411

  RAW_RETURN_CHECK(tDecodeSVCreateStbReq(&coder, &req));

×

412

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

×

414

end:

×

415

  tDecoderClear(&coder);

×

416

  RAW_LOG_END

×

417

  return code;

×

420

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

×

421

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

×

422

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

×

423

    return TSDB_CODE_INVALID_PARA;

×

425

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

×

426

  char*   sname = pCreateReq->ctb.stbName;

×

427

  char*   name = pCreateReq->name;

×

428

  SArray* tagName = pCreateReq->ctb.tagName;

×

429

  int64_t id = pCreateReq->uid;

×

430

  uint8_t tagNum = pCreateReq->ctb.tagNum;

×

431

  int32_t code = 0;

×

432

  int32_t lino = 0;

×

433

  SArray* pTagVals = NULL;

×

434

  char*   pJson = NULL;

×

435

  char*   buf = NULL;

×

436

  RAW_LOG_START

×

438

  ADD_TO_JSON_STRING(json, "tableName", name);

×

440

  if (pCreateReq->type == TSDB_VIRTUAL_CHILD_TABLE) {

×

441

    cJSON* refs = cJSON_AddArrayToObject(json, "refs");

×

442

    RAW_NULL_CHECK(refs);

×

444

    for (int i = 0; i < pCreateReq->colRef.nCols; i++) {

×

445

      SColRef* pColRef = pCreateReq->colRef.pColRef + i;

×

447

      if (!pColRef->hasRef) {

×

448

        continue;

×

450

      cJSON* ref = tmqAddObjectToArray(refs);

×

451

      RAW_NULL_CHECK(ref);

×

452

      ADD_TO_JSON_STRING(ref, "colName", pColRef->colName);

×

453

      ADD_TO_JSON_STRING(ref, "refDbName", pColRef->refDbName);

×

454

      ADD_TO_JSON_STRING(ref, "refTableName", pColRef->refTableName);

×

455

      ADD_TO_JSON_STRING(ref, "refColName", pColRef->refColName);

×

459

  ADD_TO_JSON_STRING(json, "using", sname);

×

460

  ADD_TO_JSON_NUMBER(json, "tagNum", tagNum);

×

462

  cJSON* tags = cJSON_AddArrayToObject(json, "tags");

×

463

  RAW_NULL_CHECK(tags);

×

464

  RAW_RETURN_CHECK(tTagToValArray(pTag, &pTagVals));

×

465

  if (tTagIsJson(pTag)) {

×

466

    STag* p = (STag*)pTag;

×

467

    if (p->nTag == 0) {

×

468

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

×

469

      goto end;

×

471

    parseTagDatatoJson(pTag, &pJson, NULL);

×

472

    RAW_NULL_CHECK(pJson);

×

473

    cJSON* tag = tmqAddObjectToArray(tags);

×

474

    RAW_NULL_CHECK(tag);

×

475

    STagVal* pTagVal = taosArrayGet(pTagVals, 0);

×

476

    RAW_NULL_CHECK(pTagVal);

×

477

    char* ptname = taosArrayGet(tagName, 0);

×

478

    RAW_NULL_CHECK(ptname);

×

479

    ADD_TO_JSON_STRING(tag, "name", ptname);

×

480

    ADD_TO_JSON_NUMBER(tag, "type", TSDB_DATA_TYPE_JSON);

×

481

    ADD_TO_JSON_STRING(tag, "value", pJson);

×

482

    goto end;

×

485

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

×

486

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

×

487

    RAW_NULL_CHECK(pTagVal);

×

488

    cJSON* tag = tmqAddObjectToArray(tags);

×

489

    RAW_NULL_CHECK(tag);

×

490

    char* ptname = taosArrayGet(tagName, i);

×

491

    RAW_NULL_CHECK(ptname);

×

492

    ADD_TO_JSON_STRING(tag, "name", ptname);

×

493

    ADD_TO_JSON_NUMBER(tag, "type", pTagVal->type);

×

495

    if (IS_VAR_DATA_TYPE(pTagVal->type)) {

×

496

      if (IS_STR_DATA_BLOB(pTagVal->type)) {

×

497

        goto end;

×

499

      int64_t bufSize = 0;

×

500

      if (pTagVal->type == TSDB_DATA_TYPE_VARBINARY) {

×

501

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

×

503

        bufSize = pTagVal->nData + 3;

×

505

      buf = taosMemoryCalloc(bufSize, 1);

×

506

      RAW_NULL_CHECK(buf);

×

507

      code = dataConverToStr(buf, bufSize, pTagVal->type, pTagVal->pData, pTagVal->nData, NULL);

×

508

      if (code != TSDB_CODE_SUCCESS) {

×

509

        uError("convert tag value to string failed");

×

510

        goto end;

×

513

      ADD_TO_JSON_STRING(tag, "value", buf)

×

514

      taosMemoryFreeClear(buf);

×

516

      double val = 0;

×

517

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

×

518

      ADD_TO_JSON_NUMBER(tag, "value", val)

×

522

end:

×

523

  taosMemoryFree(pJson);

×

524

  taosArrayDestroy(pTagVals);

×

525

  taosMemoryFree(buf);

×

526

  RAW_LOG_END

×

527

  return code;

×

530

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

×

531

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

×

532

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

×

533

    return TSDB_CODE_INVALID_PARA;

×

535

  int32_t code = 0;

×

536

  int32_t lino = 0;

×

537

  RAW_LOG_START

×

538

  char*  string = NULL;

×

539

  cJSON* json = cJSON_CreateObject();

×

540

  RAW_NULL_CHECK(json);

×

541

  ADD_TO_JSON_STRING(json, "type", "create");

×

542

  ADD_TO_JSON_STRING(json, "tableType", "child");

×

544

  RAW_RETURN_CHECK(buildChildElement(json, pCreateReq));

×

545

  cJSON* createList = cJSON_AddArrayToObject(json, "createList");

×

546

  RAW_NULL_CHECK(createList);

×

548

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

×

549

    cJSON* create = tmqAddObjectToArray(createList);

×

550

    RAW_NULL_CHECK(create);

×

551

    RAW_RETURN_CHECK(buildChildElement(create, pCreateReq + i));

×

554

end:

×

555

  *pJson = json;

×

556

  RAW_LOG_END

×

557

  return code;

×

560

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

×

561

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

×

562

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

×

563

    return TSDB_CODE_INVALID_PARA;

×

565

  int32_t            code = TSDB_CODE_SUCCESS;

×

566

  int32_t            lino = 0;

×

567

  SDecoder           decoder = {0};

×

568

  SVCreateTbBatchReq req = {0};

×

570

  RAW_LOG_START

×

572

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

×

573

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

×

574

  tDecoderInit(&decoder, data, len);

×

575

  RAW_RETURN_CHECK(tDecodeSVCreateTbBatchReq(&decoder, &req));

×

577

  if (req.nReqs > 0) {

×

578

    pCreateReq = req.pReqs;

×

579

    if (pCreateReq->type == TSDB_CHILD_TABLE || pCreateReq->type == TSDB_VIRTUAL_CHILD_TABLE) {

×

580

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

×

581

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

×

582

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

×

588

end:

×

589

  tDeleteSVCreateTbBatchReq(&req);

×

590

  tDecoderClear(&decoder);

×

591

  RAW_LOG_END

×

592

  return code;

×

595

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

×

596

  int32_t lino = 0;

×

597

  int32_t code = TSDB_CODE_SUCCESS;

×

598

  RAW_LOG_START

×

599

  RAW_FALSE_CHECK(rsp != NULL && string != NULL);

×

600

  SDecoder*      decoder = NULL;

×

601

  SVCreateTbReq* pCreateReq = NULL;

×

602

  RAW_FALSE_CHECK(rsp->createTableNum > 0);

×

604

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

×

605

  RAW_NULL_CHECK(decoder);

×

606

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

×

607

  RAW_NULL_CHECK(pCreateReq);

×

610

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

×

612

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

×

613

    RAW_NULL_CHECK(data);

×

614

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

×

615

    RAW_NULL_CHECK(len);

×

616

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

×

617

    RAW_RETURN_CHECK(tDecodeSVCreateTbReq(&decoder[iReq], pCreateReq + iReq));

×

619

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

×

620

      uError("%s failed. pCreateReq[iReq].type:%d invalid", __func__, pCreateReq[iReq].type);

×

621

      code = TSDB_CODE_INVALID_PARA;

×

622

      goto end;

×

625

  cJSON* pJson = NULL;

×

626

  if (pCreateReq->type == TSDB_NORMAL_TABLE) {

×

627

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

×

629

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

×

630

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

×

633

  *string = cJSON_PrintUnformatted(pJson);

×

634

  cJSON_Delete(pJson);

×

636

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

×

638

end:

×

639

  RAW_LOG_END

×

640

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

×

641

    tDecoderClear(&decoder[i]);

×

642

    taosMemoryFreeClear(pCreateReq[i].comment);

×

643

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

×

644

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

×

647

  taosMemoryFree(decoder);

×

648

  taosMemoryFree(pCreateReq);

×

649

  return code;

×

652

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

×

653

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

×

654

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

×

655

    return TSDB_CODE_INVALID_PARA;

×

657

  SDecoder     decoder = {0};

×

658

  SVAlterTbReq vAlterTbReq = {0};

×

659

  char*        string = NULL;

×

660

  cJSON*       json = NULL;

×

661

  int32_t      code = 0;

×

662

  int32_t      lino = 0;

×

663

  char*        buf = NULL;

×

664

  char*        buf1 = NULL;

×

666

  RAW_LOG_START

×

669

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

×

670

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

×

671

  tDecoderInit(&decoder, data, len);

×

672

  RAW_RETURN_CHECK(tDecodeSVAlterTbReq(&decoder, &vAlterTbReq));

×

674

  json = cJSON_CreateObject();

×

675

  RAW_NULL_CHECK(json);

×

676

  ADD_TO_JSON_STRING(json, "type", "alter");

×

678

  char* tableType = NULL;

×

679

  if (vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_TAG_VAL ||

×

680

      vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL){

×

681

    tableType = "child";

×

682

  } else if (vAlterTbReq.action == TSDB_ALTER_TABLE_ALTER_COLUMN_REF ||

×

683

             vAlterTbReq.action == TSDB_ALTER_TABLE_REMOVE_COLUMN_REF) {

×

684

    tableType = "";

×

686

    tableType = "normal";

×

689

  ADD_TO_JSON_STRING(json, "tableType", tableType);

×

690

  ADD_TO_JSON_STRING(json, "tableName", vAlterTbReq.tbName);

×

691

  ADD_TO_JSON_NUMBER(json, "alterType", vAlterTbReq.action);

×

693

  uDebug("alter table action:%d", vAlterTbReq.action);

×

694

  switch (vAlterTbReq.action) {

×

695

    case TSDB_ALTER_TABLE_ADD_COLUMN:

×

697

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.colName);

×

698

      ADD_TO_JSON_NUMBER(json, "colType", vAlterTbReq.type);

×

700

      int32_t length = getLength(vAlterTbReq.type, vAlterTbReq.bytes, vAlterTbReq.typeMod);

×

701

      if (length > 0) {

×

702

        ADD_TO_JSON_NUMBER(json, "colLength", length);

×

705

      if (vAlterTbReq.action == TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COLUMN_REF) {

×

706

        ADD_TO_JSON_STRING(json, "refDbName", vAlterTbReq.refDbName);

×

707

        ADD_TO_JSON_STRING(json, "refTbName", vAlterTbReq.refTbName);

×

708

        ADD_TO_JSON_STRING(json, "refColName", vAlterTbReq.refColName);

×

710

      break;

×

712

    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION: {

×

713

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.colName);

×

714

      ADD_TO_JSON_NUMBER(json, "colType", vAlterTbReq.type);

×

716

      int32_t length = getLength(vAlterTbReq.type, vAlterTbReq.bytes, vAlterTbReq.typeMod);

×

717

      if (length > 0) {

×

718

        ADD_TO_JSON_NUMBER(json, "colLength", length);

×

720

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

×

721

      break;

×

723

    case TSDB_ALTER_TABLE_DROP_COLUMN: {

×

724

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.colName);

×

725

      break;

×

727

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: {

×

728

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.colName);

×

729

      ADD_TO_JSON_NUMBER(json, "colType", vAlterTbReq.colModType);

×

730

      int32_t length = getLength(vAlterTbReq.colModType, vAlterTbReq.colModBytes, vAlterTbReq.typeMod);

×

731

      if (length > 0) {

×

732

        ADD_TO_JSON_NUMBER(json, "colLength", length);

×

734

      break;

×

736

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: {

×

737

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.colName);

×

738

      ADD_TO_JSON_STRING(json, "colNewName", vAlterTbReq.colNewName);

×

739

      break;

×

741

    case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: {

×

742

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.tagName);

×

744

      bool isNull = vAlterTbReq.isNull;

×

745

      if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) {

×

746

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

×

747

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

×

749

      if (!isNull) {

×

750

        if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) {

×

751

          if (!tTagIsJson(vAlterTbReq.pTagVal)) {

×

752

            code = TSDB_CODE_INVALID_PARA;

×

753

            uError("processAlterTable isJson false");

×

754

            goto end;

×

756

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

×

757

          if (buf == NULL) {

×

758

            code = TSDB_CODE_INVALID_PARA;

×

759

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

×

760

            goto end;

×

763

          int64_t bufSize = 0;

×

764

          if (vAlterTbReq.tagType == TSDB_DATA_TYPE_VARBINARY) {

×

765

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

×

767

            bufSize = vAlterTbReq.nTagVal + 32;

×

769

          buf = taosMemoryCalloc(bufSize, 1);

×

770

          RAW_NULL_CHECK(buf);

×

771

          code = dataConverToStr(buf, bufSize, vAlterTbReq.tagType, vAlterTbReq.pTagVal, vAlterTbReq.nTagVal, NULL);

×

772

          if (code != TSDB_CODE_SUCCESS) {

×

773

            uError("convert tag value to string failed");

×

774

            goto end;

×

778

        ADD_TO_JSON_STRING(json, "colValue", buf);

×

781

      ADD_TO_JSON_BOOL(json, "colValueNull", isNull);

×

782

      break;

×

784

    case TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL: {

×

785

      int32_t nTags = taosArrayGetSize(vAlterTbReq.pMultiTag);

×

786

      if (nTags <= 0) {

×

787

        code = TSDB_CODE_INVALID_PARA;

×

788

        uError("processAlterTable parse multi tags error");

×

789

        goto end;

×

792

      cJSON* tags = cJSON_AddArrayToObject(json, "tags");

×

793

      RAW_NULL_CHECK(tags);

×

795

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

×

796

        cJSON* member = tmqAddObjectToArray(tags);

×

797

        RAW_NULL_CHECK(member);

×

799

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

×

800

        ADD_TO_JSON_STRING(member, "colName", pTagVal->tagName);

×

802

        if (pTagVal->tagType == TSDB_DATA_TYPE_JSON) {

×

803

          code = TSDB_CODE_INVALID_PARA;

×

804

          uError("processAlterTable isJson false");

×

805

          goto end;

×

807

        bool isNull = pTagVal->isNull;

×

808

        if (!isNull) {

×

809

          int64_t bufSize = 0;

×

810

          if (pTagVal->tagType == TSDB_DATA_TYPE_VARBINARY) {

×

811

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

×

813

            bufSize = pTagVal->nTagVal + 3;

×

815

          buf1 = taosMemoryCalloc(bufSize, 1);

×

816

          RAW_NULL_CHECK(buf1);

×

817

          code = dataConverToStr(buf1, bufSize, pTagVal->tagType, pTagVal->pTagVal, pTagVal->nTagVal, NULL);

×

818

          if (code != TSDB_CODE_SUCCESS) {

×

819

            uError("convert tag value to string failed");

×

820

            goto end;

×

822

          ADD_TO_JSON_STRING(member, "colValue", buf1)

×

823

          taosMemoryFreeClear(buf1);

×

825

        ADD_TO_JSON_BOOL(member, "colValueNull", isNull)

×

827

      break;

×

830

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {

×

831

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.colName);

×

832

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

×

833

      break;

×

835

    case TSDB_ALTER_TABLE_ALTER_COLUMN_REF: {

×

836

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.colName);

×

837

      ADD_TO_JSON_STRING(json, "refDbName", vAlterTbReq.refDbName);

×

838

      ADD_TO_JSON_STRING(json, "refTbName", vAlterTbReq.refTbName);

×

839

      ADD_TO_JSON_STRING(json, "refColName", vAlterTbReq.refColName);

×

840

      break;

×

842

    case TSDB_ALTER_TABLE_REMOVE_COLUMN_REF:{

×

843

      ADD_TO_JSON_STRING(json, "colName", vAlterTbReq.colName);

×

844

      break;

×

846

    default:

×

847

      break;

×

850

end:

×

851

  if (vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL) {

×

852

    taosArrayDestroy(vAlterTbReq.pMultiTag);

×

854

  tDecoderClear(&decoder);

×

855

  taosMemoryFree(buf);

×

856

  taosMemoryFree(buf1);

×

857

  *pJson = json;

×

858

  RAW_LOG_END

×

859

  return code;

×

862

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

×

863

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

×

864

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

×

865

    return TSDB_CODE_INVALID_PARA;

×

867

  SDecoder     decoder = {0};

×

868

  SVDropStbReq req = {0};

×

869

  cJSON*       json = NULL;

×

870

  int32_t      code = 0;

×

871

  int32_t      lino = 0;

×

872

  RAW_LOG_START

×

875

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

×

876

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

×

877

  tDecoderInit(&decoder, data, len);

×

878

  RAW_RETURN_CHECK(tDecodeSVDropStbReq(&decoder, &req));

×

880

  json = cJSON_CreateObject();

×

881

  RAW_NULL_CHECK(json);

×

882

  ADD_TO_JSON_STRING(json, "type", "drop");

×

883

  ADD_TO_JSON_STRING(json, "tableType", "super");

×

884

  ADD_TO_JSON_STRING(json, "tableName", req.name);

×

886

end:

×

887

  tDecoderClear(&decoder);

×

888

  *pJson = json;

×

889

  RAW_LOG_END

×

890

  return code;

×

892

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

×

893

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

×

894

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

×

895

    return TSDB_CODE_INVALID_PARA;

×

897

  SDeleteRes req = {0};

×

898

  SDecoder   coder = {0};

×

899

  cJSON*     json = NULL;

×

900

  int32_t    code = 0;

×

901

  int32_t    lino = 0;

×

902

  RAW_LOG_START

×

905

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

×

906

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

×

908

  tDecoderInit(&coder, data, len);

×

909

  RAW_RETURN_CHECK(tDecodeDeleteRes(&coder, &req));

×

911

  char sql[256] = {0};

×

912

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

×

915

  json = cJSON_CreateObject();

×

916

  RAW_NULL_CHECK(json);

×

917

  ADD_TO_JSON_STRING(json, "type", "delete");

×

918

  ADD_TO_JSON_STRING(json, "sql", sql);

×

920

end:

×

921

  tDecoderClear(&coder);

×

922

  *pJson = json;

×

923

  RAW_LOG_END

×

924

  return code;

×

927

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

×

928

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

×

929

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

×

930

    return TSDB_CODE_INVALID_PARA;

×

932

  SDecoder         decoder = {0};

×

933

  SVDropTbBatchReq req = {0};

×

934

  cJSON*           json = NULL;

×

935

  int32_t          code = 0;

×

936

  int32_t          lino = 0;

×

937

  RAW_LOG_START

×

939

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

×

940

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

×

941

  tDecoderInit(&decoder, data, len);

×

942

  RAW_RETURN_CHECK(tDecodeSVDropTbBatchReq(&decoder, &req));

×

944

  json = cJSON_CreateObject();

×

945

  RAW_NULL_CHECK(json);

×

946

  ADD_TO_JSON_STRING(json, "type", "drop");

×

948

  cJSON* tableNameList = cJSON_AddArrayToObject(json, "tableNameList");

×

949

  RAW_NULL_CHECK(tableNameList);

×

951

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

×

952

    SVDropTbReq* pDropTbReq = req.pReqs + iReq;

×

953

    RAW_NULL_CHECK(tmqAddStringToArray(tableNameList, pDropTbReq->name));

×

956

end:

×

957

  tDecoderClear(&decoder);

×

958

  *pJson = json;

×

959

  RAW_LOG_END

×

960

  return code;

×

963

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

×

964

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

×

965

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

×

966

    return TSDB_CODE_INVALID_PARA;

×

968

  SVCreateStbReq req = {0};

×

969

  SDecoder       coder = {0};

×

970

  SMCreateStbReq pReq = {0};

×

971

  int32_t        code = TSDB_CODE_SUCCESS;

×

972

  int32_t        lino = 0;

×

973

  SRequestObj*   pRequest = NULL;

×

974

  SCmdMsgInfo    pCmdMsg = {0};

×

975

  RAW_LOG_START

×

977

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

×

978

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

×

979

  pRequest->syncQuery = true;

×

980

  if (!pRequest->pDb) {

×

981

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

982

    uError(LOG_ID_TAG " %s no database selected", LOG_ID_VALUE, __func__);

×

983

    goto end;

×

986

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

×

987

  uint32_t len = metaLen - sizeof(SMsgHead);

×

988

  tDecoderInit(&coder, data, len);

×

989

  RAW_RETURN_CHECK(tDecodeSVCreateStbReq(&coder, &req));

×

991

  int8_t           createDefaultCompress = 0;

×

992

  SColCmprWrapper* p = &req.colCmpr;

×

993

  if (p->nCols == 0) {

×

994

    createDefaultCompress = 1;

×

997

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

×

998

  RAW_NULL_CHECK(pReq.pColumns);

×

999

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

×

1000

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

×

1001

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

×

1002

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

×

1004

    if (createDefaultCompress) {

×

1005

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

×

1007

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

×

1008

      field.compress = pCmp->alg;

×

1010

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

×

1011

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

×

1013

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

×

1014

  RAW_NULL_CHECK(pReq.pTags);

×

1015

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

×

1016

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

×

1017

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

×

1018

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

×

1019

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

×

1022

  pReq.colVer = req.schemaRow.version;

×

1023

  pReq.tagVer = req.schemaTag.version;

×

1024

  pReq.numOfColumns = req.schemaRow.nCols;

×

1025

  pReq.numOfTags = req.schemaTag.nCols;

×

1026

  pReq.commentLen = -1;

×

1027

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

×

1028

  pReq.source = TD_REQ_FROM_TAOX;

×

1029

  pReq.igExists = true;

×

1030

  pReq.virtualStb = req.virtualStb;

×

1032

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

×

1034

  STscObj* pTscObj = pRequest->pTscObj;

×

1035

  SName    tableName = {0};

×

1036

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

×

1037

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

×

1038

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

×

1039

  pCmdMsg.msgType = TDMT_MND_CREATE_STB;

×

1040

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

×

1041

  RAW_FALSE_CHECK(pCmdMsg.msgLen > 0);

×

1042

  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);

×

1043

  RAW_NULL_CHECK(pCmdMsg.pMsg);

×

1044

  RAW_FALSE_CHECK(tSerializeSMCreateStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq) > 0);

×

1046

  SQuery pQuery = {0};

×

1047

  pQuery.execMode = QUERY_EXEC_MODE_RPC;

×

1048

  pQuery.pCmdMsg = &pCmdMsg;

×

1049

  pQuery.msgType = pQuery.pCmdMsg->msgType;

×

1050

  pQuery.stableQuery = true;

×

1052

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

×

1054

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1055

    SCatalog* pCatalog = NULL;

×

1056

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

×

1057

    RAW_RETURN_CHECK(catalogRemoveTableMeta(pCatalog, &tableName));

×

1060

  code = pRequest->code;

×

1061

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

×

1063

end:

×

1064

  destroyRequest(pRequest);

×

1065

  tFreeSMCreateStbReq(&pReq);

×

1066

  tDecoderClear(&coder);

×

1067

  taosMemoryFree(pCmdMsg.pMsg);

×

1068

  RAW_LOG_END

×

1069

  return code;

×

1072

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

×

1073

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

×

1074

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

×

1075

    return TSDB_CODE_INVALID_PARA;

×

1077

  SVDropStbReq req = {0};

×

1078

  SDecoder     coder = {0};

×

1079

  SMDropStbReq pReq = {0};

×

1080

  int32_t      code = TSDB_CODE_SUCCESS;

×

1081

  int32_t      lino = 0;

×

1082

  SRequestObj* pRequest = NULL;

×

1083

  SCmdMsgInfo  pCmdMsg = {0};

×

1085

  RAW_LOG_START

×

1086

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

×

1087

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

×

1088

  pRequest->syncQuery = true;

×

1089

  if (!pRequest->pDb) {

×

1090

    uError(LOG_ID_TAG " %s no database selected", LOG_ID_VALUE, __func__);

×

1091

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1092

    goto end;

×

1095

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

×

1096

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1097

  tDecoderInit(&coder, data, len);

×

1098

  RAW_RETURN_CHECK(tDecodeSVDropStbReq(&coder, &req));

×

1099

  SCatalog* pCatalog = NULL;

×

1100

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

×

1101

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

×

1102

                           .requestId = pRequest->requestId,

×

1103

                           .requestObjRefId = pRequest->self,

×

1104

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

×

1105

  SName            pName = {0};

×

1106

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

×

1107

  STableMeta* pTableMeta = NULL;

×

1108

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

×

1109

  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1110

    uInfo(LOG_ID_TAG " stable %s not exist, ignore drop", LOG_ID_VALUE, req.name);

×

1111

    code = TSDB_CODE_SUCCESS;

×

1112

    taosMemoryFreeClear(pTableMeta);

×

1113

    goto end;

×

1115

  RAW_RETURN_CHECK(code);

×

1116

  pReq.suid = pTableMeta->uid;

×

1117

  taosMemoryFreeClear(pTableMeta);

×

1120

  pReq.igNotExists = true;

×

1121

  pReq.source = TD_REQ_FROM_TAOX;

×

1124

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

×

1126

  STscObj* pTscObj = pRequest->pTscObj;

×

1127

  SName    tableName = {0};

×

1128

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

×

1129

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

×

1131

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

×

1132

  pCmdMsg.msgType = TDMT_MND_DROP_STB;

×

1133

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

×

1134

  RAW_FALSE_CHECK(pCmdMsg.msgLen > 0);

×

1135

  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);

×

1136

  RAW_NULL_CHECK(pCmdMsg.pMsg);

×

1137

  RAW_FALSE_CHECK(tSerializeSMDropStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq) > 0);

×

1139

  SQuery pQuery = {0};

×

1140

  pQuery.execMode = QUERY_EXEC_MODE_RPC;

×

1141

  pQuery.pCmdMsg = &pCmdMsg;

×

1142

  pQuery.msgType = pQuery.pCmdMsg->msgType;

×

1143

  pQuery.stableQuery = true;

×

1145

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

×

1146

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1148

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

×

1149

    RAW_RETURN_CHECK(catalogRemoveTableMeta(pCatalog, &tableName));

×

1152

  code = pRequest->code;

×

1153

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

×

1155

end:

×

1156

  RAW_LOG_END

×

1157

  destroyRequest(pRequest);

×

1158

  tDecoderClear(&coder);

×

1159

  return code;

×

1168

static void destroyCreateTbReqBatch(void* data) {

×

1169

  if (data == NULL) {

×

1170

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

×

1171

    return;

×

1173

  SVgroupCreateTableBatch* pTbBatch = (SVgroupCreateTableBatch*)data;

×

1174

  taosArrayDestroy(pTbBatch->req.pArray);

×

1177

static const SSchema* getNormalColSchema(const STableMeta* pTableMeta, const char* pColName) {

×

1178

  for (int32_t i = 0; i < pTableMeta->tableInfo.numOfColumns; ++i) {

×

1179

    const SSchema* pSchema = pTableMeta->schema + i;

×

1180

    if (0 == strcmp(pColName, pSchema->name)) {

×

1181

      return pSchema;

×

1184

  return NULL;

×

1187

static STableMeta* getTableMeta(SCatalog* pCatalog, SRequestConnInfo* conn, char* dbName, char* tbName, int32_t acctId){

×

1188

  SName       sName = {0};

×

1189

  toName(acctId, dbName, tbName, &sName);

×

1190

  STableMeta* pTableMeta = NULL;

×

1191

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

×

1192

  if (code != 0) {

×

1193

    uError("failed to get table meta for reference table:%s.%s", dbName, tbName);

×

1194

    taosMemoryFreeClear(pTableMeta);

×

1195

    terrno = code;

×

1196

    return NULL;

×

1198

  return pTableMeta;

×

1201

static int32_t checkColRef(STableMeta* pTableMeta, char* colName, uint8_t precision, const SSchema* pSchema) {

×

1202

  int32_t code = TSDB_CODE_SUCCESS;

×

1203

  if (pTableMeta->tableInfo.precision != precision) {

×

1204

    code = TSDB_CODE_PAR_INVALID_REF_COLUMN_TYPE;

×

1205

    uError("timestamp precision of virtual table and its reference table do not match");

×

1206

    goto end;

×

1209

  if (pTableMeta->tableInfo.numOfColumns > 1 && pTableMeta->schema[1].flags & COL_IS_KEY) {

×

1210

    code = TSDB_CODE_PAR_INVALID_REF_COLUMN;

×

1211

    uError("virtual table's column:\"%s\"'s reference can not from table with composite key", colName);

×

1212

    goto end;

×

1216

  if (pTableMeta->tableType != TSDB_NORMAL_TABLE && pTableMeta->tableType != TSDB_CHILD_TABLE) {

×

1217

    code = TSDB_CODE_PAR_INVALID_REF_COLUMN;

×

1218

    uError("virtual table's column:\"%s\"'s reference can only be normal table or child table", colName);

×

1219

    goto end;

×

1222

  const SSchema* pRefCol = getNormalColSchema(pTableMeta, colName);

×

1223

  if (NULL == pRefCol) {

×

1224

    code = TSDB_CODE_PAR_INVALID_REF_COLUMN;

×

1225

    uError("virtual table's column:\"%s\"'s reference column:\"%s\" not exist", pSchema->name, colName);

×

1226

    goto end;

×

1229

  if (pRefCol->type != pSchema->type || pRefCol->bytes != pSchema->bytes) {

×

1230

    code = TSDB_CODE_PAR_INVALID_REF_COLUMN_TYPE;

×

1231

    uError("virtual table's column:\"%s\"'s type and reference column:\"%s\"'s type not match, %d %d %d %d",

×

1233

    goto end;

×

1236

end:

×

1237

  return code;

×

1240

static int32_t checkColRefForCreate(SCatalog* pCatalog, SRequestConnInfo* conn, SColRef* pColRef, int32_t acctId, uint8_t precision, SSchema* pSchema) {

×

1241

  STableMeta* pTableMeta = getTableMeta(pCatalog, conn, pColRef->refDbName, pColRef->refTableName, acctId);

×

1242

  if (pTableMeta == NULL) {

×

1243

      return terrno;

×

1245

  int32_t code = checkColRef(pTableMeta, pColRef->refColName, precision, pSchema);

×

1246

  taosMemoryFreeClear(pTableMeta);

×

1247

  return code;

×

1250

static int32_t checkColRefForAdd(SCatalog* pCatalog, SRequestConnInfo* conn, int32_t acctId, char* dbName, char* tbName, char* colName,

×

1252

  int32_t code = 0;

×

1253

  STableMeta* pTableMeta = getTableMeta(pCatalog, conn, dbName, tbName, acctId);

×

1254

  if (pTableMeta == NULL) {

×

1255

    code = terrno;

×

1256

    goto end;

×

1258

  STableMeta* pTableMetaSrc = getTableMeta(pCatalog, conn, dbNameSrc, tbNameSrc, acctId);

×

1259

  if (pTableMetaSrc == NULL) {

×

1260

    code = terrno;

×

1261

    goto end;

×

1264

  SSchema pSchema = {.type = type, .bytes = bytes};

×

1265

  tstrncpy(pSchema.name, colNameSrc, TSDB_COL_NAME_LEN);

×

1266

  code = checkColRef(pTableMeta, colName, pTableMetaSrc->tableInfo.precision, &pSchema);

×

1268

end:

×

1269

  taosMemoryFreeClear(pTableMeta);

×

1270

  taosMemoryFreeClear(pTableMetaSrc);

×

1271

  return code;

×

1274

static int32_t checkColRefForAlter(SCatalog* pCatalog, SRequestConnInfo* conn, int32_t acctId, char* dbName, char* tbName, char* colName,

×

1276

  int32_t code = 0;

×

1277

  STableMeta* pTableMeta = getTableMeta(pCatalog, conn, dbName, tbName, acctId);

×

1278

  if (pTableMeta == NULL) {

×

1279

    code = terrno;

×

1280

    goto end;

×

1282

  STableMeta* pTableMetaSrc = getTableMeta(pCatalog, conn, dbNameSrc, tbNameSrc, acctId);

×

1283

  if (pTableMetaSrc == NULL) {

×

1284

    code = terrno;

×

1285

    goto end;

×

1287

  const SSchema* pSchema = getNormalColSchema(pTableMetaSrc, colNameSrc);

×

1288

  if (NULL == pSchema) {

×

1289

    code = TSDB_CODE_PAR_INVALID_REF_COLUMN;

×

1290

    uError("virtual table's column:\"%s\" not exist", colNameSrc);

×

1291

    goto end;

×

1294

  code = checkColRef(pTableMeta, colName, pTableMetaSrc->tableInfo.precision, pSchema);

×

1296

end:

×

1297

  taosMemoryFreeClear(pTableMeta);

×

1298

  taosMemoryFreeClear(pTableMetaSrc);

×

1299

  return code;

×

1302

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

×

1303

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

×

1304

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

×

1305

    return TSDB_CODE_INVALID_PARA;

×

1307

  SVCreateTbBatchReq req = {0};

×

1308

  SDecoder           coder = {0};

×

1309

  int32_t            code = TSDB_CODE_SUCCESS;

×

1310

  int32_t            lino = 0;

×

1311

  SRequestObj*       pRequest = NULL;

×

1312

  SQuery*            pQuery = NULL;

×

1313

  SHashObj*          pVgroupHashmap = NULL;

×

1315

  RAW_LOG_START

×

1316

  SArray* pTagList = taosArrayInit(0, POINTER_BYTES);

×

1317

  RAW_NULL_CHECK(pTagList);

×

1318

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

×

1319

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

×

1321

  pRequest->syncQuery = true;

×

1322

  if (!pRequest->pDb) {

×

1323

    uError(LOG_ID_TAG " %s no database selected", LOG_ID_VALUE, __func__);

×

1324

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1325

    goto end;

×

1328

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

×

1329

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1330

  tDecoderInit(&coder, data, len);

×

1331

  RAW_RETURN_CHECK(tDecodeSVCreateTbBatchReq(&coder, &req));

×

1332

  STscObj* pTscObj = pRequest->pTscObj;

×

1334

  SVCreateTbReq* pCreateReq = NULL;

×

1335

  SCatalog*      pCatalog = NULL;

×

1336

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

×

1337

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

×

1338

  RAW_NULL_CHECK(pVgroupHashmap);

×

1339

  taosHashSetFreeFp(pVgroupHashmap, destroyCreateTbReqBatch);

×

1341

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

×

1342

                           .requestId = pRequest->requestId,

×

1343

                           .requestObjRefId = pRequest->self,

×

1344

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

×

1346

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

×

1347

  RAW_NULL_CHECK(pRequest->tableList);

×

1349

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

×

1350

    pCreateReq = req.pReqs + iReq;

×

1352

    SVgroupInfo pInfo = {0};

×

1353

    SName       pName = {0};

×

1354

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

×

1355

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

×

1357

    pCreateReq->flags |= TD_CREATE_IF_NOT_EXISTS;

×

1359

    if (pCreateReq->type == TSDB_CHILD_TABLE || pCreateReq->type == TSDB_VIRTUAL_CHILD_TABLE) {

×

1360

      STableMeta* pTableMeta = NULL;

×

1361

      SName       sName = {0};

×

1362

      tb_uid_t    oldSuid = pCreateReq->ctb.suid;

×

1364

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

×

1365

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

×

1366

      if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1367

        uInfo(LOG_ID_TAG " super table %s not exist, ignore create child table %s", LOG_ID_VALUE,

×

1369

        code = TSDB_CODE_SUCCESS;

×

1370

        taosMemoryFreeClear(pTableMeta);

×

1371

        continue;

×

1374

      RAW_RETURN_CHECK(code);

×

1375

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

×

1377

      bool changeDB = strlen(tmqWriteRefDB) > 0;

×

1378

      for (int32_t i = 0; changeDB && i < pCreateReq->colRef.nCols; i++) {

×

1379

        SColRef* pColRef = pCreateReq->colRef.pColRef + i;

×

1380

        tstrncpy(pColRef->refDbName, tmqWriteRefDB, TSDB_DB_NAME_LEN);

×

1383

      for (int32_t i = 0; tmqWriteCheckRef && i < pCreateReq->colRef.nCols && i < pTableMeta->tableInfo.numOfColumns; i++) {

×

1384

        SColRef* pColRef = pCreateReq->colRef.pColRef + i;

×

1385

        if (!pColRef || !pColRef->hasRef) continue;

×

1386

        SSchema* pSchema = pTableMeta->schema + i;

×

1387

        RAW_RETURN_CHECK(checkColRefForCreate(pCatalog, &conn, pColRef, pTscObj->acctId, pTableMeta->tableInfo.precision, pSchema));

×

1390

      SArray* pTagVals = NULL;

×

1391

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

×

1392

      if (code != TSDB_CODE_SUCCESS) {

×

1393

        uError("create tb invalid tag data %s", pCreateReq->name);

×

1394

        taosMemoryFreeClear(pTableMeta);

×

1395

        goto end;

×

1398

      bool rebuildTag = false;

×

1399

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

×

1400

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

×

1401

        if (tName == NULL) {

×

1402

          continue;

×

1404

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

×

1405

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

×

1406

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

×

1407

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

×

1408

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

×

1409

            if (pTagVal) {

×

1410

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

×

1411

                pTagVal->cid = tag->colId;

×

1412

                rebuildTag = true;

×

1415

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

×

1421

      taosMemoryFreeClear(pTableMeta);

×

1422

      if (rebuildTag) {

×

1423

        STag* ppTag = NULL;

×

1424

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

×

1425

        taosArrayDestroy(pTagVals);

×

1426

        pTagVals = NULL;

×

1427

        if (code != TSDB_CODE_SUCCESS) {

×

1428

          goto end;

×

1430

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

×

1431

          code = terrno;

×

1432

          tTagFree(ppTag);

×

1433

          goto end;

×

1435

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

×

1437

      taosArrayDestroy(pTagVals);

×

1439

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

×

1441

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

×

1442

    if (pTableBatch == NULL) {

×

1443

      SVgroupCreateTableBatch tBatch = {0};

×

1444

      tBatch.info = pInfo;

×

1445

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

×

1447

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

×

1448

      RAW_NULL_CHECK(tBatch.req.pArray);

×

1449

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

×

1450

      tBatch.req.source = TD_REQ_FROM_TAOX;

×

1451

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

×

1453

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

×

1457

  if (taosHashGetSize(pVgroupHashmap) == 0) {

×

1458

    goto end;

×

1460

  SArray* pBufArray = NULL;

×

1461

  RAW_RETURN_CHECK(serializeVgroupsCreateTableBatch(pVgroupHashmap, &pBufArray));

×

1462

  pQuery = NULL;

×

1463

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

×

1464

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1465

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1466

  pQuery->msgType = TDMT_VND_CREATE_TABLE;

×

1467

  pQuery->stableQuery = false;

×

1468

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

×

1469

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1470

  RAW_NULL_CHECK(pQuery->pRoot);

×

1472

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pBufArray));

×

1474

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1475

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1476

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

×

1479

  code = pRequest->code;

×

1480

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

×

1482

end:

×

1483

  tDeleteSVCreateTbBatchReq(&req);

×

1485

  taosHashCleanup(pVgroupHashmap);

×

1486

  destroyRequest(pRequest);

×

1487

  tDecoderClear(&coder);

×

1488

  qDestroyQuery(pQuery);

×

1489

  taosArrayDestroyP(pTagList, NULL);

×

1490

  RAW_LOG_END

×

1491

  return code;

×

1500

static void destroyDropTbReqBatch(void* data) {

×

1501

  if (data == NULL) {

×

1502

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

×

1503

    return;

×

1505

  SVgroupDropTableBatch* pTbBatch = (SVgroupDropTableBatch*)data;

×

1506

  taosArrayDestroy(pTbBatch->req.pArray);

×

1509

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

×

1510

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

×

1511

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

×

1512

    return TSDB_CODE_INVALID_PARA;

×

1514

  SVDropTbBatchReq req = {0};

×

1515

  SDecoder         coder = {0};

×

1516

  int32_t          code = TSDB_CODE_SUCCESS;

×

1517

  int32_t          lino = 0;

×

1518

  SRequestObj*     pRequest = NULL;

×

1519

  SQuery*          pQuery = NULL;

×

1520

  SHashObj*        pVgroupHashmap = NULL;

×

1522

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

×

1523

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

×

1525

  pRequest->syncQuery = true;

×

1526

  if (!pRequest->pDb) {

×

1527

    uError(LOG_ID_TAG " %s no database selected", LOG_ID_VALUE, __func__);

×

1528

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1529

    goto end;

×

1532

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

×

1533

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1534

  tDecoderInit(&coder, data, len);

×

1535

  RAW_RETURN_CHECK(tDecodeSVDropTbBatchReq(&coder, &req));

×

1536

  STscObj* pTscObj = pRequest->pTscObj;

×

1538

  SVDropTbReq* pDropReq = NULL;

×

1539

  SCatalog*    pCatalog = NULL;

×

1540

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

×

1542

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

×

1543

  RAW_NULL_CHECK(pVgroupHashmap);

×

1544

  taosHashSetFreeFp(pVgroupHashmap, destroyDropTbReqBatch);

×

1546

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

×

1547

                           .requestId = pRequest->requestId,

×

1548

                           .requestObjRefId = pRequest->self,

×

1549

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

×

1550

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

×

1551

  RAW_NULL_CHECK(pRequest->tableList);

×

1553

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

×

1554

    pDropReq = req.pReqs + iReq;

×

1555

    pDropReq->igNotExists = true;

×

1558

    SVgroupInfo pInfo = {0};

×

1559

    SName       pName = {0};

×

1560

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

×

1561

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

×

1563

    STableMeta* pTableMeta = NULL;

×

1564

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

×

1565

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {

×

1566

      code = TSDB_CODE_SUCCESS;

×

1567

      uInfo(LOG_ID_TAG " table %s not exist, ignore drop", LOG_ID_VALUE, pDropReq->name);

×

1568

      taosMemoryFreeClear(pTableMeta);

×

1569

      continue;

×

1571

    RAW_RETURN_CHECK(code);

×

1572

    tb_uid_t oldSuid = pDropReq->suid;

×

1573

    pDropReq->suid = pTableMeta->suid;

×

1574

    taosMemoryFreeClear(pTableMeta);

×

1575

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

×

1578

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

×

1579

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

×

1580

    if (pTableBatch == NULL) {

×

1581

      SVgroupDropTableBatch tBatch = {0};

×

1582

      tBatch.info = pInfo;

×

1583

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

×

1584

      RAW_NULL_CHECK(tBatch.req.pArray);

×

1585

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

×

1586

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

×

1588

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

×

1592

  if (taosHashGetSize(pVgroupHashmap) == 0) {

×

1593

    goto end;

×

1595

  SArray* pBufArray = NULL;

×

1596

  RAW_RETURN_CHECK(serializeVgroupsDropTableBatch(pVgroupHashmap, &pBufArray));

×

1597

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

×

1598

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1599

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1600

  pQuery->msgType = TDMT_VND_DROP_TABLE;

×

1601

  pQuery->stableQuery = false;

×

1602

  pQuery->pRoot = NULL;

×

1603

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

×

1604

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1605

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pBufArray));

×

1607

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1608

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1609

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

×

1611

  code = pRequest->code;

×

1612

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

×

1614

end:

×

1615

  taosHashCleanup(pVgroupHashmap);

×

1616

  destroyRequest(pRequest);

×

1617

  tDecoderClear(&coder);

×

1618

  qDestroyQuery(pQuery);

×

1619

  RAW_LOG_END

×

1620

  return code;

×

1623

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

×

1624

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

×

1625

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

×

1626

    return TSDB_CODE_INVALID_PARA;

×

1628

  SDeleteRes req = {0};

×

1629

  SDecoder   coder = {0};

×

1630

  char       sql[256] = {0};

×

1631

  int32_t    code = TSDB_CODE_SUCCESS;

×

1632

  int32_t    lino = 0;

×

1633

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

×

1636

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

×

1637

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1638

  tDecoderInit(&coder, data, len);

×

1639

  RAW_RETURN_CHECK(tDecodeDeleteRes(&coder, &req));

×

1640

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

×

1643

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

×

1644

  RAW_NULL_CHECK(res);

×

1645

  SRequestObj* pRequest = (SRequestObj*)res;

×

1646

  code = pRequest->code;

×

1647

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

×

1648

    code = TSDB_CODE_SUCCESS;

×

1650

  taos_free_result(res);

×

1651

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

×

1653

end:

×

1654

  RAW_LOG_END

×

1655

  tDecoderClear(&coder);

×

1656

  return code;

×

1659

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

×

1660

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

×

1661

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

×

1662

    return TSDB_CODE_INVALID_PARA;

×

1664

  SVAlterTbReq   req = {0};

×

1665

  SDecoder       dcoder = {0};

×

1666

  int32_t        code = TSDB_CODE_SUCCESS;

×

1667

  int32_t        lino = 0;

×

1668

  SRequestObj*   pRequest = NULL;

×

1669

  SQuery*        pQuery = NULL;

×

1670

  SArray*        pArray = NULL;

×

1671

  SVgDataBlocks* pVgData = NULL;

×

1672

  SEncoder       coder = {0};

×

1674

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

×

1675

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

×

1676

  pRequest->syncQuery = true;

×

1677

  if (!pRequest->pDb) {

×

1678

    uError(LOG_ID_TAG " %s no database selected", LOG_ID_VALUE, __func__);

×

1679

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1680

    goto end;

×

1683

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

×

1684

  uint32_t len = metaLen - sizeof(SMsgHead);

×

1685

  tDecoderInit(&dcoder, data, len);

×

1686

  RAW_RETURN_CHECK(tDecodeSVAlterTbReq(&dcoder, &req));

×

1688

  if (req.action == TSDB_ALTER_TABLE_UPDATE_OPTIONS) {

×

1689

    uInfo(LOG_ID_TAG " alter table action is UPDATE_OPTIONS, ignore", LOG_ID_VALUE);

×

1690

    goto end;

×

1693

  STscObj*  pTscObj = pRequest->pTscObj;

×

1694

  SCatalog* pCatalog = NULL;

×

1695

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

×

1696

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

×

1697

                           .requestId = pRequest->requestId,

×

1698

                           .requestObjRefId = pRequest->self,

×

1699

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

×

1701

  SVgroupInfo pInfo = {0};

×

1702

  SName       pName = {0};

×

1703

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

×

1704

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

×

1705

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

×

1706

  RAW_NULL_CHECK(pArray);

×

1708

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

×

1709

  RAW_NULL_CHECK(pVgData);

×

1710

  pVgData->vg = pInfo;

×

1712

  int tlen = 0;

×

1713

  req.source = TD_REQ_FROM_TAOX;

×

1715

  if (strlen(tmqWriteRefDB) > 0) {

×

1716

    req.refDbName = tmqWriteRefDB;

×

1719

  if (req.action == TSDB_ALTER_TABLE_ALTER_COLUMN_REF && tmqWriteCheckRef) {

×

1720

    RAW_RETURN_CHECK(checkColRefForAlter(pCatalog, &conn, pTscObj->acctId, req.refDbName, req.refTbName, req.refColName,

×

1722

  }else if (req.action == TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COLUMN_REF && tmqWriteCheckRef) {

×

1723

    RAW_RETURN_CHECK(checkColRefForAdd(pCatalog, &conn, pTscObj->acctId, req.refDbName, req.refTbName, req.refColName,

×

1727

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

×

1728

  RAW_RETURN_CHECK(code);

×

1729

  tlen += sizeof(SMsgHead);

×

1730

  void* pMsg = taosMemoryMalloc(tlen);

×

1731

  RAW_NULL_CHECK(pMsg);

×

1732

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

×

1733

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

×

1734

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

×

1735

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

×

1736

  code = tEncodeSVAlterTbReq(&coder, &req);

×

1737

  RAW_RETURN_CHECK(code);

×

1739

  pVgData->pData = pMsg;

×

1740

  pVgData->size = tlen;

×

1742

  pVgData->numOfTables = 1;

×

1743

  RAW_NULL_CHECK(taosArrayPush(pArray, &pVgData));

×

1745

  pQuery = NULL;

×

1746

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

×

1747

  if (NULL == pQuery) goto end;

×

1748

  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;

×

1749

  pQuery->msgType = TDMT_VND_ALTER_TABLE;

×

1750

  pQuery->stableQuery = false;

×

1751

  pQuery->pRoot = NULL;

×

1752

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

×

1753

  if (TSDB_CODE_SUCCESS != code) goto end;

×

1754

  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pArray));

×

1756

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1758

  pVgData = NULL;

×

1759

  pArray = NULL;

×

1760

  code = pRequest->code;

×

1761

  if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

×

1762

    code = TSDB_CODE_SUCCESS;

×

1765

  if (pRequest->code == TSDB_CODE_SUCCESS) {

×

1766

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

×

1767

    if (pRes->res != NULL) {

×

1768

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

×

1771

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

×

1773

end:

×

1774

  taosArrayDestroy(pArray);

×

1775

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

×

1776

  taosMemoryFreeClear(pVgData);

×

1777

  destroyRequest(pRequest);

×

1778

  tDecoderClear(&dcoder);

×

1779

  qDestroyQuery(pQuery);

×

1780

  taosArrayDestroy(req.pMultiTag);

×

1781

  tEncoderClear(&coder);

×

1782

  RAW_LOG_END

×

1783

  return code;

×

1786

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

×

1788

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

×

1791

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

×

1793

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

×

1794

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

×

1795

    return TSDB_CODE_INVALID_PARA;

×

1797

  int32_t     code = TSDB_CODE_SUCCESS;

×

1798

  int32_t     lino = 0;

×

1799

  STableMeta* pTableMeta = NULL;

×

1800

  SQuery*     pQuery = NULL;

×

1801

  SHashObj*   pVgHash = NULL;

×

1803

  SRequestObj* pRequest = NULL;

×

1804

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

×

1806

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

×

1809

  pRequest->syncQuery = true;

×

1810

  if (!pRequest->pDb) {

×

1811

    uError(LOG_ID_TAG " %s no database selected", LOG_ID_VALUE, __func__);

×

1812

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1813

    goto end;

×

1816

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

×

1817

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

×

1818

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

×

1820

  struct SCatalog* pCatalog = NULL;

×

1821

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

×

1823

  SRequestConnInfo conn = {0};

×

1824

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

×

1825

  conn.requestId = pRequest->requestId;

×

1826

  conn.requestObjRefId = pRequest->self;

×

1827

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

×

1829

  SVgroupInfo vgData = {0};

×

1830

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

×

1831

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

×

1832

  RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

1833

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

×

1834

  RAW_NULL_CHECK(pVgHash);

×

1835

  RAW_RETURN_CHECK(

×

1837

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

×

1838

  RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

1840

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1841

  code = pRequest->code;

×

1842

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

×

1844

end:

×

1845

  taosMemoryFreeClear(pTableMeta);

×

1846

  qDestroyQuery(pQuery);

×

1847

  destroyRequest(pRequest);

×

1848

  taosHashCleanup(pVgHash);

×

1849

  RAW_LOG_END

×

1850

  return code;

×

1853

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

×

1854

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

×

1857

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

×

1858

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

×

1859

    return TSDB_CODE_INVALID_PARA;

×

1861

  int32_t     code = TSDB_CODE_SUCCESS;

×

1862

  int32_t     lino = 0;

×

1863

  STableMeta* pTableMeta = NULL;

×

1864

  SQuery*     pQuery = NULL;

×

1865

  SHashObj*   pVgHash = NULL;

×

1867

  SRequestObj* pRequest = NULL;

×

1868

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

×

1870

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

×

1872

  pRequest->syncQuery = true;

×

1873

  if (!pRequest->pDb) {

×

1874

    uError(LOG_ID_TAG " %s no database selected", LOG_ID_VALUE, __func__);

×

1875

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

1876

    goto end;

×

1879

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

×

1880

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

×

1881

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

×

1883

  struct SCatalog* pCatalog = NULL;

×

1884

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

×

1886

  SRequestConnInfo conn = {0};

×

1887

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

×

1888

  conn.requestId = pRequest->requestId;

×

1889

  conn.requestObjRefId = pRequest->self;

×

1890

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

×

1892

  SVgroupInfo vgData = {0};

×

1893

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

×

1894

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

×

1895

  RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

1896

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

×

1897

  RAW_NULL_CHECK(pVgHash);

×

1898

  RAW_RETURN_CHECK(

×

1900

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

×

1901

  RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

1903

  launchQueryImpl(pRequest, pQuery, true, NULL);

×

1904

  code = pRequest->code;

×

1905

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

×

1907

end:

×

1908

  taosMemoryFreeClear(pTableMeta);

×

1909

  qDestroyQuery(pQuery);

×

1910

  destroyRequest(pRequest);

×

1911

  taosHashCleanup(pVgHash);

×

1912

  RAW_LOG_END

×

1913

  return code;

×

1916

static void* getRawDataFromRes(void* pRetrieve) {

×

1917

  if (pRetrieve == NULL) {

×

1918

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

×

1919

    return NULL;

×

1921

  void* rawData = NULL;

×

1923

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

×

1924

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

×

1925

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

×

1926

             *(int64_t*)pRetrieve == RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION) {

×

1927

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

×

1929

  return rawData;

×

1932

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

×

1933

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

×

1934

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

×

1935

    return TSDB_CODE_INVALID_PARA;

×

1938

  int32_t       code = 0;

×

1939

  int32_t       lino = 0;

×

1940

  SVCreateTbReq pCreateReq = {0};

×

1941

  SDecoder      decoderTmp = {0};

×

1942

  RAW_LOG_START

×

1943

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

×

1944

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

×

1945

    RAW_NULL_CHECK(dataTmp);

×

1946

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

×

1947

    RAW_NULL_CHECK(lenTmp);

×

1949

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

×

1950

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

×

1952

    if (pCreateReq.type != TSDB_CHILD_TABLE) {

×

1953

      uError("invalid table type %d in %s", pCreateReq.type, __func__);

×

1954

      code = TSDB_CODE_INVALID_MSG;

×

1955

      goto end;

×

1957

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

×

1958

      RAW_RETURN_CHECK(

×

1961

      tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);

×

1964

    tDecoderClear(&decoderTmp);

×

1965

    pCreateReq = (SVCreateTbReq){0};

×

1968

end:

×

1969

  tDecoderClear(&decoderTmp);

×

1970

  tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);

×

1971

  RAW_LOG_END

×

1972

  return code;

×

1997

static void tmqFreeMeta(void* data) {

×

1998

  if (data == NULL) {

×

1999

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

×

2000

    return;

×

2002

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

×

2003

  taosMemoryFree(pTableMeta);

×

2006

static void freeRawCache(void* data) {

×

2007

  if (data == NULL) {

×

2008

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

×

2009

    return;

×

2011

  rawCacheInfo* pRawCache = (rawCacheInfo*)data;

×

2012

  taosHashCleanup(pRawCache->pMetaHash);

×

2013

  taosHashCleanup(pRawCache->pNameHash);

×

2014

  taosHashCleanup(pRawCache->pVgHash);

×

2017

static int32_t initRawCacheHash() {

×

2018

  if (writeRawCache == NULL) {

×

2019

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

×

2020

    if (writeRawCache == NULL) {

×

2021

      return terrno;

×

2023

    taosHashSetFreeFp(writeRawCache, freeRawCache);

×

2025

  return 0;

×

2028

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

×

2029

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

×

2030

    return false;

×

2032

  if (pTableMeta == NULL) {

×

2033

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

×

2034

    return false;

×

2036

  char* p = (char*)rawData;

×

2039

  p += sizeof(int32_t);

×

2040

  p += sizeof(int32_t);

×

2041

  p += sizeof(int32_t);

×

2042

  p += sizeof(int32_t);

×

2043

  p += sizeof(int32_t);

×

2044

  p += sizeof(uint64_t);

×

2045

  int8_t* fields = p;

×

2047

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

×

2048

    return true;

×

2051

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

×

2052

    int j = 0;

×

2053

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

×

2054

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

×

2055

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

×

2056

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

×

2058

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

×

2059

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

×

2060

          return true;

×

2062

        break;

×

2065

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

×

2067

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

×

2069

  return false;

×

2072

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

×

2073

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

×

2074

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

×

2075

    return TSDB_CODE_INVALID_PARA;

×

2077

  int32_t code = 0;

×

2078

  int32_t lino = 0;

×

2079

  RAW_LOG_START

×

2080

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

×

2081

  if (cacheInfo == NULL) {

×

2082

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

×

2083

    RAW_NULL_CHECK(*pVgHash);

×

2084

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

×

2085

    RAW_NULL_CHECK(*pNameHash);

×

2086

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

×

2087

    RAW_NULL_CHECK(*pMetaHash);

×

2088

    taosHashSetFreeFp(*pMetaHash, tmqFreeMeta);

×

2089

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

×

2090

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

×

2092

    rawCacheInfo* info = (rawCacheInfo*)cacheInfo;

×

2093

    *pVgHash = info->pVgHash;

×

2094

    *pNameHash = info->pNameHash;

×

2095

    *pMetaHash = info->pMetaHash;

×

2098

end:

×

2099

  if (code != 0) {

×

2100

    taosHashCleanup(*pMetaHash);

×

2101

    taosHashCleanup(*pNameHash);

×

2102

    taosHashCleanup(*pVgHash);

×

2104

  RAW_LOG_END

×

2105

  return code;

×

2108

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

×

2109

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

×

2110

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

×

2111

    return TSDB_CODE_INVALID_PARA;

×

2113

  int32_t code = 0;

×

2114

  int32_t lino = 0;

×

2115

  RAW_LOG_START

×

2116

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

×

2117

  (*pRequest)->syncQuery = true;

×

2118

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

×

2119

    uError("%s no database selected", __func__);

×

2120

    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;

×

2121

    goto end;

×

2124

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

×

2125

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

×

2126

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

×

2127

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

×

2128

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

×

2130

end:

×

2131

  RAW_LOG_END

×

2132

  return code;

×

2136

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

×

2138

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

×

2139

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

×

2140

    return TSDB_CODE_INVALID_PARA;

×

2142

  int8_t dataVersion = *(int8_t*)data;

×

2143

  if (dataVersion >= MQ_DATA_RSP_VERSION) {

×

2144

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

×

2145

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

×

2146

      return TSDB_CODE_INVALID_PARA;

×

2148

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

×

2151

  rspObj->resIter = -1;

×

2152

  tDecoderInit(decoder, data, dataLen);

×

2153

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

×

2154

  if (code != 0) {

×

2155

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

×

2157

  return code;

×

2160

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

×

2163

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

×

2165

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

×

2166

    return TSDB_CODE_INVALID_PARA;

×

2168

  int32_t code = 0;

×

2169

  int32_t lino = 0;

×

2170

  RAW_LOG_START

×

2171

  STableMeta* pTableMeta = NULL;

×

2172

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

×

2173

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

×

2174

    tbInfo info = {0};

×

2176

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

×

2177

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

×

2178

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

×

2180

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

×

2181

    info.uid = pTableMeta->uid;

×

2182

    if (pTableMeta->tableType == TSDB_CHILD_TABLE) {

×

2183

      info.suid = pTableMeta->suid;

×

2185

      info.suid = pTableMeta->uid;

×

2187

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

×

2188

    RAW_RETURN_CHECK(code);

×

2190

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

×

2192

    if (pCreateReqDst) {

×

2193

      pTableMeta->vgId = info.vgInfo.vgId;

×

2194

      pTableMeta->uid = pCreateReqDst->uid;

×

2195

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

×

2198

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

×

2199

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

×

2200

    RAW_RETURN_CHECK(

×

2204

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

×

2205

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

×

2206

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

×

2207

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

×

2208

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

×

2209

      RAW_RETURN_CHECK(code);

×

2210

      uDebug("put table meta to hash2, suid:%" PRId64 ", metaHashSIze:%d, nameHashSize:%d, vgHashSize:%d",

×

2213

      pTableMeta = *pTableMetaTmp;

×

2214

      pTableMeta->uid = tmpInfo->uid;

×

2215

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

×

2218

  *pMeta = pTableMeta;

×

2219

  pTableMeta = NULL;

×

2221

end:

×

2222

  taosMemoryFree(pTableMeta);

×

2223

  RAW_LOG_END

×

2224

  return code;

×

2227

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

×

2228

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

×

2229

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

×

2230

    return TSDB_CODE_INVALID_PARA;

×

2232

  int32_t   code = TSDB_CODE_SUCCESS;

×

2233

  int32_t   lino = 0;

×

2234

  SQuery*   pQuery = NULL;

×

2235

  SMqRspObj rspObj = {0};

×

2236

  SDecoder  decoder = {0};

×

2238

  SRequestObj*     pRequest = NULL;

×

2239

  SCatalog*        pCatalog = NULL;

×

2240

  SRequestConnInfo conn = {0};

×

2241

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

×

2242

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

×

2243

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

×

2245

  SHashObj* pVgHash = NULL;

×

2246

  SHashObj* pNameHash = NULL;

×

2247

  SHashObj* pMetaHash = NULL;

×

2248

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

×

2249

  int retry = 0;

×

2251

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2252

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

×

2253

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

×

2254

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

×

2255

      RAW_NULL_CHECK(tbName);

×

2256

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

×

2257

      RAW_NULL_CHECK(pSW);

×

2258

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

×

2259

      RAW_NULL_CHECK(pRetrieve);

×

2260

      void* rawData = getRawDataFromRes(pRetrieve);

×

2261

      RAW_NULL_CHECK(rawData);

×

2263

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

×

2264

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

×

2265

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

×

2266

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2268

      STableMeta* pTableMeta = NULL;

×

2269

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

×

2271

      char err[ERR_MSG_LEN] = {0};

×

2272

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

×

2273

      if (code != TSDB_CODE_SUCCESS) {

×

2274

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

×

2275

        goto end;

×

2278

    RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

2279

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2280

    code = pRequest->code;

×

2282

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

×

2283

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

×

2284

      qDestroyQuery(pQuery);

×

2285

      pQuery = NULL;

×

2286

      rspObj.resIter = -1;

×

2287

      continue;

×

2289

    break;

×

2291

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

×

2293

end:

×

2294

  tDeleteMqDataRsp(&rspObj.dataRsp);

×

2295

  tDecoderClear(&decoder);

×

2296

  qDestroyQuery(pQuery);

×

2297

  destroyRequest(pRequest);

×

2298

  RAW_LOG_END

×

2299

  return code;

×

2302

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

×

2303

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

×

2304

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

×

2305

    return TSDB_CODE_INVALID_PARA;

×

2307

  int32_t   code = TSDB_CODE_SUCCESS;

×

2308

  int32_t   lino = 0;

×

2309

  SQuery*   pQuery = NULL;

×

2310

  SMqRspObj rspObj = {0};

×

2311

  SDecoder  decoder = {0};

×

2312

  SHashObj* pCreateTbHash = NULL;

×

2314

  SRequestObj*     pRequest = NULL;

×

2315

  SCatalog*        pCatalog = NULL;

×

2316

  SRequestConnInfo conn = {0};

×

2318

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

×

2319

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

×

2320

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

×

2322

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

×

2323

  RAW_NULL_CHECK(pCreateTbHash);

×

2324

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

×

2326

  SHashObj* pVgHash = NULL;

×

2327

  SHashObj* pNameHash = NULL;

×

2328

  SHashObj* pMetaHash = NULL;

×

2329

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

×

2330

  int retry = 0;

×

2332

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2333

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

×

2334

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

×

2335

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

×

2336

      RAW_NULL_CHECK(tbName);

×

2337

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

×

2338

      RAW_NULL_CHECK(pSW);

×

2339

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

×

2340

      RAW_NULL_CHECK(pRetrieve);

×

2341

      void* rawData = getRawDataFromRes(pRetrieve);

×

2342

      RAW_NULL_CHECK(rawData);

×

2344

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

×

2345

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

×

2346

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

×

2347

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2350

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

×

2351

      STableMeta*    pTableMeta = NULL;

×

2352

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

×

2354

      char err[ERR_MSG_LEN] = {0};

×

2356

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

×

2357

      if (code != TSDB_CODE_SUCCESS) {

×

2358

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

×

2359

        goto end;

×

2362

    RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));

×

2363

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2364

    code = pRequest->code;

×

2366

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

×

2367

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

×

2368

      qDestroyQuery(pQuery);

×

2369

      pQuery = NULL;

×

2370

      rspObj.resIter = -1;

×

2371

      continue;

×

2373

    break;

×

2375

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

×

2377

end:

×

2378

  tDeleteSTaosxRsp(&rspObj.dataRsp);

×

2379

  void* pIter = taosHashIterate(pCreateTbHash, NULL);

×

2380

  while (pIter) {

×

2381

    tDestroySVCreateTbReq(pIter, TSDB_MSG_FLG_DECODE);

×

2382

    pIter = taosHashIterate(pCreateTbHash, pIter);

×

2384

  taosHashCleanup(pCreateTbHash);

×

2385

  tDecoderClear(&decoder);

×

2386

  qDestroyQuery(pQuery);

×

2387

  destroyRequest(pRequest);

×

2388

  RAW_LOG_END

×

2389

  return code;

×

2392

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

×

2393

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

×

2394

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

×

2395

    return TSDB_CODE_INVALID_PARA;

×

2397

  int32_t   code = TSDB_CODE_SUCCESS;

×

2398

  int32_t   lino = 0;

×

2399

  SQuery*   pQuery = NULL;

×

2400

  SHashObj* pVgroupHash = NULL;

×

2401

  SMqRspObj rspObj = {0};

×

2402

  SDecoder  decoder = {0};

×

2404

  SRequestObj*     pRequest = NULL;

×

2405

  SCatalog*        pCatalog = NULL;

×

2406

  SRequestConnInfo conn = {0};

×

2408

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

×

2409

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

×

2410

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

×

2412

  SHashObj* pVgHash = NULL;

×

2413

  SHashObj* pNameHash = NULL;

×

2414

  SHashObj* pMetaHash = NULL;

×

2415

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

×

2416

  int retry = 0;

×

2417

  while (1) {

×

2418

    RAW_RETURN_CHECK(smlInitHandle(&pQuery));

×

2419

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

×

2420

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

×

2421

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

×

2422

    RAW_NULL_CHECK(pVgroupHash);

×

2423

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

×

2424

    RAW_NULL_CHECK(pStmt->pVgDataBlocks);

×

2426

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

×

2427

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

×

2428

      RAW_NULL_CHECK(tbName);

×

2429

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

×

2430

      RAW_NULL_CHECK(pRetrieve);

×

2431

      void* rawData = getRawDataFromRes(pRetrieve);

×

2432

      RAW_NULL_CHECK(rawData);

×

2434

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

×

2435

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

×

2436

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

×

2437

      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);

×

2440

      STableMeta* pTableMeta = NULL;

×

2441

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

×

2443

      char err[ERR_MSG_LEN] = {0};

×

2444

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

×

2445

      if (code != TSDB_CODE_SUCCESS) {

×

2446

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

×

2447

        goto end;

×

2450

    taosHashCleanup(pVgroupHash);

×

2451

    pVgroupHash = NULL;

×

2453

    RAW_RETURN_CHECK(smlBuildOutputRaw(pQuery, pVgHash));

×

2454

    launchQueryImpl(pRequest, pQuery, true, NULL);

×

2455

    code = pRequest->code;

×

2457

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

×

2458

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

×

2459

      qDestroyQuery(pQuery);

×

2460

      pQuery = NULL;

×

2461

      rspObj.resIter = -1;

×

2462

      continue;

×

2464

    break;

×

2466

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

×

2468

end:

×

2469

  tDeleteMqDataRsp(&rspObj.dataRsp);

×

2470

  tDecoderClear(&decoder);

×

2471

  qDestroyQuery(pQuery);

×

2472

  taosHashCleanup(pVgroupHash);

×

2473

  destroyRequest(pRequest);

×

2474

  RAW_LOG_END

×

2475

  return code;

×

2478

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

×

2479

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

×

2480

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

×

2481

    return TSDB_CODE_INVALID_PARA;

×

2483

  int32_t code = 0;

×

2484

  int32_t lino = 0;

×

2485

  RAW_LOG_START

×

2486

  if (pMetaRsp->resMsgType == TDMT_VND_CREATE_STB) {

×

2487

    RAW_RETURN_CHECK(processCreateStb(pMetaRsp, meta));

×

2488

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

×

2489

    RAW_RETURN_CHECK(processAlterStb(pMetaRsp, meta));

×

2490

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

×

2491

    RAW_RETURN_CHECK(processDropSTable(pMetaRsp, meta));

×

2492

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

×

2493

    RAW_RETURN_CHECK(processCreateTable(pMetaRsp, meta));

×

2494

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

×

2495

    RAW_RETURN_CHECK(processAlterTable(pMetaRsp, meta));

×

2496

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

×

2497

    RAW_RETURN_CHECK(processDropTable(pMetaRsp, meta));

×

2498

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

×

2499

    RAW_RETURN_CHECK(processDeleteTable(pMetaRsp, meta));

×

2502

end:

×

2503

  RAW_LOG_END

×

2504

  return code;

×

2507

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

×

2508

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

×

2509

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

×

2510

    return TSDB_CODE_INVALID_PARA;

×

2512

  SDecoder        coder = {0};

×

2513

  SMqBatchMetaRsp rsp = {0};

×

2514

  int32_t         code = 0;

×

2515

  int32_t         lino = 0;

×

2516

  cJSON*          pJson = NULL;

×

2517

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

×

2518

  RAW_RETURN_CHECK(tDecodeMqBatchMetaRsp(&coder, &rsp));

×

2520

  pJson = cJSON_CreateObject();

×

2521

  RAW_NULL_CHECK(pJson);

×

2522

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

×

2523

  cJSON* pMetaArr = cJSON_AddArrayToObject(pJson, "metas");

×

2524

  RAW_NULL_CHECK(pMetaArr);

×

2526

  int32_t num = taosArrayGetSize(rsp.batchMetaReq);

×

2527

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

×

2528

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

×

2529

    RAW_NULL_CHECK(len);

×

2530

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

×

2531

    RAW_NULL_CHECK(tmpBuf);

×

2532

    SDecoder   metaCoder = {0};

×

2533

    SMqMetaRsp metaRsp = {0};

×

2534

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

×

2535

    RAW_RETURN_CHECK(tDecodeMqMetaRsp(&metaCoder, &metaRsp));

×

2536

    cJSON* pItem = NULL;

×

2537

    RAW_RETURN_CHECK(processSimpleMeta(&metaRsp, &pItem));

×

2538

    tDeleteMqMetaRsp(&metaRsp);

×

2539

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

×

2542

  char* fullStr = cJSON_PrintUnformatted(pJson);

×

2543

  *string = fullStr;

×

2545

end:

×

2546

  cJSON_Delete(pJson);

×

2547

  tDeleteMqBatchMetaRsp(&rsp);

×

2548

  RAW_LOG_END

×

2549

  return code;

×

2552

char* tmq_get_json_meta(TAOS_RES* res) {

×

2553

  int32_t code = TSDB_CODE_SUCCESS;

×

2554

  int32_t lino = 0;

×

2555

  char*   string = NULL;

×

2556

  RAW_LOG_START

×

2557

  RAW_NULL_CHECK(res);

×

2558

  RAW_FALSE_CHECK(TD_RES_TMQ_META(res) || TD_RES_TMQ_METADATA(res) || TD_RES_TMQ_BATCH_META(res));

×

2560

  SMqRspObj* rspObj = (SMqRspObj*)res;

×

2561

  if (TD_RES_TMQ_METADATA(res)) {

×

2562

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

×

2563

  } else if (TD_RES_TMQ_BATCH_META(res)) {

×

2564

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

×

2565

  } else if (TD_RES_TMQ_META(res)) {

×

2566

    cJSON* pJson = NULL;

×

2567

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

×

2568

    string = cJSON_PrintUnformatted(pJson);

×

2569

    cJSON_Delete(pJson);

×

2571

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

×

2574

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

×

2576

end:

×

2577

  RAW_LOG_END

×

2578

  return string;

×

2581

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

×

2583

static int32_t getOffSetLen(const SMqDataRsp* pRsp) {

×

2584

  if (pRsp == NULL) {

×

2585

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

×

2586

    return TSDB_CODE_INVALID_PARA;

×

2588

  int32_t pos = 0;

×

2589

  int32_t code = 0;

×

2590

  int32_t lino = 0;

×

2591

  RAW_LOG_START

×

2592

  SEncoder coder = {0};

×

2593

  tEncoderInit(&coder, NULL, 0);

×

2594

  RAW_RETURN_CHECK(tEncodeSTqOffsetVal(&coder, &pRsp->reqOffset));

×

2595

  RAW_RETURN_CHECK(tEncodeSTqOffsetVal(&coder, &pRsp->rspOffset));

×

2596

  pos = coder.pos;

×

2597

  tEncoderClear(&coder);

×

2599

end:

×

2600

  if (code != 0) {

×

2601

    uError("getOffSetLen failed, code:%d", code);

×

2602

    return code;

×

2604

    uDebug("getOffSetLen success, len:%d", pos);

×

2605

    return pos;

×

2610

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

×

2611

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

×

2612

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

×

2613

    return TSDB_CODE_INVALID_PARA;

×

2615

  uint32_t len = 0;

×

2616

  int32_t  code = 0;

×

2617

  int32_t  lino = 0;

×

2618

  SEncoder encoder = {0};

×

2619

  void*    buf = NULL;

×

2620

  tEncodeSize(encodeFunc, rspObj, len, code);

×

2621

  RAW_FALSE_CHECK(code >= 0);

×

2622

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

×

2623

  buf = taosMemoryCalloc(1, len);

×

2624

  RAW_NULL_CHECK(buf);

×

2625

  tEncoderInit(&encoder, buf, len);

×

2626

  RAW_RETURN_CHECK(tEncodeI8(&encoder, MQ_DATA_RSP_VERSION));

×

2627

  int32_t offsetLen = getOffSetLen(rspObj);

×

2628

  RAW_FALSE_CHECK(offsetLen > 0);

×

2629

  RAW_RETURN_CHECK(tEncodeI32(&encoder, offsetLen));

×

2630

  RAW_RETURN_CHECK(encodeFunc(&encoder, rspObj));

×

2632

  raw->raw = buf;

×

2633

  buf = NULL;

×

2634

  raw->raw_len = len;

×

2636

end:

×

2637

  RAW_LOG_END

×

2638

  return code;

×

2641

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

×

2642

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

×

2643

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

×

2644

    return TSDB_CODE_INVALID_PARA;

×

2646

  int32_t code = TSDB_CODE_SUCCESS;

×

2647

  int32_t lino = 0;

×

2648

  RAW_LOG_START

×

2649

  *raw = (tmq_raw_data){0};

×

2650

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

×

2651

  if (TD_RES_TMQ_META(res)) {

×

2652

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

×

2653

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

×

2654

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

×

2655

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

×

2656

  } else if (TD_RES_TMQ(res)) {

×

2657

    RAW_RETURN_CHECK(encodeMqDataRsp(tEncodeMqDataRsp, &rspObj->dataRsp, raw));

×

2658

    raw->raw_type = RES_TYPE__TMQ;

×

2659

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

×

2660

  } else if (TD_RES_TMQ_METADATA(res)) {

×

2661

    RAW_RETURN_CHECK(encodeMqDataRsp(tEncodeSTaosxRsp, &rspObj->dataRsp, raw));

×

2662

    raw->raw_type = RES_TYPE__TMQ_METADATA;

×

2663

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

×

2664

  } else if (TD_RES_TMQ_BATCH_META(res)) {

×

2665

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

×

2666

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

×

2667

    raw->raw_type = rspObj->resType;

×

2668

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

×

2669

  } else if (TD_RES_TMQ_RAW(res)) {

×

2670

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

×

2671

    rspObj->dataRsp.rawData = NULL;

×

2672

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

×

2673

    raw->raw_type = rspObj->resType;

×

2674

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

×

2676

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

×

2677

    code = TSDB_CODE_TMQ_INVALID_MSG;

×

2680

end:

×

2681

  RAW_LOG_END

×

2682

  return code;

×

2685

void tmq_free_raw(tmq_raw_data raw) {

×

2686

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

×

2687

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

×

2688

    taosMemoryFree(raw.raw);

×

2689

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

×

2690

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

×

2692

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

×

2693

×

2695

static int32_t writeRawInit() {

×

2696

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

×

2697

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

×

2698

    if (old == 0) {

×

2699

      int32_t code = initRawCacheHash();

×

2700

      if (code != 0) {

×

2701

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

×

2702

        atomic_store_8(&initedFlag, WRITE_RAW_INIT_FAIL);

×

2703

        return code;

×

2705

      atomic_store_8(&initedFlag, WRITE_RAW_INIT_OK);

×

2709

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

×

2710

    return TSDB_CODE_INTERNAL_ERROR;

×

2712

  return 0;

×

2715

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

×

2716

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

×

2717

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

×

2718

    return TSDB_CODE_INVALID_PARA;

×

2720

  if (writeRawInit() != 0) {

×

2721

    return TSDB_CODE_INTERNAL_ERROR;

×

2724

  if (type == TDMT_VND_CREATE_STB) {

×

2725

    return taosCreateStb(taos, buf, len);

×

2726

  } else if (type == TDMT_VND_ALTER_STB) {

×

2727

    return taosCreateStb(taos, buf, len);

×

2728

  } else if (type == TDMT_VND_DROP_STB) {

×

2729

    return taosDropStb(taos, buf, len);

×

2730

  } else if (type == TDMT_VND_CREATE_TABLE) {

×

2731

    return taosCreateTable(taos, buf, len);

×

2732

  } else if (type == TDMT_VND_ALTER_TABLE) {

×

2733

    return taosAlterTable(taos, buf, len);

×

2734

  } else if (type == TDMT_VND_DROP_TABLE) {

×

2735

    return taosDropTable(taos, buf, len);

×

2736

  } else if (type == TDMT_VND_DELETE) {

×

2737

    return taosDeleteData(taos, buf, len);

×

2738

  } else if (type == RES_TYPE__TMQ_METADATA) {

×

2739

    return tmqWriteRawMetaDataImpl(taos, buf, len);

×

2740

  } else if (type == RES_TYPE__TMQ_RAWDATA) {

×

2741

    return tmqWriteRawRawDataImpl(taos, buf, len);

×

2742

  } else if (type == RES_TYPE__TMQ) {

×

2743

    return tmqWriteRawDataImpl(taos, buf, len);

×

2744

  } else if (type == RES_TYPE__TMQ_BATCH_META) {

×

2745

    return tmqWriteBatchMetaDataImpl(taos, buf, len);

×

2747

  return TSDB_CODE_INVALID_PARA;

×

2750

int32_t tmq_write_raw(TAOS* taos, tmq_raw_data raw) {

×

2751

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

×

2752

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

×

2753

    return TSDB_CODE_INVALID_PARA;

×

2755

  taosClearErrMsg();  // clear global error message

×

2757

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

×

2760

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

×

2761

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

×

2762

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

×

2763

    return TSDB_CODE_INVALID_PARA;

×

2765

  SMqBatchMetaRsp rsp = {0};

×

2766

  SDecoder        coder = {0};

×

2767

  int32_t         code = TSDB_CODE_SUCCESS;

×

2768

  int32_t         lino = 0;

×

2770

  RAW_LOG_START

×

2772

  tDecoderInit(&coder, meta, metaLen);

×

2773

  RAW_RETURN_CHECK(tDecodeMqBatchMetaRsp(&coder, &rsp));

×

2774

  int32_t num = taosArrayGetSize(rsp.batchMetaReq);

×

2775

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

×

2776

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

×

2777

    RAW_NULL_CHECK(len);

×

2778

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

×

2779

    RAW_NULL_CHECK(tmpBuf);

×

2780

    SDecoder   metaCoder = {0};

×

2781

    SMqMetaRsp metaRsp = {0};

×

2782

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

×

2783

    RAW_RETURN_CHECK(tDecodeMqMetaRsp(&metaCoder, &metaRsp));

×

2784

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

×

2785

    tDeleteMqMetaRsp(&metaRsp);

×

2786

    if (code != TSDB_CODE_SUCCESS) {

×

2787

      goto end;

×

2791

end:

×

2792

  tDeleteMqBatchMetaRsp(&rsp);

×

2793

  RAW_LOG_END

×

2794

  return code;

×

taosdata / TDengine / #4986

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

taosdata / TDengine / #4986

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