From 20e2010f709cab26d7262863449fec139af99375 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Thu, 23 Mar 2023 13:35:17 +0800 Subject: [PATCH 1/6] fix:memory leak --- source/client/src/clientRawBlockWrite.c | 65 ++++++++++++++----------- 1 file changed, 36 insertions(+), 29 deletions(-) diff --git a/source/client/src/clientRawBlockWrite.c b/source/client/src/clientRawBlockWrite.c index 6bd8b01842..b1299c16e8 100644 --- a/source/client/src/clientRawBlockWrite.c +++ b/source/client/src/clientRawBlockWrite.c @@ -183,7 +183,7 @@ static char* buildAlterSTableJson(void* alterData, int32_t alterDataLen) { } string = cJSON_PrintUnformatted(json); -end: + end: cJSON_Delete(json); tFreeSMAltertbReq(&req); return string; @@ -205,7 +205,7 @@ static char* processCreateStb(SMqMetaRsp* metaRsp) { } string = buildCreateTableJson(&req.schemaRow, &req.schemaTag, req.name, req.suid, TSDB_SUPER_TABLE); uDebug("processCreateStb %s", string); -_err: + _err: tDecoderClear(&coder); return string; } @@ -227,7 +227,7 @@ static char* processAlterStb(SMqMetaRsp* metaRsp) { string = buildAlterSTableJson(req.alterOriData, req.alterOriDataLen); uDebug("processAlterStb %s", string); -_err: + _err: tDecoderClear(&coder); return string; } @@ -309,7 +309,7 @@ static void buildChildElement(cJSON* json, SVCreateTbReq* pCreateReq) { cJSON_AddItemToArray(tags, tag); } -end: + end: cJSON_AddItemToObject(json, "tags", tags); taosArrayDestroy(pTagVals); } @@ -368,7 +368,7 @@ static char* processCreateTable(SMqMetaRsp* metaRsp) { uDebug("processCreateTable :%s", string); } -_exit: + _exit: for (int32_t iReq = 0; iReq < req.nReqs; iReq++) { pCreateReq = req.pReqs + iReq; taosMemoryFreeClear(pCreateReq->comment); @@ -408,7 +408,7 @@ static char* processAutoCreateTable(STaosxRsp* rsp) { } string = buildCreateCTableJson(pCreateReq, rsp->createTableNum); uDebug("processAutoCreateTable :%s", string); -_exit: + _exit: for (int i = 0; i < rsp->createTableNum; i++) { tDecoderClear(&decoder[i]); taosMemoryFreeClear(pCreateReq[i].comment); @@ -535,7 +535,7 @@ static char* processAlterTable(SMqMetaRsp* metaRsp) { string = cJSON_PrintUnformatted(json); uDebug("processAlterTable :%s", string); -_exit: + _exit: cJSON_Delete(json); tDecoderClear(&decoder); return string; @@ -569,7 +569,7 @@ static char* processDropSTable(SMqMetaRsp* metaRsp) { string = cJSON_PrintUnformatted(json); uDebug("processDropSTable :%s", string); -_exit: + _exit: cJSON_Delete(json); tDecoderClear(&decoder); return string; @@ -609,7 +609,7 @@ static char* processDeleteTable(SMqMetaRsp* metaRsp) { string = cJSON_PrintUnformatted(json); uDebug("processDeleteTable :%s", string); -_exit: + _exit: cJSON_Delete(json); tDecoderClear(&coder); return string; @@ -652,7 +652,7 @@ static char* processDropTable(SMqMetaRsp* metaRsp) { string = cJSON_PrintUnformatted(json); uDebug("processDropTable :%s", string); -_exit: + _exit: cJSON_Delete(json); tDecoderClear(&decoder); return string; @@ -742,7 +742,7 @@ static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) { code = pRequest->code; taosMemoryFree(pCmdMsg.pMsg); -end: + end: destroyRequest(pRequest); tFreeSMCreateStbReq(&pReq); tDecoderClear(&coder); @@ -839,7 +839,7 @@ static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) { code = pRequest->code; taosMemoryFree(pCmdMsg.pMsg); -end: + end: destroyRequest(pRequest); tDecoderClear(&coder); return code; @@ -901,9 +901,9 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) { taosHashSetFreeFp(pVgroupHashmap, destroyCreateTbReqBatch); SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter, - .requestId = pRequest->requestId, - .requestObjRefId = pRequest->self, - .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; + .requestId = pRequest->requestId, + .requestObjRefId = pRequest->self, + .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; pRequest->tableList = taosArrayInit(req.nReqs, sizeof(SName)); // loop to create table @@ -987,7 +987,7 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) { code = pRequest->code; -end: + end: for (int32_t iReq = 0; iReq < req.nReqs; iReq++) { pCreateReq = req.pReqs + iReq; taosMemoryFreeClear(pCreateReq->comment); @@ -1058,9 +1058,9 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) { taosHashSetFreeFp(pVgroupHashmap, destroyDropTbReqBatch); SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter, - .requestId = pRequest->requestId, - .requestObjRefId = pRequest->self, - .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; + .requestId = pRequest->requestId, + .requestObjRefId = pRequest->self, + .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; pRequest->tableList = taosArrayInit(req.nReqs, sizeof(SName)); // loop to create table for (int32_t iReq = 0; iReq < req.nReqs; iReq++) { @@ -1132,7 +1132,7 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) { } code = pRequest->code; -end: + end: taosHashCleanup(pVgroupHashmap); destroyRequest(pRequest); tDecoderClear(&coder); @@ -1201,7 +1201,7 @@ static int32_t taosDeleteData(TAOS* taos, void* meta, int32_t metaLen) { } taos_free_result(res); -end: + end: tDecoderClear(&coder); return code; } @@ -1249,9 +1249,9 @@ static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) { } SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter, - .requestId = pRequest->requestId, - .requestObjRefId = pRequest->self, - .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; + .requestId = pRequest->requestId, + .requestObjRefId = pRequest->self, + .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; SVgroupInfo pInfo = {0}; SName pName = {0}; @@ -1311,7 +1311,7 @@ static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) { code = handleAlterTbExecRes(pRes->res, pCatalog); } } -end: + end: taosArrayDestroy(pArray); if (pVgData) taosMemoryFreeClear(pVgData->pData); taosMemoryFreeClear(pVgData); @@ -1399,7 +1399,7 @@ int taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const ch launchQueryImpl(pRequest, pQuery, true, NULL); code = pRequest->code; -end: + end: taosMemoryFreeClear(pTableMeta); qDestroyQuery(pQuery); destroyRequest(pRequest); @@ -1481,7 +1481,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) launchQueryImpl(pRequest, pQuery, true, NULL); code = pRequest->code; -end: + end: taosMemoryFreeClear(pTableMeta); qDestroyQuery(pQuery); destroyRequest(pRequest); @@ -1601,6 +1601,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) { uError("WriteRaw:rawBlockBindData failed"); goto end; } + taosMemoryFreeClear(pTableMeta); } code = smlBuildOutput(pQuery, pVgHash); @@ -1612,7 +1613,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) { launchQueryImpl(pRequest, pQuery, true, NULL); code = pRequest->code; -end: + end: tDeleteSMqDataRsp(&rspObj.rsp); tDecoderClear(&decoder); qDestroyQuery(pQuery); @@ -1707,6 +1708,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) tDecoderInit(&decoderTmp, *dataTmp, *lenTmp); if (tDecodeSVCreateTbReq(&decoderTmp, &pCreateReq) < 0) { tDecoderClear(&decoderTmp); + tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE); uError("WriteRaw: tDecodeSVCreateTbReq error"); code = TSDB_CODE_TMQ_INVALID_MSG; goto end; @@ -1715,15 +1717,19 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) if (pCreateReq.type != TSDB_CHILD_TABLE) { uError("WriteRaw:pCreateReq.type != TSDB_CHILD_TABLE. table name: %s", tbName); code = TSDB_CODE_TSC_INVALID_VALUE; + tDecoderClear(&decoderTmp); + tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE); goto end; } if (strcmp(tbName, pCreateReq.name) == 0) { cloneSVreateTbReq(&pCreateReq, &pCreateReqDst); // pCreateReqDst->ctb.suid = processSuid(pCreateReqDst->ctb.suid, pRequest->pDb); tDecoderClear(&decoderTmp); + tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE); break; } tDecoderClear(&decoderTmp); + tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE); } SVgroupInfo vg; @@ -1774,6 +1780,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) goto end; } pCreateReqDst = NULL; + taosMemoryFreeClear(pTableMeta); } code = smlBuildOutput(pQuery, pVgHash); @@ -1785,7 +1792,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) launchQueryImpl(pRequest, pQuery, true, NULL); code = pRequest->code; -end: + end: tDeleteSTaosxRsp(&rspObj.rsp); tDecoderClear(&decoder); qDestroyQuery(pQuery); From bf78f63e22e3e3863d1bfba07dafe64522ee37bb Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Fri, 24 Mar 2023 10:10:52 +0800 Subject: [PATCH 2/6] fix:send data batch if consume wal where subscribe db --- source/dnode/vnode/src/inc/tq.h | 3 +- source/dnode/vnode/src/tq/tq.c | 56 ++++++++++++++---------------- source/dnode/vnode/src/tq/tqExec.c | 27 ++------------ 3 files changed, 30 insertions(+), 56 deletions(-) diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h index c2b38f5cd1..bf390c1632 100644 --- a/source/dnode/vnode/src/inc/tq.h +++ b/source/dnode/vnode/src/inc/tq.h @@ -144,8 +144,7 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs int32_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum, uint64_t reqId); // tqExec -int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxRsp* pRsp); -// int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp* pRsp); +int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxRsp* pRsp, int32_t* totalRows); int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision); int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp, int32_t type); int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index f48b57b74b..eb38de1c9a 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -596,8 +596,6 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* } if (offset.type == TMQ_OFFSET__LOG) { - -// if (offset.type == TMQ_OFFSET__LOG) { int64_t fetchVer = offset.version + 1; pCkHead = taosMemoryMalloc(sizeof(SWalCkHead) + 2048); if (pCkHead == NULL) { @@ -605,9 +603,8 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* terrno = TSDB_CODE_OUT_OF_MEMORY; return -1; } - walSetReaderCapacity(pHandle->pWalReader, 2048); - + int totalRows = 0; while (1) { // todo refactor: this is not correct. int32_t savedEpoch = atomic_load_32(&pHandle->epoch); @@ -630,32 +627,8 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d) iter log, vgId:%d offset %" PRId64 " msgType %d", consumerId, pRequest->epoch, vgId, fetchVer, pHead->msgType); - if (pHead->msgType == TDMT_VND_SUBMIT) { - SPackedData submit = { - .msgStr = POINTER_SHIFT(pHead->body, sizeof(SSubmitReq2Msg)), - .msgLen = pHead->bodyLen - sizeof(SSubmitReq2Msg), - .ver = pHead->version, - }; - - if (tqTaosxScanLog(pTq, pHandle, submit, &taosxRsp) < 0) { - tqError("tmq poll: tqTaosxScanLog error %" PRId64 ", in vgId:%d, subkey %s", consumerId, vgId, - pRequest->subKey); - return -1; - } - - if (taosxRsp.blockNum > 0) { - tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer); - code = tqSendDataRsp(pTq, pMsg, pRequest, (SMqDataRsp*)&taosxRsp, TMQ_MSG_TYPE__TAOSX_RSP); - tDeleteSTaosxRsp(&taosxRsp); - taosMemoryFreeClear(pCkHead); - return code; - } else { - fetchVer++; - } - - } else { - /*A(pHandle->fetchMeta);*/ - /*A(IS_META_MSG(pHead->msgType));*/ + // process meta + if (pHead->msgType != TDMT_VND_SUBMIT) { tqDebug("fetch meta msg, ver:%" PRId64 ", type:%s", pHead->version, TMSG_INFO(pHead->msgType)); tqOffsetResetToLog(&metaRsp.rspOffset, fetchVer); metaRsp.resMsgType = pHead->msgType; @@ -672,6 +645,29 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* tDeleteSTaosxRsp(&taosxRsp); return code; } + + // process data + SPackedData submit = { + .msgStr = POINTER_SHIFT(pHead->body, sizeof(SSubmitReq2Msg)), + .msgLen = pHead->bodyLen - sizeof(SSubmitReq2Msg), + .ver = pHead->version, + }; + + if (tqTaosxScanLog(pTq, pHandle, submit, &taosxRsp, &totalRows) < 0) { + tqError("tmq poll: tqTaosxScanLog error %" PRId64 ", in vgId:%d, subkey %s", consumerId, vgId, + pRequest->subKey); + return -1; + } + + if (totalRows >= 4096 || taosxRsp.createTableNum > 0) { + tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer); + code = tqSendDataRsp(pTq, pMsg, pRequest, (SMqDataRsp*)&taosxRsp, TMQ_MSG_TYPE__TAOSX_RSP); + tDeleteSTaosxRsp(&taosxRsp); + taosMemoryFreeClear(pCkHead); + return code; + } else { + fetchVer++; + } } } diff --git a/source/dnode/vnode/src/tq/tqExec.c b/source/dnode/vnode/src/tq/tqExec.c index a62101eb47..796f150cbc 100644 --- a/source/dnode/vnode/src/tq/tqExec.c +++ b/source/dnode/vnode/src/tq/tqExec.c @@ -230,23 +230,15 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta return 0; } -int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxRsp* pRsp) { +int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxRsp* pRsp, int32_t* totalRows) { STqExecHandle* pExec = &pHandle->execHandle; - /*A(pExec->subType != TOPIC_SUB_TYPE__COLUMN);*/ - SArray* pBlocks = taosArrayInit(0, sizeof(SSDataBlock)); SArray* pSchemas = taosArrayInit(0, sizeof(void*)); if (pExec->subType == TOPIC_SUB_TYPE__TABLE) { STqReader* pReader = pExec->pExecReader; - /*tqReaderSetDataMsg(pReader, pReq, 0);*/ tqReaderSetSubmitReq2(pReader, submit.msgStr, submit.msgLen, submit.ver); while (tqNextDataBlock2(pReader)) { - /*SSDataBlock block = {0};*/ - /*if (tqRetrieveDataBlock(&block, pReader) < 0) {*/ - /*if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue;*/ - /*}*/ - taosArrayClear(pBlocks); taosArrayClear(pSchemas); SSubmitTbData* pSubmitTbDataRet = NULL; @@ -254,7 +246,6 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue; } if (pRsp->withTbName) { - /*int64_t uid = pExec->pExecReader->msgIter.uid;*/ int64_t uid = pExec->pExecReader->lastBlkUid; if (tqAddTbNameToRsp(pTq, uid, pRsp, taosArrayGetSize(pBlocks)) < 0) { taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes); @@ -296,6 +287,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR SSDataBlock* pBlock = taosArrayGet(pBlocks, i); tqAddBlockDataToRsp(pBlock, (SMqDataRsp*)pRsp, taosArrayGetSize(pBlock->pDataBlock), pTq->pVnode->config.tsdbCfg.precision); + totalRows += pBlock->info.rows; blockDataFreeRes(pBlock); SSchemaWrapper* pSW = taosArrayGetP(pSchemas, i); taosArrayPush(pRsp->blockSchema, &pSW); @@ -304,13 +296,8 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR } } else if (pExec->subType == TOPIC_SUB_TYPE__DB) { STqReader* pReader = pExec->pExecReader; - /*tqReaderSetDataMsg(pReader, pReq, 0);*/ tqReaderSetSubmitReq2(pReader, submit.msgStr, submit.msgLen, submit.ver); while (tqNextDataBlockFilterOut2(pReader, pExec->execDb.pFilterOutTbUid)) { - /*SSDataBlock block = {0};*/ - /*if (tqRetrieveDataBlock(&block, pReader) < 0) {*/ - /*if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue;*/ - /*}*/ taosArrayClear(pBlocks); taosArrayClear(pSchemas); SSubmitTbData* pSubmitTbDataRet = NULL; @@ -355,15 +342,11 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR tEncoderClear(&encoder); } - /*tqAddBlockDataToRsp(&block, (SMqDataRsp*)pRsp, taosArrayGetSize(block.pDataBlock),*/ - /*pTq->pVnode->config.tsdbCfg.precision);*/ - /*blockDataFreeRes(&block);*/ - /*tqAddBlockSchemaToRsp(pExec, (SMqDataRsp*)pRsp);*/ - /*pRsp->blockNum++;*/ for (int32_t i = 0; i < taosArrayGetSize(pBlocks); i++) { SSDataBlock* pBlock = taosArrayGet(pBlocks, i); tqAddBlockDataToRsp(pBlock, (SMqDataRsp*)pRsp, taosArrayGetSize(pBlock->pDataBlock), pTq->pVnode->config.tsdbCfg.precision); + totalRows += pBlock->info.rows; blockDataFreeRes(pBlock); SSchemaWrapper* pSW = taosArrayGetP(pSchemas, i); taosArrayPush(pRsp->blockSchema, &pSW); @@ -373,9 +356,5 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR } taosArrayDestroy(pBlocks); taosArrayDestroy(pSchemas); -// if (pRsp->blockNum == 0) { -// return -1; -// } - return 0; } From dcaab41a27b1a6b40921dd35250f573a4473c690 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Fri, 24 Mar 2023 13:20:03 +0800 Subject: [PATCH 3/6] fix:error in optimize consume logic --- source/libs/parser/src/parInsertUtil.c | 68 ++- utils/test/c/tmq_taosx_ci.c | 556 ++++++++++++------------- 2 files changed, 336 insertions(+), 288 deletions(-) diff --git a/source/libs/parser/src/parInsertUtil.c b/source/libs/parser/src/parInsertUtil.c index 704f381afa..c8e4d8e6b6 100644 --- a/source/libs/parser/src/parInsertUtil.c +++ b/source/libs/parser/src/parInsertUtil.c @@ -611,8 +611,35 @@ static int bindFileds(SBoundColInfo* pBoundInfo, SSchema* pSchema, TAOS_FIELD* f return code; } +static bool isSameBindFileds(SBoundColInfo* pBoundInfo, SSchema* pSchema, TAOS_FIELD* fields, int numFields) { + int16_t lastColIdx = -1; // last column found + for (int i = 0; i < numFields; i++) { + SToken token; + token.z = fields[i].name; + token.n = strlen(fields[i].name); + + int16_t t = lastColIdx + 1; + int16_t index = insFindCol(&token, t, pBoundInfo->numOfCols, pSchema); + if (index < 0 && t > 0) { + index = insFindCol(&token, 0, t, pSchema); + } + if (index < 0) { + uError("can not find column name:%s", token.z); + return false; + } else { + lastColIdx = index; + if(pBoundInfo->pColIndex[i] != index){ + return false; + } + } + } + + return true; +} + int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreateTbReq* pCreateTb, TAOS_FIELD* tFields, int numFields, bool needChangeLength) { + void* tmp = taosHashGet(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid, sizeof(pTableMeta->uid)); STableDataCxt* pTableCxt = NULL; int ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid, sizeof(pTableMeta->uid), pTableMeta, &pCreateTb, &pTableCxt, true); @@ -620,19 +647,40 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate uError("insGetTableDataCxt error"); goto end; } - if (tFields != NULL) { - ret = bindFileds(&pTableCxt->boundColsInfo, getTableColumnSchema(pTableMeta), tFields, numFields); + + do { + if (tmp != NULL){ + if(!isSameBindFileds(&pTableCxt->boundColsInfo, getTableColumnSchema(pTableMeta), tFields, numFields)){ + char* fieldNames = (char*)taosMemoryCalloc(numFields, sizeof(tFields[0].name)); + for(int i = 0; i < numFields; i++){ + memcpy(fieldNames + i * sizeof(tFields[0].name), tFields[i].name, sizeof(tFields[0].name)); + } + ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, fieldNames, + numFields * sizeof(tFields[0].name), pTableMeta, &pCreateTb, &pTableCxt, true); + taosMemoryFree(fieldNames); + if (ret != TSDB_CODE_SUCCESS) { + uError("insGetTableDataCxt inner error"); + goto end; + } + }else{ + break; + } + } + + if (tFields != NULL) { + ret = bindFileds(&pTableCxt->boundColsInfo, getTableColumnSchema(pTableMeta), tFields, numFields); + if (ret != TSDB_CODE_SUCCESS) { + uError("bindFileds error"); + goto end; + } + } + // no need to bind, because select * get all fields + ret = initTableColSubmitData(pTableCxt); if (ret != TSDB_CODE_SUCCESS) { - uError("bindFileds error"); + uError("initTableColSubmitData error"); goto end; } - } - // no need to bind, because select * get all fields - ret = initTableColSubmitData(pTableCxt); - if (ret != TSDB_CODE_SUCCESS) { - uError("initTableColSubmitData error"); - goto end; - } + }while(0); char* p = (char*)data; // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column diff --git a/utils/test/c/tmq_taosx_ci.c b/utils/test/c/tmq_taosx_ci.c index 1f25eae366..a3df1106fd 100644 --- a/utils/test/c/tmq_taosx_ci.c +++ b/utils/test/c/tmq_taosx_ci.c @@ -102,284 +102,284 @@ int buildDatabase(TAOS* pConn, TAOS_RES* pRes) { /* test for TD-20612 end*/ - pRes = taos_query(pConn, - "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " - "nchar(8), t4 bool)"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table if not exists ct0 using st1 tags(1000, \"ttt\", true)"); - if (taos_errno(pRes) != 0) { - printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into ct0 values(1626006833400, 1, 2, 'a')"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ct0, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table if not exists ct1 using st1(t1) tags(2000)"); - if (taos_errno(pRes) != 0) { - printf("failed to create child table ct1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table if not exists ct2 using st1(t1) tags(NULL)"); - if (taos_errno(pRes) != 0) { - printf("failed to create child table ct2, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into ct1 values(1626006833600, 3, 4, 'b')"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ct1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table if not exists ct3 using st1(t1) tags(3000)"); - if (taos_errno(pRes) != 0) { - printf("failed to create child table ct3, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query( - pConn, - "insert into ct3 values(1626006833600, 5, 6, 'c') ct1 values(1626006833601, 2, 3, 'sds') (1626006833602, 4, 5, " - "'ddd') ct0 values(1626006833603, 4, 3, 'hwj') ct1 values(now+5s, 23, 32, 's21ds')"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table st1 add column c4 bigint"); - if (taos_errno(pRes) != 0) { - printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table st1 modify column c3 binary(64)"); - if (taos_errno(pRes) != 0) { - printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, - "insert into ct3 values(1626006833605, 53, 63, 'cffffffffffffffffffffffffffff', 8989898899999) " - "(1626006833609, 51, 62, 'c333', 940)"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into ct3 select * from ct1"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table st1 add tag t2 binary(64)"); - if (taos_errno(pRes) != 0) { - printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table ct3 set tag t1=5000"); - if (taos_errno(pRes) != 0) { - printf("failed to slter child table ct3, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "delete from abc1 .ct3 where ts < 1626006833606"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - if (g_conf.dropTable) { - pRes = taos_query(pConn, "drop table ct3, ct1"); - if (taos_errno(pRes) != 0) { - printf("failed to drop child table ct3, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "drop table st1"); - if (taos_errno(pRes) != 0) { - printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - } - - pRes = taos_query(pConn, "create table if not exists n1(ts timestamp, c1 int, c2 nchar(4))"); - if (taos_errno(pRes) != 0) { - printf("failed to create normal table n1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table n1 add column c3 bigint"); - if (taos_errno(pRes) != 0) { - printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table n1 modify column c2 nchar(8)"); - if (taos_errno(pRes) != 0) { - printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table n1 rename column c3 cc3"); - if (taos_errno(pRes) != 0) { - printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table n1 comment 'hello'"); - if (taos_errno(pRes) != 0) { - printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "alter table n1 drop column c1"); - if (taos_errno(pRes) != 0) { - printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into n1 values(now, 'eeee', 8989898899999) (now+9s, 'c333', 940)"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into n1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - if (g_conf.dropTable) { - pRes = taos_query(pConn, "drop table n1"); - if (taos_errno(pRes) != 0) { - printf("failed to drop normal table n1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - } - - pRes = taos_query(pConn, "create table jt(ts timestamp, i int) tags(t json)"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table jt, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table jt1 using jt tags('{\"k1\":1, \"k2\":\"hello\"}')"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table jt, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table jt2 using jt tags('')"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into jt1 values(now, 1)"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table jt1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into jt2 values(now, 11)"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - if (g_conf.dropTable) { - pRes = taos_query(pConn, - "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " - "nchar(8), t4 bool)"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "drop table st1"); - if (taos_errno(pRes) != 0) { - printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - } - - pRes = taos_query(pConn, - "create stable if not exists stt (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " - "nchar(8), t4 bool)"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table stt, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, - "create stable if not exists sttb (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " - "nchar(8), t4 bool)"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table sttb, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query( - pConn, - "create table if not exists stt1 using stt tags(2, \"stt1\", true) sttb1 using sttb tags(4, \"sttb1\", true) " - "stt2 using stt tags(43, \"stt2\", false) sttb2 using sttb tags(54, \"sttb2\", true)"); - if (taos_errno(pRes) != 0) { - printf("failed to create child table stt1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = - taos_query(pConn, - "insert into stt1 values(now + 2s, 3, 2, 'stt1') stt3 using stt tags(23, \"stt3\", true) values(now + " - "1s, 1, 2, 'stt3') sttb3 using sttb tags(4, \"sttb3\", true) values(now + 2s, 13, 22, 'sttb3') " - "stt4 using stt tags(433, \"stt4\", false) values(now + 3s, 21, 21, 'stt4') sttb4 using sttb " - "tags(543, \"sttb4\", true) values(now + 4s, 16, 25, 'sttb4')"); - if (taos_errno(pRes) != 0) { - printf("failed to create child table stt1, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); +// pRes = taos_query(pConn, +// "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " +// "nchar(8), t4 bool)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "create table if not exists ct0 using st1 tags(1000, \"ttt\", true)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "insert into ct0 values(1626006833400, 1, 2, 'a')"); +// if (taos_errno(pRes) != 0) { +// printf("failed to insert into ct0, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "create table if not exists ct1 using st1(t1) tags(2000)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create child table ct1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "create table if not exists ct2 using st1(t1) tags(NULL)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create child table ct2, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "insert into ct1 values(1626006833600, 3, 4, 'b')"); +// if (taos_errno(pRes) != 0) { +// printf("failed to insert into ct1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "create table if not exists ct3 using st1(t1) tags(3000)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create child table ct3, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query( +// pConn, +// "insert into ct3 values(1626006833600, 5, 6, 'c') ct1 values(1626006833601, 2, 3, 'sds') (1626006833602, 4, 5, " +// "'ddd') ct0 values(1626006833603, 4, 3, 'hwj') ct1 values(now+5s, 23, 32, 's21ds')"); +// if (taos_errno(pRes) != 0) { +// printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table st1 add column c4 bigint"); +// if (taos_errno(pRes) != 0) { +// printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table st1 modify column c3 binary(64)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, +// "insert into ct3 values(1626006833605, 53, 63, 'cffffffffffffffffffffffffffff', 8989898899999) " +// "(1626006833609, 51, 62, 'c333', 940)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "insert into ct3 select * from ct1"); +// if (taos_errno(pRes) != 0) { +// printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table st1 add tag t2 binary(64)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table ct3 set tag t1=5000"); +// if (taos_errno(pRes) != 0) { +// printf("failed to slter child table ct3, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "delete from abc1 .ct3 where ts < 1626006833606"); +// if (taos_errno(pRes) != 0) { +// printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// if (g_conf.dropTable) { +// pRes = taos_query(pConn, "drop table ct3, ct1"); +// if (taos_errno(pRes) != 0) { +// printf("failed to drop child table ct3, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "drop table st1"); +// if (taos_errno(pRes) != 0) { +// printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// } +// +// pRes = taos_query(pConn, "create table if not exists n1(ts timestamp, c1 int, c2 nchar(4))"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create normal table n1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table n1 add column c3 bigint"); +// if (taos_errno(pRes) != 0) { +// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table n1 modify column c2 nchar(8)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table n1 rename column c3 cc3"); +// if (taos_errno(pRes) != 0) { +// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table n1 comment 'hello'"); +// if (taos_errno(pRes) != 0) { +// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "alter table n1 drop column c1"); +// if (taos_errno(pRes) != 0) { +// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "insert into n1 values(now, 'eeee', 8989898899999) (now+9s, 'c333', 940)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to insert into n1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// if (g_conf.dropTable) { +// pRes = taos_query(pConn, "drop table n1"); +// if (taos_errno(pRes) != 0) { +// printf("failed to drop normal table n1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// } +// +// pRes = taos_query(pConn, "create table jt(ts timestamp, i int) tags(t json)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table jt, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "create table jt1 using jt tags('{\"k1\":1, \"k2\":\"hello\"}')"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table jt, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "create table jt2 using jt tags('')"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "insert into jt1 values(now, 1)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table jt1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "insert into jt2 values(now, 11)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// if (g_conf.dropTable) { +// pRes = taos_query(pConn, +// "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " +// "nchar(8), t4 bool)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, "drop table st1"); +// if (taos_errno(pRes) != 0) { +// printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// } +// +// pRes = taos_query(pConn, +// "create stable if not exists stt (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " +// "nchar(8), t4 bool)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table stt, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query(pConn, +// "create stable if not exists sttb (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " +// "nchar(8), t4 bool)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table sttb, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = taos_query( +// pConn, +// "create table if not exists stt1 using stt tags(2, \"stt1\", true) sttb1 using sttb tags(4, \"sttb1\", true) " +// "stt2 using stt tags(43, \"stt2\", false) sttb2 using sttb tags(54, \"sttb2\", true)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create child table stt1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// pRes = +// taos_query(pConn, +// "insert into stt1 values(now + 2s, 3, 2, 'stt1') stt3 using stt tags(23, \"stt3\", true) values(now + " +// "1s, 1, 2, 'stt3') sttb3 using sttb tags(4, \"sttb3\", true) values(now + 2s, 13, 22, 'sttb3') " +// "stt4 using stt tags(433, \"stt4\", false) values(now + 3s, 21, 21, 'stt4') sttb4 using sttb " +// "tags(543, \"sttb4\", true) values(now + 4s, 16, 25, 'sttb4')"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create child table stt1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); return 0; } From 5d06474f2ed3b6f06115c96e6451a394702ad3d7 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Fri, 24 Mar 2023 14:43:29 +0800 Subject: [PATCH 4/6] fix:send data batch if consume wal where subscribe db --- source/common/src/tdataformat.c | 4 + source/dnode/vnode/src/tq/tq.c | 10 + source/dnode/vnode/src/tq/tqExec.c | 2 +- source/libs/parser/src/parInsertUtil.c | 146 ++----- utils/test/c/tmq_taosx_ci.c | 556 ++++++++++++------------- 5 files changed, 324 insertions(+), 394 deletions(-) diff --git a/source/common/src/tdataformat.c b/source/common/src/tdataformat.c index 2bb8708372..5c313b3b3b 100644 --- a/source/common/src/tdataformat.c +++ b/source/common/src/tdataformat.c @@ -2439,6 +2439,10 @@ _exit: int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes, int32_t nRows, char *lengthOrbitmap, char *data) { int32_t code = 0; + if(data == NULL){ + code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NONE](pColData, NULL, 0); + goto _exit; + } if (IS_VAR_DATA_TYPE(type)) { // var-length data type for (int32_t i = 0; i < nRows; ++i) { diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index eb38de1c9a..437870ebd6 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -629,6 +629,14 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* // process meta if (pHead->msgType != TDMT_VND_SUBMIT) { + if(totalRows > 0) { + tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer - 1); + code = tqSendDataRsp(pTq, pMsg, pRequest, (SMqDataRsp*)&taosxRsp, TMQ_MSG_TYPE__TAOSX_RSP); + tDeleteSTaosxRsp(&taosxRsp); + taosMemoryFreeClear(pCkHead); + return code; + } + tqDebug("fetch meta msg, ver:%" PRId64 ", type:%s", pHead->version, TMSG_INFO(pHead->msgType)); tqOffsetResetToLog(&metaRsp.rspOffset, fetchVer); metaRsp.resMsgType = pHead->msgType; @@ -656,6 +664,8 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* if (tqTaosxScanLog(pTq, pHandle, submit, &taosxRsp, &totalRows) < 0) { tqError("tmq poll: tqTaosxScanLog error %" PRId64 ", in vgId:%d, subkey %s", consumerId, vgId, pRequest->subKey); + taosMemoryFreeClear(pCkHead); + tDeleteSTaosxRsp(&taosxRsp); return -1; } diff --git a/source/dnode/vnode/src/tq/tqExec.c b/source/dnode/vnode/src/tq/tqExec.c index 796f150cbc..cbcd918094 100644 --- a/source/dnode/vnode/src/tq/tqExec.c +++ b/source/dnode/vnode/src/tq/tqExec.c @@ -346,7 +346,7 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR SSDataBlock* pBlock = taosArrayGet(pBlocks, i); tqAddBlockDataToRsp(pBlock, (SMqDataRsp*)pRsp, taosArrayGetSize(pBlock->pDataBlock), pTq->pVnode->config.tsdbCfg.precision); - totalRows += pBlock->info.rows; + *totalRows += pBlock->info.rows; blockDataFreeRes(pBlock); SSchemaWrapper* pSW = taosArrayGetP(pSchemas, i); taosArrayPush(pRsp->blockSchema, &pSW); diff --git a/source/libs/parser/src/parInsertUtil.c b/source/libs/parser/src/parInsertUtil.c index c8e4d8e6b6..f96c81cf85 100644 --- a/source/libs/parser/src/parInsertUtil.c +++ b/source/libs/parser/src/parInsertUtil.c @@ -566,75 +566,14 @@ int32_t insBuildVgDataBlocks(SHashObj* pVgroupsHashObj, SArray* pVgDataCxtList, return code; } -static int bindFileds(SBoundColInfo* pBoundInfo, SSchema* pSchema, TAOS_FIELD* fields, int numFields) { - bool* pUseCols = taosMemoryCalloc(pBoundInfo->numOfCols, sizeof(bool)); - if (NULL == pUseCols) { - return TSDB_CODE_OUT_OF_MEMORY; - } - - pBoundInfo->numOfBound = 0; - - int16_t lastColIdx = -1; // last column found - int32_t code = TSDB_CODE_SUCCESS; +static bool findFileds(SSchema* pSchema, TAOS_FIELD* fields, int numFields) { for (int i = 0; i < numFields; i++) { - SToken token; - token.z = fields[i].name; - token.n = strlen(fields[i].name); - - int16_t t = lastColIdx + 1; - int16_t index = insFindCol(&token, t, pBoundInfo->numOfCols, pSchema); - if (index < 0 && t > 0) { - index = insFindCol(&token, 0, t, pSchema); - } - if (index < 0) { - uError("can not find column name:%s", token.z); - code = TSDB_CODE_PAR_INVALID_COLUMN; - break; - } else if (pUseCols[index]) { - code = TSDB_CODE_PAR_INVALID_COLUMN; - uError("duplicated column name:%s", token.z); - break; - } else { - lastColIdx = index; - pUseCols[index] = true; - pBoundInfo->pColIndex[pBoundInfo->numOfBound] = index; - ++pBoundInfo->numOfBound; + if(strcmp(pSchema->name, fields[i].name) == 0){ + return true; } } - if (TSDB_CODE_SUCCESS == code && !pUseCols[0]) { - uError("primary timestamp column can not be null:"); - code = TSDB_CODE_PAR_INVALID_COLUMN; - } - - taosMemoryFree(pUseCols); - return code; -} - -static bool isSameBindFileds(SBoundColInfo* pBoundInfo, SSchema* pSchema, TAOS_FIELD* fields, int numFields) { - int16_t lastColIdx = -1; // last column found - for (int i = 0; i < numFields; i++) { - SToken token; - token.z = fields[i].name; - token.n = strlen(fields[i].name); - - int16_t t = lastColIdx + 1; - int16_t index = insFindCol(&token, t, pBoundInfo->numOfCols, pSchema); - if (index < 0 && t > 0) { - index = insFindCol(&token, 0, t, pSchema); - } - if (index < 0) { - uError("can not find column name:%s", token.z); - return false; - } else { - lastColIdx = index; - if(pBoundInfo->pColIndex[i] != index){ - return false; - } - } - } - - return true; + return false; } int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreateTbReq* pCreateTb, TAOS_FIELD* tFields, @@ -648,39 +587,13 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate goto end; } - do { - if (tmp != NULL){ - if(!isSameBindFileds(&pTableCxt->boundColsInfo, getTableColumnSchema(pTableMeta), tFields, numFields)){ - char* fieldNames = (char*)taosMemoryCalloc(numFields, sizeof(tFields[0].name)); - for(int i = 0; i < numFields; i++){ - memcpy(fieldNames + i * sizeof(tFields[0].name), tFields[i].name, sizeof(tFields[0].name)); - } - ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, fieldNames, - numFields * sizeof(tFields[0].name), pTableMeta, &pCreateTb, &pTableCxt, true); - taosMemoryFree(fieldNames); - if (ret != TSDB_CODE_SUCCESS) { - uError("insGetTableDataCxt inner error"); - goto end; - } - }else{ - break; - } - } - - if (tFields != NULL) { - ret = bindFileds(&pTableCxt->boundColsInfo, getTableColumnSchema(pTableMeta), tFields, numFields); - if (ret != TSDB_CODE_SUCCESS) { - uError("bindFileds error"); - goto end; - } - } - // no need to bind, because select * get all fields + if(tmp == NULL){ ret = initTableColSubmitData(pTableCxt); if (ret != TSDB_CODE_SUCCESS) { uError("initTableColSubmitData error"); goto end; } - }while(0); + } char* p = (char*)data; // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column @@ -708,35 +621,38 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate SSchema* pSchema = getTableColumnSchema(pTableCxt->pMeta); SBoundColInfo* boundInfo = &pTableCxt->boundColsInfo; - if (boundInfo->numOfBound != numOfCols) { - uError("boundInfo->numOfBound:%d != numOfCols:%d", boundInfo->numOfBound, numOfCols); + if (numFields != numOfCols) { + uError("numFields:%d != numOfCols:%d", numFields, numOfCols); ret = TSDB_CODE_INVALID_PARA; goto end; } for (int c = 0; c < boundInfo->numOfBound; ++c) { - SSchema* pColSchema = &pSchema[boundInfo->pColIndex[c]]; + SSchema* pColSchema = &pSchema[c]; SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, c); + if(findFileds(pColSchema, tFields, numFields)){ + if (*fields != pColSchema->type && *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) { + uError("type or bytes not equal"); + ret = TSDB_CODE_INVALID_PARA; + goto end; + } - if (*fields != pColSchema->type && *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) { - uError("type or bytes not equal"); - ret = TSDB_CODE_INVALID_PARA; - goto end; - } + int8_t* offset = pStart; + if (IS_VAR_DATA_TYPE(pColSchema->type)) { + pStart += numOfRows * sizeof(int32_t); + } else { + pStart += BitmapLen(numOfRows); + } + char* pData = pStart; - int8_t* offset = pStart; - if (IS_VAR_DATA_TYPE(pColSchema->type)) { - pStart += numOfRows * sizeof(int32_t); - } else { - pStart += BitmapLen(numOfRows); - } - char* pData = pStart; - - tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, offset, pData); - fields += sizeof(int8_t) + sizeof(int32_t); - if (needChangeLength) { - pStart += htonl(colLength[c]); - } else { - pStart += colLength[c]; + tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, offset, pData); + fields += sizeof(int8_t) + sizeof(int32_t); + if (needChangeLength) { + pStart += htonl(colLength[c]); + } else { + pStart += colLength[c]; + } + }else{ + tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, NULL, NULL); } } diff --git a/utils/test/c/tmq_taosx_ci.c b/utils/test/c/tmq_taosx_ci.c index a3df1106fd..1f25eae366 100644 --- a/utils/test/c/tmq_taosx_ci.c +++ b/utils/test/c/tmq_taosx_ci.c @@ -102,284 +102,284 @@ int buildDatabase(TAOS* pConn, TAOS_RES* pRes) { /* test for TD-20612 end*/ -// pRes = taos_query(pConn, -// "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " -// "nchar(8), t4 bool)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "create table if not exists ct0 using st1 tags(1000, \"ttt\", true)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "insert into ct0 values(1626006833400, 1, 2, 'a')"); -// if (taos_errno(pRes) != 0) { -// printf("failed to insert into ct0, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "create table if not exists ct1 using st1(t1) tags(2000)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create child table ct1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "create table if not exists ct2 using st1(t1) tags(NULL)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create child table ct2, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "insert into ct1 values(1626006833600, 3, 4, 'b')"); -// if (taos_errno(pRes) != 0) { -// printf("failed to insert into ct1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "create table if not exists ct3 using st1(t1) tags(3000)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create child table ct3, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query( -// pConn, -// "insert into ct3 values(1626006833600, 5, 6, 'c') ct1 values(1626006833601, 2, 3, 'sds') (1626006833602, 4, 5, " -// "'ddd') ct0 values(1626006833603, 4, 3, 'hwj') ct1 values(now+5s, 23, 32, 's21ds')"); -// if (taos_errno(pRes) != 0) { -// printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table st1 add column c4 bigint"); -// if (taos_errno(pRes) != 0) { -// printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table st1 modify column c3 binary(64)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, -// "insert into ct3 values(1626006833605, 53, 63, 'cffffffffffffffffffffffffffff', 8989898899999) " -// "(1626006833609, 51, 62, 'c333', 940)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "insert into ct3 select * from ct1"); -// if (taos_errno(pRes) != 0) { -// printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table st1 add tag t2 binary(64)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table ct3 set tag t1=5000"); -// if (taos_errno(pRes) != 0) { -// printf("failed to slter child table ct3, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "delete from abc1 .ct3 where ts < 1626006833606"); -// if (taos_errno(pRes) != 0) { -// printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// if (g_conf.dropTable) { -// pRes = taos_query(pConn, "drop table ct3, ct1"); -// if (taos_errno(pRes) != 0) { -// printf("failed to drop child table ct3, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "drop table st1"); -// if (taos_errno(pRes) != 0) { -// printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// } -// -// pRes = taos_query(pConn, "create table if not exists n1(ts timestamp, c1 int, c2 nchar(4))"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create normal table n1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table n1 add column c3 bigint"); -// if (taos_errno(pRes) != 0) { -// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table n1 modify column c2 nchar(8)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table n1 rename column c3 cc3"); -// if (taos_errno(pRes) != 0) { -// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table n1 comment 'hello'"); -// if (taos_errno(pRes) != 0) { -// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "alter table n1 drop column c1"); -// if (taos_errno(pRes) != 0) { -// printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "insert into n1 values(now, 'eeee', 8989898899999) (now+9s, 'c333', 940)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to insert into n1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// if (g_conf.dropTable) { -// pRes = taos_query(pConn, "drop table n1"); -// if (taos_errno(pRes) != 0) { -// printf("failed to drop normal table n1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// } -// -// pRes = taos_query(pConn, "create table jt(ts timestamp, i int) tags(t json)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table jt, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "create table jt1 using jt tags('{\"k1\":1, \"k2\":\"hello\"}')"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table jt, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "create table jt2 using jt tags('')"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "insert into jt1 values(now, 1)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table jt1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "insert into jt2 values(now, 11)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// if (g_conf.dropTable) { -// pRes = taos_query(pConn, -// "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " -// "nchar(8), t4 bool)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, "drop table st1"); -// if (taos_errno(pRes) != 0) { -// printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// } -// -// pRes = taos_query(pConn, -// "create stable if not exists stt (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " -// "nchar(8), t4 bool)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table stt, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query(pConn, -// "create stable if not exists sttb (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " -// "nchar(8), t4 bool)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create super table sttb, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = taos_query( -// pConn, -// "create table if not exists stt1 using stt tags(2, \"stt1\", true) sttb1 using sttb tags(4, \"sttb1\", true) " -// "stt2 using stt tags(43, \"stt2\", false) sttb2 using sttb tags(54, \"sttb2\", true)"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create child table stt1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); -// -// pRes = -// taos_query(pConn, -// "insert into stt1 values(now + 2s, 3, 2, 'stt1') stt3 using stt tags(23, \"stt3\", true) values(now + " -// "1s, 1, 2, 'stt3') sttb3 using sttb tags(4, \"sttb3\", true) values(now + 2s, 13, 22, 'sttb3') " -// "stt4 using stt tags(433, \"stt4\", false) values(now + 3s, 21, 21, 'stt4') sttb4 using sttb " -// "tags(543, \"sttb4\", true) values(now + 4s, 16, 25, 'sttb4')"); -// if (taos_errno(pRes) != 0) { -// printf("failed to create child table stt1, reason:%s\n", taos_errstr(pRes)); -// return -1; -// } -// taos_free_result(pRes); + pRes = taos_query(pConn, + "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " + "nchar(8), t4 bool)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table if not exists ct0 using st1 tags(1000, \"ttt\", true)"); + if (taos_errno(pRes) != 0) { + printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into ct0 values(1626006833400, 1, 2, 'a')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct0, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table if not exists ct1 using st1(t1) tags(2000)"); + if (taos_errno(pRes) != 0) { + printf("failed to create child table ct1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table if not exists ct2 using st1(t1) tags(NULL)"); + if (taos_errno(pRes) != 0) { + printf("failed to create child table ct2, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into ct1 values(1626006833600, 3, 4, 'b')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table if not exists ct3 using st1(t1) tags(3000)"); + if (taos_errno(pRes) != 0) { + printf("failed to create child table ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query( + pConn, + "insert into ct3 values(1626006833600, 5, 6, 'c') ct1 values(1626006833601, 2, 3, 'sds') (1626006833602, 4, 5, " + "'ddd') ct0 values(1626006833603, 4, 3, 'hwj') ct1 values(now+5s, 23, 32, 's21ds')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table st1 add column c4 bigint"); + if (taos_errno(pRes) != 0) { + printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table st1 modify column c3 binary(64)"); + if (taos_errno(pRes) != 0) { + printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, + "insert into ct3 values(1626006833605, 53, 63, 'cffffffffffffffffffffffffffff', 8989898899999) " + "(1626006833609, 51, 62, 'c333', 940)"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into ct3 select * from ct1"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table st1 add tag t2 binary(64)"); + if (taos_errno(pRes) != 0) { + printf("failed to alter super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table ct3 set tag t1=5000"); + if (taos_errno(pRes) != 0) { + printf("failed to slter child table ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "delete from abc1 .ct3 where ts < 1626006833606"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + if (g_conf.dropTable) { + pRes = taos_query(pConn, "drop table ct3, ct1"); + if (taos_errno(pRes) != 0) { + printf("failed to drop child table ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "drop table st1"); + if (taos_errno(pRes) != 0) { + printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + } + + pRes = taos_query(pConn, "create table if not exists n1(ts timestamp, c1 int, c2 nchar(4))"); + if (taos_errno(pRes) != 0) { + printf("failed to create normal table n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table n1 add column c3 bigint"); + if (taos_errno(pRes) != 0) { + printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table n1 modify column c2 nchar(8)"); + if (taos_errno(pRes) != 0) { + printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table n1 rename column c3 cc3"); + if (taos_errno(pRes) != 0) { + printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table n1 comment 'hello'"); + if (taos_errno(pRes) != 0) { + printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "alter table n1 drop column c1"); + if (taos_errno(pRes) != 0) { + printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into n1 values(now, 'eeee', 8989898899999) (now+9s, 'c333', 940)"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + if (g_conf.dropTable) { + pRes = taos_query(pConn, "drop table n1"); + if (taos_errno(pRes) != 0) { + printf("failed to drop normal table n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + } + + pRes = taos_query(pConn, "create table jt(ts timestamp, i int) tags(t json)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table jt, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table jt1 using jt tags('{\"k1\":1, \"k2\":\"hello\"}')"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table jt, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create table jt2 using jt tags('')"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into jt1 values(now, 1)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table jt1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "insert into jt2 values(now, 11)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + if (g_conf.dropTable) { + pRes = taos_query(pConn, + "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " + "nchar(8), t4 bool)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "drop table st1"); + if (taos_errno(pRes) != 0) { + printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + } + + pRes = taos_query(pConn, + "create stable if not exists stt (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " + "nchar(8), t4 bool)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table stt, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, + "create stable if not exists sttb (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " + "nchar(8), t4 bool)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table sttb, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query( + pConn, + "create table if not exists stt1 using stt tags(2, \"stt1\", true) sttb1 using sttb tags(4, \"sttb1\", true) " + "stt2 using stt tags(43, \"stt2\", false) sttb2 using sttb tags(54, \"sttb2\", true)"); + if (taos_errno(pRes) != 0) { + printf("failed to create child table stt1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = + taos_query(pConn, + "insert into stt1 values(now + 2s, 3, 2, 'stt1') stt3 using stt tags(23, \"stt3\", true) values(now + " + "1s, 1, 2, 'stt3') sttb3 using sttb tags(4, \"sttb3\", true) values(now + 2s, 13, 22, 'sttb3') " + "stt4 using stt tags(433, \"stt4\", false) values(now + 3s, 21, 21, 'stt4') sttb4 using sttb " + "tags(543, \"sttb4\", true) values(now + 4s, 16, 25, 'sttb4')"); + if (taos_errno(pRes) != 0) { + printf("failed to create child table stt1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); return 0; } From 8081572ae1a2ea068dc7f749f11faa47e33e44d1 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Sat, 25 Mar 2023 11:31:19 +0800 Subject: [PATCH 5/6] fix:modify ci case if consume too fast --- tests/system-test/7-tmq/tmqDropNtb-snapshot0.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py b/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py index 6c49fae299..8792bdd3ba 100644 --- a/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py +++ b/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py @@ -100,7 +100,7 @@ class TDTestCase: tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt)) - if not ((totalConsumeRows >= expectrowcnt * 3/4) and (totalConsumeRows < expectrowcnt)): + if not ((totalConsumeRows >= expectrowcnt * 3/4) and (totalConsumeRows <= expectrowcnt)): tdLog.exit("tmq consume rows error with snapshot = 0!") tdLog.info("wait subscriptions exit ....") From 8823997941d0630d96c34cc0495f160f6a7294fa Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Sat, 25 Mar 2023 15:56:22 +0800 Subject: [PATCH 6/6] fix:case error & add NONE for rows to col mode --- source/common/src/tdataformat.c | 4 +++- source/libs/parser/src/parInsertUtil.c | 5 +++++ tests/system-test/7-tmq/tmqDropNtb-snapshot0.py | 4 ++-- 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/source/common/src/tdataformat.c b/source/common/src/tdataformat.c index 5c313b3b3b..34808aa389 100644 --- a/source/common/src/tdataformat.c +++ b/source/common/src/tdataformat.c @@ -2440,7 +2440,9 @@ int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t byt char *data) { int32_t code = 0; if(data == NULL){ - code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NONE](pColData, NULL, 0); + for (int32_t i = 0; i < nRows; ++i) { + code = tColDataAppendValueImpl[pColData->flag][CV_FLAG_NONE](pColData, NULL, 0); + } goto _exit; } diff --git a/source/libs/parser/src/parInsertUtil.c b/source/libs/parser/src/parInsertUtil.c index f96c81cf85..4768231fb5 100644 --- a/source/libs/parser/src/parInsertUtil.c +++ b/source/libs/parser/src/parInsertUtil.c @@ -626,6 +626,11 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate ret = TSDB_CODE_INVALID_PARA; goto end; } + if (numFields > boundInfo->numOfBound) { + uError("numFields:%d > boundInfo->numOfBound:%d", numFields, boundInfo->numOfBound); + ret = TSDB_CODE_INVALID_PARA; + goto end; + } for (int c = 0; c < boundInfo->numOfBound; ++c) { SSchema* pColSchema = &pSchema[c]; SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, c); diff --git a/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py b/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py index 8792bdd3ba..c8bcdd6235 100644 --- a/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py +++ b/tests/system-test/7-tmq/tmqDropNtb-snapshot0.py @@ -131,7 +131,7 @@ class TDTestCase: 'batchNum': 100, 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 'endTs': 0, - 'pollDelay': 10, + 'pollDelay': 20, 'showMsg': 1, 'showRow': 1, 'snapshot': 0} @@ -193,7 +193,7 @@ class TDTestCase: tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt)) - if not ((totalConsumeRows >= expectrowcnt / 2 * (1 + 3/4)) and (totalConsumeRows < expectrowcnt)): + if not ((totalConsumeRows >= expectrowcnt / 2 * (1 + 3/4)) and (totalConsumeRows <= expectrowcnt)): tdLog.exit("tmq consume rows error with snapshot = 0!") tdLog.info("wait subscriptions exit ....")