[td-225] fix bugs in union query.
This commit is contained in:
parent
66dbdcdd90
commit
98e99fd45b
|
@ -263,9 +263,11 @@ int16_t tscGetJoinTagColIdByUid(STagCond* pTagCond, uint64_t uid);
|
||||||
void tscPrintSelectClause(SSqlObj* pSql, int32_t subClauseIndex);
|
void tscPrintSelectClause(SSqlObj* pSql, int32_t subClauseIndex);
|
||||||
|
|
||||||
bool hasMoreVnodesToTry(SSqlObj *pSql);
|
bool hasMoreVnodesToTry(SSqlObj *pSql);
|
||||||
|
bool hasMoreClauseToTry(SSqlObj* pSql);
|
||||||
|
|
||||||
void tscTryQueryNextVnode(SSqlObj *pSql, __async_cb_func_t fp);
|
void tscTryQueryNextVnode(SSqlObj *pSql, __async_cb_func_t fp);
|
||||||
void tscAsyncQuerySingleRowForNextVnode(void *param, TAOS_RES *tres, int numOfRows);
|
void tscAsyncQuerySingleRowForNextVnode(void *param, TAOS_RES *tres, int numOfRows);
|
||||||
void tscTryQueryNextClause(SSqlObj* pSql, void (*queryFp)());
|
void tscTryQueryNextClause(SSqlObj* pSql, __async_cb_func_t fp);
|
||||||
int tscSetMgmtIpListFromCfg(const char *first, const char *second);
|
int tscSetMgmtIpListFromCfg(const char *first, const char *second);
|
||||||
|
|
||||||
void* malloc_throw(size_t size);
|
void* malloc_throw(size_t size);
|
||||||
|
|
|
@ -169,7 +169,11 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
|
||||||
pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
|
pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
|
||||||
}
|
}
|
||||||
|
|
||||||
tscProcessSql(pSql);
|
if (pCmd->command == TSDB_SQL_TABLE_JOIN_RETRIEVE) {
|
||||||
|
tscFetchDatablockFromSubquery(pSql);
|
||||||
|
} else {
|
||||||
|
tscProcessSql(pSql);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
|
|
|
@ -485,7 +485,6 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
|
||||||
}
|
}
|
||||||
|
|
||||||
case TSDB_SQL_SELECT: {
|
case TSDB_SQL_SELECT: {
|
||||||
assert(pCmd->numOfClause == 1);
|
|
||||||
const char* msg1 = "columns in select clause not identical";
|
const char* msg1 = "columns in select clause not identical";
|
||||||
|
|
||||||
for (int32_t i = pCmd->numOfClause; i < pInfo->subclauseInfo.numOfClause; ++i) {
|
for (int32_t i = pCmd->numOfClause; i < pInfo->subclauseInfo.numOfClause; ++i) {
|
||||||
|
@ -496,16 +495,19 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
|
||||||
}
|
}
|
||||||
|
|
||||||
assert(pCmd->numOfClause == pInfo->subclauseInfo.numOfClause);
|
assert(pCmd->numOfClause == pInfo->subclauseInfo.numOfClause);
|
||||||
for (int32_t i = 0; i < pInfo->subclauseInfo.numOfClause; ++i) {
|
for (int32_t i = pCmd->clauseIndex; i < pInfo->subclauseInfo.numOfClause; ++i) {
|
||||||
SQuerySQL* pQuerySql = pInfo->subclauseInfo.pClause[i];
|
SQuerySQL* pQuerySql = pInfo->subclauseInfo.pClause[i];
|
||||||
|
tscTrace("%p start to parse %dth subclause, total:%d", pSql, i, pInfo->subclauseInfo.numOfClause);
|
||||||
if ((code = doCheckForQuery(pSql, pQuerySql, i)) != TSDB_CODE_SUCCESS) {
|
if ((code = doCheckForQuery(pSql, pQuerySql, i)) != TSDB_CODE_SUCCESS) {
|
||||||
return code;
|
return code;
|
||||||
}
|
}
|
||||||
|
|
||||||
tscPrintSelectClause(pSql, i);
|
tscPrintSelectClause(pSql, i);
|
||||||
|
pCmd->clauseIndex += 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// restore the clause index
|
||||||
|
pCmd->clauseIndex = 0;
|
||||||
// set the command/global limit parameters from the first subclause to the sqlcmd object
|
// set the command/global limit parameters from the first subclause to the sqlcmd object
|
||||||
SQueryInfo* pQueryInfo1 = tscGetQueryInfoDetail(pCmd, 0);
|
SQueryInfo* pQueryInfo1 = tscGetQueryInfoDetail(pCmd, 0);
|
||||||
pCmd->command = pQueryInfo1->command;
|
pCmd->command = pQueryInfo1->command;
|
||||||
|
@ -5867,6 +5869,8 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) {
|
||||||
pTableMetaInfo = tscAddEmptyMetaInfo(pQueryInfo);
|
pTableMetaInfo = tscAddEmptyMetaInfo(pQueryInfo);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
assert(pCmd->clauseIndex == index);
|
||||||
|
|
||||||
// too many result columns not support order by in query
|
// too many result columns not support order by in query
|
||||||
if (pQuerySql->pSelection->nExpr > TSDB_MAX_COLUMNS) {
|
if (pQuerySql->pSelection->nExpr > TSDB_MAX_COLUMNS) {
|
||||||
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg8);
|
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg8);
|
||||||
|
@ -5980,12 +5984,11 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) {
|
||||||
pQueryInfo->window.ekey = pQueryInfo->window.ekey / 1000;
|
pQueryInfo->window.ekey = pQueryInfo->window.ekey / 1000;
|
||||||
}
|
}
|
||||||
} else { // set the time rang
|
} else { // set the time rang
|
||||||
pQueryInfo->window.skey = TSKEY_INITIAL_VAL;
|
pQueryInfo->window = TSWINDOW_INITIALIZER;
|
||||||
pQueryInfo->window.ekey = INT64_MAX;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// user does not specified the query time window, twa is not allowed in such case.
|
// user does not specified the query time window, twa is not allowed in such case.
|
||||||
if ((pQueryInfo->window.skey == 0 || pQueryInfo->window.ekey == INT64_MAX ||
|
if ((pQueryInfo->window.skey == INT64_MIN || pQueryInfo->window.ekey == INT64_MAX ||
|
||||||
(pQueryInfo->window.ekey == INT64_MAX / 1000 && tinfo.precision == TSDB_TIME_PRECISION_MILLI)) && tscIsTWAQuery(pQueryInfo)) {
|
(pQueryInfo->window.ekey == INT64_MAX / 1000 && tinfo.precision == TSDB_TIME_PRECISION_MILLI)) && tscIsTWAQuery(pQueryInfo)) {
|
||||||
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg9);
|
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg9);
|
||||||
}
|
}
|
||||||
|
|
|
@ -475,7 +475,7 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
|
||||||
SRetrieveTableMsg *pRetrieveMsg = (SRetrieveTableMsg *) pSql->cmd.payload;
|
SRetrieveTableMsg *pRetrieveMsg = (SRetrieveTableMsg *) pSql->cmd.payload;
|
||||||
pRetrieveMsg->qhandle = htobe64(pSql->res.qhandle);
|
pRetrieveMsg->qhandle = htobe64(pSql->res.qhandle);
|
||||||
|
|
||||||
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
|
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, pSql->cmd.clauseIndex);
|
||||||
pRetrieveMsg->free = htons(pQueryInfo->type);
|
pRetrieveMsg->free = htons(pQueryInfo->type);
|
||||||
|
|
||||||
// todo valid the vgroupId at the client side
|
// todo valid the vgroupId at the client side
|
||||||
|
|
|
@ -424,7 +424,7 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
|
||||||
}
|
}
|
||||||
|
|
||||||
// current data set are exhausted, fetch more data from node
|
// current data set are exhausted, fetch more data from node
|
||||||
if (pRes->row >= pRes->numOfRows && (pRes->completed != true || hasMoreVnodesToTry(pSql)) &&
|
if (pRes->row >= pRes->numOfRows && (pRes->completed != true || hasMoreVnodesToTry(pSql) || hasMoreClauseToTry(pSql)) &&
|
||||||
(pCmd->command == TSDB_SQL_RETRIEVE ||
|
(pCmd->command == TSDB_SQL_RETRIEVE ||
|
||||||
pCmd->command == TSDB_SQL_RETRIEVE_LOCALMERGE ||
|
pCmd->command == TSDB_SQL_RETRIEVE_LOCALMERGE ||
|
||||||
pCmd->command == TSDB_SQL_TABLE_JOIN_RETRIEVE ||
|
pCmd->command == TSDB_SQL_TABLE_JOIN_RETRIEVE ||
|
||||||
|
|
|
@ -848,13 +848,14 @@ static void joinRetrieveFinalResCallback(void* param, TAOS_RES* tres, int numOfR
|
||||||
SSqlRes* pRes = &pSql->res;
|
SSqlRes* pRes = &pSql->res;
|
||||||
|
|
||||||
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
|
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
|
||||||
|
|
||||||
// TODO put to async res?
|
|
||||||
if (taos_errno(pSql) != TSDB_CODE_SUCCESS) {
|
if (taos_errno(pSql) != TSDB_CODE_SUCCESS) {
|
||||||
assert(numOfRows == taos_errno(pSql));
|
assert(numOfRows == taos_errno(pSql));
|
||||||
|
|
||||||
pParentSql->res.code = numOfRows;
|
pParentSql->res.code = numOfRows;
|
||||||
tscError("%p retrieve failed, index:%d, code:%s", pSql, pSupporter->subqueryIndex, tstrerror(numOfRows));
|
tscError("%p retrieve failed, index:%d, code:%s", pSql, pSupporter->subqueryIndex, tstrerror(numOfRows));
|
||||||
|
|
||||||
|
tscQueueAsyncRes(pParentSql);
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (numOfRows >= 0) {
|
if (numOfRows >= 0) {
|
||||||
|
@ -941,31 +942,22 @@ void tscFetchDatablockFromSubquery(SSqlObj* pSql) {
|
||||||
|
|
||||||
SSqlRes *pRes = &pSub->res;
|
SSqlRes *pRes = &pSub->res;
|
||||||
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSub->cmd, 0);
|
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSub->cmd, 0);
|
||||||
// STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
|
|
||||||
|
|
||||||
// if (tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0)) {
|
|
||||||
// if (pRes->row >= pRes->numOfRows && pTableMetaInfo->vgroupIndex < pTableMetaInfo->vgroupList->numOfVgroups &&
|
|
||||||
// (!tscHasReachLimitation(pQueryInfo, pRes)) && !pRes->completed) {
|
|
||||||
// numOfFetch++;
|
|
||||||
// }
|
|
||||||
// } else {
|
|
||||||
if (!tscHasReachLimitation(pQueryInfo, pRes)) {
|
|
||||||
if (pRes->row >= pRes->numOfRows) {
|
|
||||||
hasData = false;
|
|
||||||
|
|
||||||
if (!pRes->completed) {
|
if (!tscHasReachLimitation(pQueryInfo, pRes)) {
|
||||||
numOfFetch++;
|
if (pRes->row >= pRes->numOfRows) {
|
||||||
}
|
hasData = false;
|
||||||
}
|
|
||||||
} else { // has reach the limitation, no data anymore
|
if (!pRes->completed) {
|
||||||
if (pRes->row >= pRes->numOfRows) {
|
numOfFetch++;
|
||||||
hasData = false;
|
|
||||||
break;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
} else { // has reach the limitation, no data anymore
|
||||||
|
if (pRes->row >= pRes->numOfRows) {
|
||||||
|
hasData = false;
|
||||||
|
break;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
// }
|
}
|
||||||
|
|
||||||
// has data remains in client side, and continue to return data to app
|
// has data remains in client side, and continue to return data to app
|
||||||
if (hasData) {
|
if (hasData) {
|
||||||
|
@ -1026,7 +1018,7 @@ void tscSetupOutputColumnIndex(SSqlObj* pSql) {
|
||||||
SSqlCmd* pCmd = &pSql->cmd;
|
SSqlCmd* pCmd = &pSql->cmd;
|
||||||
SSqlRes* pRes = &pSql->res;
|
SSqlRes* pRes = &pSql->res;
|
||||||
|
|
||||||
tscDebug("%p all subquery response, retrieve data", pSql);
|
tscDebug("%p all subquery response, retrieve data for subclause:%d", pSql, pCmd->clauseIndex);
|
||||||
|
|
||||||
// the column transfer support struct has been built
|
// the column transfer support struct has been built
|
||||||
if (pRes->pColumnIndex != NULL) {
|
if (pRes->pColumnIndex != NULL) {
|
||||||
|
@ -1195,8 +1187,11 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSupporter
|
||||||
|
|
||||||
pNew->cmd.numOfCols = 0;
|
pNew->cmd.numOfCols = 0;
|
||||||
pNewQueryInfo->intervalTime = 0;
|
pNewQueryInfo->intervalTime = 0;
|
||||||
memset(&pNewQueryInfo->limit, 0, sizeof(SLimitVal));
|
pSupporter->limit = pNewQueryInfo->limit;
|
||||||
|
|
||||||
|
pNewQueryInfo->limit.limit = -1;
|
||||||
|
pNewQueryInfo->limit.offset = 0;
|
||||||
|
|
||||||
// backup the data and clear it in the sqlcmd object
|
// backup the data and clear it in the sqlcmd object
|
||||||
pSupporter->groupbyExpr = pNewQueryInfo->groupbyExpr;
|
pSupporter->groupbyExpr = pNewQueryInfo->groupbyExpr;
|
||||||
memset(&pNewQueryInfo->groupbyExpr, 0, sizeof(SSqlGroupbyExpr));
|
memset(&pNewQueryInfo->groupbyExpr, 0, sizeof(SSqlGroupbyExpr));
|
||||||
|
@ -1307,7 +1302,7 @@ int32_t tscHandleMasterJoinQuery(SSqlObj* pSql) {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pSql->cmd.command = (pSql->numOfSubs <= 0)? TSDB_SQL_RETRIEVE_EMPTY_RESULT:TSDB_SQL_TABLE_JOIN_RETRIEVE;
|
pSql->cmd.command = (pSql->numOfSubs <= 0)? TSDB_SQL_RETRIEVE_EMPTY_RESULT:TSDB_SQL_TABLE_JOIN_RETRIEVE;
|
||||||
|
|
||||||
return TSDB_CODE_SUCCESS;
|
return TSDB_CODE_SUCCESS;
|
||||||
|
@ -1982,88 +1977,119 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
|
||||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static char* getResultBlockPosition(SSqlCmd* pCmd, SSqlRes* pRes, int32_t columnIndex, int16_t* bytes) {
|
||||||
|
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
|
||||||
|
|
||||||
|
SFieldSupInfo* pInfo = (SFieldSupInfo*) TARRAY_GET_ELEM(pQueryInfo->fieldsInfo.pSupportInfo, columnIndex);
|
||||||
|
assert(pInfo->pSqlExpr != NULL);
|
||||||
|
|
||||||
|
*bytes = pInfo->pSqlExpr->resBytes;
|
||||||
|
char* pData = pRes->data + pInfo->pSqlExpr->offset * pRes->numOfRows;
|
||||||
|
|
||||||
|
return pData;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void doBuildResFromSubqueries(SSqlObj* pSql) {
|
||||||
|
SSqlRes* pRes = &pSql->res;
|
||||||
|
|
||||||
|
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, pSql->cmd.clauseIndex);
|
||||||
|
|
||||||
|
int32_t numOfRes = INT32_MAX;
|
||||||
|
for (int32_t i = 0; i < pSql->numOfSubs; ++i) {
|
||||||
|
if (pSql->pSubs[i] == NULL) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
numOfRes = MIN(numOfRes, pSql->pSubs[i]->res.numOfRows);
|
||||||
|
}
|
||||||
|
|
||||||
|
int32_t totalSize = tscGetResRowLength(pQueryInfo->exprList);
|
||||||
|
pRes->pRsp = realloc(pRes->pRsp, numOfRes * totalSize);
|
||||||
|
pRes->data = pRes->pRsp;
|
||||||
|
|
||||||
|
char* data = pRes->data;
|
||||||
|
int16_t bytes = 0;
|
||||||
|
|
||||||
|
size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
|
||||||
|
for(int32_t i = 0; i < numOfExprs; ++i) {
|
||||||
|
SColumnIndex* pIndex = &pRes->pColumnIndex[i];
|
||||||
|
SSqlRes *pRes1 = &pSql->pSubs[pIndex->tableIndex]->res;
|
||||||
|
SSqlCmd *pCmd1 = &pSql->pSubs[pIndex->tableIndex]->cmd;
|
||||||
|
|
||||||
|
char* pData = getResultBlockPosition(pCmd1, pRes1, pIndex->columnIndex, &bytes);
|
||||||
|
memcpy(data, pData, bytes * numOfRes);
|
||||||
|
|
||||||
|
data += bytes * numOfRes;
|
||||||
|
pRes1->row = numOfRes;
|
||||||
|
}
|
||||||
|
|
||||||
|
pRes->numOfRows = numOfRes;
|
||||||
|
pRes->numOfClauseTotal += numOfRes;
|
||||||
|
}
|
||||||
|
|
||||||
void tscBuildResFromSubqueries(SSqlObj *pSql) {
|
void tscBuildResFromSubqueries(SSqlObj *pSql) {
|
||||||
SSqlRes *pRes = &pSql->res;
|
SSqlRes* pRes = &pSql->res;
|
||||||
|
|
||||||
if (pRes->code != TSDB_CODE_SUCCESS) {
|
if (pRes->code != TSDB_CODE_SUCCESS) {
|
||||||
tscQueueAsyncRes(pSql);
|
tscQueueAsyncRes(pSql);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
while (1) {
|
if (pRes->tsrow == NULL) {
|
||||||
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, pSql->cmd.clauseIndex);
|
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, pSql->cmd.clauseIndex);
|
||||||
|
|
||||||
size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
|
size_t numOfExprs = tscSqlExprNumOfExprs(pQueryInfo);
|
||||||
|
pRes->tsrow = calloc(numOfExprs, POINTER_BYTES);
|
||||||
if (pRes->tsrow == NULL) {
|
pRes->buffer = calloc(numOfExprs, POINTER_BYTES);
|
||||||
pRes->tsrow = calloc(numOfExprs, POINTER_BYTES);
|
pRes->length = calloc(numOfExprs, sizeof(int32_t));
|
||||||
pRes->length = calloc(numOfExprs, sizeof(int32_t));
|
|
||||||
|
tscRestoreSQLFuncForSTableQuery(pQueryInfo);
|
||||||
|
}
|
||||||
|
|
||||||
|
while (1) {
|
||||||
|
if (pRes->row < pRes->numOfRows) {
|
||||||
|
assert(0);
|
||||||
}
|
}
|
||||||
|
|
||||||
bool success = false;
|
doBuildResFromSubqueries(pSql);
|
||||||
|
sem_post(&pSql->rspSem);
|
||||||
int32_t numOfTableHasRes = 0;
|
|
||||||
for (int32_t i = 0; i < pSql->numOfSubs; ++i) {
|
return;
|
||||||
if (pSql->pSubs[i] != NULL) {
|
|
||||||
numOfTableHasRes++;
|
// continue retrieve data from vnode
|
||||||
}
|
// if (!tscHasRemainDataInSubqueryResultSet(pSql)) {
|
||||||
}
|
// tscDebug("%p at least one subquery exhausted, free all other %d subqueries", pSql, pSql->numOfSubs - 1);
|
||||||
|
// SSubqueryState* pState = NULL;
|
||||||
if (numOfTableHasRes >= 2) { // do merge result
|
//
|
||||||
success = (doSetResultRowData(pSql->pSubs[0], false) != NULL) && (doSetResultRowData(pSql->pSubs[1], false) != NULL);
|
// // free all sub sqlobj
|
||||||
} else { // only one subquery
|
// for (int32_t i = 0; i < pSql->numOfSubs; ++i) {
|
||||||
SSqlObj *pSub = pSql->pSubs[0];
|
// SSqlObj* pChildObj = pSql->pSubs[i];
|
||||||
if (pSub == NULL) {
|
// if (pChildObj == NULL) {
|
||||||
pSub = pSql->pSubs[1];
|
// continue;
|
||||||
}
|
// }
|
||||||
|
//
|
||||||
success = (doSetResultRowData(pSub, false) != NULL);
|
// SJoinSupporter* pSupporter = (SJoinSupporter*)pChildObj->param;
|
||||||
}
|
// pState = pSupporter->pState;
|
||||||
|
//
|
||||||
if (success) { // current row of final output has been built, return to app
|
// tscDestroyJoinSupporter(pChildObj->param);
|
||||||
for (int32_t i = 0; i < numOfExprs; ++i) {
|
// taos_free_result(pChildObj);
|
||||||
SColumnIndex* pIndex = &pRes->pColumnIndex[i];
|
// }
|
||||||
SSqlRes *pRes1 = &pSql->pSubs[pIndex->tableIndex]->res;
|
//
|
||||||
pRes->tsrow[i] = pRes1->tsrow[pIndex->columnIndex];
|
// free(pState);
|
||||||
pRes->length[i] = pRes1->length[pIndex->columnIndex];
|
//
|
||||||
}
|
// pRes->completed = true; // set query completed
|
||||||
|
// sem_post(&pSql->rspSem);
|
||||||
pRes->numOfClauseTotal++;
|
// return;
|
||||||
break;
|
// }
|
||||||
} else { // continue retrieve data from vnode
|
|
||||||
if (!tscHasRemainDataInSubqueryResultSet(pSql)) {
|
tscFetchDatablockFromSubquery(pSql);
|
||||||
tscDebug("%p at least one subquery exhausted, free all other %d subqueries", pSql, pSql->numOfSubs - 1);
|
if (pRes->code != TSDB_CODE_SUCCESS) {
|
||||||
SSubqueryState *pState = NULL;
|
return;
|
||||||
|
|
||||||
// free all sub sqlobj
|
|
||||||
for (int32_t i = 0; i < pSql->numOfSubs; ++i) {
|
|
||||||
SSqlObj *pChildObj = pSql->pSubs[i];
|
|
||||||
if (pChildObj == NULL) {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
SJoinSupporter *pSupporter = (SJoinSupporter *)pChildObj->param;
|
|
||||||
pState = pSupporter->pState;
|
|
||||||
|
|
||||||
tscDestroyJoinSupporter(pChildObj->param);
|
|
||||||
taos_free_result(pChildObj);
|
|
||||||
}
|
|
||||||
|
|
||||||
free(pState);
|
|
||||||
|
|
||||||
pRes->completed = true; // set query completed
|
|
||||||
sem_post(&pSql->rspSem);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
tscFetchDatablockFromSubquery(pSql);
|
|
||||||
if (pRes->code != TSDB_CODE_SUCCESS) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (pSql->res.code == TSDB_CODE_SUCCESS) {
|
if (pSql->res.code == TSDB_CODE_SUCCESS) {
|
||||||
(*pSql->fp)(pSql->param, pSql, 0);
|
(*pSql->fp)(pSql->param, pSql, pRes->numOfRows);
|
||||||
} else {
|
} else {
|
||||||
tscQueueAsyncRes(pSql);
|
tscQueueAsyncRes(pSql);
|
||||||
}
|
}
|
||||||
|
@ -2117,14 +2143,6 @@ void **doSetResultRowData(SSqlObj *pSql, bool finalResult) {
|
||||||
|
|
||||||
assert(pRes->row >= 0 && pRes->row <= pRes->numOfRows);
|
assert(pRes->row >= 0 && pRes->row <= pRes->numOfRows);
|
||||||
|
|
||||||
if(pCmd->command == TSDB_SQL_TABLE_JOIN_RETRIEVE) {
|
|
||||||
if (pRes->completed) {
|
|
||||||
tfree(pRes->tsrow);
|
|
||||||
}
|
|
||||||
|
|
||||||
return pRes->tsrow;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (pRes->row >= pRes->numOfRows) { // all the results has returned to invoker
|
if (pRes->row >= pRes->numOfRows) { // all the results has returned to invoker
|
||||||
tfree(pRes->tsrow);
|
tfree(pRes->tsrow);
|
||||||
return pRes->tsrow;
|
return pRes->tsrow;
|
||||||
|
@ -2182,7 +2200,7 @@ void **doSetResultRowData(SSqlObj *pSql, bool finalResult) {
|
||||||
return pRes->tsrow;
|
return pRes->tsrow;
|
||||||
}
|
}
|
||||||
|
|
||||||
static bool tscHasRemainDataInSubqueryResultSet(SSqlObj *pSql) {
|
static UNUSED_FUNC bool tscHasRemainDataInSubqueryResultSet(SSqlObj *pSql) {
|
||||||
bool hasData = true;
|
bool hasData = true;
|
||||||
SSqlCmd *pCmd = &pSql->cmd;
|
SSqlCmd *pCmd = &pSql->cmd;
|
||||||
|
|
||||||
|
|
|
@ -1994,6 +1994,10 @@ bool hasMoreVnodesToTry(SSqlObj* pSql) {
|
||||||
(!tscHasReachLimitation(pQueryInfo, pRes)) && (pTableMetaInfo->vgroupIndex < numOfVgroups - 1);
|
(!tscHasReachLimitation(pQueryInfo, pRes)) && (pTableMetaInfo->vgroupIndex < numOfVgroups - 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
bool hasMoreClauseToTry(SSqlObj* pSql) {
|
||||||
|
return pSql->cmd.clauseIndex < pSql->cmd.numOfClause - 1;
|
||||||
|
}
|
||||||
|
|
||||||
void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
|
void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
|
||||||
SSqlCmd* pCmd = &pSql->cmd;
|
SSqlCmd* pCmd = &pSql->cmd;
|
||||||
SSqlRes* pRes = &pSql->res;
|
SSqlRes* pRes = &pSql->res;
|
||||||
|
@ -2050,7 +2054,7 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
void tscTryQueryNextClause(SSqlObj* pSql, void (*queryFp)()) {
|
void tscTryQueryNextClause(SSqlObj* pSql, __async_cb_func_t fp) {
|
||||||
SSqlCmd* pCmd = &pSql->cmd;
|
SSqlCmd* pCmd = &pSql->cmd;
|
||||||
SSqlRes* pRes = &pSql->res;
|
SSqlRes* pRes = &pSql->res;
|
||||||
|
|
||||||
|
@ -2070,17 +2074,13 @@ void tscTryQueryNextClause(SSqlObj* pSql, void (*queryFp)()) {
|
||||||
|
|
||||||
tfree(pSql->pSubs);
|
tfree(pSql->pSubs);
|
||||||
pSql->numOfSubs = 0;
|
pSql->numOfSubs = 0;
|
||||||
|
pSql->fp = fp;
|
||||||
if (pSql->fp != NULL) {
|
|
||||||
pSql->fp = queryFp;
|
|
||||||
assert(queryFp != NULL);
|
|
||||||
}
|
|
||||||
|
|
||||||
tscDebug("%p try data in the next subclause:%d, total subclause:%d", pSql, pCmd->clauseIndex, pCmd->numOfClause);
|
tscDebug("%p try data in the next subclause:%d, total subclause:%d", pSql, pCmd->clauseIndex, pCmd->numOfClause);
|
||||||
if (pCmd->command > TSDB_SQL_LOCAL) {
|
if (pCmd->command > TSDB_SQL_LOCAL) {
|
||||||
tscProcessLocalCmd(pSql);
|
tscProcessLocalCmd(pSql);
|
||||||
} else {
|
} else {
|
||||||
tscProcessSql(pSql);
|
tscDoQuery(pSql);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -1,10 +1,10 @@
|
||||||
system sh/stop_dnodes.sh
|
#system sh/stop_dnodes.sh
|
||||||
|
#
|
||||||
system sh/deploy.sh -n dnode1 -i 1
|
#system sh/deploy.sh -n dnode1 -i 1
|
||||||
system sh/cfg.sh -n dnode1 -c walLevel -v 0
|
#system sh/cfg.sh -n dnode1 -c walLevel -v 0
|
||||||
system sh/cfg.sh -n dnode1 -c debugFlag -v 135
|
#system sh/cfg.sh -n dnode1 -c debugFlag -v 135
|
||||||
system sh/cfg.sh -n dnode1 -c rpcDebugFlag -v 135
|
#system sh/cfg.sh -n dnode1 -c rpcDebugFlag -v 135
|
||||||
system sh/exec.sh -n dnode1 -s start
|
#system sh/exec.sh -n dnode1 -s start
|
||||||
sleep 1000
|
sleep 1000
|
||||||
sql connect
|
sql connect
|
||||||
|
|
||||||
|
@ -24,77 +24,77 @@ $mt = $mtPrefix . $i
|
||||||
$j = 1
|
$j = 1
|
||||||
|
|
||||||
$mt1 = $mtPrefix . $j
|
$mt1 = $mtPrefix . $j
|
||||||
|
#
|
||||||
sql drop database if exits $db -x step1
|
#sql drop database if exits $db -x step1
|
||||||
step1:
|
#step1:
|
||||||
sql create database if not exists $db maxtables 4
|
#sql create database if not exists $db maxtables 4
|
||||||
sql use $db
|
sql use $db
|
||||||
sql create table $mt (ts timestamp, c1 int, c2 float, c3 bigint, c4 smallint, c5 tinyint, c6 double, c7 bool, c8 binary(10), c9 nchar(9)) TAGS(t1 int)
|
#sql create table $mt (ts timestamp, c1 int, c2 float, c3 bigint, c4 smallint, c5 tinyint, c6 double, c7 bool, c8 binary(10), c9 nchar(9)) TAGS(t1 int)
|
||||||
|
#
|
||||||
$i = 0
|
#$i = 0
|
||||||
$t = 1578203484000
|
#$t = 1578203484000
|
||||||
|
#
|
||||||
while $i < $tbNum
|
#while $i < $tbNum
|
||||||
$tb = $tbPrefix . $i
|
# $tb = $tbPrefix . $i
|
||||||
sql create table $tb using $mt tags( $i )
|
# sql create table $tb using $mt tags( $i )
|
||||||
|
#
|
||||||
$x = 0
|
# $x = 0
|
||||||
while $x < $rowNum
|
# while $x < $rowNum
|
||||||
$ms = $x * 1000
|
# $ms = $x * 1000
|
||||||
$ms = $ms * 60
|
# $ms = $ms * 60
|
||||||
|
#
|
||||||
$c = $x / 100
|
# $c = $x / 100
|
||||||
$c = $c * 100
|
# $c = $c * 100
|
||||||
$c = $x - $c
|
# $c = $x - $c
|
||||||
$binary = 'binary . $c
|
# $binary = 'binary . $c
|
||||||
$binary = $binary . '
|
# $binary = $binary . '
|
||||||
$nchar = 'nchar . $c
|
# $nchar = 'nchar . $c
|
||||||
$nchar = $nchar . '
|
# $nchar = $nchar . '
|
||||||
|
#
|
||||||
$t1 = $t + $ms
|
# $t1 = $t + $ms
|
||||||
sql insert into $tb values ($t1 , $c , $c , $c , $c , $c , $c , $c , $binary , $nchar )
|
# sql insert into $tb values ($t1 , $c , $c , $c , $c , $c , $c , $c , $binary , $nchar )
|
||||||
$x = $x + 1
|
# $x = $x + 1
|
||||||
endw
|
# endw
|
||||||
|
#
|
||||||
$i = $i + 1
|
# $i = $i + 1
|
||||||
endw
|
#endw
|
||||||
|
#
|
||||||
sql create table $mt1 (ts timestamp, c1 int, c2 float, c3 bigint, c4 smallint, c5 tinyint, c6 double, c7 bool, c8 binary(10), c9 nchar(9)) TAGS(t1 int)
|
#sql create table $mt1 (ts timestamp, c1 int, c2 float, c3 bigint, c4 smallint, c5 tinyint, c6 double, c7 bool, c8 binary(10), c9 nchar(9)) TAGS(t1 int)
|
||||||
|
#
|
||||||
$j = 0
|
#$j = 0
|
||||||
$t = 1578203484000
|
#$t = 1578203484000
|
||||||
$rowNum = 1000
|
#$rowNum = 1000
|
||||||
$tbNum = 5
|
#$tbNum = 5
|
||||||
$i = 0
|
#$i = 0
|
||||||
|
#
|
||||||
while $i < $tbNum
|
#while $i < $tbNum
|
||||||
$tb1 = $tbPrefix1 . $j
|
# $tb1 = $tbPrefix1 . $j
|
||||||
sql create table $tb1 using $mt1 tags( $i )
|
# sql create table $tb1 using $mt1 tags( $i )
|
||||||
|
#
|
||||||
$x = 0
|
# $x = 0
|
||||||
while $x < $rowNum
|
# while $x < $rowNum
|
||||||
$ms = $x * 1000
|
# $ms = $x * 1000
|
||||||
$ms = $ms * 60
|
# $ms = $ms * 60
|
||||||
|
#
|
||||||
$c = $x / 100
|
# $c = $x / 100
|
||||||
$c = $c * 100
|
# $c = $c * 100
|
||||||
$c = $x - $c
|
# $c = $x - $c
|
||||||
$binary = 'binary . $c
|
# $binary = 'binary . $c
|
||||||
$binary = $binary . '
|
# $binary = $binary . '
|
||||||
$nchar = 'nchar . $c
|
# $nchar = 'nchar . $c
|
||||||
$nchar = $nchar . '
|
# $nchar = $nchar . '
|
||||||
|
#
|
||||||
$t1 = $t + $ms
|
# $t1 = $t + $ms
|
||||||
sql insert into $tb1 values ($t1 , $c , $c , $c , $c , $c , $c , $c , $binary , $nchar )
|
# sql insert into $tb1 values ($t1 , $c , $c , $c , $c , $c , $c , $c , $binary , $nchar )
|
||||||
$x = $x + 1
|
# $x = $x + 1
|
||||||
endw
|
# endw
|
||||||
|
#
|
||||||
$i = $i + 1
|
# $i = $i + 1
|
||||||
$j = $j + 1
|
# $j = $j + 1
|
||||||
endw
|
#endw
|
||||||
|
#
|
||||||
print sleep 1sec.
|
#print sleep 1sec.
|
||||||
sleep 1000
|
#sleep 1000
|
||||||
|
|
||||||
$i = 1
|
$i = 1
|
||||||
$tb = $tbPrefix . $i
|
$tb = $tbPrefix . $i
|
||||||
|
@ -222,7 +222,7 @@ endi
|
||||||
|
|
||||||
print ===========================================tags union
|
print ===========================================tags union
|
||||||
# two super table tag union, limit is not active during retrieve tags query
|
# two super table tag union, limit is not active during retrieve tags query
|
||||||
sql select t1 from union_mt0 union all select t1 from union_mt0 limit 1
|
sql select t1 from union_mt0 union all select t1 from union_mt0
|
||||||
if $rows != 20 then
|
if $rows != 20 then
|
||||||
return -1
|
return -1
|
||||||
endi
|
endi
|
||||||
|
@ -235,6 +235,10 @@ if $data90 != 9 then
|
||||||
return -1
|
return -1
|
||||||
endi
|
endi
|
||||||
|
|
||||||
|
#sql select t1 from union_mt0 union all select t1 from union_mt0 limit 1
|
||||||
|
#if $row != 11 then
|
||||||
|
# return -1
|
||||||
|
#endi
|
||||||
#========================================== two super table join subclause
|
#========================================== two super table join subclause
|
||||||
print ================two super table join subclause
|
print ================two super table join subclause
|
||||||
sql select avg(union_mt0.c1) as c from union_mt0 interval(1h) limit 10 union all select union_mt1.ts, union_mt1.c1/1.0 as c from union_mt0, union_mt1 where union_mt1.ts=union_mt0.ts and union_mt1.t1=union_mt0.t1 limit 5;
|
sql select avg(union_mt0.c1) as c from union_mt0 interval(1h) limit 10 union all select union_mt1.ts, union_mt1.c1/1.0 as c from union_mt0, union_mt1 where union_mt1.ts=union_mt0.ts and union_mt1.t1=union_mt0.t1 limit 5;
|
||||||
|
|
Loading…
Reference in New Issue