From 213b97b79ac2722db50cc029f363a88a84f0ac7d Mon Sep 17 00:00:00 2001 From: Alex Duan <417921451@qq.com> Date: Fri, 2 Dec 2022 23:20:19 +0800 Subject: [PATCH] feat(api): taos_affected_rows64 add and modfiy tmsg.h --- include/libs/executor/dataSinkMgt.h | 2 +- source/client/src/clientMsgHandler.c | 2 +- source/dnode/vnode/src/tq/tqExec.c | 2 +- source/libs/command/src/command.c | 2 +- source/libs/command/src/explain.c | 2 +- source/libs/qworker/src/qwMsg.c | 2 +- source/libs/qworker/src/qworker.c | 6 +++--- source/libs/scheduler/src/schJob.c | 2 +- source/libs/scheduler/src/schRemote.c | 2 +- source/libs/stream/src/streamDispatch.c | 4 ++-- 10 files changed, 13 insertions(+), 13 deletions(-) diff --git a/include/libs/executor/dataSinkMgt.h b/include/libs/executor/dataSinkMgt.h index 8a02f372d1..ed7cbc8125 100644 --- a/include/libs/executor/dataSinkMgt.h +++ b/include/libs/executor/dataSinkMgt.h @@ -68,7 +68,7 @@ typedef struct SInputData { typedef struct SOutputData { int32_t numOfBlocks; - int32_t numOfRows; + int64_t numOfRows; // int32_t changed to int64_t int32_t numOfCols; int8_t compressed; char* pData; diff --git a/source/client/src/clientMsgHandler.c b/source/client/src/clientMsgHandler.c index 4bd74a842f..d5a567075c 100644 --- a/source/client/src/clientMsgHandler.c +++ b/source/client/src/clientMsgHandler.c @@ -450,7 +450,7 @@ static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) { (*pRsp)->precision = 0; (*pRsp)->compressed = 0; (*pRsp)->compLen = 0; - (*pRsp)->numOfRows = htonl(pBlock->info.rows); + (*pRsp)->numOfRows = htobe64((int64_t)pBlock->info.rows); (*pRsp)->numOfCols = htonl(SHOW_VARIABLES_RESULT_COLS); int32_t len = blockEncode(pBlock, (*pRsp)->data, SHOW_VARIABLES_RESULT_COLS); diff --git a/source/dnode/vnode/src/tq/tqExec.c b/source/dnode/vnode/src/tq/tqExec.c index 2514190035..093186ebbb 100644 --- a/source/dnode/vnode/src/tq/tqExec.c +++ b/source/dnode/vnode/src/tq/tqExec.c @@ -25,7 +25,7 @@ int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t pRetrieve->precision = precision; pRetrieve->compressed = 0; pRetrieve->completed = 1; - pRetrieve->numOfRows = htonl(pBlock->info.rows); + pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows); int32_t actualLen = blockEncode(pBlock, pRetrieve->data, numOfCols); actualLen += sizeof(SRetrieveTableRsp); diff --git a/source/libs/command/src/command.c b/source/libs/command/src/command.c index d58c4dc6d3..7661944379 100644 --- a/source/libs/command/src/command.c +++ b/source/libs/command/src/command.c @@ -36,7 +36,7 @@ static int32_t buildRetrieveTableRsp(SSDataBlock* pBlock, int32_t numOfCols, SRe (*pRsp)->precision = 0; (*pRsp)->compressed = 0; (*pRsp)->compLen = 0; - (*pRsp)->numOfRows = htonl(pBlock->info.rows); + (*pRsp)->numOfRows = htobe64((int64_t)pBlock->info.rows); (*pRsp)->numOfCols = htonl(numOfCols); int32_t len = blockEncode(pBlock, (*pRsp)->data, numOfCols); diff --git a/source/libs/command/src/explain.c b/source/libs/command/src/explain.c index 410e62a18b..dd48023679 100644 --- a/source/libs/command/src/explain.c +++ b/source/libs/command/src/explain.c @@ -1615,7 +1615,7 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) { } rsp->completed = 1; - rsp->numOfRows = htonl(rowNum); + rsp->numOfRows = htobe64((int64_t)rowNum); int32_t len = blockEncode(pBlock, rsp->data, taosArrayGetSize(pBlock->pDataBlock)); ASSERT(len == rspSize - sizeof(SRetrieveTableRsp)); diff --git a/source/libs/qworker/src/qwMsg.c b/source/libs/qworker/src/qwMsg.c index d9a7cea411..7154e22300 100644 --- a/source/libs/qworker/src/qwMsg.c +++ b/source/libs/qworker/src/qwMsg.c @@ -37,7 +37,7 @@ void qwBuildFetchRsp(void *msg, SOutputData *input, int32_t len, bool qComplete) rsp->precision = input->precision; rsp->compressed = input->compressed; rsp->compLen = htonl(len); - rsp->numOfRows = htonl(input->numOfRows); + rsp->numOfRows = htobe64(input->numOfRows); rsp->numOfCols = htonl(input->numOfCols); rsp->numOfBlocks = htonl(input->numOfBlocks); } diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index 0890d10b65..fe2acb805b 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -275,7 +275,7 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, QW_ERR_RET(code); } - QW_TASK_DLOG("no more data in sink and query end, fetched blocks %d rows %d", pOutput->numOfBlocks, + QW_TASK_DLOG("no more data in sink and query end, fetched blocks %d rows %"PRId64, pOutput->numOfBlocks, pOutput->numOfRows); qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCC); @@ -320,7 +320,7 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, pOutput->numOfBlocks++; if (DS_BUF_EMPTY == pOutput->bufStatus && pOutput->queryEnd) { - QW_TASK_DLOG("task all data fetched and done, fetched blocks %d rows %d", pOutput->numOfBlocks, + QW_TASK_DLOG("task all data fetched and done, fetched blocks %d rows %" PRId64, pOutput->numOfBlocks, pOutput->numOfRows); qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCC); break; @@ -332,7 +332,7 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, } if (pOutput->numOfRows >= QW_MIN_RES_ROWS) { - QW_TASK_DLOG("task fetched blocks %d rows %d reaches the min rows", pOutput->numOfBlocks, pOutput->numOfRows); + QW_TASK_DLOG("task fetched blocks %d rows %" PRId64 " reaches the min rows", pOutput->numOfBlocks, pOutput->numOfRows); break; } } diff --git a/source/libs/scheduler/src/schJob.c b/source/libs/scheduler/src/schJob.c index ea49b0c4e0..b9062859c8 100644 --- a/source/libs/scheduler/src/schJob.c +++ b/source/libs/scheduler/src/schJob.c @@ -526,7 +526,7 @@ int32_t schProcessOnJobPartialSuccess(SSchJob *pJob) { void schProcessOnDataFetched(SSchJob *pJob) { schPostJobRes(pJob, SCH_OP_FETCH); } int32_t schProcessOnExplainDone(SSchJob *pJob, SSchTask *pTask, SRetrieveTableRsp *pRsp) { - SCH_TASK_DLOG("got explain rsp, rows:%d, complete:%d", htonl(pRsp->numOfRows), pRsp->completed); + SCH_TASK_DLOG("got explain rsp, rows:%" PRId64 ", complete:%d", htobe64(pRsp->numOfRows), pRsp->completed); atomic_store_64(&pJob->resNumOfRows, htobe64(pRsp->numOfRows)); atomic_store_ptr(&pJob->fetchRes, pRsp); diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c index 76ac0a2e9f..30622f96bb 100644 --- a/source/libs/scheduler/src/schRemote.c +++ b/source/libs/scheduler/src/schRemote.c @@ -114,7 +114,7 @@ int32_t schProcessFetchRsp(SSchJob *pJob, SSchTask *pTask, char *msg, int32_t rs SCH_SET_TASK_STATUS(pTask, JOB_TASK_STATUS_SUCC); } - SCH_TASK_DLOG("got fetch rsp, rows:%d, complete:%d", htonl(rsp->numOfRows), rsp->completed); + SCH_TASK_DLOG("got fetch rsp, rows:%" PRId64 ", complete:%d", htobe64(rsp->numOfRows), rsp->completed); msg = NULL; schProcessOnDataFetched(pJob); diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 2c36c299ee..4e0b0630bc 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -112,7 +112,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock) pRetrieve->compressed = 0; pRetrieve->completed = 1; pRetrieve->streamBlockType = pBlock->info.type; - pRetrieve->numOfRows = htonl(pBlock->info.rows); + pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows); pRetrieve->numOfCols = htonl(numOfCols); pRetrieve->skey = htobe64(pBlock->info.window.skey); pRetrieve->ekey = htobe64(pBlock->info.window.ekey); @@ -189,7 +189,7 @@ static int32_t streamAddBlockToDispatchMsg(const SSDataBlock* pBlock, SStreamDis pRetrieve->compressed = 0; pRetrieve->completed = 1; pRetrieve->streamBlockType = pBlock->info.type; - pRetrieve->numOfRows = htonl(pBlock->info.rows); + pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows); pRetrieve->skey = htobe64(pBlock->info.window.skey); pRetrieve->ekey = htobe64(pBlock->info.window.ekey); pRetrieve->version = htobe64(pBlock->info.version);