Merge branch 'develop' into feature/sangshuduo/ci-change-timezone-to-gmt8
This commit is contained in:
Shuduo Sang
commit
71970e4c17
|
|
@ -12,12 +12,10 @@
|
|||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
#include <taosmsg.h>
|
||||
#include "os.h"
|
||||
#include "qfill.h"
|
||||
|
||||
#include "hash.h"
|
||||
#include "hashfunc.h"
|
||||
#include "qExecutor.h"
|
||||
#include "qUtil.h"
|
||||
#include "qast.h"
|
||||
|
|
@ -25,7 +23,6 @@
|
|||
#include "query.h"
|
||||
#include "queryLog.h"
|
||||
#include "taosmsg.h"
|
||||
#include "tdataformat.h"
|
||||
#include "tlosertree.h"
|
||||
#include "tscUtil.h" // todo move the function to common module
|
||||
#include "tscompression.h"
|
||||
|
|
@ -91,6 +88,9 @@ typedef struct {
|
|||
} SQueryStatusInfo;
|
||||
|
||||
#define CLEAR_QUERY_STATUS(q, st) ((q)->status &= (~(st)))
|
||||
#define GET_NUM_OF_TABLEGROUP(q) taosArrayGetSize((q)->tableqinfoGroupInfo.pGroupList)
|
||||
#define GET_TABLEGROUP(q, _index) ((SArray*) taosArrayGetP((q)->tableqinfoGroupInfo.pGroupList, (_index)))
|
||||
|
||||
static void setQueryStatus(SQuery *pQuery, int8_t status);
|
||||
|
||||
static bool isIntervalQuery(SQuery *pQuery) { return pQuery->intervalTime > 0; }
|
||||
|
|
@ -1708,7 +1708,23 @@ static bool onlyFirstQuery(SQuery *pQuery) { return onlyOneQueryType(pQuery, TSD
|
|||
|
||||
static bool onlyLastQuery(SQuery *pQuery) { return onlyOneQueryType(pQuery, TSDB_FUNC_LAST, TSDB_FUNC_LAST_DST); }
|
||||
|
||||
static void changeExecuteScanOrder(SQuery *pQuery, bool stableQuery) {
|
||||
// todo refactor, add iterator
|
||||
static void doExchangeTimeWindow(SQInfo* pQInfo) {
|
||||
size_t t = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
for(int32_t i = 0; i < t; ++i) {
|
||||
SArray* p1 = GET_TABLEGROUP(pQInfo, i);
|
||||
|
||||
size_t len = taosArrayGetSize(p1);
|
||||
for(int32_t j = 0; j < len; ++j) {
|
||||
STableQueryInfo* pTableQueryInfo = (STableQueryInfo*) taosArrayGetP(p1, j);
|
||||
SWAP(pTableQueryInfo->win.skey, pTableQueryInfo->win.ekey, TSKEY);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
|
||||
SQuery* pQuery = pQInfo->runtimeEnv.pQuery;
|
||||
|
||||
// in case of point-interpolation query, use asc order scan
|
||||
char msg[] = "QInfo:%p scan order changed for %s query, old:%d, new:%d, qrange exchanged, old qrange:%" PRId64
|
||||
"-%" PRId64 ", new qrange:%" PRId64 "-%" PRId64;
|
||||
|
|
@ -1748,6 +1764,7 @@ static void changeExecuteScanOrder(SQuery *pQuery, bool stableQuery) {
|
|||
pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
|
||||
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
doExchangeTimeWindow(pQInfo);
|
||||
}
|
||||
|
||||
pQuery->order.order = TSDB_ORDER_ASC;
|
||||
|
|
@ -1757,6 +1774,7 @@ static void changeExecuteScanOrder(SQuery *pQuery, bool stableQuery) {
|
|||
pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
|
||||
|
||||
SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
|
||||
doExchangeTimeWindow(pQInfo);
|
||||
}
|
||||
|
||||
pQuery->order.order = TSDB_ORDER_DESC;
|
||||
|
|
@ -2489,10 +2507,10 @@ int32_t mergeIntoGroupResult(SQInfo *pQInfo) {
|
|||
int64_t st = taosGetTimestampMs();
|
||||
int32_t ret = TSDB_CODE_SUCCESS;
|
||||
|
||||
int32_t numOfGroups = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
int32_t numOfGroups = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
|
||||
while (pQInfo->groupIndex < numOfGroups) {
|
||||
SArray *group = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, pQInfo->groupIndex);
|
||||
SArray *group = GET_TABLEGROUP(pQInfo, pQInfo->groupIndex);
|
||||
ret = mergeIntoGroupResultImpl(pQInfo, group);
|
||||
if (ret < 0) { // not enough disk space to save the data into disk
|
||||
return -1;
|
||||
|
|
@ -2525,7 +2543,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
|
|||
}
|
||||
|
||||
// check if all results has been sent to client
|
||||
int32_t numOfGroup = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
int32_t numOfGroup = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
if (pQInfo->numOfGroupResultPages == 0 && pQInfo->groupIndex == numOfGroup) {
|
||||
pQInfo->tableIndex = pQInfo->tableqinfoGroupInfo.numOfTables; // set query completed
|
||||
return;
|
||||
|
|
@ -2859,10 +2877,10 @@ void disableFuncInReverseScan(SQInfo *pQInfo) {
|
|||
}
|
||||
}
|
||||
|
||||
int32_t numOfGroups = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
int32_t numOfGroups = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
|
||||
for(int32_t i = 0; i < numOfGroups; ++i) {
|
||||
SArray *group = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, i);
|
||||
SArray *group = GET_TABLEGROUP(pQInfo, i);
|
||||
|
||||
size_t t = taosArrayGetSize(group);
|
||||
for (int32_t j = 0; j < t; ++j) {
|
||||
|
|
@ -3349,7 +3367,7 @@ void setWindowResOutputBufInitCtx(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *
|
|||
SQLFunctionCtx *pCtx = &pRuntimeEnv->pCtx[i];
|
||||
|
||||
pCtx->resultInfo = &pResult->resultInfo[i];
|
||||
if (pCtx->resultInfo->complete) {
|
||||
if (pCtx->resultInfo->initialized && pCtx->resultInfo->complete) {
|
||||
continue;
|
||||
}
|
||||
|
||||
|
|
@ -3479,7 +3497,7 @@ static int32_t getNumOfSubset(SQInfo *pQInfo) {
|
|||
if (isGroupbyNormalCol(pQuery->pGroupbyExpr) || (isIntervalQuery(pQuery))) {
|
||||
totalSubset = numOfClosedTimeWindow(&pQInfo->runtimeEnv.windowResInfo);
|
||||
} else {
|
||||
totalSubset = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
totalSubset = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
}
|
||||
|
||||
return totalSubset;
|
||||
|
|
@ -3619,36 +3637,40 @@ void stableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *
|
|||
bool queryHasRemainResults(SQueryRuntimeEnv* pRuntimeEnv) {
|
||||
SQuery *pQuery = pRuntimeEnv->pQuery;
|
||||
SFillInfo *pFillInfo = pRuntimeEnv->pFillInfo;
|
||||
|
||||
// todo refactor
|
||||
if (pQuery->fillType == TSDB_FILL_NONE || (pQuery->fillType != TSDB_FILL_NONE && isPointInterpoQuery(pQuery))) {
|
||||
assert(pFillInfo == NULL);
|
||||
|
||||
if (pQuery->limit.limit > 0 && pQuery->rec.total >= pQuery->limit.limit) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (pQuery->limit.limit > 0 && pQuery->rec.rows >= pQuery->limit.limit) {
|
||||
return false;
|
||||
}
|
||||
if (pQuery->fillType != TSDB_FILL_NONE && !isPointInterpoQuery(pQuery)) {
|
||||
// There are results not returned to client yet, so filling operation applied to the remain result is required
|
||||
// in the first place.
|
||||
int32_t remain = taosNumOfRemainRows(pFillInfo);
|
||||
if (remain > 0) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// There are results not returned to client, fill operation applied to the remain result set in the
|
||||
// first place is required.
|
||||
int32_t remain = taosNumOfRemainRows(pFillInfo);
|
||||
if (remain > 0) {
|
||||
return true;
|
||||
}
|
||||
|
||||
/*
|
||||
* While the code reaches here, there are no results returned to client now.
|
||||
* If query is not completed yet, the gaps between two results blocks need to be handled after next data block
|
||||
* is retrieved from TSDB.
|
||||
*
|
||||
* NOTE: If the result set is not the first block, the gap in front of the result set will be filled. If the result
|
||||
* set is the FIRST result block, the gap between the start time of query time window and the timestamp of the
|
||||
* first result row in the actual result set will fill nothing.
|
||||
*/
|
||||
if (Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED)) {
|
||||
int32_t numOfTotal = getFilledNumOfRes(pFillInfo, pQuery->window.ekey, pQuery->rec.capacity);
|
||||
return numOfTotal > 0;
|
||||
/*
|
||||
* While the code reaches here, there are no results remains now.
|
||||
* If query is not completed yet, the gaps between two results blocks need to be handled after next data block
|
||||
* is retrieved from TSDB.
|
||||
*
|
||||
* NOTE: If the result set is not the first block, the gap in front of the result set will be filled. If the result
|
||||
* set is the FIRST result block, the gap between the start time of query time window and the timestamp of the
|
||||
* first result row in the actual result set will fill nothing.
|
||||
*/
|
||||
if (Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED)) {
|
||||
int32_t numOfTotal = getFilledNumOfRes(pFillInfo, pQuery->window.ekey, pQuery->rec.capacity);
|
||||
return numOfTotal > 0;
|
||||
}
|
||||
|
||||
} else {
|
||||
// there are results waiting for returned to client.
|
||||
if (Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED) &&
|
||||
(isGroupbyNormalCol(pQuery->pGroupbyExpr) || isIntervalQuery(pQuery)) &&
|
||||
(pRuntimeEnv->windowResInfo.size > 0)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
|
|
@ -3690,7 +3712,7 @@ static void doCopyQueryResultToMsg(SQInfo *pQInfo, int32_t numOfRows, char *data
|
|||
}
|
||||
}
|
||||
|
||||
int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int32_t *numOfInterpo) {
|
||||
int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int32_t *numOfFilled) {
|
||||
SQInfo* pQInfo = GET_QINFO_ADDR(pRuntimeEnv);
|
||||
SQuery *pQuery = pRuntimeEnv->pQuery;
|
||||
SFillInfo* pFillInfo = pRuntimeEnv->pFillInfo;
|
||||
|
|
@ -3995,7 +4017,7 @@ static void setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery) {
|
|||
&& (!isGroupbyNormalCol(pQuery->pGroupbyExpr))
|
||||
&& (!isFixedOutputQuery(pQuery))
|
||||
) {
|
||||
SArray* pa = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, 0);
|
||||
SArray* pa = GET_TABLEGROUP(pQInfo, 0);
|
||||
STableQueryInfo* pCheckInfo = taosArrayGetP(pa, 0);
|
||||
cond.twindow = pCheckInfo->win;
|
||||
}
|
||||
|
|
@ -4039,7 +4061,7 @@ int32_t doInitQInfo(SQInfo *pQInfo, void *param, void *tsdb, int32_t vgId, bool
|
|||
pQuery->precision = tsdbGetCfg(tsdb)->precision;
|
||||
|
||||
setScanLimitationByResultBuffer(pQuery);
|
||||
changeExecuteScanOrder(pQuery, false);
|
||||
changeExecuteScanOrder(pQInfo, false);
|
||||
setupQueryHandle(tsdb, pQInfo, isSTableQuery);
|
||||
|
||||
pQInfo->tsdb = tsdb;
|
||||
|
|
@ -4142,9 +4164,9 @@ static int64_t scanMultiTableDataBlocks(SQInfo *pQInfo) {
|
|||
STableQueryInfo *pTableQueryInfo = NULL;
|
||||
|
||||
// todo opt performance using hash table
|
||||
size_t numOfGroup = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
size_t numOfGroup = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
for (int32_t i = 0; i < numOfGroup; ++i) {
|
||||
SArray *group = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, i);
|
||||
SArray *group = GET_TABLEGROUP(pQInfo, i);
|
||||
|
||||
size_t num = taosArrayGetSize(group);
|
||||
for (int32_t j = 0; j < num; ++j) {
|
||||
|
|
@ -4197,7 +4219,7 @@ static bool multiTableMultioutputHelper(SQInfo *pQInfo, int32_t index) {
|
|||
SQuery * pQuery = pRuntimeEnv->pQuery;
|
||||
|
||||
setQueryStatus(pQuery, QUERY_NOT_COMPLETED);
|
||||
SArray *group = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, 0);
|
||||
SArray *group = GET_TABLEGROUP(pQInfo, 0);
|
||||
STableQueryInfo* pCheckInfo = taosArrayGetP(group, index);
|
||||
|
||||
setTagVal(pRuntimeEnv, pCheckInfo->pTable, pQInfo->tsdb);
|
||||
|
|
@ -4261,7 +4283,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
|
|||
SQuery * pQuery = pRuntimeEnv->pQuery;
|
||||
setQueryStatus(pQuery, QUERY_COMPLETED);
|
||||
|
||||
size_t numOfGroups = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
size_t numOfGroups = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
|
||||
if (isPointInterpoQuery(pQuery) || isFirstLastRowQuery(pQuery)) {
|
||||
resetCtxOutputBuf(pRuntimeEnv);
|
||||
|
|
@ -4311,7 +4333,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
|
|||
taosArrayDestroy(s);
|
||||
|
||||
// here we simply set the first table as current table
|
||||
pQuery->current = (STableQueryInfo*) taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, 0);
|
||||
pQuery->current = (STableQueryInfo*) GET_TABLEGROUP(pQInfo, 0);
|
||||
scanOneTableDataBlocks(pRuntimeEnv, pQuery->current->lastKey);
|
||||
|
||||
int64_t numOfRes = getNumOfResult(pRuntimeEnv);
|
||||
|
|
@ -4424,7 +4446,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
|
|||
resetCtxOutputBuf(pRuntimeEnv);
|
||||
resetTimeWindowInfo(pRuntimeEnv, &pRuntimeEnv->windowResInfo);
|
||||
|
||||
SArray *group = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, 0);
|
||||
SArray *group = GET_TABLEGROUP(pQInfo, 0);
|
||||
assert(taosArrayGetSize(group) == pQInfo->tableqinfoGroupInfo.numOfTables &&
|
||||
1 == taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList));
|
||||
|
||||
|
|
@ -4575,9 +4597,9 @@ static void doCloseAllTimeWindowAfterScan(SQInfo *pQInfo) {
|
|||
SQuery *pQuery = pQInfo->runtimeEnv.pQuery;
|
||||
|
||||
if (isIntervalQuery(pQuery)) {
|
||||
size_t numOfGroup = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
size_t numOfGroup = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
for (int32_t i = 0; i < numOfGroup; ++i) {
|
||||
SArray *group = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, i);
|
||||
SArray *group = GET_TABLEGROUP(pQInfo, i);
|
||||
|
||||
size_t num = taosArrayGetSize(group);
|
||||
for (int32_t j = 0; j < num; ++j) {
|
||||
|
|
@ -4794,7 +4816,7 @@ static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
|
|||
SQuery *pQuery = pRuntimeEnv->pQuery;
|
||||
pQuery->current = pTableInfo;
|
||||
|
||||
int32_t numOfInterpo = 0;
|
||||
int32_t numOfFilled = 0;
|
||||
TSKEY newStartKey = TSKEY_INITIAL_VAL;
|
||||
|
||||
// skip blocks without load the actual data block from file if no filter condition present
|
||||
|
|
@ -4822,9 +4844,9 @@ static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
|
|||
} else {
|
||||
taosFillSetStartInfo(pRuntimeEnv->pFillInfo, pQuery->rec.rows, pQuery->window.ekey);
|
||||
taosFillCopyInputDataFromFilePage(pRuntimeEnv->pFillInfo, (tFilePage**) pQuery->sdata);
|
||||
numOfInterpo = 0;
|
||||
numOfFilled = 0;
|
||||
|
||||
pQuery->rec.rows = doFillGapsInResults(pRuntimeEnv, (tFilePage **)pQuery->sdata, &numOfInterpo);
|
||||
pQuery->rec.rows = doFillGapsInResults(pRuntimeEnv, (tFilePage **)pQuery->sdata, &numOfFilled);
|
||||
if (pQuery->rec.rows > 0 || Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED)) {
|
||||
limitResults(pRuntimeEnv);
|
||||
break;
|
||||
|
|
@ -4843,7 +4865,7 @@ static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
|
|||
clearFirstNTimeWindow(pRuntimeEnv, pQInfo->groupIndex);
|
||||
}
|
||||
|
||||
pQInfo->pointsInterpo += numOfInterpo;
|
||||
pQInfo->pointsInterpo += numOfFilled;
|
||||
}
|
||||
|
||||
static void tableQueryImpl(SQInfo *pQInfo) {
|
||||
|
|
@ -4851,45 +4873,41 @@ static void tableQueryImpl(SQInfo *pQInfo) {
|
|||
SQuery * pQuery = pRuntimeEnv->pQuery;
|
||||
|
||||
if (queryHasRemainResults(pRuntimeEnv)) {
|
||||
/*
|
||||
* There are remain results that are not returned due to result interpolation
|
||||
* So, we do keep in this procedure instead of launching retrieve procedure for next results.
|
||||
*/
|
||||
int32_t numOfInterpo = 0;
|
||||
pQuery->rec.rows = doFillGapsInResults(pRuntimeEnv, (tFilePage **)pQuery->sdata, &numOfInterpo);
|
||||
|
||||
if (pQuery->rec.rows > 0) {
|
||||
limitResults(pRuntimeEnv);
|
||||
}
|
||||
|
||||
qDebug("QInfo:%p current:%" PRId64 " returned, total:%" PRId64, pQInfo, pQuery->rec.rows, pQuery->rec.total);
|
||||
return;
|
||||
}
|
||||
|
||||
// here we have scan all qualified data in both data file and cache
|
||||
if (Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED)) {
|
||||
// continue to get push data from the group result
|
||||
if (isGroupbyNormalCol(pQuery->pGroupbyExpr) ||
|
||||
((isIntervalQuery(pQuery) && pQuery->rec.total < pQuery->limit.limit))) {
|
||||
// todo limit the output for interval query?
|
||||
if (pQuery->fillType != TSDB_FILL_NONE) {
|
||||
/*
|
||||
* There are remain results that are not returned due to result interpolation
|
||||
* So, we do keep in this procedure instead of launching retrieve procedure for next results.
|
||||
*/
|
||||
int32_t numOfFilled = 0;
|
||||
pQuery->rec.rows = doFillGapsInResults(pRuntimeEnv, (tFilePage **)pQuery->sdata, &numOfFilled);
|
||||
|
||||
if (pQuery->rec.rows > 0) {
|
||||
limitResults(pRuntimeEnv);
|
||||
}
|
||||
|
||||
qDebug("QInfo:%p current:%" PRId64 " returned, total:%" PRId64, pQInfo, pQuery->rec.rows, pQuery->rec.total);
|
||||
return;
|
||||
} else {
|
||||
pQuery->rec.rows = 0;
|
||||
pQInfo->groupIndex = 0; // always start from 0
|
||||
|
||||
if (pRuntimeEnv->windowResInfo.size > 0) {
|
||||
copyFromWindowResToSData(pQInfo, pRuntimeEnv->windowResInfo.pResult);
|
||||
pQuery->rec.rows += pQuery->rec.rows;
|
||||
|
||||
clearFirstNTimeWindow(pRuntimeEnv, pQInfo->groupIndex);
|
||||
|
||||
if (pQuery->rec.rows > 0) {
|
||||
qDebug("QInfo:%p %"PRId64" rows returned from group results, total:%"PRId64"", pQInfo, pQuery->rec.rows, pQuery->rec.total);
|
||||
|
||||
// there are not data remains
|
||||
if (pRuntimeEnv->windowResInfo.size <= 0) {
|
||||
qDebug("QInfo:%p query over, %"PRId64" rows are returned", pQInfo, pQuery->rec.total);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
qDebug("QInfo:%p query over, %"PRId64" rows are returned", pQInfo, pQuery->rec.total);
|
||||
return;
|
||||
}
|
||||
|
||||
// number of points returned during this query
|
||||
|
|
@ -4897,7 +4915,7 @@ static void tableQueryImpl(SQInfo *pQInfo) {
|
|||
int64_t st = taosGetTimestampUs();
|
||||
|
||||
assert(pQInfo->tableqinfoGroupInfo.numOfTables == 1);
|
||||
SArray* g = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, 0);
|
||||
SArray* g = GET_TABLEGROUP(pQInfo, 0);
|
||||
STableQueryInfo* item = taosArrayGetP(g, 0);
|
||||
|
||||
// group by normal column, sliding window query, interval query are handled by interval query processor
|
||||
|
|
@ -5658,26 +5676,6 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
|
|||
return pQInfo;
|
||||
|
||||
_cleanup:
|
||||
//tfree(pQuery->fillVal);
|
||||
|
||||
//if (pQuery->sdata != NULL) {
|
||||
// for (int16_t col = 0; col < pQuery->numOfOutput; ++col) {
|
||||
// tfree(pQuery->sdata[col]);
|
||||
// }
|
||||
//}
|
||||
|
||||
//
|
||||
//tfree(pQuery->sdata);
|
||||
//tfree(pQuery->pFilterInfo);
|
||||
//tfree(pQuery->colList);
|
||||
|
||||
//tfree(pExprs);
|
||||
//tfree(pGroupbyExpr);
|
||||
|
||||
//taosArrayDestroy(pQInfo->arrTableIdInfo);
|
||||
//tsdbDestoryTableGroup(&pQInfo->tableGroupInfo);
|
||||
//
|
||||
//tfree(pQInfo);
|
||||
freeQInfo(pQInfo);
|
||||
|
||||
return NULL;
|
||||
|
|
@ -5712,7 +5710,6 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ
|
|||
UNUSED(ret);
|
||||
}
|
||||
|
||||
// only the successful complete requries the sem_post/over = 1 operations.
|
||||
if ((QUERY_IS_ASC_QUERY(pQuery) && (pQuery->window.skey > pQuery->window.ekey)) ||
|
||||
(!QUERY_IS_ASC_QUERY(pQuery) && (pQuery->window.ekey > pQuery->window.skey))) {
|
||||
qDebug("QInfo:%p no result in time range %" PRId64 "-%" PRId64 ", order %d", pQInfo, pQuery->window.skey,
|
||||
|
|
@ -5722,7 +5719,7 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ
|
|||
sem_post(&pQInfo->dataReady);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
if (pQInfo->tableqinfoGroupInfo.numOfTables == 0) {
|
||||
qDebug("QInfo:%p no table qualified for tag filter, abort query", pQInfo);
|
||||
setQueryStatus(pQuery, QUERY_COMPLETED);
|
||||
|
|
@ -5784,9 +5781,9 @@ static void freeQInfo(SQInfo *pQInfo) {
|
|||
}
|
||||
|
||||
// todo refactor, extract method to destroytableDataInfo
|
||||
int32_t numOfGroups = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
int32_t numOfGroups = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
for (int32_t i = 0; i < numOfGroups; ++i) {
|
||||
SArray *p = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, i);
|
||||
SArray *p = GET_TABLEGROUP(pQInfo, i);;
|
||||
|
||||
size_t num = taosArrayGetSize(p);
|
||||
for(int32_t j = 0; j < num; ++j) {
|
||||
|
|
@ -6053,6 +6050,11 @@ void qTableQuery(qinfo_t qinfo, void (*fp)(void*), void* param) {
|
|||
return;
|
||||
}
|
||||
|
||||
if (pQInfo->tableqinfoGroupInfo.numOfTables == 0) {
|
||||
qDebug("QInfo:%p no table exists for query, abort", pQInfo);
|
||||
return;
|
||||
}
|
||||
|
||||
qDebug("QInfo:%p query task is launched", pQInfo);
|
||||
|
||||
if (onlyQueryTags(pQInfo->runtimeEnv.pQuery)) {
|
||||
|
|
@ -6175,14 +6177,14 @@ static void buildTagQueryResult(SQInfo* pQInfo) {
|
|||
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
|
||||
SQuery * pQuery = pRuntimeEnv->pQuery;
|
||||
|
||||
size_t numOfGroup = taosArrayGetSize(pQInfo->tableqinfoGroupInfo.pGroupList);
|
||||
size_t numOfGroup = GET_NUM_OF_TABLEGROUP(pQInfo);
|
||||
assert(numOfGroup == 0 || numOfGroup == 1);
|
||||
|
||||
if (numOfGroup == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
SArray* pa = taosArrayGetP(pQInfo->tableqinfoGroupInfo.pGroupList, 0);
|
||||
SArray* pa = GET_TABLEGROUP(pQInfo, 0);
|
||||
|
||||
size_t num = taosArrayGetSize(pa);
|
||||
assert(num == pQInfo->tableqinfoGroupInfo.numOfTables);
|
||||
|
|
|
|||
|
|
@ -133,7 +133,6 @@ void clearFirstNTimeWindow(SQueryRuntimeEnv *pRuntimeEnv, int32_t num) {
|
|||
}
|
||||
|
||||
pWindowResInfo->size = remain;
|
||||
printf("---------------size:%ld\n", taosHashGetSize(pWindowResInfo->hashList));
|
||||
for (int32_t k = 0; k < pWindowResInfo->size; ++k) {
|
||||
SWindowResult *pResult = &pWindowResInfo->pResult[k];
|
||||
int32_t *p = (int32_t *)taosHashGet(pWindowResInfo->hashList, (const char *)&pResult->window.skey,
|
||||
|
|
|
|||
|
|
@ -121,7 +121,7 @@ int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup) {
|
|||
if (tsendfile(pHelper->files.nHeadF.fd, pHelper->files.headF.fd, NULL, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) {
|
||||
tsdbError("vgId:%d failed to sendfile %d bytes from file %s to %s since %s", REPO_ID(pHelper->pRepo),
|
||||
TSDB_FILE_HEAD_SIZE, pHelper->files.headF.fname, pHelper->files.nHeadF.fname, strerror(errno));
|
||||
errno = TAOS_SYSTEM_ERROR(errno);
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
goto _err;
|
||||
}
|
||||
|
||||
|
|
@ -144,7 +144,7 @@ int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup) {
|
|||
return tsdbLoadCompIdx(pHelper, NULL);
|
||||
|
||||
_err:
|
||||
return -1;
|
||||
return terrno;
|
||||
}
|
||||
|
||||
int tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError) {
|
||||
|
|
|
|||
|
|
@ -148,7 +148,7 @@ TsdbQueryHandleT* tsdbQueryTables(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, STab
|
|||
pQueryHandle->type = TSDB_QUERY_TYPE_ALL;
|
||||
pQueryHandle->cur.fid = -1;
|
||||
pQueryHandle->cur.win = TSWINDOW_INITIALIZER;
|
||||
pQueryHandle->checkFiles = true;//ASCENDING_TRAVERSE(pQueryHandle->order);
|
||||
pQueryHandle->checkFiles = true;
|
||||
pQueryHandle->activeIndex = 0; // current active table index
|
||||
pQueryHandle->qinfo = qinfo;
|
||||
pQueryHandle->outputCapacity = ((STsdbRepo*)tsdb)->config.maxRowsPerFileBlock;
|
||||
|
|
@ -475,11 +475,15 @@ static int32_t binarySearchForBlock(SCompBlock* pBlock, int32_t numOfBlocks, TSK
|
|||
}
|
||||
|
||||
static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlocks, int32_t type) {
|
||||
// todo check open file failed
|
||||
SFileGroup* fileGroup = pQueryHandle->pFileGroup;
|
||||
|
||||
assert(fileGroup->files[TSDB_FILE_TYPE_HEAD].fname > 0);
|
||||
tsdbSetAndOpenHelperFile(&pQueryHandle->rhelper, fileGroup);
|
||||
|
||||
int32_t code = tsdbSetAndOpenHelperFile(&pQueryHandle->rhelper, fileGroup);
|
||||
|
||||
//open file failed, return error code to client
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
return code;
|
||||
}
|
||||
|
||||
// load all the comp offset value for all tables in this file
|
||||
*numOfBlocks = 0;
|
||||
|
|
@ -593,8 +597,7 @@ static bool doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlo
|
|||
if (pCheckInfo->pDataCols == NULL) {
|
||||
STsdbMeta* pMeta = tsdbGetMeta(pRepo);
|
||||
// TODO
|
||||
pCheckInfo->pDataCols =
|
||||
tdNewDataCols(pMeta->maxRowBytes, pMeta->maxCols, pRepo->config.maxRowsPerFileBlock);
|
||||
pCheckInfo->pDataCols = tdNewDataCols(pMeta->maxRowBytes, pMeta->maxCols, pRepo->config.maxRowsPerFileBlock);
|
||||
}
|
||||
|
||||
tdInitDataCols(pCheckInfo->pDataCols, tsdbGetTableSchema(pCheckInfo->pTableObj));
|
||||
|
|
@ -1359,16 +1362,18 @@ static int32_t createDataBlocksInfo(STsdbQueryHandle* pQueryHandle, int32_t numO
|
|||
}
|
||||
|
||||
// todo opt for only one table case
|
||||
static bool getDataBlocksInFilesImpl(STsdbQueryHandle* pQueryHandle) {
|
||||
static int32_t getDataBlocksInFilesImpl(STsdbQueryHandle* pQueryHandle, bool* exists) {
|
||||
pQueryHandle->numOfBlocks = 0;
|
||||
SQueryFilePos* cur = &pQueryHandle->cur;
|
||||
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
|
||||
int32_t numOfBlocks = 0;
|
||||
int32_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
|
||||
|
||||
while ((pQueryHandle->pFileGroup = tsdbGetFileGroupNext(&pQueryHandle->fileIter)) != NULL) {
|
||||
int32_t type = ASCENDING_TRAVERSE(pQueryHandle->order)? QUERY_RANGE_GREATER_EQUAL:QUERY_RANGE_LESS_EQUAL;
|
||||
if (getFileCompInfo(pQueryHandle, &numOfBlocks, type) != TSDB_CODE_SUCCESS) {
|
||||
if ((code = getFileCompInfo(pQueryHandle, &numOfBlocks, type)) != TSDB_CODE_SUCCESS) {
|
||||
break;
|
||||
}
|
||||
|
||||
|
|
@ -1393,20 +1398,25 @@ static bool getDataBlocksInFilesImpl(STsdbQueryHandle* pQueryHandle) {
|
|||
|
||||
// no data in file anymore
|
||||
if (pQueryHandle->numOfBlocks <= 0) {
|
||||
assert(pQueryHandle->pFileGroup == NULL);
|
||||
if (code == TSDB_CODE_SUCCESS) {
|
||||
assert(pQueryHandle->pFileGroup == NULL);
|
||||
}
|
||||
|
||||
cur->fid = -1; // denote that there are no data in file anymore
|
||||
|
||||
return false;
|
||||
*exists = false;
|
||||
return code;
|
||||
}
|
||||
|
||||
cur->slot = ASCENDING_TRAVERSE(pQueryHandle->order)? 0:pQueryHandle->numOfBlocks-1;
|
||||
cur->fid = pQueryHandle->pFileGroup->fileId;
|
||||
|
||||
STableBlockInfo* pBlockInfo = &pQueryHandle->pDataBlockInfo[cur->slot];
|
||||
return loadFileDataBlock(pQueryHandle, pBlockInfo->compBlock, pBlockInfo->pTableCheckInfo);
|
||||
*exists = loadFileDataBlock(pQueryHandle, pBlockInfo->compBlock, pBlockInfo->pTableCheckInfo);
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
|
||||
static int32_t getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle, bool* exists) {
|
||||
STsdbFileH* pFileHandle = tsdbGetFile(pQueryHandle->pTsdb);
|
||||
SQueryFilePos* cur = &pQueryHandle->cur;
|
||||
|
||||
|
|
@ -1419,7 +1429,7 @@ static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
|
|||
tsdbInitFileGroupIter(pFileHandle, &pQueryHandle->fileIter, pQueryHandle->order);
|
||||
tsdbSeekFileGroupIter(&pQueryHandle->fileIter, fid);
|
||||
|
||||
return getDataBlocksInFilesImpl(pQueryHandle);
|
||||
return getDataBlocksInFilesImpl(pQueryHandle, exists);
|
||||
} else {
|
||||
// check if current file block is all consumed
|
||||
STableBlockInfo* pBlockInfo = &pQueryHandle->pDataBlockInfo[cur->slot];
|
||||
|
|
@ -1430,7 +1440,7 @@ static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
|
|||
if ((cur->slot == pQueryHandle->numOfBlocks - 1 && ASCENDING_TRAVERSE(pQueryHandle->order)) ||
|
||||
(cur->slot == 0 && !ASCENDING_TRAVERSE(pQueryHandle->order))) {
|
||||
// all data blocks in current file has been checked already, try next file if exists
|
||||
return getDataBlocksInFilesImpl(pQueryHandle);
|
||||
return getDataBlocksInFilesImpl(pQueryHandle, exists);
|
||||
} else {
|
||||
// next block of the same file
|
||||
int32_t step = ASCENDING_TRAVERSE(pQueryHandle->order) ? 1 : -1;
|
||||
|
|
@ -1440,11 +1450,15 @@ static bool getDataBlocksInFiles(STsdbQueryHandle* pQueryHandle) {
|
|||
cur->blockCompleted = false;
|
||||
|
||||
STableBlockInfo* pNext = &pQueryHandle->pDataBlockInfo[cur->slot];
|
||||
return loadFileDataBlock(pQueryHandle, pNext->compBlock, pNext->pTableCheckInfo);
|
||||
*exists = loadFileDataBlock(pQueryHandle, pNext->compBlock, pNext->pTableCheckInfo);
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
} else {
|
||||
handleDataMergeIfNeeded(pQueryHandle, pBlockInfo->compBlock, pCheckInfo);
|
||||
return pQueryHandle->realNumOfRows > 0;
|
||||
*exists = pQueryHandle->realNumOfRows > 0;
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1576,8 +1590,14 @@ bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) {
|
|||
}
|
||||
|
||||
if (pQueryHandle->checkFiles) {
|
||||
if (getDataBlocksInFiles(pQueryHandle)) {
|
||||
return true;
|
||||
bool exists = true;
|
||||
int32_t code = getDataBlocksInFiles(pQueryHandle, &exists);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (exists) {
|
||||
return exists;
|
||||
}
|
||||
|
||||
pQueryHandle->activeIndex = 0;
|
||||
|
|
|
|||
Loading…
Reference in New Issue