diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index 3407d2fe4bdb456c8b7b2a0f1ef3ab5b92be14a3..ef5ec5808b66b3a100fd72714d7e0972eb574298 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -652,7 +652,16 @@ int32_t parseIntervalClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) { 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) { @@ -675,6 +684,8 @@ int32_t setSlidingClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) { if (pQueryInfo->nSlidingTime > pQueryInfo->nAggTimeInterval) { return invalidSqlErrMsg(pQueryInfo->msg, msg1); } + } else { + pSliding->n = pQueryInfo->nAggTimeInterval; } 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 if (setMeterID(pMeterMetaInfo, pzTableName, pSql) != TSDB_CODE_SUCCESS) { return invalidSqlErrMsg(pQueryInfo->msg, msg1); @@ -5581,29 +5588,29 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) { return invalidSqlErrMsg(pQueryInfo->msg, msg2); } - // set sliding value, the query time range needs to be decide in the first place - SSQLToken* pSliding = &pQuerySql->sliding; - if (pSliding->n != 0) { - if (!tscEmbedded && pCmd->inStream == 0 && hasDefaultQueryTimeRange(pQueryInfo)) { // sliding only allowed in stream - const char* msg = "time range expected for sliding window query"; - return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg); - } - - getTimestampInUsFromStr(pSliding->z, pSliding->n, &pQueryInfo->nSlidingTime); - if (pMeterMetaInfo->pMeterMeta->precision == TSDB_TIME_PRECISION_MILLI) { - pQueryInfo->nSlidingTime /= 1000; - } - - if (pQueryInfo->nSlidingTime < tsMinSlidingTime) { - return invalidSqlErrMsg(pQueryInfo->msg, msg3); - } - - if (pQueryInfo->nSlidingTime > pQueryInfo->nAggTimeInterval) { - return invalidSqlErrMsg(pQueryInfo->msg, msg4); - } - } else { - pQueryInfo->nSlidingTime = -1; - } +// // set sliding value, the query time range needs to be decide in the first place +// SSQLToken* pSliding = &pQuerySql->sliding; +// if (pSliding->n != 0) { +// if (!tscEmbedded && pCmd->inStream == 0 && hasDefaultQueryTimeRange(pQueryInfo)) { // sliding only allowed in stream +// const char* msg = "time range expected for sliding window query"; +// return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg); +// } +// +// getTimestampInUsFromStr(pSliding->z, pSliding->n, &pQueryInfo->nSlidingTime); +// if (pMeterMetaInfo->pMeterMeta->precision == TSDB_TIME_PRECISION_MILLI) { +// pQueryInfo->nSlidingTime /= 1000; +// } +// +// if (pQueryInfo->nSlidingTime < tsMinSlidingTime) { +// return invalidSqlErrMsg(pQueryInfo->msg, msg3); +// } +// +// if (pQueryInfo->nSlidingTime > pQueryInfo->nAggTimeInterval) { +// return invalidSqlErrMsg(pQueryInfo->msg, msg4); +// } +// } else { +// pQueryInfo->nSlidingTime = -1; +// } // in case of join query, time range is required. if (QUERY_IS_JOIN_QUERY(pQueryInfo->type)) { diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 361fb16f697a8398a95815df286121430e2f88c7..492cc5cda2b77fa00ce7b2c4e4df6c122404ce6d 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -1110,8 +1110,7 @@ static void tscHandleSubRetrievalError(SRetrieveSupport *trsupport, SSqlObj *pSq } // all subqueries are failed - tscError("%p retrieve from %d vnode(s) completed,code:%d.FAILED.", pPObj, pState->numOfTotal, - pState->code); + tscError("%p retrieve from %d vnode(s) completed,code:%d.FAILED.", pPObj, pState->numOfTotal, pState->code); pPObj->res.code = -(pState->code); // release allocated resource diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 2b72c24c94ba986e71cc8bca73d630a18e9785a8..6a03278a077bab0aa078685acd8a0ecd3402acce 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -754,6 +754,8 @@ void taos_free_result_imp(TAOS_RES* res, int keepCmd) { pSql->pStream == NULL && pMeterMetaInfo->pMeterMeta != NULL))) { 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; if (fp != NULL) { pSql->freed = 1; diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c index 7314f55144a61ed2f79c16c0610866da7ec0b391..9fc9706dd9fb87179fabcc484f7ae56afb231914 100644 --- a/src/client/src/tscStream.c +++ b/src/client/src/tscStream.c @@ -411,7 +411,9 @@ static void tscSetSlidingWindowInfo(SSqlObj *pSql, SSqlStream *pStream) { } pStream->slidingTime = pQueryInfo->nSlidingTime; + 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) { diff --git a/src/client/src/tscSyntaxtreefunction.c b/src/client/src/tscSyntaxtreefunction.c index 914053f2f17461e0f5e7ffbd56691e7fd206cd49..1d82b0f239572676c204025fe535704c192e4da6 100644 --- a/src/client/src/tscSyntaxtreefunction.c +++ b/src/client/src/tscSyntaxtreefunction.c @@ -26,7 +26,7 @@ int32_t step = ((_ord) == TSQL_SO_ASC) ? 1 : -1; \ \ 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)) { \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ continue; \ @@ -34,7 +34,7 @@ *(out) = (double)(left)[i] op(right)[i]; \ } \ } 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)) { \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ continue; \ @@ -42,7 +42,7 @@ *(out) = (double)(left)[0] op(right)[i]; \ } \ } 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)) { \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ continue; \ @@ -58,7 +58,7 @@ int32_t step = (_ord == TSQL_SO_ASC) ? 1 : -1; \ \ 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)) { \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ continue; \ @@ -66,7 +66,7 @@ *(out) = (double)(left)[i] - ((int64_t)(((double)(left)[i]) / (right)[i])) * (right)[i]; \ } \ } 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)) { \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ continue; \ @@ -74,7 +74,7 @@ *(out) = (double)(left)[0] - ((int64_t)(((double)(left)[0]) / (right)[i])) * (right)[i]; \ } \ } 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)) { \ setNull((char *)(out), _res_type, tDataTypeDesc[_res_type].nSize); \ 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; 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)) { setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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; 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)) { setNull((char *)&(pOutput[i]), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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; 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)) { setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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; 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)) { setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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; 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)) { setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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; 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)) { setNull((char *)(pOutput), TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); 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]; } } 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)) { setNull((char *)pOutput, TSDB_DATA_TYPE_DOUBLE, tDataTypeDesc[TSDB_DATA_TYPE_DOUBLE].nSize); continue; diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 8c5953c10c4c429a720b229f1ecd069ea9e5c6ce..e9395d7dde46e478ae5058e6842579eb33079b57 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -214,6 +214,10 @@ bool tscIsTwoStageMergeMetricQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) { if (pMeterMetaInfo == NULL || pMeterMetaInfo->pMetricMeta == NULL) { 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 if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, tableIndex)) { diff --git a/src/system/detail/src/vnodeQueryImpl.c b/src/system/detail/src/vnodeQueryImpl.c index ce38ada8bfef01d63e5c0d36799544558b28a967..81fce50bd6f950c2f50fc60efb4ce251481855bb 100644 --- a/src/system/detail/src/vnodeQueryImpl.c +++ b/src/system/detail/src/vnodeQueryImpl.c @@ -49,8 +49,6 @@ enum { #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, int32_t size); @@ -70,12 +68,12 @@ static int32_t getNextDataFileCompInfo(SQueryRuntimeEnv *pRuntimeEnv, SMeterObj static void setGroupOutputBuffer(SQueryRuntimeEnv *pRuntimeEnv, SOutputRes *pResult); 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, __block_search_fn_t searchFn); 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, __block_search_fn_t searchFn); @@ -1728,6 +1726,8 @@ int32_t numOfClosedSlidingWindow(SSlidingWindowInfo *pSlidingWindowInfo) { return i; } } + + return 0; } void closeSlidingWindow(SSlidingWindowInfo* pSlidingWindowInfo, int32_t slot) { @@ -2112,7 +2112,7 @@ static int32_t rowwiseApplyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, int32_t * SSlidingWindowInfo *pSlidingWindowInfo = &pRuntimeEnv->swindowResInfo; // 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))) { closeAllSlidingWindow(pSlidingWindowInfo); @@ -5027,7 +5027,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) { static void updatelastkey(SQuery *pQuery, SMeterQueryInfo *pMeterQInfo) { pMeterQInfo->lastKey = pQuery->lastKey; } 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) { /* cache blocks may be assign to other meter, abort */ 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 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 - updatelastkey(pQuery, pDataHeadInfoEx->pMeterQInfo); + updatelastkey(pQuery, pMeterDataInfo->pMeterQInfo); if (pRuntimeEnv->pTSBuf != NULL) { - pDataHeadInfoEx->pMeterQInfo->cur = tsBufGetCursor(pRuntimeEnv->pTSBuf); + pMeterDataInfo->pMeterQInfo->cur = tsBufGetCursor(pRuntimeEnv->pTSBuf); } } else { - applyIntervalQueryOnBlock(pSupporter, pDataHeadInfoEx, primaryKeys, pBlockBasicInfo, blockStatus, pFields, - searchFn); + applyIntervalQueryOnBlock(pSupporter, pMeterDataInfo, pBlockBasicInfo, blockStatus, pFields, searchFn); } } @@ -5265,13 +5264,13 @@ static tFilePage *getFilePage(SMeterQuerySupportObj *pSupporter, int32_t pageId) return (tFilePage *)(pSupporter->meterOutputMMapBuf + DEFAULT_INTERN_BUF_SIZE * pageId); } -static tFilePage *getMeterDataPage(SMeterQuerySupportObj *pSupporter, SMeterDataInfo *pInfoEx, int32_t pageId) { - SMeterQueryInfo *pInfo = pInfoEx->pMeterQInfo; - if (pageId >= pInfo->numOfPages) { +static tFilePage *getMeterDataPage(SMeterQuerySupportObj *pSupporter, SMeterDataInfo *pMeterDataInfo, int32_t pageId) { + SMeterQueryInfo *pMeterQueryInfo = pMeterDataInfo->pMeterQInfo; + if (pageId >= pMeterQueryInfo->numOfPages) { return NULL; } - int32_t realId = pInfo->pageList[pageId]; + int32_t realId = pMeterQueryInfo->pageList[pageId]; return getFilePage(pSupporter, realId); } @@ -5281,14 +5280,14 @@ typedef struct Position { } Position; typedef struct SCompSupporter { - SMeterDataInfo ** pInfoEx; + SMeterDataInfo ** pMeterDataInfo; Position * pPosition; SMeterQuerySupportObj *pSupporter; } SCompSupporter; int64_t getCurrentTimestamp(SCompSupporter *pSupportor, int32_t 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); } @@ -5311,10 +5310,10 @@ int32_t meterResultComparator(const void *pLeft, const void *pRight, void *param 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); - 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); if (leftTimestamp == rightTimestamp) { @@ -5465,17 +5464,17 @@ int32_t doMergeMetersResultsToGroupRes(SMeterQuerySupportObj *pSupporter, SQuery 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 - 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); if (newPage->numOfElems <= 0) { // if current source data page is null, it must be the last page of source output page 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 - 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].rowIdx = -1; @@ -5635,7 +5634,7 @@ int32_t doCloseAllOpenedResults(SMeterQuerySupportObj *pSupporter) { void disableFunctForSuppleScan(SQueryRuntimeEnv *pRuntimeEnv, int32_t order) { 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) { 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, SMeterDataBlockInfoEx **pDataBlockInfoEx, int32_t numOfCompBlocks, - int32_t *nAllocBlocksInfoSize, int64_t addr) { + int32_t *numOfAllocBlocks, int64_t addr) { // 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); if (tmp == NULL) { tfree(*pDataBlockInfoEx); @@ -6706,7 +6705,7 @@ int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMet *pDataBlockInfoEx = (SMeterDataBlockInfoEx *)tmp; memset((*pDataBlockInfoEx), 0, sizeof(SMeterDataBlockInfoEx) * numOfCompBlocks); - *nAllocBlocksInfoSize = numOfCompBlocks; + *numOfAllocBlocks = numOfCompBlocks; } SBlockOrderSupporter supporter = {0}; @@ -6740,14 +6739,14 @@ int32_t createDataBlocksInfoEx(SMeterDataInfo **pMeterDataInfo, int32_t numOfMet supporter.pDataBlockInfoEx[numOfQualMeters] = (SMeterDataBlockInfoEx *)buf; for (int32_t k = 0; k < pMeterDataInfo[j]->numOfBlocks; ++k) { - SMeterDataBlockInfoEx *pInfoEx = &supporter.pDataBlockInfoEx[numOfQualMeters][k]; - - pInfoEx->pBlock.compBlock = &pBlock[k]; - pInfoEx->pBlock.fields = NULL; - - pInfoEx->pMeterDataInfo = pMeterDataInfo[j]; - pInfoEx->groupIdx = pMeterDataInfo[j]->groupIdx; // set the group index - pInfoEx->blockIndex = pMeterDataInfo[j]->start + k; // set the block index in original meter + SMeterDataBlockInfoEx *pBlockInfoEx = &supporter.pDataBlockInfoEx[numOfQualMeters][k]; + + pBlockInfoEx->pBlock.compBlock = &pBlock[k]; + pBlockInfoEx->pBlock.fields = NULL; + + pBlockInfoEx->pMeterDataInfo = pMeterDataInfo[j]; + pBlockInfoEx->groupIdx = pMeterDataInfo[j]->groupIdx; // set the group index + pBlockInfoEx->blockIndex = pMeterDataInfo[j]->start + k; // set the block index in original meter cnt++; } @@ -7013,7 +7012,7 @@ int32_t setIntervalQueryExecutionContext(SMeterQuerySupportObj *pSupporter, int3 return 0; } -static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pInfo, +static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMeterQueryInfo *pMeterQueryInfo, SBlockInfo *pBlockInfo, int64_t *pPrimaryCol, SField *pFields, __block_search_fn_t searchFn) { SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv; @@ -7029,10 +7028,10 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete assert(steps > 0); // NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range - if (pInfo->lastResRows == 0) { - pInfo->lastResRows = numOfRes; + if (pMeterQueryInfo->lastResRows == 0) { + pMeterQueryInfo->lastResRows = numOfRes; } else { - assert(pInfo->lastResRows == 1); + assert(pMeterQueryInfo->lastResRows == 1); } 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) { /* whole query completed, save result and abort */ 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. - saveIntervalQueryRange(pRuntimeEnv, pInfo); + saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo); } else if (pQuery->ekey == pBlockInfo->keyLast) { /* current interval query is completed, set the next query range on other data blocks if exist */ int64_t prevEKey = pQuery->ekey; getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey); - saveIntervalQueryRange(pRuntimeEnv, pInfo); + saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo); assert(queryCompleted && prevEKey < pQuery->skey); - if (pInfo->lastResRows > 0) { - saveResult(pSupporter, pInfo, pInfo->lastResRows); + if (pMeterQueryInfo->lastResRows > 0) { + saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows); } } 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, * 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); /* @@ -7103,7 +7102,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete * merge with other meters in the same group */ 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)) { /* whole query completed, save result and abort */ 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. */ - saveIntervalQueryRange(pRuntimeEnv, pInfo); + saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo); } else if (pQuery->ekey == pBlockInfo->keyFirst) { // current interval query is completed, set the next query range on other data blocks if exist int64_t prevEKey = pQuery->ekey; getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey); - saveIntervalQueryRange(pRuntimeEnv, pInfo); + saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo); assert(queryCompleted && prevEKey > pQuery->skey); - if (pInfo->lastResRows > 0) { - saveResult(pSupporter, pInfo, pInfo->lastResRows); + if (pMeterQueryInfo->lastResRows > 0) { + saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows); } } 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, * 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); /* @@ -7154,7 +7153,7 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete * flag, and merge with other meters in the same group */ if (queryCompleted) { - saveResult(pSupporter, pInfo, pInfo->lastResRows); + saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows); } } @@ -7163,14 +7162,14 @@ static void doApplyIntervalQueryOnBlock(SMeterQuerySupportObj *pSupporter, SMete } assert(queryCompleted); - saveResult(pSupporter, pInfo, pInfo->lastResRows); + saveResult(pSupporter, pMeterQueryInfo, pMeterQueryInfo->lastResRows); assert((nextKey >= pQuery->lastKey && QUERY_IS_ASC_QUERY(pQuery)) || (nextKey <= pQuery->lastKey && !QUERY_IS_ASC_QUERY(pQuery))); /* still in the same block to query */ 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); 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, __block_search_fn_t searchFn) { SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv; SQuery * pQuery = pRuntimeEnv->pQuery; int32_t factor = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); - int64_t nextKey = -1; - bool completed = false; - while (1) { + int64_t nextKey = -1; int32_t numOfRes = 0; + int32_t steps = applyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, pPrimaryCol, pFields, searchFn, &numOfRes); assert(steps > 0); // NOTE: in case of stable query, only ONE(or ZERO) row of result generated for each query range - if (pInfo->lastResRows == 0) { - pInfo->lastResRows = numOfRes; + if (pMeterQueryInfo->lastResRows == 0) { + pMeterQueryInfo->lastResRows = numOfRes; } else { - assert(pInfo->lastResRows == 1); + assert(pMeterQueryInfo->lastResRows == 1); } 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 + bool completed = false; if (QUERY_IS_ASC_QUERY(pQuery)) { completed = (pQuery->lastKey > pQuery->ekey); } 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, * current interval query being completed or not can be decided. */ - saveIntervalQueryRange(pRuntimeEnv, pInfo); + saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo); if (QUERY_IS_ASC_QUERY(pQuery)) { 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) { /* 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. - saveIntervalQueryRange(pRuntimeEnv, pInfo); + saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo); return; } else if ((QUERY_IS_ASC_QUERY(pQuery) && pQuery->ekey == pBlockInfo->keyLast) || - (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 */ -// int64_t prevEKey = pQuery->ekey; -// -// getAlignedIntervalQueryRange(pRuntimeEnv, pQuery->lastKey, pSupporter->rawSKey, pSupporter->rawEKey); - saveIntervalQueryRange(pRuntimeEnv, pInfo); -// -// assert(prevEKey < pQuery->skey); -// if (pInfo->lastResRows > 0) { -// saveResult(pSupporter, pInfo, pInfo->lastResRows); -// } -// + (!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 */ + saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo); 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)) || (nextKey <= pQuery->lastKey && !QUERY_IS_ASC_QUERY(pQuery))); /* still in the same block to query */ 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); assert(newPos == pQuery->pos + steps * factor); @@ -7691,19 +7682,20 @@ void copyFromGroupBuf(SQInfo *pQInfo, SOutputRes *result) { 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, __block_search_fn_t searchFn) { SQueryRuntimeEnv *pRuntimeEnv = &pSupporter->runtimeEnv; 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 * or not is based on the start key of current block. */ - TSKEY key = getNextAccessedKeyInData(pQuery, pPrimaryData, pBlockInfo, blockStatus); - setIntervalQueryRange(pInfoEx->pMeterQInfo, pSupporter, key); + TSKEY key = getNextAccessedKeyInData(pQuery, pPrimaryKey, pBlockInfo, blockStatus); + setIntervalQueryRange(pMeterDataInfo->pMeterQInfo, pSupporter, key); if (((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))) { int32_t numOfRes = 0; /* 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); - if (pInfo->lastResRows == 0) { - pInfo->lastResRows = numOfRes; + if (pMeterQueryInfo->lastResRows == 0) { + pMeterQueryInfo->lastResRows = numOfRes; } else { - assert(pInfo->lastResRows == 1); + assert(pMeterQueryInfo->lastResRows == 1); } - saveIntervalQueryRange(pRuntimeEnv, pInfo); + saveIntervalQueryRange(pRuntimeEnv, pMeterQueryInfo); } else { - doApplyIntervalQueryOnBlock(pSupporter, pInfo, pBlockInfo, pPrimaryData, pFields, searchFn); + doApplyIntervalQueryOnBlock(pSupporter, pMeterQueryInfo, pBlockInfo, pPrimaryKey, pFields, searchFn); } }