diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c index ffe42de2362d3d8a837290799d656b9d32fe96bd..4e05e1ce4c7fea9719e6249bd4b8ed48e5dce4d4 100644 --- a/src/client/src/tscAsync.c +++ b/src/client/src/tscAsync.c @@ -45,6 +45,7 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const pSql->pTscObj = pObj; pSql->maxRetry = TSDB_MAX_REPLICA; pSql->fp = fp; + pSql->fetchFp = fp; pSql->sqlstr = calloc(1, sqlLen + 1); if (pSql->sqlstr == NULL) { @@ -159,7 +160,7 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo pRes->code = numOfRows; } - tscQueueAsyncError(pSql->fetchFp, param, pRes->code); + tscQueueAsyncRes(pSql); return; } @@ -167,6 +168,7 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo if (pCmd->command != TSDB_SQL_RETRIEVE_LOCALMERGE && pCmd->command < TSDB_SQL_LOCAL) { pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH; } + tscProcessSql(pSql); } @@ -346,31 +348,32 @@ void tscProcessFetchRow(SSchedMsg *pMsg) { void tscProcessAsyncRes(SSchedMsg *pMsg) { SSqlObj *pSql = (SSqlObj *)pMsg->ahandle; - SSqlCmd *pCmd = &pSql->cmd; +// SSqlCmd *pCmd = &pSql->cmd; SSqlRes *pRes = &pSql->res; - void *taosres = pSql; +// void *taosres = pSql; // pCmd may be released, so cache pCmd->command - int cmd = pCmd->command; - int code = pRes->code; +// int cmd = pCmd->command; +// int code = pRes->code; // in case of async insert, restore the user specified callback function - bool shouldFree = tscShouldBeFreed(pSql); - - if (cmd == TSDB_SQL_INSERT) { - assert(pSql->fp != NULL); - pSql->fp = pSql->fetchFp; - } - - if (pSql->fp) { - (*pSql->fp)(pSql->param, taosres, code); - } - - if (shouldFree) { - tscDebug("%p sqlObj is automatically freed in async res", pSql); - tscFreeSqlObj(pSql); - } +// bool shouldFree = tscShouldBeFreed(pSql); + +// if (pCmd->command == TSDB_SQL_INSERT) { +// assert(pSql->fp != NULL); + assert(pSql->fp != NULL && pSql->fetchFp != NULL); +// } + +// if (pSql->fp) { + pSql->fp = pSql->fetchFp; + (*pSql->fp)(pSql->param, pSql, pRes->code); +// } + +// if (shouldFree) { +// tscDebug("%p sqlObj is automatically freed in async res", pSql); +// tscFreeSqlObj(pSql); +// } } static void tscProcessAsyncError(SSchedMsg *pMsg) { @@ -420,15 +423,15 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { SSqlCmd *pCmd = &pSql->cmd; SSqlRes *pRes = &pSql->res; + pRes->code = code; if (code != TSDB_CODE_SUCCESS) { - pRes->code = code; - tscQueueAsyncRes(pSql); - return; + tscError("%p ge tableMeta failed, code:%s", pSql, tstrerror(code)); + goto _error; + } else { + tscDebug("%p get tableMeta successfully", pSql); } - tscDebug("%p get tableMeta successfully", pSql); - if (pSql->pStream == NULL) { SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); @@ -453,11 +456,9 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { assert(pParObj->signature == pParObj && trs->subqueryIndex == pTableMetaInfo->vgroupIndex && pTableMetaInfo->vgroupIndex >= 0 && pTableMetaInfo->vgroupList != NULL); - if ((code = tscProcessSql(pSql)) == TSDB_CODE_SUCCESS) { - return; - } - - goto _error; + // tscProcessSql can add error into async res + tscProcessSql(pSql); + return; } else { // continue to process normal async query if (pCmd->parseFinished) { tscDebug("%p update table meta in local cache, continue to process sql and send corresponding query", pSql); @@ -481,26 +482,21 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { return; + } else if (code != TSDB_CODE_SUCCESS) { + goto _error; } - if (code == TSDB_CODE_SUCCESS) { - /* - * Discard previous built submit blocks, and then parse the sql string again and build up all submit blocks, - * and send the required submit block according to index value in supporter to server. - */ - pSql->fp = pSql->fetchFp; // restore the fp - if ((code = tscHandleInsertRetry(pSql)) == TSDB_CODE_SUCCESS) { - return; - } - } - + /* + * Discard previous built submit blocks, and then parse the sql string again and build up all submit blocks, + * and send the required submit block according to index value in supporter to server. + */ + pSql->fp = pSql->fetchFp; // restore the fp + tscHandleInsertRetry(pSql); } else {// in case of other query type, continue - if ((code = tscProcessSql(pSql)) == TSDB_CODE_SUCCESS) { - return; - } + tscProcessSql(pSql); } - goto _error; + return; } else { tscDebug("%p continue parse sql after get table meta", pSql); @@ -538,7 +534,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { goto _error; } - if (code == TSDB_CODE_SUCCESS && UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { + if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) { code = tscGetSTableVgroupInfo(pSql, pCmd->clauseIndex); if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { return; diff --git a/src/client/src/tscFunctionImpl.c b/src/client/src/tscFunctionImpl.c index 06e5ff73bf76d7a2d794fa23544a4967d90d5028..1ec84f023ad74cc8c29654bf14f55781485b691d 100644 --- a/src/client/src/tscFunctionImpl.c +++ b/src/client/src/tscFunctionImpl.c @@ -330,10 +330,6 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI return TSDB_CODE_SUCCESS; } -bool stableQueryFunctChanged(int32_t funcId) { - return (aAggs[funcId].stableFuncId != funcId); -} - /** * the numOfRes should be kept, since it may be used later * and allow the ResultInfo to be re initialized @@ -361,7 +357,6 @@ static bool function_setup(SQLFunctionCtx *pCtx) { } memset(pCtx->aOutputBuf, 0, (size_t)pCtx->outputBytes); - initResultInfo(pResInfo); return true; } @@ -675,16 +670,16 @@ static void sum_func_second_merge(SQLFunctionCtx *pCtx) { } } -static int32_t precal_req_load_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { +static int32_t statisRequired(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { return BLK_DATA_STATIS_NEEDED; } -static int32_t data_req_load_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { +static int32_t dataBlockRequired(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { return BLK_DATA_ALL_NEEDED; } // todo: if column in current data block are null, opt for this case -static int32_t first_data_req_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { +static int32_t firstFuncRequired(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { if (pCtx->order == TSDB_ORDER_DESC) { return BLK_DATA_NO_NEEDED; } @@ -697,7 +692,7 @@ static int32_t first_data_req_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, } } -static int32_t last_data_req_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { +static int32_t lastFuncRequired(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { if (pCtx->order != pCtx->param[0].i64Key) { return BLK_DATA_NO_NEEDED; } @@ -709,34 +704,40 @@ static int32_t last_data_req_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, } } -static int32_t first_dist_data_req_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { +static int32_t firstDistFuncRequired(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { if (pCtx->order == TSDB_ORDER_DESC) { return BLK_DATA_NO_NEEDED; } + + // not initialized yet, it is the first block, load it. + if (pCtx->aOutputBuf == NULL) { + return BLK_DATA_ALL_NEEDED; + } - // result buffer has not been set yet. - return BLK_DATA_ALL_NEEDED; - //todo optimize the filter info -// SFirstLastInfo *pInfo = (SFirstLastInfo*) (pCtx->aOutputBuf + pCtx->inputBytes); -// if (pInfo->hasResult != DATA_SET_FLAG) { -// return BLK_DATA_ALL_NEEDED; -// } else { // data in current block is not earlier than current result -// return (pInfo->ts <= start) ? BLK_DATA_NO_NEEDED : BLK_DATA_ALL_NEEDED; -// } + SFirstLastInfo *pInfo = (SFirstLastInfo*) (pCtx->aOutputBuf + pCtx->inputBytes); + if (pInfo->hasResult != DATA_SET_FLAG) { + return BLK_DATA_ALL_NEEDED; + } else { // data in current block is not earlier than current result + return (pInfo->ts <= start) ? BLK_DATA_NO_NEEDED : BLK_DATA_ALL_NEEDED; + } } -static int32_t last_dist_data_req_info(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { +static int32_t lastDistFuncRequired(SQLFunctionCtx *pCtx, TSKEY start, TSKEY end, int32_t colId) { if (pCtx->order != pCtx->param[0].i64Key) { return BLK_DATA_NO_NEEDED; } - - return BLK_DATA_ALL_NEEDED; -// SFirstLastInfo *pInfo = (SFirstLastInfo*) (pCtx->aOutputBuf + pCtx->inputBytes); -// if (pInfo->hasResult != DATA_SET_FLAG) { -// return BLK_DATA_ALL_NEEDED; -// } else { -// return (pInfo->ts > end) ? BLK_DATA_NO_NEEDED : BLK_DATA_ALL_NEEDED; -// } + + // not initialized yet, it is the first block, load it. + if (pCtx->aOutputBuf == NULL) { + return BLK_DATA_ALL_NEEDED; + } + + SFirstLastInfo *pInfo = (SFirstLastInfo*) (pCtx->aOutputBuf + pCtx->inputBytes); + if (pInfo->hasResult != DATA_SET_FLAG) { + return BLK_DATA_ALL_NEEDED; + } else { + return (pInfo->ts > end) ? BLK_DATA_NO_NEEDED : BLK_DATA_ALL_NEEDED; + } } ////////////////////////////////////////////////////////////////////////////////////////////// @@ -1549,6 +1550,8 @@ static void first_data_assign_impl(SQLFunctionCtx *pCtx, char *pData, int32_t in * to decide if the value is earlier than current intermediate result */ static void first_dist_function(SQLFunctionCtx *pCtx) { + assert(pCtx->size > 0); + if (pCtx->size == 0) { return; } @@ -1563,7 +1566,12 @@ static void first_dist_function(SQLFunctionCtx *pCtx) { } int32_t notNullElems = 0; - + + // data block is discard, not loaded, do not need to check it + if (!pCtx->preAggVals.dataBlockLoaded) { + return; + } + // find the first not null value for (int32_t i = 0; i < pCtx->size; ++i) { char *data = GET_INPUT_CHAR_INDEX(pCtx, i); @@ -1584,10 +1592,6 @@ static void first_dist_function(SQLFunctionCtx *pCtx) { } static void first_dist_function_f(SQLFunctionCtx *pCtx, int32_t index) { - if (pCtx->size == 0) { - return; - } - char *pData = GET_INPUT_CHAR_INDEX(pCtx, index); if (pCtx->hasNull && isNull(pData, pCtx->inputType)) { return; @@ -1715,10 +1719,6 @@ static void last_data_assign_impl(SQLFunctionCtx *pCtx, char *pData, int32_t ind } static void last_dist_function(SQLFunctionCtx *pCtx) { - if (pCtx->size == 0) { - return; - } - /* * 1. for scan data in asc order, no need to check data * 2. for data blocks that are not loaded, no need to check data @@ -1726,7 +1726,12 @@ static void last_dist_function(SQLFunctionCtx *pCtx) { if (pCtx->order != pCtx->param[0].i64Key) { return; } - + + // data block is discard, not loaded, do not need to check it + if (!pCtx->preAggVals.dataBlockLoaded) { + return; + } + int32_t notNullElems = 0; for (int32_t i = pCtx->size - 1; i >= 0; --i) { @@ -2123,74 +2128,72 @@ static void copyTopBotRes(SQLFunctionCtx *pCtx, int32_t type) { tfree(pData); } -bool top_bot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, char *minval, char *maxval) { - STopBotInfo *pTopBotInfo = (STopBotInfo *)GET_RES_INFO(pCtx)->interResultBuf; - - int32_t numOfExistsRes = pTopBotInfo->num; +/* + * Parameters values: + * 1. param[0]: maximum allowable results + * 2. param[1]: order by type (time or value) + * 3. param[2]: asc/desc order + * + * top/bottom use the intermediate result buffer to keep the intermediate result + */ +static STopBotInfo *getTopBotOutputInfo(SQLFunctionCtx *pCtx) { + SResultInfo *pResInfo = GET_RES_INFO(pCtx); + + // only the first_stage_merge is directly written data into final output buffer + if (pResInfo->superTableQ && pCtx->currentStage != SECONDARY_STAGE_MERGE) { + return (STopBotInfo*) pCtx->aOutputBuf; + } else { // during normal table query and super table at the secondary_stage, result is written to intermediate buffer + return pResInfo->interResultBuf; + } +} + +bool topbot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, const char *minval, const char *maxval) { + STopBotInfo *pTopBotInfo = getTopBotOutputInfo(pCtx); // required number of results are not reached, continue load data block - if (numOfExistsRes < pCtx->param[0].i64Key) { + if (pTopBotInfo->num < pCtx->param[0].i64Key) { return true; } - tValuePair *pRes = (tValuePair*) pTopBotInfo->res; + tValuePair **pRes = (tValuePair**) pTopBotInfo->res; if (functionId == TSDB_FUNC_TOP) { switch (pCtx->inputType) { case TSDB_DATA_TYPE_TINYINT: - return GET_INT8_VAL(maxval) > pRes[0].v.i64Key; + return GET_INT8_VAL(maxval) > pRes[0]->v.i64Key; case TSDB_DATA_TYPE_SMALLINT: - return GET_INT16_VAL(maxval) > pRes[0].v.i64Key; + return GET_INT16_VAL(maxval) > pRes[0]->v.i64Key; case TSDB_DATA_TYPE_INT: - return GET_INT32_VAL(maxval) > pRes[0].v.i64Key; + return GET_INT32_VAL(maxval) > pRes[0]->v.i64Key; case TSDB_DATA_TYPE_BIGINT: - return GET_INT64_VAL(maxval) > pRes[0].v.i64Key; + return GET_INT64_VAL(maxval) > pRes[0]->v.i64Key; case TSDB_DATA_TYPE_FLOAT: - return GET_FLOAT_VAL(maxval) > pRes[0].v.dKey; + return GET_FLOAT_VAL(maxval) > pRes[0]->v.dKey; case TSDB_DATA_TYPE_DOUBLE: - return GET_DOUBLE_VAL(maxval) > pRes[0].v.dKey; + return GET_DOUBLE_VAL(maxval) > pRes[0]->v.dKey; default: return true; } } else { switch (pCtx->inputType) { case TSDB_DATA_TYPE_TINYINT: - return GET_INT8_VAL(minval) < pRes[0].v.i64Key; + return GET_INT8_VAL(minval) < pRes[0]->v.i64Key; case TSDB_DATA_TYPE_SMALLINT: - return GET_INT16_VAL(minval) < pRes[0].v.i64Key; + return GET_INT16_VAL(minval) < pRes[0]->v.i64Key; case TSDB_DATA_TYPE_INT: - return GET_INT32_VAL(minval) < pRes[0].v.i64Key; + return GET_INT32_VAL(minval) < pRes[0]->v.i64Key; case TSDB_DATA_TYPE_BIGINT: - return GET_INT64_VAL(minval) < pRes[0].v.i64Key; + return GET_INT64_VAL(minval) < pRes[0]->v.i64Key; case TSDB_DATA_TYPE_FLOAT: - return GET_FLOAT_VAL(minval) < pRes[0].v.dKey; + return GET_FLOAT_VAL(minval) < pRes[0]->v.dKey; case TSDB_DATA_TYPE_DOUBLE: - return GET_DOUBLE_VAL(minval) < pRes[0].v.dKey; + return GET_DOUBLE_VAL(minval) < pRes[0]->v.dKey; default: return true; } } } -/* - * Parameters values: - * 1. param[0]: maximum allowable results - * 2. param[1]: order by type (time or value) - * 3. param[2]: asc/desc order - * - * top/bottom use the intermediate result buffer to keep the intermediate result - */ -static STopBotInfo *getTopBotOutputInfo(SQLFunctionCtx *pCtx) { - SResultInfo *pResInfo = GET_RES_INFO(pCtx); - - // only the first_stage_merge is directly written data into final output buffer - if (pResInfo->superTableQ && pCtx->currentStage != SECONDARY_STAGE_MERGE) { - return (STopBotInfo*) pCtx->aOutputBuf; - } else { // during normal table query and super table at the secondary_stage, result is written to intermediate buffer - return pResInfo->interResultBuf; - } -} - /* * keep the intermediate results during scan data blocks in the format of: * +-----------------------------------+-------------one value pair-----------+------------next value pair-----------+ @@ -3376,7 +3379,7 @@ static void spread_function(SQLFunctionCtx *pCtx) { SResultInfo *pResInfo = GET_RES_INFO(pCtx); SSpreadInfo *pInfo = pResInfo->interResultBuf; - int32_t numOfElems = pCtx->size; + int32_t numOfElems = 0; // todo : opt with pre-calculated result // column missing cause the hasNull to be true @@ -4412,7 +4415,7 @@ static void sumrate_finalizer(SQLFunctionCtx *pCtx) { * e.g., count/sum/avg/min/max/stddev/percentile/apercentile/first/last... * */ -int32_t funcCompatDefList[] = { +int32_t functionCompatList[] = { // count, sum, avg, min, max, stddev, percentile, apercentile, first, last 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, // last_row, top, bottom, spread, twa, leastsqr, ts, ts_dummy, tag_dummy, ts_z @@ -4451,7 +4454,7 @@ SQLAggFuncElem aAggs[] = {{ function_finalizer, sum_func_merge, sum_func_second_merge, - precal_req_load_info, + statisRequired, }, { // 2 @@ -4466,7 +4469,7 @@ SQLAggFuncElem aAggs[] = {{ avg_finalizer, avg_func_merge, avg_func_second_merge, - precal_req_load_info, + statisRequired, }, { // 3 @@ -4481,7 +4484,7 @@ SQLAggFuncElem aAggs[] = {{ function_finalizer, min_func_merge, min_func_second_merge, - precal_req_load_info, + statisRequired, }, { // 4 @@ -4496,7 +4499,7 @@ SQLAggFuncElem aAggs[] = {{ function_finalizer, max_func_merge, max_func_second_merge, - precal_req_load_info, + statisRequired, }, { // 5 @@ -4511,7 +4514,7 @@ SQLAggFuncElem aAggs[] = {{ stddev_finalizer, noop1, noop1, - data_req_load_info, + dataBlockRequired, }, { // 6 @@ -4526,7 +4529,7 @@ SQLAggFuncElem aAggs[] = {{ percentile_finalizer, noop1, noop1, - data_req_load_info, + dataBlockRequired, }, { // 7 @@ -4541,7 +4544,7 @@ SQLAggFuncElem aAggs[] = {{ apercentile_finalizer, apercentile_func_merge, apercentile_func_second_merge, - data_req_load_info, + dataBlockRequired, }, { // 8 @@ -4556,7 +4559,7 @@ SQLAggFuncElem aAggs[] = {{ function_finalizer, noop1, noop1, - first_data_req_info, + firstFuncRequired, }, { // 9 @@ -4571,7 +4574,7 @@ SQLAggFuncElem aAggs[] = {{ function_finalizer, noop1, noop1, - last_data_req_info, + lastFuncRequired, }, { // 10 @@ -4587,7 +4590,7 @@ SQLAggFuncElem aAggs[] = {{ last_row_finalizer, noop1, last_dist_func_second_merge, - data_req_load_info, + dataBlockRequired, }, { // 11 @@ -4603,7 +4606,7 @@ SQLAggFuncElem aAggs[] = {{ top_bottom_func_finalizer, top_func_merge, top_func_second_merge, - data_req_load_info, + dataBlockRequired, }, { // 12 @@ -4619,7 +4622,7 @@ SQLAggFuncElem aAggs[] = {{ top_bottom_func_finalizer, bottom_func_merge, bottom_func_second_merge, - data_req_load_info, + dataBlockRequired, }, { // 13 @@ -4649,7 +4652,7 @@ SQLAggFuncElem aAggs[] = {{ twa_function_finalizer, twa_func_merge, twa_function_copy, - data_req_load_info, + dataBlockRequired, }, { // 15 @@ -4664,7 +4667,7 @@ SQLAggFuncElem aAggs[] = {{ leastsquares_finalizer, noop1, noop1, - data_req_load_info, + dataBlockRequired, }, { // 16 @@ -4694,7 +4697,7 @@ SQLAggFuncElem aAggs[] = {{ doFinalizer, copy_function, copy_function, - data_req_load_info, + dataBlockRequired, }, { // 18 @@ -4724,7 +4727,7 @@ SQLAggFuncElem aAggs[] = {{ ts_comp_finalize, copy_function, copy_function, - data_req_load_info, + dataBlockRequired, }, { // 20 @@ -4754,7 +4757,7 @@ SQLAggFuncElem aAggs[] = {{ doFinalizer, copy_function, copy_function, - data_req_load_info, + dataBlockRequired, }, { // 22, multi-output, tag function has only one result @@ -4784,7 +4787,7 @@ SQLAggFuncElem aAggs[] = {{ doFinalizer, copy_function, copy_function, - data_req_load_info, + dataBlockRequired, }, { // 24 @@ -4799,7 +4802,7 @@ SQLAggFuncElem aAggs[] = {{ doFinalizer, noop1, noop1, - data_req_load_info, + dataBlockRequired, }, // distributed version used in two-stage aggregation processes { @@ -4815,7 +4818,7 @@ SQLAggFuncElem aAggs[] = {{ function_finalizer, first_dist_func_merge, first_dist_func_second_merge, - first_dist_data_req_info, + firstDistFuncRequired, }, { // 26 @@ -4830,7 +4833,7 @@ SQLAggFuncElem aAggs[] = {{ function_finalizer, last_dist_func_merge, last_dist_func_second_merge, - last_dist_data_req_info, + lastDistFuncRequired, }, { // 27 @@ -4845,7 +4848,7 @@ SQLAggFuncElem aAggs[] = {{ doFinalizer, noop1, copy_function, - data_req_load_info, + dataBlockRequired, }, { // 28 @@ -4860,7 +4863,7 @@ SQLAggFuncElem aAggs[] = {{ rate_finalizer, rate_func_merge, rate_func_copy, - data_req_load_info, + dataBlockRequired, }, { // 29 @@ -4875,7 +4878,7 @@ SQLAggFuncElem aAggs[] = {{ rate_finalizer, rate_func_merge, rate_func_copy, - data_req_load_info, + dataBlockRequired, }, { // 30 @@ -4890,7 +4893,7 @@ SQLAggFuncElem aAggs[] = {{ sumrate_finalizer, sumrate_func_merge, sumrate_func_second_merge, - data_req_load_info, + dataBlockRequired, }, { // 31 @@ -4905,7 +4908,7 @@ SQLAggFuncElem aAggs[] = {{ sumrate_finalizer, sumrate_func_merge, sumrate_func_second_merge, - data_req_load_info, + dataBlockRequired, }, { // 32 @@ -4920,7 +4923,7 @@ SQLAggFuncElem aAggs[] = {{ sumrate_finalizer, sumrate_func_merge, sumrate_func_second_merge, - data_req_load_info, + dataBlockRequired, }, { // 33 @@ -4935,7 +4938,7 @@ SQLAggFuncElem aAggs[] = {{ sumrate_finalizer, sumrate_func_merge, sumrate_func_second_merge, - data_req_load_info, + dataBlockRequired, }, { // 34 @@ -4950,5 +4953,5 @@ SQLAggFuncElem aAggs[] = {{ noop1, noop1, noop1, - data_req_load_info, + dataBlockRequired, }}; diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index ca6dfcee9efb03058ad2401ed57f2dce6f581d59..3bcbe37b81055a609c88e06429002a96f2e4df1e 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -2471,7 +2471,7 @@ static bool functionCompatibleCheck(SQueryInfo* pQueryInfo) { startIdx++; } - int32_t factor = funcCompatDefList[tscSqlExprGet(pQueryInfo, startIdx)->functionId]; + int32_t factor = functionCompatList[tscSqlExprGet(pQueryInfo, startIdx)->functionId]; // diff function cannot be executed with other function // arithmetic function can be executed with other arithmetic functions @@ -2489,7 +2489,7 @@ static bool functionCompatibleCheck(SQueryInfo* pQueryInfo) { continue; } - if (funcCompatDefList[functionId] != factor) { + if (functionCompatList[functionId] != factor) { return false; } } diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 88fcc3828ec8756a4ac8926bb03fee8712103ee7..564d5ae23f75a22a36d6538e56ef5a0f4655f162 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -339,7 +339,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) { } if (rpcMsg->code != TSDB_CODE_TSC_ACTION_IN_PROGRESS) { - rpcMsg->code = (pRes->code == TSDB_CODE_SUCCESS) ? pRes->numOfRows: pRes->code; + rpcMsg->code = (pRes->code == TSDB_CODE_SUCCESS)? pRes->numOfRows: pRes->code; bool shouldFree = tscShouldBeFreed(pSql); (*pSql->fp)(pSql->param, pSql, rpcMsg->code); @@ -412,7 +412,7 @@ int tscProcessSql(SSqlObj *pSql) { return pSql->res.code; } } else if (pCmd->command < TSDB_SQL_LOCAL) { - pSql->ipList = tscMgmtIpSet; //? + pSql->ipList = tscMgmtIpSet; } else { // local handler return (*tscProcessMsgRsp[pCmd->command])(pSql); } @@ -476,6 +476,8 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) { int32_t vgIndex = pTableMetaInfo->vgroupIndex; SVgroupsInfo* pVgroupInfo = pTableMetaInfo->vgroupList; + assert(pVgroupInfo->vgroups[vgIndex].vgId > 0 && vgIndex < pTableMetaInfo->vgroupList->numOfVgroups); + pRetrieveMsg->header.vgId = htonl(pVgroupInfo->vgroups[vgIndex].vgId); } else { STableMeta* pTableMeta = pTableMetaInfo->pTableMeta; @@ -549,6 +551,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char assert(index >= 0); if (pTableMetaInfo->vgroupList->numOfVgroups > 0) { + assert(index < pTableMetaInfo->vgroupList->numOfVgroups); pVgroupInfo = &pTableMetaInfo->vgroupList->vgroups[index]; } tscDebug("%p query on stable, vgIndex:%d, numOfVgroups:%d", pSql, index, pTableMetaInfo->vgroupList->numOfVgroups); @@ -1372,7 +1375,6 @@ static int tscLocalResultCommonBuilder(SSqlObj *pSql, int32_t numOfRes) { SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); pRes->code = TSDB_CODE_SUCCESS; - if (pRes->rspType == 0) { pRes->numOfRows = numOfRes; pRes->row = 0; diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 73b93435ad1e8224707bdcf6f029d77ee4d51737..8048a2389b150801d2f34c7bccd9ca11e474fd19 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -481,25 +481,14 @@ static bool tscFreeQhandleInVnode(SSqlObj* pSql) { if (pRes->code == TSDB_CODE_SUCCESS && pRes->completed == false && !tscIsTwoStageSTableQuery(pQueryInfo, 0) && (pCmd->command == TSDB_SQL_SELECT || - pCmd->command == TSDB_SQL_SHOW || - pCmd->command == TSDB_SQL_RETRIEVE || - pCmd->command == TSDB_SQL_FETCH) && - (pCmd->command == TSDB_SQL_SELECT && pSql->pStream == NULL && pTableMetaInfo->pTableMeta != NULL)) { + pCmd->command == TSDB_SQL_SHOW || + pCmd->command == TSDB_SQL_RETRIEVE || + pCmd->command == TSDB_SQL_FETCH) && + (pSql->pStream == NULL && pTableMetaInfo->pTableMeta != NULL)) { pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH; - tscDebug("%p send msg to dnode to free qhandle ASAP, command:%s", pSql, sqlCmd[pCmd->command]); + tscDebug("%p send msg to dnode to free qhandle ASAP, command:%s, ", pSql, sqlCmd[pCmd->command]); tscProcessSql(pSql); - - // in case of sync model query, waits for response and then goes on -// if (pSql->fp == waitForQueryRsp || pSql->fp == waitForRetrieveRsp) { -// sem_wait(&pSql->rspSem); - -// tscFreeSqlObj(pSql); -// tscDebug("%p sqlObj is freed by app", pSql); -// } else { - tscDebug("%p sqlObj will be freed while rsp received", pSql); -// } - return true; } diff --git a/src/client/src/tscSubquery.c b/src/client/src/tscSubquery.c index 5e2ba46fbcca85f4e9e758ce0f655cc0b3accddf..dd4ed991ed9845b5fdabc3a0e07727bfcd1756ef 100644 --- a/src/client/src/tscSubquery.c +++ b/src/client/src/tscSubquery.c @@ -1895,9 +1895,11 @@ int32_t tscHandleInsertRetry(SSqlObj* pSql) { assert(pSupporter->index < pSupporter->pState->numOfTotal); STableDataBlocks* pTableDataBlock = taosArrayGetP(pCmd->pDataBlocks, pSupporter->index); - pRes->code = tscCopyDataBlockToPayload(pSql, pTableDataBlock); - if (pRes->code != TSDB_CODE_SUCCESS) { - return pRes->code; + int32_t code = tscCopyDataBlockToPayload(pSql, pTableDataBlock); + + if ((pRes->code = code)!= TSDB_CODE_SUCCESS) { + tscQueueAsyncRes(pSql); + return code; // here the pSql may have been released already. } return tscProcessSql(pSql); diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index eea5d60f64cedd0442d69ac96f21bc920bae2c90..3e0fe0b4bec2c392c48fcbec84508ec5b1d3303a 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -1648,6 +1648,7 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cm } pNew->fp = fp; + pNew->fetchFp = fp; pNew->param = param; pNew->maxRetry = TSDB_MAX_REPLICA; @@ -1803,6 +1804,8 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void } pNew->fp = fp; + pNew->fetchFp = fp; + pNew->param = param; pNew->maxRetry = TSDB_MAX_REPLICA; @@ -2005,7 +2008,7 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) { STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); int32_t totalVgroups = pTableMetaInfo->vgroupList->numOfVgroups; - while (++pTableMetaInfo->vgroupIndex < totalVgroups) { + if (++pTableMetaInfo->vgroupIndex < totalVgroups) { tscDebug("%p results from vgroup index:%d completed, try next:%d. total vgroups:%d. current numOfRes:%" PRId64, pSql, pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, pRes->numOfClauseTotal); @@ -2041,11 +2044,9 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) { // set the callback function pSql->fp = fp; - int32_t ret = tscProcessSql(pSql); - if (ret == TSDB_CODE_SUCCESS) { - return; - } else {// todo check for failure - } + tscProcessSql(pSql); + } else { + tscDebug("%p try all %d vnodes, query complete. current numOfRes:%" PRId64, pSql, totalVgroups, pRes->numOfClauseTotal); } } diff --git a/src/query/inc/qExecutor.h b/src/query/inc/qExecutor.h index 1faad6b141733b917893e02e10bfd7fa4eaff2d9..cdf5f9612cf192cc5ba445f232e408531cfc3afa 100644 --- a/src/query/inc/qExecutor.h +++ b/src/query/inc/qExecutor.h @@ -121,6 +121,7 @@ typedef struct SQueryCostInfo { uint32_t loadBlockStatis; uint32_t discardBlocks; uint64_t elapsedTime; + uint64_t ioTime; uint64_t computTime; } SQueryCostInfo; diff --git a/src/query/inc/tsqlfunction.h b/src/query/inc/tsqlfunction.h index 0c69bc02d39495cc6b77f00bd28524036cfef211..e57cb2645666af6887dc3253a76ce603314afbdf 100644 --- a/src/query/inc/tsqlfunction.h +++ b/src/query/inc/tsqlfunction.h @@ -125,7 +125,8 @@ typedef struct SArithmeticSupport { } SArithmeticSupport; typedef struct SQLPreAggVal { - bool isSet; + bool isSet; // statistics info set or not + bool dataBlockLoaded; // data block is loaded or not SDataStatis statis; } SQLPreAggVal; @@ -224,25 +225,14 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI #define IS_SINGLEOUTPUT(x) (((x)&TSDB_FUNCSTATE_SO) != 0) #define IS_OUTER_FORWARD(x) (((x)&TSDB_FUNCSTATE_OF) != 0) -/* - * the status of one block, used in metric query. all blocks are mixed together, - * we need the status to decide if one block is a first/end/inter block of one meter - */ -enum { - BLK_FILE_BLOCK = 0x1, - BLK_BLOCK_LOADED = 0x2, - BLK_CACHE_BLOCK = 0x4, // in case of cache block, block must be loaded -}; - /* determine the real data need to calculated the result */ enum { - BLK_DATA_NO_NEEDED = 0x0, + BLK_DATA_NO_NEEDED = 0x0, BLK_DATA_STATIS_NEEDED = 0x1, - BLK_DATA_ALL_NEEDED = 0x3, + BLK_DATA_ALL_NEEDED = 0x3, + BLK_DATA_DISCARD = 0x4, // discard current data block since it is not qualified for filter }; -#define SET_DATA_BLOCK_NOT_LOADED(x) ((x) &= (~BLK_BLOCK_LOADED)); - typedef struct STwaInfo { TSKEY lastKey; int8_t hasResult; // flag to denote has value @@ -264,12 +254,9 @@ typedef struct STwaInfo { /* global sql function array */ extern struct SQLAggFuncElem aAggs[]; -/* compatible check array list */ -extern int32_t funcCompatDefList[]; - -bool top_bot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, char *minval, char *maxval); +extern int32_t functionCompatList[]; // compatible check array list -bool stableQueryFunctChanged(int32_t funcId); +bool topbot_datablock_filter(SQLFunctionCtx *pCtx, int32_t functionId, const char *minval, const char *maxval); void resetResultInfo(SResultInfo *pResInfo); void setResultInfoBuf(SResultInfo *pResInfo, int32_t size, bool superTable, char* buf); diff --git a/src/query/src/qExecutor.c b/src/query/src/qExecutor.c index a12db2b206aec81d7d85608342f70af916508947..7f36288569dd6bb9bf39b7fab4773d76ccf12121 100644 --- a/src/query/src/qExecutor.c +++ b/src/query/src/qExecutor.c @@ -1358,6 +1358,8 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void* inputData, TSKEY pCtx->preAggVals.isSet = false; } + pCtx->preAggVals.dataBlockLoaded = (inputData != NULL); + // limit/offset query will affect this value pCtx->startOffset = QUERY_IS_ASC_QUERY(pQuery) ? pQuery->pos:0; pCtx->size = QUERY_IS_ASC_QUERY(pQuery) ? pBlockInfo->rows - pQuery->pos : pQuery->pos + 1; @@ -1928,73 +1930,46 @@ char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int32_t columnIndex, SWi pQuery->pSelectExpr[columnIndex].bytes * realRowId; } -/** - * decrease the refcount for each table involved in this query - * @param pQInfo - */ -UNUSED_FUNC void vnodeDecMeterRefcnt(SQInfo *pQInfo) { - if (pQInfo != NULL) { - // assert(taosHashGetSize(pQInfo->tableqinfoGroupInfo) >= 1); - } - -#if 0 - if (pQInfo == NULL || pQInfo->tableqinfoGroupInfo.numOfTables == 1) { - atomic_fetch_sub_32(&pQInfo->pObj->numOfQueries, 1); - qDebug("QInfo:%p vid:%d sid:%d meterId:%s, query is over, numOfQueries:%d", pQInfo, pQInfo->pObj->vnode, - pQInfo->pObj->sid, pQInfo->pObj->meterId, pQInfo->pObj->numOfQueries); - } else { - int32_t num = 0; - for (int32_t i = 0; i < pQInfo->tableqinfoGroupInfo.numOfTables; ++i) { - SMeterObj *pMeter = getMeterObj(pQInfo->tableqinfoGroupInfo, pQInfo->pSidSet->pTableIdList[i]->sid); - atomic_fetch_sub_32(&(pMeter->numOfQueries), 1); - - if (pMeter->numOfQueries > 0) { - qDebug("QInfo:%p vid:%d sid:%d meterId:%s, query is over, numOfQueries:%d", pQInfo, pMeter->vnode, pMeter->sid, - pMeter->meterId, pMeter->numOfQueries); - num++; - } - } - - /* - * in order to reduce log output, for all meters of which numOfQueries count are 0, - * we do not output corresponding information - */ - num = pQInfo->tableqinfoGroupInfo.numOfTables - num; - qDebug("QInfo:%p metric query is over, dec query ref for %d meters, numOfQueries on %d meters are 0", pQInfo, - pQInfo->tableqinfoGroupInfo.numOfTables, num); - } -#endif -} +#define IS_PREFILTER_TYPE(_t) ((_t) != TSDB_DATA_TYPE_BINARY && (_t) != TSDB_DATA_TYPE_NCHAR) -static bool needToLoadDataBlock(SQuery *pQuery, SDataStatis *pDataStatis, SQLFunctionCtx *pCtx, - int32_t numOfTotalPoints) { - if (pDataStatis == NULL) { +static bool needToLoadDataBlock(SQueryRuntimeEnv* pRuntimeEnv, SDataStatis *pDataStatis, SQLFunctionCtx *pCtx, + int32_t numOfRows) { + SQuery* pQuery = pRuntimeEnv->pQuery; + if (pDataStatis == NULL || (pQuery->numOfFilterCols == 0 && (!pRuntimeEnv->topBotQuery))) { return true; } -#if 0 for (int32_t k = 0; k < pQuery->numOfFilterCols; ++k) { SSingleColumnFilterInfo *pFilterInfo = &pQuery->pFilterInfo[k]; - int32_t colIndex = pFilterInfo->info.colIndex; - // this column not valid in current data block - if (colIndex < 0 || pDataStatis[colIndex].colId != pFilterInfo->info.data.colId) { - continue; + int32_t index = -1; + for(int32_t i = 0; i < pQuery->numOfCols; ++i) { + if (pDataStatis[i].colId == pFilterInfo->info.colId) { + index = i; + break; + } + } + + // no statistics data + if (index == -1) { + return true; } // not support pre-filter operation on binary/nchar data type - if (!vnodeSupportPrefilter(pFilterInfo->info.data.type)) { - continue; + if (!IS_PREFILTER_TYPE(pFilterInfo->info.type)) { + return true; } // all points in current column are NULL, no need to check its boundary value - if (pDataStatis[colIndex].numOfNull == numOfTotalPoints) { + if (pDataStatis[index].numOfNull == numOfRows) { continue; } - if (pFilterInfo->info.info.type == TSDB_DATA_TYPE_FLOAT) { - float minval = *(double *)(&pDataStatis[colIndex].min); - float maxval = *(double *)(&pDataStatis[colIndex].max); + SDataStatis* pDataBlockst = &pDataStatis[index]; + + if (pFilterInfo->info.type == TSDB_DATA_TYPE_FLOAT) { + float minval = *(double *)(&pDataBlockst->min); + float maxval = *(double *)(&pDataBlockst->max); for (int32_t i = 0; i < pFilterInfo->numOfFilters; ++i) { if (pFilterInfo->pFilters[i].fp(&pFilterInfo->pFilters[i], (char *)&minval, (char *)&maxval)) { @@ -2003,53 +1978,50 @@ static bool needToLoadDataBlock(SQuery *pQuery, SDataStatis *pDataStatis, SQLFun } } else { for (int32_t i = 0; i < pFilterInfo->numOfFilters; ++i) { - if (pFilterInfo->pFilters[i].fp(&pFilterInfo->pFilters[i], (char *)&pDataStatis[colIndex].min, - (char *)&pDataStatis[colIndex].max)) { + if (pFilterInfo->pFilters[i].fp(&pFilterInfo->pFilters[i], (char *)&pDataBlockst->min, (char *)&pDataBlockst->max)) { return true; } } } } - // todo disable this opt code block temporarily - // for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { - // int32_t functId = pQuery->pSelectExpr[i].base.functionId; - // if (functId == TSDB_FUNC_TOP || functId == TSDB_FUNC_BOTTOM) { - // return top_bot_datablock_filter(&pCtx[i], functId, (char *)&pField[i].min, (char *)&pField[i].max); - // } - // } + if (pRuntimeEnv->topBotQuery) { + for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { + int32_t functionId = pQuery->pSelectExpr[i].base.functionId; + if (functionId == TSDB_FUNC_TOP || functionId == TSDB_FUNC_BOTTOM) { + return topbot_datablock_filter(&pCtx[i], functionId, (char *)&pDataStatis[i].min, (char *)&pDataStatis[i].max); + } + } + } -#endif - return true; + return false; } -SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle, SDataBlockInfo* pBlockInfo, SDataStatis **pStatis) { +int32_t loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle, SDataBlockInfo* pBlockInfo, SDataStatis **pStatis, SArray** pDataBlock) { SQuery *pQuery = pRuntimeEnv->pQuery; - uint32_t r = 0; - SArray * pDataBlock = NULL; - + uint32_t status = 0; if (pQuery->numOfFilterCols > 0) { - r = BLK_DATA_ALL_NEEDED; - } else { - // check if this data block is required to load + status = BLK_DATA_ALL_NEEDED; + } else { // check if this data block is required to load for (int32_t i = 0; i < pQuery->numOfOutput; ++i) { SSqlFuncMsg* pSqlFunc = &pQuery->pSelectExpr[i].base; int32_t functionId = pSqlFunc->functionId; int32_t colId = pSqlFunc->colInfo.colId; - r |= aAggs[functionId].dataReqFunc(&pRuntimeEnv->pCtx[i], pQuery->window.skey, pQuery->window.ekey, colId); + status |= aAggs[functionId].dataReqFunc(&pRuntimeEnv->pCtx[i], pBlockInfo->window.skey, pBlockInfo->window.ekey, colId); } if (pRuntimeEnv->pTSBuf > 0 || QUERY_IS_INTERVAL_QUERY(pQuery)) { - r |= BLK_DATA_ALL_NEEDED; + status |= BLK_DATA_ALL_NEEDED; } } - if (r == BLK_DATA_NO_NEEDED) { - qDebug("QInfo:%p data block discard, rows:%d", GET_QINFO_ADDR(pRuntimeEnv), pBlockInfo->rows); + if (status == BLK_DATA_NO_NEEDED) { + qDebug("QInfo:%p data block discard, brange:%"PRId64 "-%"PRId64", rows:%d", GET_QINFO_ADDR(pRuntimeEnv), + pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); pRuntimeEnv->summary.discardBlocks += 1; - } else if (r == BLK_DATA_STATIS_NEEDED) { + } else if (status == BLK_DATA_STATIS_NEEDED) { if (tsdbRetrieveDataBlockStatisInfo(pQueryHandle, pStatis) != TSDB_CODE_SUCCESS) { // return DISK_DATA_LOAD_FAILED; } @@ -2057,32 +2029,34 @@ SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle, pRuntimeEnv->summary.loadBlockStatis += 1; if (*pStatis == NULL) { // data block statistics does not exist, load data block - pDataBlock = tsdbRetrieveDataBlock(pQueryHandle, NULL); + *pDataBlock = tsdbRetrieveDataBlock(pQueryHandle, NULL); pRuntimeEnv->summary.totalCheckedRows += pBlockInfo->rows; } } else { - assert(r == BLK_DATA_ALL_NEEDED); + assert(status == BLK_DATA_ALL_NEEDED); // load the data block statistics to perform further filter - pRuntimeEnv->summary.loadBlockStatis +=1; + pRuntimeEnv->summary.loadBlockStatis += 1; if (tsdbRetrieveDataBlockStatisInfo(pQueryHandle, pStatis) != TSDB_CODE_SUCCESS) { } - if (!needToLoadDataBlock(pQuery,*pStatis, pRuntimeEnv->pCtx, pBlockInfo->rows)) { + if (!needToLoadDataBlock(pRuntimeEnv, *pStatis, pRuntimeEnv->pCtx, pBlockInfo->rows)) { #if defined(_DEBUG_VIEW) qDebug("QInfo:%p block discarded by per-filter", GET_QINFO_ADDR(pRuntimeEnv)); #endif // current block has been discard due to filter applied pRuntimeEnv->summary.discardBlocks += 1; - // return DISK_DATA_DISCARDED; + qDebug("QInfo:%p data block discard, brange:%"PRId64 "-%"PRId64", rows:%d", GET_QINFO_ADDR(pRuntimeEnv), + pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); + return BLK_DATA_DISCARD; } pRuntimeEnv->summary.totalCheckedRows += pBlockInfo->rows; pRuntimeEnv->summary.loadBlocks += 1; - pDataBlock = tsdbRetrieveDataBlock(pQueryHandle, NULL); + *pDataBlock = tsdbRetrieveDataBlock(pQueryHandle, NULL); } - return pDataBlock; + return TSDB_CODE_SUCCESS; } int32_t binarySearchForKey(char *pValue, int num, TSKEY key, int order) { @@ -2225,13 +2199,13 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) { pQuery->order.order); TsdbQueryHandleT pQueryHandle = IS_MASTER_SCAN(pRuntimeEnv)? pRuntimeEnv->pQueryHandle : pRuntimeEnv->pSecQueryHandle; + int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); SDataBlockInfo blockInfo = SDATA_BLOCK_INITIALIZER; while (tsdbNextDataBlock(pQueryHandle)) { summary->totalBlocks += 1; if (IS_QUERY_KILLED(GET_QINFO_ADDR(pRuntimeEnv))) { - finalizeQueryResult(pRuntimeEnv); // clean up allocated resource during query longjmp(pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED); } @@ -2259,7 +2233,11 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) { ensureOutputBuffer(pRuntimeEnv, &blockInfo); SDataStatis *pStatis = NULL; - SArray *pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis); + SArray *pDataBlock = NULL; + if (loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis, &pDataBlock) == BLK_DATA_DISCARD) { + pQuery->current->lastKey = QUERY_IS_ASC_QUERY(pQuery)? blockInfo.window.ekey + step:blockInfo.window.skey + step; + continue; + } // query start position can not move into tableApplyFunctionsOnBlock due to limit/offset condition pQuery->pos = QUERY_IS_ASC_QUERY(pQuery)? 0 : blockInfo.rows - 1; @@ -2282,8 +2260,6 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) { if (QUERY_IS_INTERVAL_QUERY(pQuery) && IS_MASTER_SCAN(pRuntimeEnv)) { if (Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED)) { -// int32_t step = QUERY_IS_ASC_QUERY(pQuery) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP; - closeAllTimeWindow(&pRuntimeEnv->windowResInfo); // removeRedundantWindow(&pRuntimeEnv->windowResInfo, pTableQueryInfo->lastKey - step, step); pRuntimeEnv->windowResInfo.curIndex = pRuntimeEnv->windowResInfo.size - 1; // point to the last time window @@ -3393,12 +3369,10 @@ void destroyTableQueryInfo(STableQueryInfo *pTableQueryInfo, int32_t numOfCols) cleanupTimeWindowInfo(&pTableQueryInfo->windowResInfo, numOfCols); } -#define SET_CURRENT_QUERY_TABLE_INFO(_runtime, _tableInfo) \ - do { \ - SQuery *_query = (_runtime)->pQuery; \ - _query->current = _tableInfo; \ - assert((((_tableInfo)->lastKey >= (_tableInfo)->win.skey) && QUERY_IS_ASC_QUERY(_query)) || \ - (((_tableInfo)->lastKey <= (_tableInfo)->win.skey) && !QUERY_IS_ASC_QUERY(_query))); \ +#define CHECK_QUERY_TIME_RANGE(_q, _tableInfo) \ + do { \ + assert((((_tableInfo)->lastKey >= (_tableInfo)->win.skey) && QUERY_IS_ASC_QUERY(_q)) || \ + (((_tableInfo)->lastKey <= (_tableInfo)->win.skey) && !QUERY_IS_ASC_QUERY(_q))); \ } while (0) /** @@ -3710,7 +3684,7 @@ static void updateWindowResNumOfRes(SQueryRuntimeEnv *pRuntimeEnv) { } } -void stableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pDataBlockInfo, SDataStatis *pStatis, +static void stableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pDataBlockInfo, SDataStatis *pStatis, SArray *pDataBlock, __block_search_fn_t searchFn) { SQuery * pQuery = pRuntimeEnv->pQuery; STableQueryInfo* pTableQueryInfo = pQuery->current; @@ -3869,9 +3843,10 @@ static void queryCostStatis(SQInfo *pQInfo) { // pQInfo, pSummary->readDiskBlocks, pSummary->totalBlockSize, pSummary->loadBlocksUs / 1000.0, // pSummary->skippedFileBlocks, pSummary->totalGenData); - qDebug("QInfo:%p :cost summary: elpased time:%"PRId64" us, total blocks:%d, use block statis:%d, use block data:%d, " - "total rows:%"PRId64 ", check rows:%"PRId64, pQInfo, pSummary->elapsedTime, pSummary->totalBlocks, - pSummary->loadBlockStatis, pSummary->loadBlocks, pSummary->totalRows, pSummary->totalCheckedRows); + qDebug("QInfo:%p :cost summary: elapsed time:%"PRId64" us, io time:%"PRId64" us, total blocks:%d, load block statis:%d," + " load data block:%d, total rows:%"PRId64 ", check rows:%"PRId64, + pQInfo, pSummary->elapsedTime, pSummary->ioTime, pSummary->totalBlocks, pSummary->loadBlockStatis, + pSummary->loadBlocks, pSummary->totalRows, pSummary->totalCheckedRows); // qDebug("QInfo:%p cost: temp file:%d Bytes", pQInfo, pSummary->tmpBufferInDisk); // @@ -4253,6 +4228,23 @@ static void enableExecutionForNextTable(SQueryRuntimeEnv *pRuntimeEnv) { } } +static FORCE_INLINE void setEnvForEachBlock(SQInfo* pQInfo, STableQueryInfo* pTableQueryInfo, SDataBlockInfo* pBlockInfo) { + SQueryRuntimeEnv* pRuntimeEnv = &pQInfo->runtimeEnv; + SQuery* pQuery = pQInfo->runtimeEnv.pQuery; + int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); + + if (!QUERY_IS_INTERVAL_QUERY(pQuery)) { + setExecutionContext(pQInfo, pTableQueryInfo->groupIndex, pBlockInfo->window.ekey + step); + } else { // interval query + TSKEY nextKey = pBlockInfo->window.skey; + setIntervalQueryRange(pQInfo, nextKey); + + if (pRuntimeEnv->hasTagResults || pRuntimeEnv->pTSBuf != NULL) { + setAdditionalInfo(pQInfo, pTableQueryInfo->pTable, pTableQueryInfo); + } + } +} + static int64_t scanMultiTableDataBlocks(SQInfo *pQInfo) { SQueryRuntimeEnv *pRuntimeEnv = &pQInfo->runtimeEnv; SQuery* pQuery = pRuntimeEnv->pQuery; @@ -4263,10 +4255,12 @@ static int64_t scanMultiTableDataBlocks(SQInfo *pQInfo) { TsdbQueryHandleT pQueryHandle = IS_MASTER_SCAN(pRuntimeEnv)? pRuntimeEnv->pQueryHandle : pRuntimeEnv->pSecQueryHandle; SDataBlockInfo blockInfo = SDATA_BLOCK_INITIALIZER; + int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); + while (tsdbNextDataBlock(pQueryHandle)) { summary->totalBlocks += 1; + if (IS_QUERY_KILLED(pQInfo)) { - finalizeQueryResult(pRuntimeEnv); // clean up allocated resource during query longjmp(pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED); } @@ -4276,24 +4270,19 @@ static int64_t scanMultiTableDataBlocks(SQInfo *pQInfo) { break; } - assert(*pTableQueryInfo != NULL); - SET_CURRENT_QUERY_TABLE_INFO(pRuntimeEnv, *pTableQueryInfo); + pQuery->current = *pTableQueryInfo; + CHECK_QUERY_TIME_RANGE(pQuery, *pTableQueryInfo); + + if (!pRuntimeEnv->groupbyNormalCol) { + setEnvForEachBlock(pQInfo, *pTableQueryInfo, &blockInfo); + } SDataStatis *pStatis = NULL; - SArray *pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis); + SArray *pDataBlock = NULL; - if (!pRuntimeEnv->groupbyNormalCol) { - if (!QUERY_IS_INTERVAL_QUERY(pQuery)) { - int32_t step = QUERY_IS_ASC_QUERY(pQuery)? 1:-1; - setExecutionContext(pQInfo, (*pTableQueryInfo)->groupIndex, blockInfo.window.ekey + step); - } else { // interval query - TSKEY nextKey = blockInfo.window.skey; - setIntervalQueryRange(pQInfo, nextKey); - - if (pRuntimeEnv->hasTagResults || pRuntimeEnv->pTSBuf != NULL) { - setAdditionalInfo(pQInfo, (*pTableQueryInfo)->pTable, *pTableQueryInfo); - } - } + if (loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis, &pDataBlock) == BLK_DATA_DISCARD) { + pQuery->current->lastKey = QUERY_IS_ASC_QUERY(pQuery)? blockInfo.window.ekey + step:blockInfo.window.skey + step; + continue; } summary->totalRows += blockInfo.rows; @@ -4553,7 +4542,6 @@ static void sequentialTableProcess(SQInfo *pQInfo) { while (pQInfo->tableIndex < pQInfo->tableqinfoGroupInfo.numOfTables) { if (IS_QUERY_KILLED(pQInfo)) { - finalizeQueryResult(pRuntimeEnv); // clean up allocated resource during query longjmp(pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED); } @@ -5051,6 +5039,7 @@ static void stableQueryImpl(SQInfo *pQInfo) { isFirstLastRowQuery(pQuery) || pRuntimeEnv->groupbyNormalCol); sequentialTableProcess(pQInfo); + } // record the total elapsed time @@ -6176,11 +6165,6 @@ _over: //pQInfo already freed in initQInfo, but *pQInfo may not pointer to null; if (code != TSDB_CODE_SUCCESS) { *pQInfo = NULL; - } else { -// SQInfo* pq = (SQInfo*) (*pQInfo); - -// T_REF_INC(pq); -// T_REF_INC(pq); } // if failed to add ref for all meters in this query, abort current query diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c index 9e5da903b1402fa048e40a07666c19530e73c105..2fdaf387a65302077887c7800e7fb114705a4edb 100644 --- a/src/tsdb/src/tsdbRead.c +++ b/src/tsdb/src/tsdbRead.c @@ -1801,7 +1801,8 @@ int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT* pQueryHandle, SDataSta } tsdbLoadCompData(&pHandle->rhelper, pBlockInfo->compBlock, NULL); - + + // todo opt perf size_t numOfCols = QH_GET_NUM_OF_COLS(pHandle); for(int32_t i = 0; i < numOfCols; ++i) { SDataStatis* st = &pHandle->statis[i]; @@ -1820,6 +1821,13 @@ int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT* pQueryHandle, SDataSta if (pHandle->statis[i].numOfNull == -1) { // set the column data are all NULL pHandle->statis[i].numOfNull = pBlockInfo->compBlock->numOfRows; } + + // todo opt perf + SColumnInfo* pColInfo = taosArrayGet(pHandle->pColumns, i); + if (pColInfo->type == TSDB_DATA_TYPE_TIMESTAMP) { + pHandle->statis[i].min = pBlockInfo->compBlock->keyFirst; + pHandle->statis[i].max = pBlockInfo->compBlock->keyLast; + } } return TSDB_CODE_SUCCESS; diff --git a/src/util/src/tcache.c b/src/util/src/tcache.c index d3c622633de7cdebfe43a7492f7a882823dec0e2..df63d567c795c69b031c6ab97112a1b51d0c2e94 100644 --- a/src/util/src/tcache.c +++ b/src/util/src/tcache.c @@ -413,57 +413,90 @@ void taosCacheRelease(SCacheObj *pCacheObj, void **data, bool _remove) { *data = NULL; // note: extend lifespan before dec ref count - if (pCacheObj->extendLifespan) { + bool inTrashCan = pNode->inTrashCan; + + if (pCacheObj->extendLifespan && (!inTrashCan)) { atomic_store_64(&pNode->expireTime, pNode->lifespan + taosGetTimestampMs()); uDebug("cache:%s data:%p extend life time to %"PRId64 " before release", pCacheObj->name, pNode->data, pNode->expireTime); } - bool inTrashCan = pNode->inTrashCan; - uDebug("cache:%s, key:%p, %p is released, refcnt:%d", pCacheObj->name, pNode->key, pNode->data, T_REF_VAL_GET(pNode) - 1); - - // NOTE: once refcount is decrease, pNode may be free by other thread immediately. - int32_t ref = T_REF_DEC(pNode); - - if (inTrashCan) { - // Remove it if the ref count is 0. - // The ref count does not need to load and check again after lock acquired, since ref count can not be increased when - // the node is in trashcan. - if (ref == 0) { - __cache_wr_lock(pCacheObj); - assert(pNode->pTNodeHeader->pData == pNode); - taosRemoveFromTrashCan(pCacheObj, pNode->pTNodeHeader); - __cache_unlock(pCacheObj); + if (_remove) { + __cache_wr_lock(pCacheObj); + + // NOTE: once refcount is decrease, pNode may be freed by other thread immediately. + int32_t ref = T_REF_DEC(pNode); + uDebug("cache:%s, key:%p, %p is released, refcnt:%d", pCacheObj->name, pNode->key, pNode->data, ref); + + /* + * If it is not referenced by other users, remove it immediately. Otherwise move this node to trashcan wait for all users + * releasing this resources. + * + * NOTE: previous ref is 0, and current ref is still 0, remove it. If previous is not 0, there is another thread + * that tries to do the same thing. + */ + if (pNode->inTrashCan) { + if (ref == 0) { + assert(pNode->pTNodeHeader->pData == pNode); + taosRemoveFromTrashCan(pCacheObj, pNode->pTNodeHeader); + } + } else { + if (ref > 0) { + assert(pNode->pTNodeHeader == NULL); + taosCacheMoveToTrash(pCacheObj, pNode); + } else { + taosCacheReleaseNode(pCacheObj, pNode); + } } + __cache_unlock(pCacheObj); + } else { - assert(pNode->pTNodeHeader == NULL); - - if (_remove) { // not in trash can, but need to remove it - __cache_wr_lock(pCacheObj); - - /* - * If not referenced by other users. Otherwise move this node to trashcan wait for all users - * releasing this resources. - * - * NOTE: previous ref is 0, and current ref is still 0, remove it. If previous is not 0, there is another thread - * that tries to do the same thing. - */ + uDebug("cache:%s, key:%p, %p is released, refcnt:%d", pCacheObj->name, pNode->key, pNode->data, T_REF_VAL_GET(pNode) - 1); + + // NOTE: once refcount is decrease, pNode may be freed by other thread immediately. + int32_t ref = T_REF_DEC(pNode); + + if (inTrashCan) { + // Remove it if the ref count is 0. + // The ref count does not need to load and check again after lock acquired, since ref count can not be increased when + // the node is in trashcan. if (ref == 0) { - if (T_REF_VAL_GET(pNode) == 0) { - taosCacheReleaseNode(pCacheObj, pNode); - } else { - taosCacheMoveToTrash(pCacheObj, pNode); - } + __cache_wr_lock(pCacheObj); + assert(pNode->pTNodeHeader->pData == pNode); + taosRemoveFromTrashCan(pCacheObj, pNode->pTNodeHeader); + __cache_unlock(pCacheObj); } - __cache_unlock(pCacheObj); -// } else { // extend its life time -// if (pCacheObj->extendLifespan) { -// atomic_store_64(&pNode->expireTime, pNode->lifespan + taosGetTimestampMs()); -// uDebug("cache:%s data:%p extend life time to %"PRId64 " after release", pCacheObj->name, pNode->data, pNode->expireTime); -// } } } + +// else { +// if (_remove) { // not in trash can, but need to remove it +// __cache_wr_lock(pCacheObj); +// +// /* +// * If not referenced by other users. Otherwise move this node to trashcan wait for all users +// * releasing this resources. +// * +// * NOTE: previous ref is 0, and current ref is still 0, remove it. If previous is not 0, there is another thread +// * that tries to do the same thing. +// */ +// if (ref == 0) { +// if (T_REF_VAL_GET(pNode) == 0) { +// taosCacheReleaseNode(pCacheObj, pNode); +// } else { +// taosCacheMoveToTrash(pCacheObj, pNode); +// } +// } else if (ref > 0) { +// if (!pNode->inTrashCan) { +// assert(pNode->pTNodeHeader == NULL); +// taosCacheMoveToTrash(pCacheObj, pNode); +// } +// } +// +// __cache_unlock(pCacheObj); +// } +// } } void taosCacheEmpty(SCacheObj *pCacheObj) { diff --git a/src/vnode/src/vnodeRead.c b/src/vnode/src/vnodeRead.c index 4c7970076c4197b573d7dcaa7d6afe2e73f1a51f..987783711b2452b17c56568fe4c63ed506c51e30 100644 --- a/src/vnode/src/vnodeRead.c +++ b/src/vnode/src/vnodeRead.c @@ -110,6 +110,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { if (handle == NULL) { // failed to register qhandle pRsp->code = TSDB_CODE_QRY_INVALID_QHANDLE; qDestroyQueryInfo(pQInfo); // destroy it directly + vError("vgId:%d QInfo:%p register qhandle failed, return to app, code:%s", pVnode->vgId, (void*) pQInfo, tstrerror(pRsp->code)); } else { assert(*handle == pQInfo); pRsp->qhandle = htobe64((uint64_t) pQInfo); @@ -125,12 +126,14 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { } else { assert(pQInfo == NULL); } + if (handle != NULL) { + vDebug("vgId:%d, QInfo:%p, dnode query msg disposed, register qhandle and return to app", vgId, *handle); + dnodePutItemIntoReadQueue(pVnode, *handle); qReleaseQInfo(pVnode->qMgmt, (void**) &handle, false); } - vDebug("vgId:%d, QInfo:%p, dnode query msg disposed", vgId, pQInfo); } else { assert(pCont != NULL); handle = qAcquireQInfo(pVnode->qMgmt, (uint64_t) pCont); @@ -138,12 +141,13 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) { vWarn("QInfo:%p invalid qhandle in continuing exec query, conn:%p", (void*) pCont, pReadMsg->rpcMsg.handle); code = TSDB_CODE_QRY_INVALID_QHANDLE; } else { - vDebug("vgId:%d, QInfo:%p, dnode query msg in progress", pVnode->vgId, (void*) pCont); + vDebug("vgId:%d, QInfo:%p, dnode continue exec query", pVnode->vgId, (void*) pCont); code = TSDB_CODE_VND_ACTION_IN_PROGRESS; qTableQuery(*handle); // do execute query } qReleaseQInfo(pVnode->qMgmt, (void**) &handle, false); } + return code; }