From 4b5807c7f020a6bc2c10d146145a39ac9da2fe61 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Wed, 26 Oct 2022 15:54:00 +0800 Subject: [PATCH] fix(tmq): set precision --- examples/c/tmq.c | 4 ++-- source/dnode/vnode/src/inc/tq.h | 2 +- source/dnode/vnode/src/tq/tqExec.c | 15 +++++++++------ source/dnode/vnode/src/tq/tqPush.c | 2 +- 4 files changed, 13 insertions(+), 10 deletions(-) diff --git a/examples/c/tmq.c b/examples/c/tmq.c index d3fc803c94..1147671ea1 100644 --- a/examples/c/tmq.c +++ b/examples/c/tmq.c @@ -47,7 +47,7 @@ static int32_t msg_process(TAOS_RES* msg) { int32_t precision = taos_result_precision(msg); rows++; taos_print_row(buf, row, fields, numOfFields); - printf("row content: %s\n", buf); + printf("precision: %d, row content: %s\n", precision, buf); } return rows; @@ -70,7 +70,7 @@ static int32_t init_env() { taos_free_result(pRes); // create database - pRes = taos_query(pConn, "create database tmqdb"); + pRes = taos_query(pConn, "create database tmqdb precision 'ns'"); if (taos_errno(pRes) != 0) { printf("error in create tmqdb, reason:%s\n", taos_errstr(pRes)); return -1; diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h index edd23c80be..ef54adf0d6 100644 --- a/source/dnode/vnode/src/inc/tq.h +++ b/source/dnode/vnode/src/inc/tq.h @@ -155,7 +155,7 @@ int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHea // tqExec int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp* pRsp); -int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols); +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 tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry); diff --git a/source/dnode/vnode/src/tq/tqExec.c b/source/dnode/vnode/src/tq/tqExec.c index 30f6f81aa9..a6e8767a4d 100644 --- a/source/dnode/vnode/src/tq/tqExec.c +++ b/source/dnode/vnode/src/tq/tqExec.c @@ -15,14 +15,14 @@ #include "tq.h" -int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols) { +int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision) { int32_t dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock); void* buf = taosMemoryCalloc(1, dataStrLen); if (buf == NULL) return -1; SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf; pRetrieve->useconds = 0; - pRetrieve->precision = TSDB_DEFAULT_PRECISION; + pRetrieve->precision = precision; pRetrieve->compressed = 0; pRetrieve->completed = 1; pRetrieve->numOfRows = htonl(pBlock->info.rows); @@ -95,7 +95,7 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs break; } - tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols); + tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision); pRsp->blockNum++; if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) { @@ -174,7 +174,8 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMeta } } - tqAddBlockDataToRsp(pDataBlock, (SMqDataRsp*)pRsp, taosArrayGetSize(pDataBlock->pDataBlock)); + tqAddBlockDataToRsp(pDataBlock, (SMqDataRsp*)pRsp, taosArrayGetSize(pDataBlock->pDataBlock), + pTq->pVnode->config.tsdbCfg.precision); pRsp->blockNum++; if (pOffset->type == TMQ_OFFSET__LOG) { continue; @@ -256,7 +257,8 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp pRsp->createTableNum++; } } - tqAddBlockDataToRsp(&block, (SMqDataRsp*)pRsp, taosArrayGetSize(block.pDataBlock)); + tqAddBlockDataToRsp(&block, (SMqDataRsp*)pRsp, taosArrayGetSize(block.pDataBlock), + pTq->pVnode->config.tsdbCfg.precision); blockDataFreeRes(&block); tqAddBlockSchemaToRsp(pExec, (SMqDataRsp*)pRsp); pRsp->blockNum++; @@ -291,7 +293,8 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp pRsp->createTableNum++; } } - tqAddBlockDataToRsp(&block, (SMqDataRsp*)pRsp, taosArrayGetSize(block.pDataBlock)); + tqAddBlockDataToRsp(&block, (SMqDataRsp*)pRsp, taosArrayGetSize(block.pDataBlock), + pTq->pVnode->config.tsdbCfg.precision); blockDataFreeRes(&block); tqAddBlockSchemaToRsp(pExec, (SMqDataRsp*)pRsp); pRsp->blockNum++; diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c index 35cb305042..12d5b4112b 100644 --- a/source/dnode/vnode/src/tq/tqPush.c +++ b/source/dnode/vnode/src/tq/tqPush.c @@ -270,7 +270,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) break; } - tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols); + tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision); pRsp->blockNum++; } -- GitLab