diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c
index d30ee32d67c2a9294b3a3aa0b2bc9144829a364c..657d8b78481d20813edabef254285121722aa2bf 100644
--- a/src/client/src/tscServer.c
+++ b/src/client/src/tscServer.c
@@ -13,7 +13,10 @@
* along with this program. If not, see .
*/
+#include
#include "os.h"
+#include "qPlan.h"
+#include "qTableMeta.h"
#include "tcmdtype.h"
#include "tlockfree.h"
#include "trpc.h"
@@ -21,10 +24,8 @@
#include "tscLog.h"
#include "tscProfile.h"
#include "tscUtil.h"
-#include "qTableMeta.h"
#include "tsclient.h"
#include "ttimer.h"
-#include "qPlan.h"
int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo) = {0};
@@ -2048,16 +2049,27 @@ int tscProcessMultiTableMetaRsp(SSqlObj *pSql) {
}
SSqlCmd *pParentCmd = &pParentSql->cmd;
-
SHashObj *pSet = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
char* pMsg = pMultiMeta->meta;
+ char* buf = NULL;
+ if (pMultiMeta->compressed) {
+ buf = malloc(pMultiMeta->rawLen - sizeof(SMultiTableMeta));
+ int32_t len = tsDecompressString(pMultiMeta->meta, pMultiMeta->contLen - sizeof(SMultiTableMeta), 1,
+ buf, pMultiMeta->rawLen - sizeof(SMultiTableMeta), ONE_STAGE_COMP, NULL, 0);
+ assert(len == pMultiMeta->rawLen - sizeof(SMultiTableMeta));
+
+ pMsg = buf;
+ }
+
for (int32_t i = 0; i < pMultiMeta->numOfTables; i++) {
STableMetaMsg *pMetaMsg = (STableMetaMsg *)pMsg;
int32_t code = tableMetaMsgConvert(pMetaMsg);
if (code != TSDB_CODE_SUCCESS) {
taosHashCleanup(pSet);
taosReleaseRef(tscObjRef, pParentSql->self);
+
+ tfree(buf);
return code;
}
@@ -2066,6 +2078,8 @@ int tscProcessMultiTableMetaRsp(SSqlObj *pSql) {
tscError("0x%"PRIx64" invalid table meta from mnode, name:%s", pSql->self, pMetaMsg->tableFname);
taosHashCleanup(pSet);
taosReleaseRef(tscObjRef, pParentSql->self);
+
+ tfree(buf);
return TSDB_CODE_TSC_INVALID_VALUE;
}
@@ -2115,6 +2129,8 @@ int tscProcessMultiTableMetaRsp(SSqlObj *pSql) {
taosHashCleanup(pSet);
taosReleaseRef(tscObjRef, pParentSql->self);
+
+ tfree(buf);
return TSDB_CODE_SUCCESS;
}
diff --git a/src/inc/taosmsg.h b/src/inc/taosmsg.h
index f20e1535bab725532bf8785cabdbe186448acc15..4e76b6dcc1fb0682e69de5a0e7e0f35440e7e40b 100644
--- a/src/inc/taosmsg.h
+++ b/src/inc/taosmsg.h
@@ -760,10 +760,12 @@ typedef struct STableMetaMsg {
} STableMetaMsg;
typedef struct SMultiTableMeta {
- int32_t numOfTables;
- int32_t numOfVgroup;
- int32_t contLen;
- char meta[];
+ int32_t numOfTables;
+ int32_t numOfVgroup;
+ uint32_t contLen:31;
+ uint8_t compressed:1; // denote if compressed or not
+ uint32_t rawLen; // size before compress
+ char meta[];
} SMultiTableMeta;
typedef struct {
diff --git a/src/mnode/src/mnodeTable.c b/src/mnode/src/mnodeTable.c
index beeff372aa75a34c4be1857782a76c2426748140..6dc2f8ad2851bbc9fdf6ff8a24b8cf4de502d603 100644
--- a/src/mnode/src/mnodeTable.c
+++ b/src/mnode/src/mnodeTable.c
@@ -2892,7 +2892,7 @@ static SMultiTableMeta* ensureMsgBufferSpace(SMultiTableMeta *pMultiMeta, SArray
(*totalMallocLen) *= 2;
}
- pMultiMeta = rpcReallocCont(pMultiMeta, *totalMallocLen);
+ pMultiMeta = realloc(pMultiMeta, *totalMallocLen);
if (pMultiMeta == NULL) {
return NULL;
}
@@ -2923,8 +2923,8 @@ static int32_t mnodeProcessMultiTableMetaMsg(SMnodeMsg *pMsg) {
}
// first malloc 80KB, subsequent reallocation will expand the size as twice of the original size
- int32_t totalMallocLen = sizeof(STableMetaMsg) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16);
- pMultiMeta = rpcMallocCont(totalMallocLen);
+ int32_t totalMallocLen = sizeof(SMultiTableMeta) + sizeof(STableMetaMsg) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16);
+ pMultiMeta = calloc(1, totalMallocLen);
if (pMultiMeta == NULL) {
code = TSDB_CODE_MND_OUT_OF_MEMORY;
goto _end;
@@ -2957,7 +2957,7 @@ static int32_t mnodeProcessMultiTableMetaMsg(SMnodeMsg *pMsg) {
int remain = totalMallocLen - pMultiMeta->contLen;
if (remain <= sizeof(STableMetaMsg) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16)) {
totalMallocLen *= 2;
- pMultiMeta = rpcReallocCont(pMultiMeta, totalMallocLen);
+ pMultiMeta = realloc(pMultiMeta, totalMallocLen);
if (pMultiMeta == NULL) {
mnodeDecTableRef(pMsg->pTable);
code = TSDB_CODE_MND_OUT_OF_MEMORY;
@@ -3027,16 +3027,41 @@ static int32_t mnodeProcessMultiTableMetaMsg(SMnodeMsg *pMsg) {
pMsg->rpcRsp.len = pMultiMeta->contLen;
code = TSDB_CODE_SUCCESS;
+ char* tmp = rpcMallocCont(pMultiMeta->contLen + 2);
+ if (tmp == NULL) {
+ code = TSDB_CODE_MND_OUT_OF_MEMORY;
+ goto _end;
+ }
+
+ int32_t len = tsCompressString(pMultiMeta->meta, (int32_t)pMultiMeta->contLen - sizeof(SMultiTableMeta), 1,
+ tmp + sizeof(SMultiTableMeta), (int32_t)pMultiMeta->contLen - sizeof(SMultiTableMeta) + 2, ONE_STAGE_COMP, NULL, 0);
+
+ pMultiMeta->rawLen = pMultiMeta->contLen;
+ if (len == -1 || len + sizeof(SMultiTableMeta) >= pMultiMeta->contLen + 2) { // compress failed, do not compress this binary data
+ pMultiMeta->compressed = 0;
+ memcpy(tmp, pMultiMeta, sizeof(SMultiTableMeta) + pMultiMeta->contLen);
+ } else {
+ pMultiMeta->compressed = 1;
+ pMultiMeta->contLen = sizeof(SMultiTableMeta) + len;
+
+ // copy the header and the compressed payload
+ memcpy(tmp, pMultiMeta, sizeof(SMultiTableMeta));
+ }
+
+ pMsg->rpcRsp.rsp = tmp;
+ pMsg->rpcRsp.len = pMultiMeta->contLen;
+
+ SMultiTableMeta* p = (SMultiTableMeta*) tmp;
+
+ mDebug("multiTable info build completed, original:%d, compressed:%d, comp:%d", p->rawLen, p->contLen, p->compressed);
+
_end:
tfree(str);
tfree(nameList);
taosArrayDestroy(pList);
pMsg->pTable = NULL;
pMsg->pVgroup = NULL;
-
- if (code != TSDB_CODE_SUCCESS) {
- rpcFreeCont(pMultiMeta);
- }
+ tfree(pMultiMeta);
return code;
}
diff --git a/src/query/inc/qExecutor.h b/src/query/inc/qExecutor.h
index 9348606d0cde43106b348e71db3f6f67082ef380..c4276bfe3781a97d91041fedfde7d67fcaa5a5d3 100644
--- a/src/query/inc/qExecutor.h
+++ b/src/query/inc/qExecutor.h
@@ -589,7 +589,7 @@ int32_t createIndirectQueryFuncExprFromMsg(SQueryTableMsg *pQueryMsg, int32_t nu
SGroupbyExpr *createGroupbyExprFromMsg(SQueryTableMsg *pQueryMsg, SColIndex *pColIndex, int32_t *code);
SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
- SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, int32_t vgId, char* sql, uint64_t *qId);
+ SExprInfo *pSecExprs, STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, int32_t vgId, char* sql, uint64_t qId);
int32_t initQInfo(STsBufInfo* pTsBufInfo, void* tsdb, void* sourceOptr, SQInfo* pQInfo, SQueryParam* param, char* start,
int32_t prevResultLen, void* merger);
diff --git a/src/query/src/qExecutor.c b/src/query/src/qExecutor.c
index fa2ddb05b8fe6de84f64d5652f88eb1c71ecbfab..ecbfe36ee3d57cdf5d3de0d3b06ede98fc790050 100644
--- a/src/query/src/qExecutor.c
+++ b/src/query/src/qExecutor.c
@@ -2268,10 +2268,11 @@ static int32_t updateBlockLoadStatus(SQueryAttr *pQuery, int32_t status) {
return status;
}
-static void doExchangeTimeWindow(SQInfo* pQInfo, STimeWindow* win) {
- SQueryAttr* pQueryAttr = &pQInfo->query;
- size_t t = taosArrayGetSize(pQueryAttr->tableGroupInfo.pGroupList);
- for(int32_t i = 0; i < t; ++i) {
+static void doUpdateLastKey(SQueryAttr* pQueryAttr) {
+ STimeWindow* win = &pQueryAttr->window;
+
+ size_t num = taosArrayGetSize(pQueryAttr->tableGroupInfo.pGroupList);
+ for(int32_t i = 0; i < num; ++i) {
SArray* p1 = taosArrayGetP(pQueryAttr->tableGroupInfo.pGroupList, i);
size_t len = taosArrayGetSize(p1);
@@ -2286,7 +2287,7 @@ static void doExchangeTimeWindow(SQInfo* pQInfo, STimeWindow* win) {
}
}
-static void changeExecuteScanOrder(SQInfo *pQInfo, SQueryTableMsg* pQueryMsg, bool stableQuery) {
+static void updateDataCheckOrder(SQInfo *pQInfo, SQueryTableMsg* pQueryMsg, bool stableQuery) {
SQueryAttr* pQueryAttr = pQInfo->runtimeEnv.pQueryAttr;
// in case of point-interpolation query, use asc order scan
@@ -2303,6 +2304,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, SQueryTableMsg* pQueryMsg, bo
SWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
}
+ pQueryAttr->needReverseScan = false;
return;
}
@@ -2312,7 +2314,8 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, SQueryTableMsg* pQueryMsg, bo
SWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
}
- doExchangeTimeWindow(pQInfo, &pQueryAttr->window);
+ pQueryAttr->needReverseScan = false;
+ doUpdateLastKey(pQueryAttr);
return;
}
@@ -2333,20 +2336,22 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, SQueryTableMsg* pQueryMsg, bo
pQueryAttr->window.ekey, pQueryAttr->window.ekey, pQueryAttr->window.skey);
SWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
- doExchangeTimeWindow(pQInfo, &pQueryAttr->window);
+ doUpdateLastKey(pQueryAttr);
}
pQueryAttr->order.order = TSDB_ORDER_ASC;
+ pQueryAttr->needReverseScan = false;
} else if (onlyLastQuery(pQueryAttr) && notContainSessionOrStateWindow(pQueryAttr)) {
if (QUERY_IS_ASC_QUERY(pQueryAttr)) {
qDebug(msg, pQInfo, "only-last", pQueryAttr->order.order, TSDB_ORDER_DESC, pQueryAttr->window.skey,
pQueryAttr->window.ekey, pQueryAttr->window.ekey, pQueryAttr->window.skey);
SWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
- doExchangeTimeWindow(pQInfo, &pQueryAttr->window);
+ doUpdateLastKey(pQueryAttr);
}
pQueryAttr->order.order = TSDB_ORDER_DESC;
+ pQueryAttr->needReverseScan = false;
}
} else { // interval query
@@ -2357,20 +2362,22 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, SQueryTableMsg* pQueryMsg, bo
pQueryAttr->window.skey, pQueryAttr->window.ekey, pQueryAttr->window.ekey, pQueryAttr->window.skey);
SWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
- doExchangeTimeWindow(pQInfo, &pQueryAttr->window);
+ doUpdateLastKey(pQueryAttr);
}
pQueryAttr->order.order = TSDB_ORDER_ASC;
+ pQueryAttr->needReverseScan = false;
} else if (onlyLastQuery(pQueryAttr)) {
if (QUERY_IS_ASC_QUERY(pQueryAttr)) {
qDebug(msg, pQInfo, "only-last stable", pQueryAttr->order.order, TSDB_ORDER_DESC,
pQueryAttr->window.skey, pQueryAttr->window.ekey, pQueryAttr->window.ekey, pQueryAttr->window.skey);
SWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
- doExchangeTimeWindow(pQInfo, &pQueryAttr->window);
+ doUpdateLastKey(pQueryAttr);
}
pQueryAttr->order.order = TSDB_ORDER_DESC;
+ pQueryAttr->needReverseScan = false;
}
}
}
@@ -2388,9 +2395,6 @@ static void getIntermediateBufInfo(SQueryRuntimeEnv* pRuntimeEnv, int32_t* ps, i
while(((*rowsize) * MIN_ROWS_PER_PAGE) > (*ps) - overhead) {
*ps = ((*ps) << 1u);
}
-
-// pRuntimeEnv->numOfRowsPerPage = ((*ps) - sizeof(tFilePage)) / (*rowsize);
-// assert(pRuntimeEnv->numOfRowsPerPage <= MAX_ROWS_PER_RESBUF_PAGE);
}
#define IS_PREFILTER_TYPE(_t) ((_t) != TSDB_DATA_TYPE_BINARY && (_t) != TSDB_DATA_TYPE_NCHAR)
@@ -4382,7 +4386,7 @@ int32_t doInitQInfo(SQInfo* pQInfo, STSBuf* pTsBuf, void* tsdb, void* sourceOptr
break;
}
case OP_DataBlocksOptScan: {
- pRuntimeEnv->proot = createDataBlocksOptScanInfo(pRuntimeEnv->pQueryHandle, pRuntimeEnv, getNumOfScanTimes(pQueryAttr), 1);
+ pRuntimeEnv->proot = createDataBlocksOptScanInfo(pRuntimeEnv->pQueryHandle, pRuntimeEnv, getNumOfScanTimes(pQueryAttr), pQueryAttr->needReverseScan? 1:0);
break;
}
case OP_TableScan: {
@@ -4420,8 +4424,10 @@ int32_t doInitQInfo(SQInfo* pQInfo, STSBuf* pTsBuf, void* tsdb, void* sourceOptr
if (pQInfo->summary.queryProfEvents == NULL) {
qDebug("QInfo:0x%"PRIx64" failed to allocate query prof events array", pQInfo->qId);
}
+
pQInfo->summary.operatorProfResults =
taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_TINYINT), true, HASH_NO_LOCK);
+
if (pQInfo->summary.operatorProfResults == NULL) {
qDebug("QInfo:0x%"PRIx64" failed to allocate operator prof results hash", pQInfo->qId);
}
@@ -4814,7 +4820,6 @@ SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, SQueryRuntime
pInfo->reverseTimes = reverseTime;
pInfo->current = 0;
pInfo->order = pRuntimeEnv->pQueryAttr->order.order;
-// pInfo->prevGroupId = -1;
SOperatorInfo* pOptr = calloc(1, sizeof(SOperatorInfo));
pOptr->name = "DataBlocksOptimizedScanOperator";
@@ -7366,7 +7371,7 @@ FORCE_INLINE bool checkQIdEqual(void *qHandle, uint64_t qId) {
SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SGroupbyExpr* pGroupbyExpr, SExprInfo* pExprs,
SExprInfo* pSecExprs, STableGroupInfo* pTableGroupInfo, SColumnInfo* pTagCols, int32_t vgId,
- char* sql, uint64_t *qId) {
+ char* sql, uint64_t qId) {
int16_t numOfCols = pQueryMsg->numOfCols;
int16_t numOfOutput = pQueryMsg->numOfOutput;
@@ -7375,7 +7380,7 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SGroupbyExpr* pGroupbyExpr, S
goto _cleanup_qinfo;
}
- pQInfo->qId = *qId;
+ pQInfo->qId = qId;
// to make sure third party won't overwrite this structure
pQInfo->signature = pQInfo;
@@ -7485,7 +7490,7 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SGroupbyExpr* pGroupbyExpr, S
tsem_init(&pQInfo->ready, 0, 0);
pQueryAttr->window = pQueryMsg->window;
- changeExecuteScanOrder(pQInfo, pQueryMsg, pQueryAttr->stableQuery);
+ updateDataCheckOrder(pQInfo, pQueryMsg, pQueryAttr->stableQuery);
SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv;
STimeWindow window = pQueryAttr->window;
diff --git a/src/query/src/queryMain.c b/src/query/src/queryMain.c
index 787cb2f7d1a34f8958977eb85cd3c2621ff9a047..d4aa523bf8ed41a4e0e015d6ee0b4bbb67963754 100644
--- a/src/query/src/queryMain.c
+++ b/src/query/src/queryMain.c
@@ -162,7 +162,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi
assert(pQueryMsg->stableQuery == isSTableQuery);
(*pQInfo) = createQInfoImpl(pQueryMsg, param.pGroupbyExpr, param.pExprs, param.pSecExprs, &tableGroupInfo,
- param.pTagColumnInfo, vgId, param.sql, qId);
+ param.pTagColumnInfo, vgId, param.sql, *qId);
param.sql = NULL;
param.pExprs = NULL;
diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c
index b17aa755a5ee18cfc61ab05fbbf36d9463cfc0dd..1eafb5e23371da93454e8b6b8700341c02d38441 100644
--- a/src/tsdb/src/tsdbRead.c
+++ b/src/tsdb/src/tsdbRead.c
@@ -98,6 +98,8 @@ typedef struct SIOCostSummary {
int64_t blockLoadTime;
int64_t statisInfoLoadTime;
int64_t checkForNextTime;
+ int64_t headFileLoad;
+ int64_t headFileLoadTime;
} SIOCostSummary;
typedef struct STsdbQueryHandle {
@@ -1045,15 +1047,21 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
int32_t code = TSDB_CODE_SUCCESS;
*numOfBlocks = 0;
+ pQueryHandle->cost.headFileLoad += 1;
+ int64_t s = taosGetTimestampUs();
+
size_t numOfTables = 0;
if (pQueryHandle->loadType == BLOCK_LOAD_TABLE_SEQ_ORDER) {
- code = loadBlockInfo(pQueryHandle, pQueryHandle->activeIndex, numOfBlocks);
+ code = loadBlockInfo(pQueryHandle, pQueryHandle->activeIndex, numOfBlocks);
} else if (pQueryHandle->loadType == BLOCK_LOAD_OFFSET_SEQ_ORDER) {
numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
for (int32_t i = 0; i < numOfTables; ++i) {
code = loadBlockInfo(pQueryHandle, i, numOfBlocks);
if (code != TSDB_CODE_SUCCESS) {
+ int64_t e = taosGetTimestampUs();
+
+ pQueryHandle->cost.headFileLoadTime += (e - s);
return code;
}
}
@@ -1061,6 +1069,8 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
assert(0);
}
+ int64_t e = taosGetTimestampUs();
+ pQueryHandle->cost.headFileLoadTime += (e - s);
return code;
}
@@ -3731,8 +3741,8 @@ void tsdbCleanupQueryHandle(TsdbQueryHandleT queryHandle) {
pQueryHandle->next = doFreeColumnInfoData(pQueryHandle->next);
SIOCostSummary* pCost = &pQueryHandle->cost;
- tsdbDebug("%p :io-cost summary: statis-info:%"PRId64" us, datablock:%" PRId64" us, check data:%"PRId64" us, 0x%"PRIx64,
- pQueryHandle, pCost->statisInfoLoadTime, pCost->blockLoadTime, pCost->checkForNextTime, pQueryHandle->qId);
+ tsdbDebug("%p :io-cost summary: head-file read cnt:%"PRIu64", head-file time:%"PRIu64" us, statis-info:%"PRId64" us, datablock:%" PRId64" us, check data:%"PRId64" us, 0x%"PRIx64,
+ pQueryHandle, pCost->headFileLoad, pCost->headFileLoadTime, pCost->statisInfoLoadTime, pCost->blockLoadTime, pCost->checkForNextTime, pQueryHandle->qId);
tfree(pQueryHandle);
}