提交 e534490a 编写于 作者: H Haojun Liao

[TD-225]refactor.

上级 03f361ec
...@@ -152,7 +152,7 @@ typedef struct SQuery { ...@@ -152,7 +152,7 @@ typedef struct SQuery {
int16_t precision; int16_t precision;
int16_t numOfOutput; int16_t numOfOutput;
int16_t fillType; int16_t fillType;
int16_t checkBuffer; // check if the buffer is full during scan each block int16_t checkResultBuf; // check if the buffer is full during scan each block
SLimitVal limit; SLimitVal limit;
int32_t rowSize; int32_t rowSize;
SSqlGroupbyExpr* pGroupbyExpr; SSqlGroupbyExpr* pGroupbyExpr;
......
...@@ -1708,7 +1708,7 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl ...@@ -1708,7 +1708,7 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl
numOfRes = (int32_t) getNumOfResult(pRuntimeEnv); numOfRes = (int32_t) getNumOfResult(pRuntimeEnv);
// update the number of output result // update the number of output result
if (numOfRes > 0 && pQuery->checkBuffer == 1) { if (numOfRes > 0 && pQuery->checkResultBuf == 1) {
assert(numOfRes >= pQuery->rec.rows); assert(numOfRes >= pQuery->rec.rows);
pQuery->rec.rows = numOfRes; pQuery->rec.rows = numOfRes;
...@@ -2222,9 +2222,9 @@ void getAlignQueryTimeWindow(SQuery *pQuery, int64_t key, int64_t keyFirst, int6 ...@@ -2222,9 +2222,9 @@ void getAlignQueryTimeWindow(SQuery *pQuery, int64_t key, int64_t keyFirst, int6
static void setScanLimitationByResultBuffer(SQuery *pQuery) { static void setScanLimitationByResultBuffer(SQuery *pQuery) {
if (isTopBottomQuery(pQuery)) { if (isTopBottomQuery(pQuery)) {
pQuery->checkBuffer = 0; pQuery->checkResultBuf = 0;
} else if (isGroupbyNormalCol(pQuery->pGroupbyExpr)) { } else if (isGroupbyNormalCol(pQuery->pGroupbyExpr)) {
pQuery->checkBuffer = 0; pQuery->checkResultBuf = 0;
} else { } else {
bool hasMultioutput = false; bool hasMultioutput = false;
for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { for (int32_t i = 0; i < pQuery->numOfOutput; ++i) {
...@@ -2239,7 +2239,7 @@ static void setScanLimitationByResultBuffer(SQuery *pQuery) { ...@@ -2239,7 +2239,7 @@ static void setScanLimitationByResultBuffer(SQuery *pQuery) {
} }
} }
pQuery->checkBuffer = hasMultioutput ? 1 : 0; pQuery->checkResultBuf = hasMultioutput ? 1 : 0;
} }
} }
...@@ -4761,20 +4761,21 @@ static void enableExecutionForNextTable(SQueryRuntimeEnv *pRuntimeEnv) { ...@@ -4761,20 +4761,21 @@ static void enableExecutionForNextTable(SQueryRuntimeEnv *pRuntimeEnv) {
} }
} }
// TODO refactor: setAdditionalInfo
static FORCE_INLINE void setEnvForEachBlock(SQInfo* pQInfo, STableQueryInfo* pTableQueryInfo, SDataBlockInfo* pBlockInfo) { static FORCE_INLINE void setEnvForEachBlock(SQInfo* pQInfo, STableQueryInfo* pTableQueryInfo, SDataBlockInfo* pBlockInfo) {
SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv; SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery* pQuery = pQInfo->runtimeEnv.pQuery; SQuery* pQuery = pQInfo->runtimeEnv.pQuery;
int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
if (!QUERY_IS_INTERVAL_QUERY(pQuery)) { if (QUERY_IS_INTERVAL_QUERY(pQuery)) {
setExecutionContext(pQInfo, pTableQueryInfo->groupIndex, pBlockInfo->window.ekey + step);
} else { // interval query
TSKEY nextKey = pBlockInfo->window.skey; TSKEY nextKey = pBlockInfo->window.skey;
setIntervalQueryRange(pQInfo, nextKey); setIntervalQueryRange(pQInfo, nextKey);
if (pRuntimeEnv->hasTagResults || pRuntimeEnv->pTsBuf != NULL) { if (pRuntimeEnv->hasTagResults || pRuntimeEnv->pTsBuf != NULL) {
setAdditionalInfo(pQInfo, pTableQueryInfo->pTable, pTableQueryInfo); setAdditionalInfo(pQInfo, pTableQueryInfo->pTable, pTableQueryInfo);
} }
} else { // non-interval query
setExecutionContext(pQInfo, pTableQueryInfo->groupIndex, pBlockInfo->window.ekey + step);
} }
} }
...@@ -5630,8 +5631,6 @@ static void tableAggregationProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) ...@@ -5630,8 +5631,6 @@ static void tableAggregationProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo)
return; return;
} }
pQuery->current = pTableInfo; // set current query table info
scanOneTableDataBlocks(pRuntimeEnv, pTableInfo->lastKey); scanOneTableDataBlocks(pRuntimeEnv, pTableInfo->lastKey);
finalizeQueryResult(pRuntimeEnv); finalizeQueryResult(pRuntimeEnv);
...@@ -5650,10 +5649,8 @@ static void tableAggregationProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) ...@@ -5650,10 +5649,8 @@ static void tableAggregationProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo)
static void tableProjectionProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) { static void tableProjectionProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv; SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv;
SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableInfo;
// for ts_comp query, re-initialized is not allowed // for ts_comp query, re-initialized is not allowed
SQuery *pQuery = pRuntimeEnv->pQuery;
if (!isTSCompQuery(pQuery)) { if (!isTSCompQuery(pQuery)) {
resetDefaultResInfoOutputBuf(pRuntimeEnv); resetDefaultResInfoOutputBuf(pRuntimeEnv);
} }
...@@ -5705,9 +5702,7 @@ static void tableProjectionProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) ...@@ -5705,9 +5702,7 @@ static void tableProjectionProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo)
// handle time interval query on table // handle time interval query on table
static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) { static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
SQueryRuntimeEnv *pRuntimeEnv = &(pQInfo->runtimeEnv); SQueryRuntimeEnv *pRuntimeEnv = &(pQInfo->runtimeEnv);
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
pQuery->current = pTableInfo;
TSKEY newStartKey = QUERY_IS_ASC_QUERY(pQuery)? INT64_MIN:INT64_MAX; TSKEY newStartKey = QUERY_IS_ASC_QUERY(pQuery)? INT64_MIN:INT64_MAX;
...@@ -5777,7 +5772,6 @@ static void tableQueryImpl(SQInfo *pQInfo) { ...@@ -5777,7 +5772,6 @@ static void tableQueryImpl(SQInfo *pQInfo) {
} }
qDebug("QInfo:%p current:%" PRId64 " returned, total:%" PRId64, pQInfo, pQuery->rec.rows, pQuery->rec.total); qDebug("QInfo:%p current:%" PRId64 " returned, total:%" PRId64, pQInfo, pQuery->rec.rows, pQuery->rec.total);
return;
} else { } else {
pQuery->rec.rows = 0; pQuery->rec.rows = 0;
assert(pRuntimeEnv->windowResInfo.size > 0); assert(pRuntimeEnv->windowResInfo.size > 0);
...@@ -5795,9 +5789,9 @@ static void tableQueryImpl(SQInfo *pQInfo) { ...@@ -5795,9 +5789,9 @@ static void tableQueryImpl(SQInfo *pQInfo) {
if (pQuery->rec.rows <= 0 || pRuntimeEnv->windowResInfo.size <= pQInfo->groupIndex) { if (pQuery->rec.rows <= 0 || pRuntimeEnv->windowResInfo.size <= pQInfo->groupIndex) {
qDebug("QInfo:%p query over, %" PRId64 " rows are returned", pQInfo, pQuery->rec.total); qDebug("QInfo:%p query over, %" PRId64 " rows are returned", pQInfo, pQuery->rec.total);
} }
return;
} }
return;
} }
// number of points returned during this query // number of points returned during this query
...@@ -5806,7 +5800,9 @@ static void tableQueryImpl(SQInfo *pQInfo) { ...@@ -5806,7 +5800,9 @@ static void tableQueryImpl(SQInfo *pQInfo) {
assert(pQInfo->tableqinfoGroupInfo.numOfTables == 1); assert(pQInfo->tableqinfoGroupInfo.numOfTables == 1);
SArray* g = GET_TABLEGROUP(pQInfo, 0); SArray* g = GET_TABLEGROUP(pQInfo, 0);
STableQueryInfo* item = taosArrayGetP(g, 0); STableQueryInfo* item = taosArrayGetP(g, 0);
pQuery->current = item;
// group by normal column, sliding window query, interval query are handled by interval query processor // group by normal column, sliding window query, interval query are handled by interval query processor
if (QUERY_IS_INTERVAL_QUERY(pQuery) || pRuntimeEnv->groupbyColumn) { // interval (down sampling operation) if (QUERY_IS_INTERVAL_QUERY(pQuery) || pRuntimeEnv->groupbyColumn) { // interval (down sampling operation)
...@@ -5814,7 +5810,7 @@ static void tableQueryImpl(SQInfo *pQInfo) { ...@@ -5814,7 +5810,7 @@ static void tableQueryImpl(SQInfo *pQInfo) {
} else if (isFixedOutputQuery(pRuntimeEnv)) { } else if (isFixedOutputQuery(pRuntimeEnv)) {
tableAggregationProcess(pQInfo, item); tableAggregationProcess(pQInfo, item);
} else { // diff/add/multiply/subtract/division } else { // diff/add/multiply/subtract/division
assert(pQuery->checkBuffer == 1); assert(pQuery->checkResultBuf == 1);
tableProjectionProcess(pQInfo, item); tableProjectionProcess(pQInfo, item);
} }
...@@ -5834,7 +5830,7 @@ static void stableQueryImpl(SQInfo *pQInfo) { ...@@ -5834,7 +5830,7 @@ static void stableQueryImpl(SQInfo *pQInfo) {
(isFixedOutputQuery(pRuntimeEnv) && (!isPointInterpoQuery(pQuery)) && (!pRuntimeEnv->groupbyColumn))) { (isFixedOutputQuery(pRuntimeEnv) && (!isPointInterpoQuery(pQuery)) && (!pRuntimeEnv->groupbyColumn))) {
multiTableQueryProcess(pQInfo); multiTableQueryProcess(pQInfo);
} else { } else {
assert((pQuery->checkBuffer == 1 && pQuery->interval.interval == 0) || isPointInterpoQuery(pQuery) || assert((pQuery->checkResultBuf == 1 && pQuery->interval.interval == 0) || isPointInterpoQuery(pQuery) ||
pRuntimeEnv->groupbyColumn); pRuntimeEnv->groupbyColumn);
sequentialTableProcess(pQInfo); sequentialTableProcess(pQInfo);
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册