diff --git a/include/client/taos.h b/include/client/taos.h index 9aaf15e8b05876e95f46ac7355811f9986a953c7..2940f1dfd01864cfa7f2107a6ed421fbb0bc20f8 100644 --- a/include/client/taos.h +++ b/include/client/taos.h @@ -307,6 +307,7 @@ DLL_EXPORT tmq_res_t tmq_get_res_type(TAOS_RES *res); DLL_EXPORT int32_t tmq_get_raw(TAOS_RES *res, tmq_raw_data *raw); DLL_EXPORT int32_t tmq_write_raw(TAOS *taos, tmq_raw_data raw); DLL_EXPORT int taos_write_raw_block(TAOS *taos, int numOfRows, char *pData, const char *tbname); +DLL_EXPORT int taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const char* tbname, TAOS_FIELD *fields, int numFields); DLL_EXPORT void tmq_free_raw(tmq_raw_data raw); // Returning null means error. Returned result need to be freed by tmq_free_json_meta DLL_EXPORT char *tmq_get_json_meta(TAOS_RES *res); diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 124a504f7bf85aa154d48d57f4582483844e41c6..846accf51b73e23932837c8dfaa3dc4077b83ace 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -1645,7 +1645,7 @@ int32_t getVersion1BlockMetaSize(const char* p, int32_t numOfCols) { static int32_t estimateJsonLen(SReqResultInfo* pResultInfo, int32_t numOfCols, int32_t numOfRows) { char* p = (char*)pResultInfo->pData; - // version + length + numOfRows + numOfCol + groupId + flag_segment + column_info + // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column length | int32_t len = getVersion1BlockMetaSize(p, numOfCols); int32_t* colLength = (int32_t*)(p + len); len += sizeof(int32_t) * numOfCols; diff --git a/source/client/src/clientRawBlockWrite.c b/source/client/src/clientRawBlockWrite.c index 6a6a5d56d752d6ad6686e05162bb946f40902999..5583c8ae007400d252691ddfc27dda546a1c5ded 100644 --- a/source/client/src/clientRawBlockWrite.c +++ b/source/client/src/clientRawBlockWrite.c @@ -1211,6 +1211,208 @@ static void destroyVgHash(void* data) { taosMemoryFreeClear(vgData->data); } +int taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const char* tbname, TAOS_FIELD *fields, int numFields){ + int32_t code = TSDB_CODE_SUCCESS; + STableMeta* pTableMeta = NULL; + SQuery* pQuery = NULL; + SSubmitReq* subReq = NULL; + + SRequestObj* pRequest = (SRequestObj*)createRequest(*(int64_t*)taos, TSDB_SQL_INSERT, 0); + if (!pRequest) { + uError("WriteRaw:createRequest error request is null"); + code = terrno; + goto end; + } + + pRequest->syncQuery = true; + if (!pRequest->pDb) { + uError("WriteRaw:not use db"); + code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; + goto end; + } + + SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}}; + tstrncpy(pName.dbname, pRequest->pDb, sizeof(pName.dbname)); + tstrncpy(pName.tname, tbname, sizeof(pName.tname)); + + struct SCatalog* pCatalog = NULL; + code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog); + if (code != TSDB_CODE_SUCCESS) { + uError("WriteRaw: get gatlog error"); + goto end; + } + + SRequestConnInfo conn = {0}; + conn.pTrans = pRequest->pTscObj->pAppInfo->pTransporter; + conn.requestId = pRequest->requestId; + conn.requestObjRefId = pRequest->self; + conn.mgmtEps = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp); + + SVgroupInfo vgData = {0}; + code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &vgData); + if (code != TSDB_CODE_SUCCESS) { + uError("WriteRaw:catalogGetTableHashVgroup failed. table name: %s", tbname); + goto end; + } + + code = catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta); + if (code != TSDB_CODE_SUCCESS) { + uError("WriteRaw:catalogGetTableMeta failed. table name: %s", tbname); + goto end; + } + uint64_t suid = (TSDB_NORMAL_TABLE == pTableMeta->tableType ? 0 : pTableMeta->suid); + uint64_t uid = pTableMeta->uid; + int32_t numOfCols = pTableMeta->tableInfo.numOfColumns; + + uint16_t fLen = 0; + int32_t rowSize = 0; + int16_t nVar = 0; + for (int i = 0; i < pTableMeta->tableInfo.numOfColumns; i++) { + SSchema* schema = pTableMeta->schema + i; + fLen += TYPE_BYTES[schema->type]; + rowSize += schema->bytes; + if (IS_VAR_DATA_TYPE(schema->type)) { + nVar++; + } + } + + fLen -= sizeof(TSKEY); + + int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) + + (int32_t)TD_BITMAP_BYTES(numOfCols - 1); + int32_t schemaLen = 0; + int32_t submitLen = sizeof(SSubmitBlk) + schemaLen + rows * extendedRowSize; + + int32_t totalLen = sizeof(SSubmitReq) + submitLen; + subReq = taosMemoryCalloc(1, totalLen); + SSubmitBlk* blk = POINTER_SHIFT(subReq, sizeof(SSubmitReq)); + void* blkSchema = POINTER_SHIFT(blk, sizeof(SSubmitBlk)); + STSRow* rowData = POINTER_SHIFT(blkSchema, schemaLen); + + SRowBuilder rb = {0}; + tdSRowInit(&rb, pTableMeta->sversion); + tdSRowSetTpInfo(&rb, numOfCols, fLen); + int32_t dataLen = 0; + + // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column length | + char* pStart = pData + getVersion1BlockMetaSize(pData, numFields); + int32_t* colLength = (int32_t*)pStart; + pStart += sizeof(int32_t) * numFields; + + SResultColumn* pCol = taosMemoryCalloc(numFields, sizeof(SResultColumn)); + + for (int32_t i = 0; i < numFields; ++i) { + if (IS_VAR_DATA_TYPE(fields[i].type)) { + pCol[i].offset = (int32_t*)pStart; + pStart += rows * sizeof(int32_t); + } else { + pCol[i].nullbitmap = pStart; + pStart += BitmapLen(rows); + } + + pCol[i].pData = pStart; + pStart += colLength[i]; + } + + SHashObj* schemaHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK); + for (int i = 0; i < numFields; i++) { + TAOS_FIELD* schema = &fields[i]; + taosHashPut(schemaHash, schema->name, strlen(schema->name), &i, sizeof(int32_t)); + } + + for (int32_t j = 0; j < rows; j++) { + tdSRowResetBuf(&rb, rowData); + int32_t offset = 0; + for (int32_t k = 0; k < numOfCols; k++) { + const SSchema* pColumn = &pTableMeta->schema[k]; + int32_t* index = taosHashGet(schemaHash, pColumn->name, strlen(pColumn->name)); + if (!index) { // add none + tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NONE, NULL, false, offset, k); + }else{ + if (IS_VAR_DATA_TYPE(pColumn->type)) { + if (pCol[*index].offset[j] != -1) { + char* data = pCol[*index].pData + pCol[*index].offset[j]; + tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, data, true, offset, k); + } else { + tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, offset, k); + } + } else { + if (!colDataIsNull_f(pCol[*index].nullbitmap, j)) { + char* data = pCol[*index].pData + pColumn->bytes * j; + tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, data, true, offset, k); + } else { + tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, offset, k); + } + } + } + + if (pColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID) { + offset += TYPE_BYTES[pColumn->type]; + } + } + tdSRowEnd(&rb); + int32_t rowLen = TD_ROW_LEN(rowData); + rowData = POINTER_SHIFT(rowData, rowLen); + dataLen += rowLen; + } + + taosHashCleanup(schemaHash); + taosMemoryFree(pCol); + + blk->uid = htobe64(uid); + blk->suid = htobe64(suid); + blk->sversion = htonl(pTableMeta->sversion); + blk->schemaLen = htonl(schemaLen); + blk->numOfRows = htonl(rows); + blk->dataLen = htonl(dataLen); + subReq->length = sizeof(SSubmitReq) + sizeof(SSubmitBlk) + schemaLen + dataLen; + subReq->numOfBlocks = 1; + + pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY); + if (NULL == pQuery) { + uError("create SQuery error"); + code = TSDB_CODE_OUT_OF_MEMORY; + goto end; + } + pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE; + pQuery->haveResultSet = false; + pQuery->msgType = TDMT_VND_SUBMIT; + pQuery->pRoot = (SNode*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT); + if (NULL == pQuery->pRoot) { + uError("create pQuery->pRoot error"); + code = TSDB_CODE_OUT_OF_MEMORY; + goto end; + } + SVnodeModifOpStmt* nodeStmt = (SVnodeModifOpStmt*)(pQuery->pRoot); + nodeStmt->pDataBlocks = taosArrayInit(1, POINTER_BYTES); + + SVgDataBlocks* dst = taosMemoryCalloc(1, sizeof(SVgDataBlocks)); + if (NULL == dst) { + code = TSDB_CODE_TSC_OUT_OF_MEMORY; + goto end; + } + dst->vg = vgData; + dst->numOfTables = subReq->numOfBlocks; + dst->size = subReq->length; + dst->pData = (char*)subReq; + subReq->header.vgId = htonl(dst->vg.vgId); + subReq->version = htonl(1); + subReq->header.contLen = htonl(subReq->length); + subReq->length = htonl(subReq->length); + subReq->numOfBlocks = htonl(subReq->numOfBlocks); + subReq = NULL; // no need free + taosArrayPush(nodeStmt->pDataBlocks, &dst); + + launchQueryImpl(pRequest, pQuery, true, NULL); + code = pRequest->code; + + end: + taosMemoryFreeClear(pTableMeta); + qDestroyQuery(pQuery); + taosMemoryFree(subReq); + return code; +} + int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) { int32_t code = TSDB_CODE_SUCCESS; STableMeta* pTableMeta = NULL; @@ -1293,6 +1495,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) tdSRowSetTpInfo(&rb, numOfCols, fLen); int32_t dataLen = 0; + // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column length | char* pStart = pData + getVersion1BlockMetaSize(pData, numOfCols); int32_t* colLength = (int32_t*)pStart; pStart += sizeof(int32_t) * numOfCols; @@ -1577,7 +1780,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) { const SSchema* pColumn = &pTableMeta->schema[k]; int32_t* index = taosHashGet(schemaHash, pColumn->name, strlen(pColumn->name)); if (!index) { - tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, offset, k); + tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NONE, NULL, false, offset, k); } else { char* colData = rspObj.resInfo.row[*index]; if (!colData) { @@ -1670,6 +1873,7 @@ end: return code; } + static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) { int32_t code = TSDB_CODE_SUCCESS; SHashObj* pVgHash = NULL; @@ -1882,7 +2086,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) const SSchema* pColumn = &pTableMeta->schema[k]; int32_t* index = taosHashGet(schemaHash, pColumn->name, strlen(pColumn->name)); if (!index) { - tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, offset, k); + tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NONE, NULL, false, offset, k); } else { char* colData = rspObj.resInfo.row[*index]; if (!colData) { @@ -1963,7 +2167,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); rspObj.resInfo.pRspMsg = NULL; doFreeReqResultInfo(&rspObj.resInfo); diff --git a/utils/test/c/tmq_taosx_ci.c b/utils/test/c/tmq_taosx_ci.c index 6540fdac4cdba1aa9fe422719f91b13ce365c7d9..5af9ba68b260ae5c6b80e673cda800c0281691c5 100644 --- a/utils/test/c/tmq_taosx_ci.c +++ b/utils/test/c/tmq_taosx_ci.c @@ -76,6 +76,32 @@ static void msg_process(TAOS_RES* msg) { } int buildDatabase(TAOS* pConn, TAOS_RES* pRes){ + + /* test for TD-20612 start*/ +// pRes = taos_query(pConn,"create table tb1 (ts timestamp, c1 int, c2 int)"); +// 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,"insert into tb1 (ts, c1) values(1669092069069, 0)"); +// 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,"insert into tb1 (ts, c2) values(1669092069069, 1)"); +// if (taos_errno(pRes) != 0) { +// printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); +// return -1; +// } +// taos_free_result(pRes); +// +// return 0; + /* 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)");