[td-32] refactor codes, change the prepare qinfo functions
This commit is contained in:
parent
f19c974227
commit
1d5bab8d46
|
@ -85,7 +85,7 @@ int32_t tscGetDataBlockFromList(void* pHashList, SDataBlockList* pDataBlockList,
|
||||||
STableDataBlocks** dataBlocks);
|
STableDataBlocks** dataBlocks);
|
||||||
|
|
||||||
SVnodeSidList* tscGetVnodeSidList(SSuperTableMeta* pMetricmeta, int32_t vnodeIdx);
|
SVnodeSidList* tscGetVnodeSidList(SSuperTableMeta* pMetricmeta, int32_t vnodeIdx);
|
||||||
STableSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx);
|
STableIdInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
*
|
*
|
||||||
|
|
|
@ -318,15 +318,16 @@ static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
|
||||||
SVnodeSidList *pSidList = (SVnodeSidList *)((char *)pMetricMeta + pMetricMeta->list[i]);
|
SVnodeSidList *pSidList = (SVnodeSidList *)((char *)pMetricMeta + pMetricMeta->list[i]);
|
||||||
|
|
||||||
for (int32_t j = 0; j < pSidList->numOfSids; ++j) {
|
for (int32_t j = 0; j < pSidList->numOfSids; ++j) {
|
||||||
STableSidExtInfo *pSidExt = tscGetMeterSidInfo(pSidList, j);
|
STableIdInfo *pSidExt = tscGetMeterSidInfo(pSidList, j);
|
||||||
|
|
||||||
for (int32_t k = 0; k < pQueryInfo->fieldsInfo.numOfOutputCols; ++k) {
|
for (int32_t k = 0; k < pQueryInfo->fieldsInfo.numOfOutputCols; ++k) {
|
||||||
SColIndexEx *pColIndex = &tscSqlExprGet(pQueryInfo, k)->colInfo;
|
SColIndexEx *pColIndex = &tscSqlExprGet(pQueryInfo, k)->colInfo;
|
||||||
int16_t offsetId = pColIndex->colIdx;
|
int16_t offsetId = pColIndex->colIdx;
|
||||||
|
|
||||||
assert((pColIndex->flag & TSDB_COL_TAG) != 0);
|
assert((pColIndex->flag & TSDB_COL_TAG) != 0);
|
||||||
|
assert(0);
|
||||||
|
|
||||||
char * val = pSidExt->tags + vOffset[offsetId];
|
char * val = NULL;//pSidExt->tags + vOffset[offsetId];
|
||||||
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, k);
|
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, k);
|
||||||
|
|
||||||
memcpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, k) * totalNumOfResults + pField->bytes * rowIdx, val,
|
memcpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, k) * totalNumOfResults + pField->bytes * rowIdx, val,
|
||||||
|
|
|
@ -598,7 +598,7 @@ static int32_t tscEstimateQueryMsgSize(SSqlCmd *pCmd, int32_t clauseIndex) {
|
||||||
SSuperTableMeta *pMetricMeta = pTableMetaInfo->pMetricMeta;
|
SSuperTableMeta *pMetricMeta = pTableMetaInfo->pMetricMeta;
|
||||||
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->vnodeIndex);
|
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->vnodeIndex);
|
||||||
|
|
||||||
int32_t meterInfoSize = (pMetricMeta->tagLen + sizeof(STableSidExtInfo)) * pVnodeSidList->numOfSids;
|
int32_t meterInfoSize = (pMetricMeta->tagLen + sizeof(STableIdInfo)) * pVnodeSidList->numOfSids;
|
||||||
int32_t outputColumnSize = pQueryInfo->exprsInfo.numOfExprs * sizeof(SSqlFuncExprMsg);
|
int32_t outputColumnSize = pQueryInfo->exprsInfo.numOfExprs * sizeof(SSqlFuncExprMsg);
|
||||||
|
|
||||||
int32_t size = meterInfoSize + outputColumnSize + srcColListSize + exprSize + MIN_QUERY_MSG_PKT_SIZE;
|
int32_t size = meterInfoSize + outputColumnSize + srcColListSize + exprSize + MIN_QUERY_MSG_PKT_SIZE;
|
||||||
|
@ -620,26 +620,23 @@ static char *doSerializeTableInfo(SSqlObj *pSql, int32_t numOfTables, int32_t vn
|
||||||
#ifdef _DEBUG_VIEW
|
#ifdef _DEBUG_VIEW
|
||||||
tscTrace("%p sid:%d, uid:%" PRIu64, pSql, pTableMetaInfo->pTableMeta->sid, pTableMetaInfo->pTableMeta->uid);
|
tscTrace("%p sid:%d, uid:%" PRIu64, pSql, pTableMetaInfo->pTableMeta->sid, pTableMetaInfo->pTableMeta->uid);
|
||||||
#endif
|
#endif
|
||||||
STableSidExtInfo *pTableMetaInfo = (STableSidExtInfo *)pMsg;
|
STableIdInfo *pTableMetaInfo = (STableIdInfo *)pMsg;
|
||||||
pTableMetaInfo->sid = htonl(pTableMeta->sid);
|
pTableMetaInfo->sid = htonl(pTableMeta->sid);
|
||||||
pTableMetaInfo->uid = htobe64(pTableMeta->uid);
|
pTableMetaInfo->uid = htobe64(pTableMeta->uid);
|
||||||
pTableMetaInfo->key = htobe64(tscGetSubscriptionProgress(pSql->pSubscription, pTableMeta->uid));
|
pTableMetaInfo->key = htobe64(tscGetSubscriptionProgress(pSql->pSubscription, pTableMeta->uid));
|
||||||
pMsg += sizeof(STableSidExtInfo);
|
pMsg += sizeof(STableIdInfo);
|
||||||
} else {
|
} else {
|
||||||
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->vnodeIndex);
|
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->vnodeIndex);
|
||||||
|
|
||||||
for (int32_t i = 0; i < numOfTables; ++i) {
|
for (int32_t i = 0; i < numOfTables; ++i) {
|
||||||
STableSidExtInfo *pTableMetaInfo = (STableSidExtInfo *)pMsg;
|
STableIdInfo *pTableIdInfo = (STableIdInfo *)pMsg;
|
||||||
STableSidExtInfo *pQueryMeterInfo = tscGetMeterSidInfo(pVnodeSidList, i);
|
STableIdInfo *pQueryMeterInfo = tscGetMeterSidInfo(pVnodeSidList, i);
|
||||||
|
|
||||||
pTableMetaInfo->sid = htonl(pQueryMeterInfo->sid);
|
pTableIdInfo->sid = htonl(pQueryMeterInfo->sid);
|
||||||
pTableMetaInfo->uid = htobe64(pQueryMeterInfo->uid);
|
pTableIdInfo->uid = htobe64(pQueryMeterInfo->uid);
|
||||||
pTableMetaInfo->key = htobe64(tscGetSubscriptionProgress(pSql->pSubscription, pQueryMeterInfo->uid));
|
pTableIdInfo->key = htobe64(tscGetSubscriptionProgress(pSql->pSubscription, pQueryMeterInfo->uid));
|
||||||
|
|
||||||
pMsg += sizeof(STableSidExtInfo);
|
pMsg += sizeof(STableIdInfo);
|
||||||
|
|
||||||
memcpy(pMsg, pQueryMeterInfo->tags, pMetricMeta->tagLen);
|
|
||||||
pMsg += pMetricMeta->tagLen;
|
|
||||||
|
|
||||||
#ifdef _DEBUG_VIEW
|
#ifdef _DEBUG_VIEW
|
||||||
tscTrace("%p sid:%d, uid:%" PRId64, pSql, pQueryMeterInfo->sid, pQueryMeterInfo->uid);
|
tscTrace("%p sid:%d, uid:%" PRId64, pSql, pQueryMeterInfo->sid, pQueryMeterInfo->uid);
|
||||||
|
@ -2067,7 +2064,7 @@ int tscProcessMetricMetaRsp(SSqlObj *pSql) {
|
||||||
pMeta->numOfVnodes = htonl(pMeta->numOfVnodes);
|
pMeta->numOfVnodes = htonl(pMeta->numOfVnodes);
|
||||||
pMeta->tagLen = htons(pMeta->tagLen);
|
pMeta->tagLen = htons(pMeta->tagLen);
|
||||||
|
|
||||||
size += pMeta->numOfVnodes * sizeof(SVnodeSidList *) + pMeta->numOfTables * sizeof(STableSidExtInfo *);
|
size += pMeta->numOfVnodes * sizeof(SVnodeSidList *) + pMeta->numOfTables * sizeof(STableIdInfo *);
|
||||||
|
|
||||||
char *pBuf = calloc(1, size);
|
char *pBuf = calloc(1, size);
|
||||||
if (pBuf == NULL) {
|
if (pBuf == NULL) {
|
||||||
|
@ -2093,16 +2090,16 @@ int tscProcessMetricMetaRsp(SSqlObj *pSql) {
|
||||||
|
|
||||||
tscTrace("%p metricmeta:vid:%d,numOfTables:%d", pSql, i, pLists->numOfSids);
|
tscTrace("%p metricmeta:vid:%d,numOfTables:%d", pSql, i, pLists->numOfSids);
|
||||||
|
|
||||||
pBuf += sizeof(SVnodeSidList) + sizeof(STableSidExtInfo *) * pSidLists->numOfSids;
|
pBuf += sizeof(SVnodeSidList) + sizeof(STableIdInfo *) * pSidLists->numOfSids;
|
||||||
rsp += sizeof(SVnodeSidList);
|
rsp += sizeof(SVnodeSidList);
|
||||||
|
|
||||||
size_t elemSize = sizeof(STableSidExtInfo) + pNewMetricMeta->tagLen;
|
size_t elemSize = sizeof(STableIdInfo) + pNewMetricMeta->tagLen;
|
||||||
for (int32_t j = 0; j < pSidLists->numOfSids; ++j) {
|
for (int32_t j = 0; j < pSidLists->numOfSids; ++j) {
|
||||||
pLists->pSidExtInfoList[j] = pBuf - (char *)pLists;
|
pLists->pSidExtInfoList[j] = pBuf - (char *)pLists;
|
||||||
memcpy(pBuf, rsp, elemSize);
|
memcpy(pBuf, rsp, elemSize);
|
||||||
|
|
||||||
((STableSidExtInfo *)pBuf)->uid = htobe64(((STableSidExtInfo *)pBuf)->uid);
|
((STableIdInfo *)pBuf)->uid = htobe64(((STableIdInfo *)pBuf)->uid);
|
||||||
((STableSidExtInfo *)pBuf)->sid = htonl(((STableSidExtInfo *)pBuf)->sid);
|
((STableIdInfo *)pBuf)->sid = htonl(((STableIdInfo *)pBuf)->sid);
|
||||||
|
|
||||||
rsp += elemSize;
|
rsp += elemSize;
|
||||||
pBuf += elemSize;
|
pBuf += elemSize;
|
||||||
|
|
|
@ -202,7 +202,7 @@ int tscUpdateSubscription(STscObj* pObj, SSub* pSub) {
|
||||||
for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) {
|
for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) {
|
||||||
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i);
|
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i);
|
||||||
for (int32_t j = 0; j < pVnodeSidList->numOfSids; j++) {
|
for (int32_t j = 0; j < pVnodeSidList->numOfSids; j++) {
|
||||||
STableSidExtInfo *pTableMetaInfo = tscGetMeterSidInfo(pVnodeSidList, j);
|
STableIdInfo *pTableMetaInfo = tscGetMeterSidInfo(pVnodeSidList, j);
|
||||||
int64_t uid = pTableMetaInfo->uid;
|
int64_t uid = pTableMetaInfo->uid;
|
||||||
progress[numOfTables].uid = uid;
|
progress[numOfTables].uid = uid;
|
||||||
progress[numOfTables++].key = tscGetSubscriptionProgress(pSub, uid);
|
progress[numOfTables++].key = tscGetSubscriptionProgress(pSub, uid);
|
||||||
|
|
|
@ -191,7 +191,7 @@ SVnodeSidList* tscGetVnodeSidList(SSuperTableMeta* pMetricmeta, int32_t vnodeIdx
|
||||||
return (SVnodeSidList*)(pMetricmeta->list[vnodeIdx] + (char*)pMetricmeta);
|
return (SVnodeSidList*)(pMetricmeta->list[vnodeIdx] + (char*)pMetricmeta);
|
||||||
}
|
}
|
||||||
|
|
||||||
STableSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx) {
|
STableIdInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx) {
|
||||||
if (pSidList == NULL) {
|
if (pSidList == NULL) {
|
||||||
tscError("illegal sidlist");
|
tscError("illegal sidlist");
|
||||||
return 0;
|
return 0;
|
||||||
|
@ -206,7 +206,7 @@ STableSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx) {
|
||||||
|
|
||||||
assert(pSidList->pSidExtInfoList[idx] >= 0);
|
assert(pSidList->pSidExtInfoList[idx] >= 0);
|
||||||
|
|
||||||
return (STableSidExtInfo*)(pSidList->pSidExtInfoList[idx] + (char*)pSidList);
|
return (STableIdInfo*)(pSidList->pSidExtInfoList[idx] + (char*)pSidList);
|
||||||
}
|
}
|
||||||
|
|
||||||
bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) {
|
bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) {
|
||||||
|
|
|
@ -15,13 +15,16 @@
|
||||||
|
|
||||||
#define _DEFAULT_SOURCE
|
#define _DEFAULT_SOURCE
|
||||||
#include "os.h"
|
#include "os.h"
|
||||||
|
|
||||||
#include "taoserror.h"
|
#include "taoserror.h"
|
||||||
#include "taosmsg.h"
|
#include "taosmsg.h"
|
||||||
#include "tlog.h"
|
#include "tlog.h"
|
||||||
#include "tqueue.h"
|
#include "tqueue.h"
|
||||||
#include "trpc.h"
|
#include "trpc.h"
|
||||||
#include "dnodeRead.h"
|
|
||||||
#include "dnodeMgmt.h"
|
#include "dnodeMgmt.h"
|
||||||
|
#include "dnodeRead.h"
|
||||||
|
#include "queryExecutor.h"
|
||||||
|
|
||||||
typedef struct {
|
typedef struct {
|
||||||
int32_t code;
|
int32_t code;
|
||||||
|
@ -223,8 +226,12 @@ static void dnodeProcessReadResult(SReadMsg *pRead) {
|
||||||
}
|
}
|
||||||
|
|
||||||
static void dnodeProcessQueryMsg(SReadMsg *pMsg) {
|
static void dnodeProcessQueryMsg(SReadMsg *pMsg) {
|
||||||
void *pQInfo = (void*)100;
|
SQueryTableMsg* pQueryTableMsg = (SQueryTableMsg*) pMsg->pCont;
|
||||||
dTrace("query msg is disposed, qInfo:%p", pQInfo);
|
|
||||||
|
SQInfo* pQInfo = NULL;
|
||||||
|
int32_t ret = qCreateQueryInfo(pQueryTableMsg, &pQInfo);
|
||||||
|
|
||||||
|
dTrace("query msg is disposed, qInfo:%p", pQueryTableMsg);
|
||||||
|
|
||||||
SQueryTableRsp *pRsp = (SQueryTableRsp *) rpcMallocCont(sizeof(SQueryTableRsp));
|
SQueryTableRsp *pRsp = (SQueryTableRsp *) rpcMallocCont(sizeof(SQueryTableRsp));
|
||||||
pRsp->code = 0;
|
pRsp->code = 0;
|
||||||
|
|
|
@ -434,15 +434,11 @@ typedef struct SColumnInfo {
|
||||||
SColumnFilterInfo *filters;
|
SColumnFilterInfo *filters;
|
||||||
} SColumnInfo;
|
} SColumnInfo;
|
||||||
|
|
||||||
/*
|
typedef struct STableIdInfo {
|
||||||
* enable vnode to understand how to group several tables with different tag;
|
|
||||||
*/
|
|
||||||
typedef struct STableSidExtInfo {
|
|
||||||
int32_t sid;
|
int32_t sid;
|
||||||
int64_t uid;
|
int64_t uid;
|
||||||
TSKEY key; // key for subscription
|
TSKEY key; // last accessed ts, for subscription
|
||||||
char tags[];
|
} STableIdInfo;
|
||||||
} STableSidExtInfo;
|
|
||||||
|
|
||||||
typedef struct STimeWindow {
|
typedef struct STimeWindow {
|
||||||
TSKEY skey;
|
TSKEY skey;
|
||||||
|
@ -670,7 +666,7 @@ typedef struct {
|
||||||
SVnodeDesc vpeerDesc[TSDB_VNODES_SUPPORT];
|
SVnodeDesc vpeerDesc[TSDB_VNODES_SUPPORT];
|
||||||
int16_t index; // used locally
|
int16_t index; // used locally
|
||||||
int32_t numOfSids;
|
int32_t numOfSids;
|
||||||
int32_t pSidExtInfoList[]; // offset value of STableSidExtInfo
|
int32_t pSidExtInfoList[]; // offset value of STableIdInfo
|
||||||
} SVnodeSidList;
|
} SVnodeSidList;
|
||||||
|
|
||||||
typedef struct {
|
typedef struct {
|
||||||
|
|
|
@ -124,7 +124,7 @@ typedef struct tTagSchema {
|
||||||
typedef struct tSidSet {
|
typedef struct tSidSet {
|
||||||
int32_t numOfSids;
|
int32_t numOfSids;
|
||||||
int32_t numOfSubSet;
|
int32_t numOfSubSet;
|
||||||
STableSidExtInfo **pSids;
|
STableIdInfo **pSids;
|
||||||
int32_t * starterPos; // position of each subgroup, generated according to
|
int32_t * starterPos; // position of each subgroup, generated according to
|
||||||
|
|
||||||
SColumnModel *pColumnModel;
|
SColumnModel *pColumnModel;
|
||||||
|
|
|
@ -152,7 +152,7 @@ typedef struct SQueryCostSummary {
|
||||||
typedef struct SQueryRuntimeEnv {
|
typedef struct SQueryRuntimeEnv {
|
||||||
SResultInfo* resultInfo; // todo refactor to merge with SWindowResInfo
|
SResultInfo* resultInfo; // todo refactor to merge with SWindowResInfo
|
||||||
SQuery* pQuery;
|
SQuery* pQuery;
|
||||||
void* pTabObj;
|
// void* pTabObj;
|
||||||
SData** pInterpoBuf;
|
SData** pInterpoBuf;
|
||||||
SQLFunctionCtx* pCtx;
|
SQLFunctionCtx* pCtx;
|
||||||
int16_t numOfRowsPerPage;
|
int16_t numOfRowsPerPage;
|
||||||
|
@ -204,7 +204,7 @@ typedef struct SQInfo {
|
||||||
* @param pQInfo
|
* @param pQInfo
|
||||||
* @return
|
* @return
|
||||||
*/
|
*/
|
||||||
int32_t qCreateQueryInfo(void* pReadMsg, SQInfo** pQInfo);
|
int32_t qCreateQueryInfo(SQueryTableMsg* pQueryTableMsg, SQInfo** pQInfo);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* query on single table
|
* query on single table
|
||||||
|
|
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue