xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #1187 from taosdata/feature/liaohj 

Feature/liaohj
This commit is contained in:
plum-lihui 2020-02-05 08:24:17 +08:00 committed by GitHub
parent 45c10cb572 a81ea81618
commit f0df8cc6e8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 150 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
src/client/src/tscSQLParser.c
View File

@ -652,7 +652,16 @@ int32_t parseIntervalClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
SColumnList ids = getColumnList(1, 0, PRIMARYKEY_TIMESTAMP_COL_INDEX); SColumnList ids = getColumnList(1, 0, PRIMARYKEY_TIMESTAMP_COL_INDEX);
return insertResultField(pQueryInfo, 0, &ids, TSDB_KEYSIZE, TSDB_DATA_TYPE_TIMESTAMP, aAggs[TSDB_FUNC_TS].aName); int32_t ret = insertResultField(pQueryInfo, 0, &ids, TSDB_KEYSIZE, TSDB_DATA_TYPE_TIMESTAMP, aAggs[TSDB_FUNC_TS].aName);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
}
if (setSlidingClause(pQueryInfo, pQuerySql) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_INVALID_SQL;
}
return TSDB_CODE_SUCCESS;
} }
int32_t setSlidingClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) { int32_t setSlidingClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
@ -675,6 +684,8 @@ int32_t setSlidingClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
if (pQueryInfo->nSlidingTime > pQueryInfo->nAggTimeInterval) { if (pQueryInfo->nSlidingTime > pQueryInfo->nAggTimeInterval) {
return invalidSqlErrMsg(pQueryInfo->msg, msg1); return invalidSqlErrMsg(pQueryInfo->msg, msg1);
} }
} else {
pSliding->n = pQueryInfo->nAggTimeInterval;
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -5394,10 +5405,6 @@ int32_t doCheckForStream(SSqlObj* pSql, SSqlInfo* pInfo) {
} }
} }
if (setSlidingClause(pQueryInfo, pQuerySql) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_INVALID_SQL;
}
// set the created table[stream] name // set the created table[stream] name
if (setMeterID(pMeterMetaInfo, pzTableName, pSql) != TSDB_CODE_SUCCESS) { if (setMeterID(pMeterMetaInfo, pzTableName, pSql) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(pQueryInfo->msg, msg1); return invalidSqlErrMsg(pQueryInfo->msg, msg1);
@ -5581,29 +5588,29 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) {
return invalidSqlErrMsg(pQueryInfo->msg, msg2); return invalidSqlErrMsg(pQueryInfo->msg, msg2);
} }
// set sliding value, the query time range needs to be decide in the first place // // set sliding value, the query time range needs to be decide in the first place
SSQLToken* pSliding = &pQuerySql->sliding; // SSQLToken* pSliding = &pQuerySql->sliding;
if (pSliding->n != 0) { // if (pSliding->n != 0) {
if (!tscEmbedded && pCmd->inStream == 0 && hasDefaultQueryTimeRange(pQueryInfo)) { // sliding only allowed in stream // if (!tscEmbedded && pCmd->inStream == 0 && hasDefaultQueryTimeRange(pQueryInfo)) { // sliding only allowed in stream
const char* msg = "time range expected for sliding window query"; // const char* msg = "time range expected for sliding window query";
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg); // return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg);
} // }
//
getTimestampInUsFromStr(pSliding->z, pSliding->n, &pQueryInfo->nSlidingTime); // getTimestampInUsFromStr(pSliding->z, pSliding->n, &pQueryInfo->nSlidingTime);
if (pMeterMetaInfo->pMeterMeta->precision == TSDB_TIME_PRECISION_MILLI) { // if (pMeterMetaInfo->pMeterMeta->precision == TSDB_TIME_PRECISION_MILLI) {
pQueryInfo->nSlidingTime /= 1000; // pQueryInfo->nSlidingTime /= 1000;
} // }
//
if (pQueryInfo->nSlidingTime < tsMinSlidingTime) { // if (pQueryInfo->nSlidingTime < tsMinSlidingTime) {
return invalidSqlErrMsg(pQueryInfo->msg, msg3); // return invalidSqlErrMsg(pQueryInfo->msg, msg3);
} // }
//
if (pQueryInfo->nSlidingTime > pQueryInfo->nAggTimeInterval) { // if (pQueryInfo->nSlidingTime > pQueryInfo->nAggTimeInterval) {
return invalidSqlErrMsg(pQueryInfo->msg, msg4); // return invalidSqlErrMsg(pQueryInfo->msg, msg4);
} // }
} else { // } else {
pQueryInfo->nSlidingTime = -1; // pQueryInfo->nSlidingTime = -1;
} // }
// in case of join query, time range is required. // in case of join query, time range is required.
if (QUERY_IS_JOIN_QUERY(pQueryInfo->type)) { if (QUERY_IS_JOIN_QUERY(pQueryInfo->type)) {

3
src/client/src/tscServer.c
View File

@ -1110,8 +1110,7 @@ static void tscHandleSubRetrievalError(SRetrieveSupport *trsupport, SSqlObj *pSq
} }
// all subqueries are failed // all subqueries are failed
tscError("%p retrieve from %d vnode(s) completed,code:%d.FAILED.", pPObj, pState->numOfTotal, tscError("%p retrieve from %d vnode(s) completed,code:%d.FAILED.", pPObj, pState->numOfTotal, pState->code);
pState->code);
pPObj->res.code = -(pState->code); pPObj->res.code = -(pState->code);
// release allocated resource // release allocated resource

2
src/client/src/tscSql.c
View File

@ -754,6 +754,8 @@ void taos_free_result_imp(TAOS_RES* res, int keepCmd) {
pSql->pStream == NULL && pMeterMetaInfo->pMeterMeta != NULL))) { pSql->pStream == NULL && pMeterMetaInfo->pMeterMeta != NULL))) {
pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH; pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
tscTrace("%p code:%d, numOfRows:%d, command:%d", pSql, pRes->code, pRes->numOfRows, pCmd->command);
void *fp = pSql->fp; void *fp = pSql->fp;
if (fp != NULL) { if (fp != NULL) {
pSql->freed = 1; pSql->freed = 1;

2
src/client/src/tscStream.c
View File

@ -411,7 +411,9 @@ static void tscSetSlidingWindowInfo(SSqlObj *pSql, SSqlStream *pStream) {
} }
pStream->slidingTime = pQueryInfo->nSlidingTime; pStream->slidingTime = pQueryInfo->nSlidingTime;
pQueryInfo->nAggTimeInterval = 0; // clear the interval value to avoid the force time window split by query processor pQueryInfo->nAggTimeInterval = 0; // clear the interval value to avoid the force time window split by query processor
pQueryInfo->nSlidingTime = 0;
} }
static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, int64_t stime) { static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, int64_t stime) {

48
src/client/src/tscSyntaxtreefunction.c
View File

@ -26,7 +26,7 @@
int32_t step = ((_ord) == TSQL_SO_ASC) ? 1 : -1; \ int32_t step = ((_ord) == TSQL_SO_ASC) ? 1 : -1; \
\ \
if ((len1) == (len2)) { \ if ((len1) == (len2)) { \
for (; i < (len2) && i >= 0; i += step, (out) += step) { \ for (; i < (len2) && i >= 0; i += step, (out) += 1) { \
if (isNull((char *)&((left)[i]), _left_type) || isNull((char *)&((right)[i]), _right_type)) { \ if (isNull((char *)&((left)[i]), _left_type) || isNull((char *)&((right)[i]), _right_type)) { \
setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \
continue; \ continue; \
@ -34,7 +34,7 @@
*(out) = (double)(left)[i] op(right)[i]; \ *(out) = (double)(left)[i] op(right)[i]; \
} \ } \
} else if ((len1) == 1) { \ } else if ((len1) == 1) { \
for (; i >= 0 && i < (len2); i += step, (out) += step) { \ for (; i >= 0 && i < (len2); i += step, (out) += 1) { \
if (isNull((char *)(left), _left_type) || isNull((char *)&(right)[i], _right_type)) { \ if (isNull((char *)(left), _left_type) || isNull((char *)&(right)[i], _right_type)) { \
setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \
continue; \ continue; \
@ -42,7 +42,7 @@
*(out) = (double)(left)[0] op(right)[i]; \ *(out) = (double)(left)[0] op(right)[i]; \
} \ } \
} else if ((len2) == 1) { \ } else if ((len2) == 1) { \
for (; i >= 0 && i < (len1); i += step, (out) += step) { \ for (; i >= 0 && i < (len1); i += step, (out) += 1) { \
if (isNull((char *)&(left)[i], _left_type) || isNull((char *)(right), _right_type)) { \ if (isNull((char *)&(left)[i], _left_type) || isNull((char *)(right), _right_type)) { \
setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \
continue; \ continue; \
@ -58,7 +58,7 @@
int32_t step = (_ord == TSQL_SO_ASC) ? 1 : -1; \ int32_t step = (_ord == TSQL_SO_ASC) ? 1 : -1; \
\ \
if (len1 == (len2)) { \ if (len1 == (len2)) { \
for (; i >= 0 && i < (len2); i += step, (out) += step) { \ for (; i >= 0 && i < (len2); i += step, (out) += 1) { \
if (isNull((char *)&(left[i]), _left_type) || isNull((char *)&(right[i]), _right_type)) { \ if (isNull((char *)&(left[i]), _left_type) || isNull((char *)&(right[i]), _right_type)) { \
setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \
continue; \ continue; \
@ -66,7 +66,7 @@
*(out) = (double)(left)[i] - ((int64_t)(((double)(left)[i]) / (right)[i])) * (right)[i]; \ *(out) = (double)(left)[i] - ((int64_t)(((double)(left)[i]) / (right)[i])) * (right)[i]; \
} \ } \
} else if (len1 == 1) { \ } else if (len1 == 1) { \
for (; i >= 0 && i < (len2); i += step, (out) += step) { \ for (; i >= 0 && i < (len2); i += step, (out) += 1) { \
if (isNull((char *)(left), _left_type) || isNull((char *)&((right)[i]), _right_type)) { \ if (isNull((char *)(left), _left_type) || isNull((char *)&((right)[i]), _right_type)) { \
setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \
continue; \ continue; \
@ -74,7 +74,7 @@
*(out) = (double)(left)[0] - ((int64_t)(((double)(left)[0]) / (right)[i])) * (right)[i]; \ *(out) = (double)(left)[0] - ((int64_t)(((double)(left)[0]) / (right)[i])) * (right)[i]; \
} \ } \
} else if ((len2) == 1) { \ } else if ((len2) == 1) { \
for (; i >= 0 && i < len1; i += step, (out) += step) { \ for (; i >= 0 && i < len1; i += step, (out) += 1) { \
if (isNull((char *)&((left)[i]), _left_type) || isNull((char *)(right), _right_type)) { \ if (isNull((char *)&((left)[i]), _left_type) || isNull((char *)(right), _right_type)) { \
setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \
continue; \ continue; \
@ -112,7 +112,7 @@ void calc_fn_i32_i32_add(void *left, void *right, int32_t numLeft, int32_t numRi
int32_t step = (order == TSQL_SO_ASC) ? 1 : -1; int32_t step = (order == TSQL_SO_ASC) ? 1 : -1;
if (numLeft == numRight) { if (numLeft == numRight) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -121,7 +121,7 @@ void calc_fn_i32_i32_add(void *left, void *right, int32_t numLeft, int32_t numRi
*pOutput = (double)pLeft[i] + pRight[i]; *pOutput = (double)pLeft[i] + pRight[i];
} }
} else if (numLeft == 1) { } else if (numLeft == 1) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -130,7 +130,7 @@ void calc_fn_i32_i32_add(void *left, void *right, int32_t numLeft, int32_t numRi
*pOutput = (double)pLeft[0] + pRight[i]; *pOutput = (double)pLeft[0] + pRight[i];
} }
} else if (numRight == 1) { } else if (numRight == 1) {
for (; i >= 0 && i < numLeft; i += step, pOutput += step) { for (; i >= 0 && i < numLeft; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -310,7 +310,7 @@ void calc_fn_i32_i32_sub(void *left, void *right, int32_t numLeft, int32_t numRi
int32_t step = (order == TSQL_SO_ASC) ? 1 : -1; int32_t step = (order == TSQL_SO_ASC) ? 1 : -1;
if (numLeft == numRight) { if (numLeft == numRight) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)&(pOutput[i]), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)&(pOutput[i]), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -318,7 +318,7 @@ void calc_fn_i32_i32_sub(void *left, void *right, int32_t numLeft, int32_t numRi
*pOutput = (double)pLeft[i] - pRight[i]; *pOutput = (double)pLeft[i] - pRight[i];
} }
} else if (numLeft == 1) { } else if (numLeft == 1) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -326,7 +326,7 @@ void calc_fn_i32_i32_sub(void *left, void *right, int32_t numLeft, int32_t numRi
*pOutput = (double)pLeft[0] - pRight[i]; *pOutput = (double)pLeft[0] - pRight[i];
} }
} else if (numRight == 1) { } else if (numRight == 1) {
for (; i >= 0 && i < numLeft; i += step, pOutput += step) { for (; i >= 0 && i < numLeft; i += step, pOutput += 1) {
if (isNull((char *)&pLeft[i], TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&pLeft[i], TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) {
setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -521,7 +521,7 @@ void calc_fn_i32_i32_multi(void *left, void *right, int32_t numLeft, int32_t num
int32_t step = (order == TSQL_SO_ASC) ? 1 : -1; int32_t step = (order == TSQL_SO_ASC) ? 1 : -1;
if (numLeft == numRight) { if (numLeft == numRight) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -530,7 +530,7 @@ void calc_fn_i32_i32_multi(void *left, void *right, int32_t numLeft, int32_t num
*pOutput = (double)pLeft[i] * pRight[i]; *pOutput = (double)pLeft[i] * pRight[i];
} }
} else if (numLeft == 1) { } else if (numLeft == 1) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -539,7 +539,7 @@ void calc_fn_i32_i32_multi(void *left, void *right, int32_t numLeft, int32_t num
*pOutput = (double)pLeft[0] * pRight[i]; *pOutput = (double)pLeft[0] * pRight[i];
} }
} else if (numRight == 1) { } else if (numRight == 1) {
for (; i >= 0 && i < numLeft; i += step, pOutput += step) { for (; i >= 0 && i < numLeft; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -719,7 +719,7 @@ void calc_fn_i32_i32_div(void *left, void *right, int32_t numLeft, int32_t numRi
int32_t step = (order == TSQL_SO_ASC) ? 1 : -1; int32_t step = (order == TSQL_SO_ASC) ? 1 : -1;
if (numLeft == numRight) { if (numLeft == numRight) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -728,7 +728,7 @@ void calc_fn_i32_i32_div(void *left, void *right, int32_t numLeft, int32_t numRi
*pOutput = (double)pLeft[i] / pRight[i]; *pOutput = (double)pLeft[i] / pRight[i];
} }
} else if (numLeft == 1) { } else if (numLeft == 1) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -737,7 +737,7 @@ void calc_fn_i32_i32_div(void *left, void *right, int32_t numLeft, int32_t numRi
*pOutput = (double)pLeft[0] / pRight[i]; *pOutput = (double)pLeft[0] / pRight[i];
} }
} else if (numRight == 1) { } else if (numRight == 1) {
for (; i >= 0 && i < numLeft; i += step, pOutput += step) { for (; i >= 0 && i < numLeft; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -933,7 +933,7 @@ void calc_fn_i32_i32_rem(void *left, void *right, int32_t numLeft, int32_t numRi
int32_t step = (order == TSQL_SO_ASC) ? 1 : -1; int32_t step = (order == TSQL_SO_ASC) ? 1 : -1;
if (numLeft == numRight) { if (numLeft == numRight) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -942,7 +942,7 @@ void calc_fn_i32_i32_rem(void *left, void *right, int32_t numLeft, int32_t numRi
*pOutput = (double)pLeft[i] - ((int64_t)(((double)pLeft[i]) / pRight[i])) * pRight[i]; *pOutput = (double)pLeft[i] - ((int64_t)(((double)pLeft[i]) / pRight[i])) * pRight[i];
} }
} else if (numLeft == 1) { } else if (numLeft == 1) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -951,7 +951,7 @@ void calc_fn_i32_i32_rem(void *left, void *right, int32_t numLeft, int32_t numRi
*pOutput = (double)pLeft[0] - ((int64_t)(((double)pLeft[0]) / pRight[i])) * pRight[i]; *pOutput = (double)pLeft[0] - ((int64_t)(((double)pLeft[0]) / pRight[i])) * pRight[i];
} }
} else if (numRight == 1) { } else if (numRight == 1) {
for (; i >= 0 && i < numLeft; i += step, pOutput += step) { for (; i >= 0 && i < numLeft; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -991,7 +991,7 @@ void calc_fn_i32_d_rem(void *left, void *right, int32_t numLeft, int32_t numRigh
int32_t step = (order == TSQL_SO_ASC) ? 1 : -1; int32_t step = (order == TSQL_SO_ASC) ? 1 : -1;
if (numLeft == numRight) { if (numLeft == numRight) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -1000,7 +1000,7 @@ void calc_fn_i32_d_rem(void *left, void *right, int32_t numLeft, int32_t numRigh
*pOutput = (double)pLeft[i] - ((int64_t)(((double)pLeft[i]) / pRight[i])) * pRight[i]; *pOutput = (double)pLeft[i] - ((int64_t)(((double)pLeft[i]) / pRight[i])) * pRight[i];
} }
} else if (numLeft == 1) { } else if (numLeft == 1) {
for (; i >= 0 && i < numRight; i += step, pOutput += step) { for (; i >= 0 && i < numRight; i += step, pOutput += 1) {
if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) { if (isNull((char *)(pLeft), TSDB_DATA_TYPE_INT) || isNull((char *)&(pRight[i]), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;
@ -1009,7 +1009,7 @@ void calc_fn_i32_d_rem(void *left, void *right, int32_t numLeft, int32_t numRigh
*pOutput = (double)pLeft[0] - ((int64_t)(((double)pLeft[0]) / pRight[i])) * pRight[i]; *pOutput = (double)pLeft[0] - ((int64_t)(((double)pLeft[0]) / pRight[i])) * pRight[i];
} }
} else if (numRight == 1) { } else if (numRight == 1) {
for (; i >= 0 && i < numLeft; i += step, pOutput += step) { for (; i >= 0 && i < numLeft; i += step, pOutput += 1) {
if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) { if (isNull((char *)&(pLeft[i]), TSDB_DATA_TYPE_INT) || isNull((char *)(pRight), TSDB_DATA_TYPE_INT)) {
setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize);
continue; continue;

4
src/client/src/tscUtil.c
View File

@ -215,6 +215,10 @@ bool tscIsTwoStageMergeMetricQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) {
return false; return false;
} }
if ((pQueryInfo->type & TSDB_QUERY_TYPE_FREE_RESOURCE) == TSDB_QUERY_TYPE_FREE_RESOURCE) {
return false;
}
// for ordered projection query, iterate all qualified vnodes sequentially // for ordered projection query, iterate all qualified vnodes sequentially
if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, tableIndex)) { if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, tableIndex)) {
return false; return false;

166
src/system/detail/src/vnodeQueryImpl.c
View File

@ -49,8 +49,6 @@ enum {
#define IS_DISK_DATA_BLOCK(q) ((q)->fileId >= 0) #define IS_DISK_DATA_BLOCK(q) ((q)->fileId >= 0)
// static int32_t copyDataFromMMapBuffer(int fd, SQInfo *pQInfo, SQueryFilesInfo *pQueryFile, char *buf, uint64_t
// offset, int32_t size);
static int32_t readDataFromDiskFile(int fd, SQInfo *pQInfo, SQueryFilesInfo *pQueryFile, char *buf, uint64_t offset, static int32_t readDataFromDiskFile(int fd, SQInfo *pQInfo, SQueryFilesInfo *pQueryFile, char *buf, uint64_t offset,
int32_t size); int32_t size);
@ -70,12 +68,12 @@ static int32_t getNextDataFileCompInfo(SQueryRuntimeEnv *pRuntimeEnv, SMeterObj
static void setGroupOutputBuffer(SQueryRuntimeEnv *pRuntimeEnv, SOutputRes *pResult); static void setGroupOutputBuffer(SQueryRuntimeEnv *pRuntimeEnv, SOutputRes *pResult);
static void getAlignedIntervalQueryRange(SQueryRuntimeEnv *pRuntimeEnv, TSKEY keyInData, TSKEY skey, TSKEY ekey); static void getAlignedIntervalQueryRange(SQueryRuntimeEnv *pRuntimeEnv, TSKEY keyInData, TSKEY skey, TSKEY ekey);
static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pInfo, static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pMeterQueryInfo,
SBlockInfo *pBlockInfo, int64_t *pPrimaryCol, SField *pFields, SBlockInfo *pBlockInfo, int64_t *pPrimaryCol, SField *pFields,
__block_search_fn_t searchFn); __block_search_fn_t searchFn);
static int32_t saveResult(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pMeterQueryInfo, int32_t numOfResult); static int32_t saveResult(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pMeterQueryInfo, int32_t numOfResult);
static void applyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterDataInfo *pInfoEx, int64_t *pPrimaryData, static void applyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterDataInfo *pMeterDataInfo,
SBlockInfo *pBlockInfo, int32_t blockStatus, SField *pFields, SBlockInfo *pBlockInfo, int32_t blockStatus, SField *pFields,
__block_search_fn_t searchFn); __block_search_fn_t searchFn);
@ -1728,6 +1726,8 @@ int32_t numOfClosedSlidingWindow(SSlidingWindowInfo *pSlidingWindowInfo) {
return i; return i;
} }
} }
return 0;
} }
void closeSlidingWindow(SSlidingWindowInfo* pSlidingWindowInfo, int32_t slot) { void closeSlidingWindow(SSlidingWindowInfo* pSlidingWindowInfo, int32_t slot) {
@ -2112,7 +2112,7 @@ static int32_t rowwiseApplyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, int32_t *
SSlidingWindowInfo *pSlidingWindowInfo = &pRuntimeEnv->swindowResInfo; SSlidingWindowInfo *pSlidingWindowInfo = &pRuntimeEnv->swindowResInfo;
// query completed // query completed
if (lastKey >= pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery) || if ((lastKey >= pQuery->ekey && QUERY_IS_ASC_QUERY(pQuery)) ||
(lastKey <= pQuery->ekey && !QUERY_IS_ASC_QUERY(pQuery))) { (lastKey <= pQuery->ekey && !QUERY_IS_ASC_QUERY(pQuery))) {
closeAllSlidingWindow(pSlidingWindowInfo); closeAllSlidingWindow(pSlidingWindowInfo);
@ -5027,7 +5027,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
static void updatelastkey(SQuery *pQuery, SMeterQueryInfo *pMeterQInfo) { pMeterQInfo->lastKey = pQuery->lastKey; } static void updatelastkey(SQuery *pQuery, SMeterQueryInfo *pMeterQInfo) { pMeterQInfo->lastKey = pQuery->lastKey; }
void queryOnBlock(SMeterQuerySupportObj *pSupporter, int64_t *primaryKeys, int32_t blockStatus, void queryOnBlock(SMeterQuerySupportObj *pSupporter, int64_t *primaryKeys, int32_t blockStatus,
SBlockInfo *pBlockBasicInfo, SMeterDataInfo *pDataHeadInfoEx, SField *pFields, SBlockInfo *pBlockBasicInfo, SMeterDataInfo *pMeterDataInfo, SField *pFields,
__block_search_fn_t searchFn) { __block_search_fn_t searchFn) {
/* cache blocks may be assign to other meter, abort */ /* cache blocks may be assign to other meter, abort */
if (pBlockBasicInfo->size <= 0) { if (pBlockBasicInfo->size <= 0) {
@ -5043,18 +5043,17 @@ void queryOnBlock(SMeterQuerySupportObj *pSupporter, int64_t *primaryKeys, int32
// note: only fixed number of output for each group by operation // note: only fixed number of output for each group by operation
if (numOfRes > 0) { if (numOfRes > 0) {
pSupporter->pResult[pDataHeadInfoEx->groupIdx].numOfRows = numOfRes; pSupporter->pResult[pMeterDataInfo->groupIdx].numOfRows = numOfRes;
} }
// used to decide the correct start position in cache after check all data in files // used to decide the correct start position in cache after check all data in files
updatelastkey(pQuery, pDataHeadInfoEx->pMeterQInfo); updatelastkey(pQuery, pMeterDataInfo->pMeterQInfo);
if (pRuntimeEnv->pTSBuf != NULL) { if (pRuntimeEnv->pTSBuf != NULL) {
pDataHeadInfoEx->pMeterQInfo->cur = tsBufGetCursor(pRuntimeEnv->pTSBuf); pMeterDataInfo->pMeterQInfo->cur = tsBufGetCursor(pRuntimeEnv->pTSBuf);
} }
} else { } else {
applyIntervalQueryOnBlock(pSupporter, pDataHeadInfoEx, primaryKeys, pBlockBasicInfo, blockStatus, pFields, applyIntervalQueryOnBlock(pSupporter, pMeterDataInfo, pBlockBasicInfo, blockStatus, pFields, searchFn);
searchFn);
} }
} }
@ -5265,13 +5264,13 @@ static tFilePage *getFilePage(SMeterQuerySupportObj *pSupporter, int32_t pageId)
return (tFilePage *)(pSupporter->meterOutputMMapBuf + DEFAULT_INTERN_BUF_SIZE * pageId); return (tFilePage *)(pSupporter->meterOutputMMapBuf + DEFAULT_INTERN_BUF_SIZE * pageId);
} }
static tFilePage *getMeterDataPage(SMeterQuerySupportObj *pSupporter, SMeterDataInfo *pInfoEx, int32_t pageId) { static tFilePage *getMeterDataPage(SMeterQuerySupportObj *pSupporter, SMeterDataInfo *pMeterDataInfo, int32_t pageId) {
SMeterQueryInfo *pInfo = pInfoEx->pMeterQInfo; SMeterQueryInfo *pMeterQueryInfo = pMeterDataInfo->pMeterQInfo;
if (pageId >= pInfo->numOfPages) { if (pageId >= pMeterQueryInfo->numOfPages) {
return NULL; return NULL;
} }
int32_t realId = pInfo->pageList[pageId]; int32_t realId = pMeterQueryInfo->pageList[pageId];
return getFilePage(pSupporter, realId); return getFilePage(pSupporter, realId);
} }
@ -5281,14 +5280,14 @@ typedef struct Position {
} Position; } Position;
typedef struct SCompSupporter { typedef struct SCompSupporter {
SMeterDataInfo ** pInfoEx; SMeterDataInfo ** pMeterDataInfo;
Position * pPosition; Position * pPosition;
SMeterQuerySupportObj *pSupporter; SMeterQuerySupportObj *pSupporter;
} SCompSupporter; } SCompSupporter;
int64_t getCurrentTimestamp(SCompSupporter *pSupportor, int32_t meterIdx) { int64_t getCurrentTimestamp(SCompSupporter *pSupportor, int32_t meterIdx) {
Position * pPos = &pSupportor->pPosition[meterIdx]; Position * pPos = &pSupportor->pPosition[meterIdx];
tFilePage *pPage = getMeterDataPage(pSupportor->pSupporter, pSupportor->pInfoEx[meterIdx], pPos->pageIdx); tFilePage *pPage = getMeterDataPage(pSupportor->pSupporter, pSupportor->pMeterDataInfo[meterIdx], pPos->pageIdx);
return *(int64_t *)(pPage->data + TSDB_KEYSIZE * pPos->rowIdx); return *(int64_t *)(pPage->data + TSDB_KEYSIZE * pPos->rowIdx);
} }
@ -5311,10 +5310,10 @@ int32_t meterResultComparator(const void *pLeft, const void *pRight, void *param
return -1; return -1;
} }
tFilePage *pPageLeft = getMeterDataPage(supportor->pSupporter, supportor->pInfoEx[left], leftPos.pageIdx); tFilePage *pPageLeft = getMeterDataPage(supportor->pSupporter, supportor->pMeterDataInfo[left], leftPos.pageIdx);
int64_t leftTimestamp = *(int64_t *)(pPageLeft->data + TSDB_KEYSIZE * leftPos.rowIdx); int64_t leftTimestamp = *(int64_t *)(pPageLeft->data + TSDB_KEYSIZE * leftPos.rowIdx);
tFilePage *pPageRight = getMeterDataPage(supportor->pSupporter, supportor->pInfoEx[right], rightPos.pageIdx); tFilePage *pPageRight = getMeterDataPage(supportor->pSupporter, supportor->pMeterDataInfo[right], rightPos.pageIdx);
int64_t rightTimestamp = *(int64_t *)(pPageRight->data + TSDB_KEYSIZE * rightPos.rowIdx); int64_t rightTimestamp = *(int64_t *)(pPageRight->data + TSDB_KEYSIZE * rightPos.rowIdx);
if (leftTimestamp == rightTimestamp) { if (leftTimestamp == rightTimestamp) {
@ -5465,17 +5464,17 @@ int32_t doMergeMetersResultsToGroupRes(SMeterQuerySupportObj *pSupporter, SQuery
cs.pPosition[pos].pageIdx += 1; // try next page cs.pPosition[pos].pageIdx += 1; // try next page
// check if current page is empty or not. if it is empty, ignore it and try next // check if current page is empty or not. if it is empty, ignore it and try next
if (cs.pPosition[pos].pageIdx <= cs.pInfoEx[pos]->pMeterQInfo->numOfPages - 1) { if (cs.pPosition[pos].pageIdx <= cs.pMeterDataInfo[pos]->pMeterQInfo->numOfPages - 1) {
tFilePage *newPage = getMeterDataPage(cs.pSupporter, pValidMeter[pos], position->pageIdx); tFilePage *newPage = getMeterDataPage(cs.pSupporter, pValidMeter[pos], position->pageIdx);
if (newPage->numOfElems <= 0) { if (newPage->numOfElems <= 0) {
// if current source data page is null, it must be the last page of source output page // if current source data page is null, it must be the last page of source output page
cs.pPosition[pos].pageIdx += 1; cs.pPosition[pos].pageIdx += 1;
assert(cs.pPosition[pos].pageIdx >= cs.pInfoEx[pos]->pMeterQInfo->numOfPages - 1); assert(cs.pPosition[pos].pageIdx >= cs.pMeterDataInfo[pos]->pMeterQInfo->numOfPages - 1);
} }
} }
// the following code must be executed if current source pages are exhausted // the following code must be executed if current source pages are exhausted
if (cs.pPosition[pos].pageIdx >= cs.pInfoEx[pos]->pMeterQInfo->numOfPages) { if (cs.pPosition[pos].pageIdx >= cs.pMeterDataInfo[pos]->pMeterQInfo->numOfPages) {
cs.pPosition[pos].pageIdx = -1; cs.pPosition[pos].pageIdx = -1;
cs.pPosition[pos].rowIdx = -1; cs.pPosition[pos].rowIdx = -1;
@ -5635,7 +5634,7 @@ int32_t doCloseAllOpenedResults(SMeterQuerySupportObj *pSupporter) {
void disableFunctForSuppleScan(SQueryRuntimeEnv *pRuntimeEnv, int32_t order) { void disableFunctForSuppleScan(SQueryRuntimeEnv *pRuntimeEnv, int32_t order) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
if (isGroupbyNormalCol(pQuery->pGroupbyExpr) || pQuery->slidingTime > 0 && pQuery->nAggTimeInterval > 0) { if (isGroupbyNormalCol(pQuery->pGroupbyExpr) || (pQuery->slidingTime > 0 && pQuery->nAggTimeInterval > 0)) {
for (int32_t i = 0; i < pQuery->numOfOutputCols; ++i) { for (int32_t i = 0; i < pQuery->numOfOutputCols; ++i) {
pRuntimeEnv->pCtx[i].order = (pRuntimeEnv->pCtx[i].order) ^ 1; pRuntimeEnv->pCtx[i].order = (pRuntimeEnv->pCtx[i].order) ^ 1;
} }
@ -6693,11 +6692,11 @@ void cleanBlockOrderSupporter(SBlockOrderSupporter *pSupporter, int32_t numOfTab
int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMeters, int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMeters,
SMeterDataBlockInfoEx **pDataBlockInfoEx, int32_t numOfCompBlocks, SMeterDataBlockInfoEx **pDataBlockInfoEx, int32_t numOfCompBlocks,
int32_t *nAllocBlocksInfoSize, int64_t addr) { int32_t *numOfAllocBlocks, int64_t addr) {
// release allocated memory first // release allocated memory first
freeDataBlockFieldInfo(*pDataBlockInfoEx, *nAllocBlocksInfoSize); freeDataBlockFieldInfo(*pDataBlockInfoEx, *numOfAllocBlocks);
if (*nAllocBlocksInfoSize == 0 || *nAllocBlocksInfoSize < numOfCompBlocks) { if (*numOfAllocBlocks == 0 || *numOfAllocBlocks < numOfCompBlocks) {
char *tmp = realloc((*pDataBlockInfoEx), sizeof(SMeterDataBlockInfoEx) * numOfCompBlocks); char *tmp = realloc((*pDataBlockInfoEx), sizeof(SMeterDataBlockInfoEx) * numOfCompBlocks);
if (tmp == NULL) { if (tmp == NULL) {
tfree(*pDataBlockInfoEx); tfree(*pDataBlockInfoEx);
@ -6706,7 +6705,7 @@ int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMet
*pDataBlockInfoEx = (SMeterDataBlockInfoEx *)tmp; *pDataBlockInfoEx = (SMeterDataBlockInfoEx *)tmp;
memset((*pDataBlockInfoEx), 0, sizeof(SMeterDataBlockInfoEx) * numOfCompBlocks); memset((*pDataBlockInfoEx), 0, sizeof(SMeterDataBlockInfoEx) * numOfCompBlocks);
*nAllocBlocksInfoSize = numOfCompBlocks; *numOfAllocBlocks = numOfCompBlocks;
} }
SBlockOrderSupporter supporter = {0}; SBlockOrderSupporter supporter = {0};
@ -6740,14 +6739,14 @@ int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMet
supporter.pDataBlockInfoEx[numOfQualMeters] = (SMeterDataBlockInfoEx *)buf; supporter.pDataBlockInfoEx[numOfQualMeters] = (SMeterDataBlockInfoEx *)buf;
for (int32_t k = 0; k < pMeterDataInfo[j]->numOfBlocks; ++k) { for (int32_t k = 0; k < pMeterDataInfo[j]->numOfBlocks; ++k) {
SMeterDataBlockInfoEx *pInfoEx = &supporter.pDataBlockInfoEx[numOfQualMeters][k]; SMeterDataBlockInfoEx *pBlockInfoEx = &supporter.pDataBlockInfoEx[numOfQualMeters][k];
pInfoEx->pBlock.compBlock = &pBlock[k]; pBlockInfoEx->pBlock.compBlock = &pBlock[k];
pInfoEx->pBlock.fields = NULL; pBlockInfoEx->pBlock.fields = NULL;
pInfoEx->pMeterDataInfo = pMeterDataInfo[j]; pBlockInfoEx->pMeterDataInfo = pMeterDataInfo[j];
pInfoEx->groupIdx = pMeterDataInfo[j]->groupIdx; // set the group index pBlockInfoEx->groupIdx = pMeterDataInfo[j]->groupIdx; // set the group index
pInfoEx->blockIndex = pMeterDataInfo[j]->start + k; // set the block index in original meter pBlockInfoEx->blockIndex = pMeterDataInfo[j]->start + k; // set the block index in original meter
cnt++; cnt++;
} }
@ -7013,7 +7012,7 @@ int32_t setIntervalQueryExecutionContext(SMeterQuerySupportObj *pSupporter, int3
return 0; return 0;
} }
static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pInfo, static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pMeterQueryInfo,
SBlockInfo *pBlockInfo, int64_t *pPrimaryCol, SField *pFields, SBlockInfo *pBlockInfo, int64_t *pPrimaryCol, SField *pFields,
__block_search_fn_t searchFn) { __block_search_fn_t searchFn) {
SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv; SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv;
@ -7029,10 +7028,10 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
assert(steps > 0); assert(steps > 0);
// NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range // NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range
if (pInfo->lastResRows == 0) { if (pMeterQueryInfo->lastResRows == 0) {
pInfo->lastResRows = numOfRes; pMeterQueryInfo->lastResRows = numOfRes;
} else { } else {
assert(pInfo->lastResRows == 1); assert(pMeterQueryInfo->lastResRows == 1);
} }
int32_t pos = pQuery->pos + steps * factor; int32_t pos = pQuery->pos + steps * factor;
@ -7071,20 +7070,20 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
if (pQuery->lastKey > pSupporter->rawEKey || nextKey > pSupporter->rawEKey) { if (pQuery->lastKey > pSupporter->rawEKey || nextKey > pSupporter->rawEKey) {
/* whole query completed, save result and abort */ /* whole query completed, save result and abort */
assert(queryCompleted); assert(queryCompleted);
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
// save the pQuery->lastKey for retrieve data in cache, actually, there will be no qualified data in cache. // save the pQuery->lastKey for retrieve data in cache, actually, there will be no qualified data in cache.
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
} else if (pQuery->ekey == pBlockInfo->keyLast) { } else if (pQuery->ekey == pBlockInfo->keyLast) {
/* current interval query is completed, set the next query range on other data blocks if exist */ /* current interval query is completed, set the next query range on other data blocks if exist */
int64_t prevEKey = pQuery->ekey; int64_t prevEKey = pQuery->ekey;
getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey); getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey);
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
assert(queryCompleted && prevEKey < pQuery->skey); assert(queryCompleted && prevEKey < pQuery->skey);
if (pInfo->lastResRows > 0) { if (pMeterQueryInfo->lastResRows > 0) {
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
} }
} else { } else {
/* /*
@ -7095,7 +7094,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
* With the information of the directly next data block, whether locates in cache or disk, * With the information of the directly next data block, whether locates in cache or disk,
* current interval query being completed or not can be decided. * current interval query being completed or not can be decided.
*/ */
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
assert(pQuery->lastKey > pBlockInfo->keyLast && pQuery->lastKey <= pQuery->ekey); assert(pQuery->lastKey > pBlockInfo->keyLast && pQuery->lastKey <= pQuery->ekey);
/* /*
@ -7103,7 +7102,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
* merge with other meters in the same group * merge with other meters in the same group
*/ */
if (queryCompleted) { if (queryCompleted) {
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
} }
} }
@ -7119,23 +7118,23 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
if (pQuery->lastKey < pSupporter->rawEKey || (nextKey < pSupporter->rawEKey && nextKey != -1)) { if (pQuery->lastKey < pSupporter->rawEKey || (nextKey < pSupporter->rawEKey && nextKey != -1)) {
/* whole query completed, save result and abort */ /* whole query completed, save result and abort */
assert(queryCompleted); assert(queryCompleted);
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
/* /*
* save the pQuery->lastKey for retrieve data in cache, actually, * save the pQuery->lastKey for retrieve data in cache, actually,
* there will be no qualified data in cache. * there will be no qualified data in cache.
*/ */
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
} else if (pQuery->ekey == pBlockInfo->keyFirst) { } else if (pQuery->ekey == pBlockInfo->keyFirst) {
// current interval query is completed, set the next query range on other data blocks if exist // current interval query is completed, set the next query range on other data blocks if exist
int64_t prevEKey = pQuery->ekey; int64_t prevEKey = pQuery->ekey;
getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey); getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey);
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
assert(queryCompleted && prevEKey > pQuery->skey); assert(queryCompleted && prevEKey > pQuery->skey);
if (pInfo->lastResRows > 0) { if (pMeterQueryInfo->lastResRows > 0) {
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
} }
} else { } else {
/* /*
@ -7146,7 +7145,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
* With the information of the directly next data block, whether locates in cache or disk, * With the information of the directly next data block, whether locates in cache or disk,
* current interval query being completed or not can be decided. * current interval query being completed or not can be decided.
*/ */
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
assert(pQuery->lastKey < pBlockInfo->keyFirst && pQuery->lastKey >= pQuery->ekey); assert(pQuery->lastKey < pBlockInfo->keyFirst && pQuery->lastKey >= pQuery->ekey);
/* /*
@ -7154,7 +7153,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
* flag, and merge with other meters in the same group * flag, and merge with other meters in the same group
*/ */
if (queryCompleted) { if (queryCompleted) {
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
} }
} }
@ -7163,14 +7162,14 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
} }
assert(queryCompleted); assert(queryCompleted);
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
assert((nextKey >= pQuery->lastKey && QUERY_IS_ASC_QUERY(pQuery)) || assert((nextKey >= pQuery->lastKey && QUERY_IS_ASC_QUERY(pQuery)) ||
(nextKey <= pQuery->lastKey && !QUERY_IS_ASC_QUERY(pQuery))); (nextKey <= pQuery->lastKey && !QUERY_IS_ASC_QUERY(pQuery)));
/* still in the same block to query */ /* still in the same block to query */
getAlignedIntervalQueryRange(pRuntimeEnv, nextKey, pSupporter->rawSKey, pSupporter->rawEKey); getAlignedIntervalQueryRange(pRuntimeEnv, nextKey, pSupporter->rawSKey, pSupporter->rawEKey);
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
int32_t newPos = searchFn((char *)pPrimaryCol, pBlockInfo->size, pQuery->skey, pQuery->order.order); int32_t newPos = searchFn((char *)pPrimaryCol, pBlockInfo->size, pQuery->skey, pQuery->order.order);
assert(newPos == pQuery->pos + steps * factor); assert(newPos == pQuery->pos + steps * factor);
@ -7179,26 +7178,25 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete
} }
} }
static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pInfo, static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pMeterQueryInfo,
SBlockInfo *pBlockInfo, int64_t *pPrimaryCol, SField *pFields, SBlockInfo *pBlockInfo, int64_t *pPrimaryCol, SField *pFields,
__block_search_fn_t searchFn) { __block_search_fn_t searchFn) {
SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv; SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv;
SQuery * pQuery = pRuntimeEnv->pQuery; SQuery * pQuery = pRuntimeEnv->pQuery;
int32_t factor = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); int32_t factor = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
int64_t nextKey = -1;
bool completed = false;
while (1) { while (1) {
int64_t nextKey = -1;
int32_t numOfRes = 0; int32_t numOfRes = 0;
int32_t steps = applyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, pPrimaryCol, pFields, searchFn, &numOfRes); int32_t steps = applyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, pPrimaryCol, pFields, searchFn, &numOfRes);
assert(steps > 0); assert(steps > 0);
// NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range // NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range
if (pInfo->lastResRows == 0) { if (pMeterQueryInfo->lastResRows == 0) {
pInfo->lastResRows = numOfRes; pMeterQueryInfo->lastResRows = numOfRes;
} else { } else {
assert(pInfo->lastResRows == 1); assert(pMeterQueryInfo->lastResRows == 1);
} }
int32_t pos = pQuery->pos + steps * factor; int32_t pos = pQuery->pos + steps * factor;
@ -7212,6 +7210,7 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
} }
// all data satisfy current query are checked, query completed // all data satisfy current query are checked, query completed
bool completed = false;
if (QUERY_IS_ASC_QUERY(pQuery)) { if (QUERY_IS_ASC_QUERY(pQuery)) {
completed = (pQuery->lastKey > pQuery->ekey); completed = (pQuery->lastKey > pQuery->ekey);
} else { } else {
@ -7232,7 +7231,7 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
* With the information of the directly next data block, whether locates in cache or disk, * With the information of the directly next data block, whether locates in cache or disk,
* current interval query being completed or not can be decided. * current interval query being completed or not can be decided.
*/ */
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
if (QUERY_IS_ASC_QUERY(pQuery)) { if (QUERY_IS_ASC_QUERY(pQuery)) {
assert(pQuery->lastKey > pBlockInfo->keyLast && pQuery->lastKey <= pQuery->ekey); assert(pQuery->lastKey > pBlockInfo->keyLast && pQuery->lastKey <= pQuery->ekey);
@ -7247,36 +7246,28 @@ static void doApplyIntervalQueryOnBlock_rv(SMeterQuerySupportObj *pSupporter, SM
if (pQuery->ekey == pSupporter->rawEKey) { if (pQuery->ekey == pSupporter->rawEKey) {
/* whole query completed, save result and abort */ /* whole query completed, save result and abort */
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
// save the pQuery->lastKey for retrieve data in cache, actually, there will be no qualified data in cache. // save the pQuery->lastKey for retrieve data in cache, actually, there will be no qualified data in cache.
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
return; return;
} else if ((QUERY_IS_ASC_QUERY(pQuery) && pQuery->ekey == pBlockInfo->keyLast) || } else if ((QUERY_IS_ASC_QUERY(pQuery) && pQuery->ekey == pBlockInfo->keyLast) ||
(QUERY_IS_ASC_QUERY(pQuery) && pQuery->ekey == pBlockInfo->keyFirst)) { (!QUERY_IS_ASC_QUERY(pQuery) && pQuery->ekey == pBlockInfo->keyFirst)) {
// /* current interval query is completed, set the next query range on other data blocks if exist */ /* current interval query is completed, set the next query range on other data blocks if exist */
// int64_t prevEKey = pQuery->ekey; saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
//
// getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey);
saveIntervalQueryRange(pRuntimeEnv, pInfo);
//
// assert(prevEKey < pQuery->skey);
// if (pInfo->lastResRows > 0) {
// saveResult(pSupporter, pInfo, pInfo->lastResRows);
// }
//
return; return;
} }
saveResult(pSupporter, pInfo, pInfo->lastResRows); saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows);
assert(pos >= 0 && pos < pBlockInfo->size);
assert((nextKey >= pQuery->lastKey && QUERY_IS_ASC_QUERY(pQuery)) || assert((nextKey >= pQuery->lastKey && QUERY_IS_ASC_QUERY(pQuery)) ||
(nextKey <= pQuery->lastKey && !QUERY_IS_ASC_QUERY(pQuery))); (nextKey <= pQuery->lastKey && !QUERY_IS_ASC_QUERY(pQuery)));
/* still in the same block to query */ /* still in the same block to query */
getAlignedIntervalQueryRange(pRuntimeEnv, nextKey, pSupporter->rawSKey, pSupporter->rawEKey); getAlignedIntervalQueryRange(pRuntimeEnv, nextKey, pSupporter->rawSKey, pSupporter->rawEKey);
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
int32_t newPos = searchFn((char *)pPrimaryCol, pBlockInfo->size, pQuery->skey, pQuery->order.order); int32_t newPos = searchFn((char *)pPrimaryCol, pBlockInfo->size, pQuery->skey, pQuery->order.order);
assert(newPos == pQuery->pos + steps * factor); assert(newPos == pQuery->pos + steps * factor);
@ -7691,19 +7682,20 @@ void copyFromGroupBuf(SQInfo *pQInfo, SOutputRes *result) {
assert(pQuery->pointsRead <= pQuery->pointsToRead); assert(pQuery->pointsRead <= pQuery->pointsToRead);
} }
static void applyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterDataInfo *pInfoEx, int64_t *pPrimaryData, static void applyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterDataInfo *pMeterDataInfo,
SBlockInfo *pBlockInfo, int32_t blockStatus, SField *pFields, SBlockInfo *pBlockInfo, int32_t blockStatus, SField *pFields,
__block_search_fn_t searchFn) { __block_search_fn_t searchFn) {
SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv; SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv;
SQuery * pQuery = pRuntimeEnv->pQuery; SQuery * pQuery = pRuntimeEnv->pQuery;
SMeterQueryInfo * pInfo = pInfoEx->pMeterQInfo; SMeterQueryInfo * pMeterQueryInfo = pMeterDataInfo->pMeterQInfo;
int64_t* pPrimaryKey = (int64_t*) pRuntimeEnv->primaryColBuffer->data;
/* /*
* for each block, we need to handle the previous query, since the determination of previous query being completed * for each block, we need to handle the previous query, since the determination of previous query being completed
* or not is based on the start key of current block. * or not is based on the start key of current block.
*/ */
TSKEY key = getNextAccessedKeyInData(pQuery, pPrimaryData, pBlockInfo, blockStatus); TSKEY key = getNextAccessedKeyInData(pQuery, pPrimaryKey, pBlockInfo, blockStatus);
setIntervalQueryRange(pInfoEx->pMeterQInfo, pSupporter, key); setIntervalQueryRange(pMeterDataInfo->pMeterQInfo, pSupporter, key);
if (((pQuery->skey > pQuery->ekey) && QUERY_IS_ASC_QUERY(pQuery)) || if (((pQuery->skey > pQuery->ekey) && QUERY_IS_ASC_QUERY(pQuery)) ||
((pQuery->skey < pQuery->ekey) && !QUERY_IS_ASC_QUERY(pQuery))) { ((pQuery->skey < pQuery->ekey) && !QUERY_IS_ASC_QUERY(pQuery))) {
@ -7714,18 +7706,18 @@ static void applyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterD
((pBlockInfo->keyFirst > pQuery->ekey) && !QUERY_IS_ASC_QUERY(pQuery))) { ((pBlockInfo->keyFirst > pQuery->ekey) && !QUERY_IS_ASC_QUERY(pQuery))) {
int32_t numOfRes = 0; int32_t numOfRes = 0;
/* current block is included in this interval */ /* current block is included in this interval */
int32_t steps = applyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, pPrimaryData, pFields, searchFn, &numOfRes); int32_t steps = applyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, pPrimaryKey, pFields, searchFn, &numOfRes);
assert(numOfRes <= 1 && numOfRes >= 0 && steps > 0); assert(numOfRes <= 1 && numOfRes >= 0 && steps > 0);
if (pInfo->lastResRows == 0) { if (pMeterQueryInfo->lastResRows == 0) {
pInfo->lastResRows = numOfRes; pMeterQueryInfo->lastResRows = numOfRes;
} else { } else {
assert(pInfo->lastResRows == 1); assert(pMeterQueryInfo->lastResRows == 1);
} }
saveIntervalQueryRange(pRuntimeEnv, pInfo); saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo);
} else { } else {
doApplyIntervalQueryOnBlock(pSupporter, pInfo, pBlockInfo, pPrimaryData, pFields, searchFn); doApplyIntervalQueryOnBlock(pSupporter, pMeterQueryInfo, pBlockInfo, pPrimaryKey, pFields, searchFn);
} }
} }