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

[td-3902]update the log.

This commit is contained in:
Haojun Liao 2021-04-21 22:38:34 +08:00
parent d37fad65dd
commit 5e1b9faa87
10 changed files with 175 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/client/src/tscAsync.c
View File

@ -283,7 +283,7 @@ void tscQueueAsyncError(void(*fp), void *param, int32_t code) {
static void tscAsyncResultCallback(SSchedMsg *pMsg) { static void tscAsyncResultCallback(SSchedMsg *pMsg) {
SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, (int64_t)pMsg->ahandle); SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, (int64_t)pMsg->ahandle);
if (pSql == NULL || pSql->signature != pSql) { if (pSql == NULL || pSql->signature != pSql) {
tscDebug("%p SqlObj is freed, not add into queue async res", pSql); tscDebug("0x%"PRIx64" SqlObj is freed, not add into queue async res", pSql->self);
return; return;
} }
@ -372,13 +372,13 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
goto _error; goto _error;
} }
tscDebug("%p get %s successfully", pSql, msg); tscDebug("0x%"PRIx64" get %s successfully", pSql->self, msg);
if (pSql->pStream == NULL) { if (pSql->pStream == NULL) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
// check if it is a sub-query of super table query first, if true, enter another routine // check if it is a sub-query of super table query first, if true, enter another routine
if (TSDB_QUERY_HAS_TYPE(pQueryInfo->type, (TSDB_QUERY_TYPE_STABLE_SUBQUERY|TSDB_QUERY_TYPE_SUBQUERY|TSDB_QUERY_TYPE_TAG_FILTER_QUERY))) { if (TSDB_QUERY_HAS_TYPE(pQueryInfo->type, (TSDB_QUERY_TYPE_STABLE_SUBQUERY|TSDB_QUERY_TYPE_SUBQUERY|TSDB_QUERY_TYPE_TAG_FILTER_QUERY))) {
tscDebug("%p update local table meta, continue to process sql and send the corresponding query", pSql); tscDebug("0x%"PRIx64" update local table meta, continue to process sql and send the corresponding query", pSql->self);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
@ -402,7 +402,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
return; return;
} else { // continue to process normal async query } else { // continue to process normal async query
if (pCmd->parseFinished) { if (pCmd->parseFinished) {
tscDebug("%p update local table meta, continue to process sql and send corresponding query", pSql); tscDebug("0x%"PRIx64" update local table meta, continue to process sql and send corresponding query", pSql->self);
STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0); STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
code = tscGetTableMeta(pSql, pTableMetaInfo); code = tscGetTableMeta(pSql, pTableMetaInfo);
@ -416,7 +416,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
assert(pCmd->command != TSDB_SQL_INSERT); assert(pCmd->command != TSDB_SQL_INSERT);
if (pCmd->command == TSDB_SQL_SELECT) { if (pCmd->command == TSDB_SQL_SELECT) {
tscDebug("%p redo parse sql string and proceed", pSql); tscDebug("0x%"PRIx64" redo parse sql string and proceed", pSql->self);
pCmd->parseFinished = false; pCmd->parseFinished = false;
tscResetSqlCmd(pCmd, true); tscResetSqlCmd(pCmd, true);
@ -436,7 +436,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
taosReleaseRef(tscObjRef, pSql->self); taosReleaseRef(tscObjRef, pSql->self);
return; return;
} else { } else {
tscDebug("%p continue parse sql after get table meta", pSql); tscDebug("0x%"PRIx64" continue parse sql after get table meta", pSql->self);
code = tsParseSql(pSql, false); code = tsParseSql(pSql, false);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
@ -486,7 +486,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
} }
} }
tscDebug("%p stream:%p meta is updated, start new query, command:%d", pSql, pSql->pStream, pSql->cmd.command); tscDebug("0x%"PRIx64" stream:%p meta is updated, start new query, command:%d", pSql->self, pSql->pStream, pSql->cmd.command);
if (!pSql->cmd.parseFinished) { if (!pSql->cmd.parseFinished) {
tsParseSql(pSql, false); tsParseSql(pSql, false);
} }

14
src/client/src/tscLocalMerge.c
View File

@ -193,7 +193,7 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
for (int32_t i = 0; i < numOfBuffer; ++i) { for (int32_t i = 0; i < numOfBuffer; ++i) {
int32_t len = pMemBuffer[i]->fileMeta.flushoutData.nLength; int32_t len = pMemBuffer[i]->fileMeta.flushoutData.nLength;
if (len == 0) { if (len == 0) {
tscDebug("%p no data retrieved from orderOfVnode:%d", pSql, i + 1); tscDebug("0x%"PRIx64" no data retrieved from orderOfVnode:%d", pSql->self, i + 1);
continue; continue;
} }
@ -203,7 +203,7 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
if (numOfFlush == 0 || numOfBuffer == 0) { if (numOfFlush == 0 || numOfBuffer == 0) {
tscLocalReducerEnvDestroy(pMemBuffer, pDesc, finalmodel, pFFModel, numOfBuffer); tscLocalReducerEnvDestroy(pMemBuffer, pDesc, finalmodel, pFFModel, numOfBuffer);
pCmd->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT; // no result, set the result empty pCmd->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT; // no result, set the result empty
tscDebug("%p retrieved no data", pSql); tscDebug("0x%"PRIx64" retrieved no data", pSql->self);
return; return;
} }
@ -235,7 +235,7 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
pReducer->numOfVnode = numOfBuffer; pReducer->numOfVnode = numOfBuffer;
pReducer->pDesc = pDesc; pReducer->pDesc = pDesc;
tscDebug("%p the number of merged leaves is: %d", pSql, pReducer->numOfBuffer); tscDebug("0x%"PRIx64" the number of merged leaves is: %d", pSql->self, pReducer->numOfBuffer);
int32_t idx = 0; int32_t idx = 0;
for (int32_t i = 0; i < numOfBuffer; ++i) { for (int32_t i = 0; i < numOfBuffer; ++i) {
@ -258,7 +258,7 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
ds->pageId = 0; ds->pageId = 0;
ds->rowIdx = 0; ds->rowIdx = 0;
tscDebug("%p load data from disk into memory, orderOfVnode:%d, total:%d", pSql, i + 1, idx + 1); tscDebug("0x%"PRIx64" load data from disk into memory, orderOfVnode:%d, total:%d", pSql->self, i + 1, idx + 1);
tExtMemBufferLoadData(pMemBuffer[i], &(ds->filePage), j, 0); tExtMemBufferLoadData(pMemBuffer[i], &(ds->filePage), j, 0);
#ifdef _DEBUG_VIEW #ifdef _DEBUG_VIEW
printf("load data page into mem for build loser tree: %" PRIu64 " rows\n", ds->filePage.num); printf("load data page into mem for build loser tree: %" PRIu64 " rows\n", ds->filePage.num);
@ -272,7 +272,7 @@ void tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrde
#endif #endif
if (ds->filePage.num == 0) { // no data in this flush, the index does not increase if (ds->filePage.num == 0) { // no data in this flush, the index does not increase
tscDebug("%p flush data is empty, ignore %d flush record", pSql, idx); tscDebug("0x%"PRIx64" flush data is empty, ignore %d flush record", pSql->self, idx);
tfree(ds); tfree(ds);
continue; continue;
} }
@ -547,10 +547,10 @@ void tscDestroyLocalMerger(SSqlObj *pSql) {
pLocalMerge->numOfCompleted = 0; pLocalMerge->numOfCompleted = 0;
free(pLocalMerge); free(pLocalMerge);
} else { } else {
tscDebug("%p already freed or another free function is invoked", pSql); tscDebug("0x%"PRIx64" already freed or another free function is invoked", pSql->self);
} }
tscDebug("%p free local reducer finished", pSql); tscDebug("0x%"PRIx64" free local reducer finished", pSql->self);
} }
static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SSqlCmd *pCmd, SColumnModel *pModel) { static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SSqlCmd *pCmd, SColumnModel *pModel) {

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

@ -1089,7 +1089,7 @@ int tsParseInsertSql(SSqlObj *pSql) {
str = pCmd->curSql; str = pCmd->curSql;
} }
tscDebug("%p create data block list hashList:%p", pSql, pCmd->pTableBlockHashList); tscDebug("0x%"PRIx64" create data block list hashList:%p", pSql->self, pCmd->pTableBlockHashList);
while (1) { while (1) {
int32_t index = 0; int32_t index = 0;
@ -1303,7 +1303,7 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
SSqlCmd* pCmd = &pSql->cmd; SSqlCmd* pCmd = &pSql->cmd;
if ((!pCmd->parseFinished) && (!initial)) { if ((!pCmd->parseFinished) && (!initial)) {
tscDebug("%p resume to parse sql: %s", pSql, pCmd->curSql); tscDebug("0x%"PRIx64" resume to parse sql: %s", pSql->self, pCmd->curSql);
} }
ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE); ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);

10
src/client/src/tscProfile.c
View File

@ -61,7 +61,7 @@ void tscAddIntoSqlList(SSqlObj *pSql) {
pSql->stime = taosGetTimestampMs(); pSql->stime = taosGetTimestampMs();
pSql->listed = 1; pSql->listed = 1;
tscDebug("%p added into sqlList", pSql); tscDebug("0x%"PRIx64" added into sqlList", pSql->self);
} }
void tscSaveSlowQueryFpCb(void *param, TAOS_RES *result, int code) { void tscSaveSlowQueryFpCb(void *param, TAOS_RES *result, int code) {
@ -99,7 +99,7 @@ void tscSaveSlowQuery(SSqlObj *pSql) {
return; return;
} }
tscDebug("%p query time:%" PRId64 " sql:%s", pSql, pSql->res.useconds, pSql->sqlstr); tscDebug("0x%"PRIx64" query time:%" PRId64 " sql:%s", pSql->self, pSql->res.useconds, pSql->sqlstr);
int32_t sqlSize = (int32_t)(TSDB_SLOW_QUERY_SQL_LEN + size); int32_t sqlSize = (int32_t)(TSDB_SLOW_QUERY_SQL_LEN + size);
char *sql = malloc(sqlSize); char *sql = malloc(sqlSize);
@ -141,7 +141,7 @@ void tscRemoveFromSqlList(SSqlObj *pSql) {
pSql->listed = 0; pSql->listed = 0;
tscSaveSlowQuery(pSql); tscSaveSlowQuery(pSql);
tscDebug("%p removed from sqlList", pSql); tscDebug("0x%"PRIx64" removed from sqlList", pSql->self);
} }
void tscKillQuery(STscObj *pObj, uint32_t killId) { void tscKillQuery(STscObj *pObj, uint32_t killId) {
@ -158,7 +158,7 @@ void tscKillQuery(STscObj *pObj, uint32_t killId) {
if (pSql == NULL) { if (pSql == NULL) {
tscError("failed to kill query, id:%d, it may have completed/terminated", killId); tscError("failed to kill query, id:%d, it may have completed/terminated", killId);
} else { } else {
tscDebug("%p query is killed, queryId:%d", pSql, killId); tscDebug("0x%"PRIx64" query is killed, queryId:%d", pSql->self, killId);
taos_stop_query(pSql); taos_stop_query(pSql);
} }
} }
@ -213,7 +213,7 @@ void tscKillStream(STscObj *pObj, uint32_t killId) {
pthread_mutex_unlock(&pObj->mutex); pthread_mutex_unlock(&pObj->mutex);
if (pStream) { if (pStream) {
tscDebug("%p stream:%p is killed, streamId:%d", pStream->pSql, pStream, killId); tscDebug("0x%"PRIx64" stream:%p is killed, streamId:%d", pStream->pSql->self, pStream, killId);
if (pStream->callback) { if (pStream->callback) {
pStream->callback(pStream->param); pStream->callback(pStream->param);
} }

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

@ -5508,15 +5508,15 @@ int32_t parseLimitClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t clauseIn
pQueryInfo->clauseLimit = pQueryInfo->limit.limit; pQueryInfo->clauseLimit = pQueryInfo->limit.limit;
pQueryInfo->slimit = pQuerySqlNode->slimit; pQueryInfo->slimit = pQuerySqlNode->slimit;
tscDebug("%p limit:%" PRId64 ", offset:%" PRId64 " slimit:%" PRId64 ", soffset:%" PRId64, pSql, pQueryInfo->limit.limit, tscDebug("0x%"PRIx64" limit:%" PRId64 ", offset:%" PRId64 " slimit:%" PRId64 ", soffset:%" PRId64, pSql->self,
pQueryInfo->limit.offset, pQueryInfo->slimit.limit, pQueryInfo->slimit.offset); pQueryInfo->limit.limit, pQueryInfo->limit.offset, pQueryInfo->slimit.limit, pQueryInfo->slimit.offset);
if (pQueryInfo->slimit.offset < 0 || pQueryInfo->limit.offset < 0) { if (pQueryInfo->slimit.offset < 0 || pQueryInfo->limit.offset < 0) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg0); return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg0);
} }
if (pQueryInfo->limit.limit == 0) { if (pQueryInfo->limit.limit == 0) {
tscDebug("%p limit 0, no output result", pSql); tscDebug("0x%"PRIx64" limit 0, no output result", pSql->self);
pQueryInfo->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT; pQueryInfo->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -5538,7 +5538,7 @@ int32_t parseLimitClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t clauseIn
} }
if (pQueryInfo->slimit.limit == 0) { if (pQueryInfo->slimit.limit == 0) {
tscDebug("%p slimit 0, no output result", pSql); tscDebug("0x%"PRIx64" slimit 0, no output result", pSql->self);
pQueryInfo->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT; pQueryInfo->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -5556,7 +5556,7 @@ int32_t parseLimitClause(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t clauseIn
// No tables included. No results generated. Query results are empty. // No tables included. No results generated. Query results are empty.
if (pTableMetaInfo->vgroupList->numOfVgroups == 0) { if (pTableMetaInfo->vgroupList->numOfVgroups == 0) {
tscDebug("%p no table in super table, no output result", pSql); tscDebug("0x%"PRIx64" no table in super table, no output result", pSql->self);
pQueryInfo->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT; pQueryInfo->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -6351,7 +6351,7 @@ void tscPrintSelectClause(SSqlObj* pSql, int32_t subClauseIndex) {
assert(offset < totalBufSize); assert(offset < totalBufSize);
str[offset] = ']'; str[offset] = ']';
assert(offset < totalBufSize); assert(offset < totalBufSize);
tscDebug("%p select clause:%s", pSql, str); tscDebug("0x%"PRIx64" select clause:%s", pSql->self, str);
} }
int32_t doCheckForCreateTable(SSqlObj* pSql, int32_t subClauseIndex, SSqlInfo* pInfo) { int32_t doCheckForCreateTable(SSqlObj* pSql, int32_t subClauseIndex, SSqlInfo* pInfo) {

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

@ -147,7 +147,7 @@ static void tscUpdateVgroupInfo(SSqlObj *pSql, SRpcEpSet *pEpSet) {
// Update the local cached epSet info cached by SqlObj // Update the local cached epSet info cached by SqlObj
int32_t inUse = pSql->epSet.inUse; int32_t inUse = pSql->epSet.inUse;
tscDumpEpSetFromVgroupInfo(&pSql->epSet, &vgroupInfo); tscDumpEpSetFromVgroupInfo(&pSql->epSet, &vgroupInfo);
tscDebug("%p update the epSet in SqlObj, in use before:%d, after:%d", pSql, inUse, pSql->epSet.inUse); tscDebug("0x%"PRIx64" update the epSet in SqlObj, in use before:%d, after:%d", pSql->self, inUse, pSql->epSet.inUse);
} }
@ -245,11 +245,11 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
if (pObj->hbrid != 0) { if (pObj->hbrid != 0) {
int32_t waitingDuring = tsShellActivityTimer * 500; int32_t waitingDuring = tsShellActivityTimer * 500;
tscDebug("%p send heartbeat in %dms", pSql, waitingDuring); tscDebug("0x%"PRIx64" send heartbeat in %dms", pSql->self, waitingDuring);
taosTmrReset(tscProcessActivityTimer, waitingDuring, (void *)pObj->rid, tscTmr, &pObj->pTimer); taosTmrReset(tscProcessActivityTimer, waitingDuring, (void *)pObj->rid, tscTmr, &pObj->pTimer);
} else { } else {
tscDebug("%p start to close tscObj:%p, not send heartbeat again", pSql, pObj); tscDebug("0x%"PRIx64" start to close tscObj:%p, not send heartbeat again", pSql->self, pObj);
} }
} }
@ -326,7 +326,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
pSql->rpcRid = -1; pSql->rpcRid = -1;
if (pObj->signature != pObj) { if (pObj->signature != pObj) {
tscDebug("%p DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command, pObj, pObj->signature); tscDebug("0x%"PRIx64" DB connection is closed, cmd:%d pObj:%p signature:%p", pSql->self, pCmd->command, pObj, pObj->signature);
taosRemoveRef(tscObjRef, handle); taosRemoveRef(tscObjRef, handle);
taosReleaseRef(tscObjRef, handle); taosReleaseRef(tscObjRef, handle);
@ -336,8 +336,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
if (pQueryInfo != NULL && pQueryInfo->type == TSDB_QUERY_TYPE_FREE_RESOURCE) { if (pQueryInfo != NULL && pQueryInfo->type == TSDB_QUERY_TYPE_FREE_RESOURCE) {
tscDebug("%p sqlObj needs to be released or DB connection is closed, cmd:%d type:%d, pObj:%p signature:%p", tscDebug("0x%"PRIx64" sqlObj needs to be released or DB connection is closed, cmd:%d type:%d, pObj:%p signature:%p",
pSql, pCmd->command, pQueryInfo->type, pObj, pObj->signature); pSql->self, pCmd->command, pQueryInfo->type, pObj, pObj->signature);
taosRemoveRef(tscObjRef, handle); taosRemoveRef(tscObjRef, handle);
taosReleaseRef(tscObjRef, handle); taosReleaseRef(tscObjRef, handle);
@ -396,13 +396,13 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
pRes->rspLen = 0; pRes->rspLen = 0;
if (pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED) { if (pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED) {
tscDebug("%p query is cancelled, code:%s", pSql, tstrerror(pRes->code)); tscDebug("0x%"PRIx64" query is cancelled, code:%s", pSql->self, tstrerror(pRes->code));
} else { } else {
pRes->code = rpcMsg->code; pRes->code = rpcMsg->code;
} }
if (pRes->code == TSDB_CODE_SUCCESS) { if (pRes->code == TSDB_CODE_SUCCESS) {
tscDebug("%p reset retry counter to be 0 due to success rsp, old:%d", pSql, pSql->retry); tscDebug("0x%"PRIx64" reset retry counter to be 0 due to success rsp, old:%d", pSql->self, pSql->retry);
pSql->retry = 0; pSql->retry = 0;
} }
@ -437,10 +437,10 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
pMsg->numOfFailedBlocks = htonl(pMsg->numOfFailedBlocks); pMsg->numOfFailedBlocks = htonl(pMsg->numOfFailedBlocks);
pRes->numOfRows += pMsg->affectedRows; pRes->numOfRows += pMsg->affectedRows;
tscDebug("%p SQL cmd:%s, code:%s inserted rows:%d rspLen:%d", pSql, sqlCmd[pCmd->command], tscDebug("0x%"PRIx64" SQL cmd:%s, code:%s inserted rows:%d rspLen:%d", pSql->self, sqlCmd[pCmd->command],
tstrerror(pRes->code), pMsg->affectedRows, pRes->rspLen); tstrerror(pRes->code), pMsg->affectedRows, pRes->rspLen);
} else { } else {
tscDebug("%p SQL cmd:%s, code:%s rspLen:%d", pSql, sqlCmd[pCmd->command], tstrerror(pRes->code), pRes->rspLen); tscDebug("0x%"PRIx64" SQL cmd:%s, code:%s rspLen:%d", pSql->self, sqlCmd[pCmd->command], tstrerror(pRes->code), pRes->rspLen);
} }
} }
@ -461,7 +461,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
if (shouldFree) { // in case of table-meta/vgrouplist query, automatically free it if (shouldFree) { // in case of table-meta/vgrouplist query, automatically free it
taosRemoveRef(tscObjRef, handle); taosRemoveRef(tscObjRef, handle);
tscDebug("%p sqlObj is automatically freed", pSql); tscDebug("0x%"PRIx64" sqlObj is automatically freed", pSql->self);
} }
taosReleaseRef(tscObjRef, handle); taosReleaseRef(tscObjRef, handle);
@ -522,7 +522,7 @@ int tscProcessSql(SSqlObj *pSql) {
assert((pQueryInfo->numOfTables == 0 && pQueryInfo->command == TSDB_SQL_HB) || pQueryInfo->numOfTables > 0); assert((pQueryInfo->numOfTables == 0 && pQueryInfo->command == TSDB_SQL_HB) || pQueryInfo->numOfTables > 0);
} }
tscDebug("%p SQL cmd:%s will be processed, name:%s, type:%d", pSql, sqlCmd[pCmd->command], name, type); tscDebug("0x%"PRIx64" SQL cmd:%s will be processed, name:%s, type:%d", pSql->self, sqlCmd[pCmd->command], name, type);
if (pCmd->command < TSDB_SQL_MGMT) { // the pTableMetaInfo cannot be NULL if (pCmd->command < TSDB_SQL_MGMT) { // the pTableMetaInfo cannot be NULL
if (pTableMetaInfo == NULL) { if (pTableMetaInfo == NULL) {
pSql->res.code = TSDB_CODE_TSC_APP_ERROR; pSql->res.code = TSDB_CODE_TSC_APP_ERROR;
@ -562,11 +562,11 @@ int tscBuildFetchMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
} }
pRetrieveMsg->header.vgId = htonl(vgId); pRetrieveMsg->header.vgId = htonl(vgId);
tscDebug("%p build fetch msg from vgId:%d, vgIndex:%d, qId:%" PRIu64, pSql, vgId, vgIndex, pSql->res.qId); tscDebug("0x%"PRIx64" build fetch msg from vgId:%d, vgIndex:%d, qId:%" PRIu64, pSql->self, vgId, vgIndex, pSql->res.qId);
} else { } else {
STableMeta* pTableMeta = pTableMetaInfo->pTableMeta; STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
pRetrieveMsg->header.vgId = htonl(pTableMeta->vgId); pRetrieveMsg->header.vgId = htonl(pTableMeta->vgId);
tscDebug("%p build fetch msg from only one vgroup, vgId:%d, qId:%" PRIu64, pSql, pTableMeta->vgId, pSql->res.qId); tscDebug("0x%"PRIx64" build fetch msg from only one vgroup, vgId:%d, qId:%" PRIu64, pSql->self, pTableMeta->vgId, pSql->res.qId);
} }
pSql->cmd.payloadLen = sizeof(SRetrieveTableMsg); pSql->cmd.payloadLen = sizeof(SRetrieveTableMsg);
@ -605,7 +605,7 @@ int tscBuildSubmitMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
taosHashGetClone(tscVgroupMap, &pTableMeta->vgId, sizeof(pTableMeta->vgId), NULL, &vgroupInfo, sizeof(SNewVgroupInfo)); taosHashGetClone(tscVgroupMap, &pTableMeta->vgId, sizeof(pTableMeta->vgId), NULL, &vgroupInfo, sizeof(SNewVgroupInfo));
tscDumpEpSetFromVgroupInfo(&pSql->epSet, &vgroupInfo); tscDumpEpSetFromVgroupInfo(&pSql->epSet, &vgroupInfo);
tscDebug("%p build submit msg, vgId:%d numOfTables:%d numberOfEP:%d", pSql, pTableMeta->vgId, pSql->cmd.numOfTablesInSubmit, tscDebug("0x%"PRIx64" build submit msg, vgId:%d numOfTables:%d numberOfEP:%d", pSql->self, pTableMeta->vgId, pSql->cmd.numOfTablesInSubmit,
pSql->epSet.numOfEps); pSql->epSet.numOfEps);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -671,7 +671,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
vgId = pVgroupInfo->vgId; vgId = pVgroupInfo->vgId;
tscSetDnodeEpSet(&pSql->epSet, pVgroupInfo); tscSetDnodeEpSet(&pSql->epSet, pVgroupInfo);
tscDebug("%p query on stable, vgIndex:%d, numOfVgroups:%d", pSql, index, pTableMetaInfo->vgroupList->numOfVgroups); tscDebug("0x%"PRIx64" query on stable, vgIndex:%d, numOfVgroups:%d", pSql->self, index, pTableMetaInfo->vgroupList->numOfVgroups);
} else { } else {
vgId = pTableMeta->vgId; vgId = pTableMeta->vgId;
@ -704,7 +704,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
int32_t numOfTables = (int32_t)taosArrayGetSize(pTableIdList->itemList); int32_t numOfTables = (int32_t)taosArrayGetSize(pTableIdList->itemList);
pQueryMsg->numOfTables = htonl(numOfTables); // set the number of tables pQueryMsg->numOfTables = htonl(numOfTables); // set the number of tables
tscDebug("%p query on stable, vgId:%d, numOfTables:%d, vgIndex:%d, numOfVgroups:%d", pSql, tscDebug("0x%"PRIx64" query on stable, vgId:%d, numOfTables:%d, vgIndex:%d, numOfVgroups:%d", pSql->self,
pTableIdList->vgInfo.vgId, numOfTables, index, numOfVgroups); pTableIdList->vgInfo.vgId, numOfTables, index, numOfVgroups);
// serialize each table id info // serialize each table id info
@ -722,7 +722,7 @@ static char *doSerializeTableInfo(SQueryTableMsg* pQueryMsg, SSqlObj *pSql, char
char n[TSDB_TABLE_FNAME_LEN] = {0}; char n[TSDB_TABLE_FNAME_LEN] = {0};
tNameExtractFullName(&pTableMetaInfo->name, n); tNameExtractFullName(&pTableMetaInfo->name, n);
tscDebug("%p vgId:%d, query on table:%s, tid:%d, uid:%" PRIu64, pSql, htonl(pQueryMsg->head.vgId), n, pTableMeta->id.tid, pTableMeta->id.uid); tscDebug("0x%"PRIx64" vgId:%d, query on table:%s, tid:%d, uid:%" PRIu64, pSql->self, htonl(pQueryMsg->head.vgId), n, pTableMeta->id.tid, pTableMeta->id.uid);
return pMsg; return pMsg;
} }
@ -1139,7 +1139,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
int32_t msgLen = (int32_t)(pMsg - pCmd->payload); int32_t msgLen = (int32_t)(pMsg - pCmd->payload);
tscDebug("%p msg built success, len:%d bytes", pSql, msgLen); tscDebug("0x%"PRIx64" msg built success, len:%d bytes", pSql->self, msgLen);
pCmd->payloadLen = msgLen; pCmd->payloadLen = msgLen;
pSql->cmd.msgType = TSDB_MSG_TYPE_QUERY; pSql->cmd.msgType = TSDB_MSG_TYPE_QUERY;
@ -1837,7 +1837,7 @@ int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
assert(pCmd->payloadLen + minMsgSize() <= pCmd->allocSize); assert(pCmd->payloadLen + minMsgSize() <= pCmd->allocSize);
tscDebug("%p build load multi-metermeta msg completed, numOfTables:%d, msg size:%d", pSql, pCmd->count, tscDebug("0x%"PRIx64" build load multi-metermeta msg completed, numOfTables:%d, msg size:%d", pSql->self, pCmd->count,
pCmd->payloadLen); pCmd->payloadLen);
return pCmd->payloadLen; return pCmd->payloadLen;
@ -2017,7 +2017,7 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) {
} }
} }
tscDebug("%p recv table meta, uid:%" PRIu64 ", tid:%d, name:%s", pSql, pTableMeta->id.uid, pTableMeta->id.tid, tscDebug("0x%"PRIx64" recv table meta, uid:%" PRIu64 ", tid:%d, name:%s", pSql->self, pTableMeta->id.uid, pTableMeta->id.tid,
tNameGetTableName(&pTableMetaInfo->name)); tNameGetTableName(&pTableMetaInfo->name));
free(pTableMeta); free(pTableMeta);
@ -2124,7 +2124,7 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
pSql->res.code = TSDB_CODE_SUCCESS; pSql->res.code = TSDB_CODE_SUCCESS;
pSql->res.numOfTotal = i; pSql->res.numOfTotal = i;
tscDebug("%p load multi-metermeta resp from complete num:%d", pSql, pSql->res.numOfTotal); tscDebug("0x%"PRIx64" load multi-metermeta resp from complete num:%d", pSql->self, pSql->res.numOfTotal);
#endif #endif
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -2292,7 +2292,7 @@ static void createHbObj(STscObj* pObj) {
pSql->signature = pSql; pSql->signature = pSql;
registerSqlObj(pSql); registerSqlObj(pSql);
tscDebug("%p HB is allocated, pObj:%p", pSql, pObj); tscDebug("0x%"PRIx64" HB is allocated, pObj:%p", pSql->self, pObj);
pObj->hbrid = pSql->self; pObj->hbrid = pSql->self;
} }
@ -2318,7 +2318,7 @@ int tscProcessConnectRsp(SSqlObj *pSql) {
tscUpdateMgmtEpSet(pSql, &pConnect->epSet); tscUpdateMgmtEpSet(pSql, &pConnect->epSet);
for (int i = 0; i < pConnect->epSet.numOfEps; ++i) { for (int i = 0; i < pConnect->epSet.numOfEps; ++i) {
tscDebug("%p epSet.fqdn[%d]: %s, pObj:%p", pSql, i, pConnect->epSet.fqdn[i], pObj); tscDebug("0x%"PRIx64" epSet.fqdn[%d]: %s, pObj:%p", pSql->self, i, pConnect->epSet.fqdn[i], pObj);
} }
} }
@ -2362,7 +2362,7 @@ int tscProcessDropTableRsp(SSqlObj *pSql) {
tNameExtractFullName(&pTableMetaInfo->name, name); tNameExtractFullName(&pTableMetaInfo->name, name);
taosHashRemove(tscTableMetaInfo, name, strnlen(name, TSDB_TABLE_FNAME_LEN)); taosHashRemove(tscTableMetaInfo, name, strnlen(name, TSDB_TABLE_FNAME_LEN));
tscDebug("%p remove table meta after drop table:%s, numOfRemain:%d", pSql, name, (int32_t) taosHashGetSize(tscTableMetaInfo)); tscDebug("0x%"PRIx64" remove table meta after drop table:%s, numOfRemain:%d", pSql->self, name, (int32_t) taosHashGetSize(tscTableMetaInfo));
pTableMetaInfo->pTableMeta = NULL; pTableMetaInfo->pTableMeta = NULL;
return 0; return 0;
@ -2374,7 +2374,7 @@ int tscProcessAlterTableMsgRsp(SSqlObj *pSql) {
char name[TSDB_TABLE_FNAME_LEN] = {0}; char name[TSDB_TABLE_FNAME_LEN] = {0};
tNameExtractFullName(&pTableMetaInfo->name, name); tNameExtractFullName(&pTableMetaInfo->name, name);
tscDebug("%p remove tableMeta in hashMap after alter-table: %s", pSql, name); tscDebug("0x%"PRIx64" remove tableMeta in hashMap after alter-table: %s", pSql->self, name);
bool isSuperTable = UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo); bool isSuperTable = UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo);
taosHashRemove(tscTableMetaInfo, name, strnlen(name, TSDB_TABLE_FNAME_LEN)); taosHashRemove(tscTableMetaInfo, name, strnlen(name, TSDB_TABLE_FNAME_LEN));
@ -2405,7 +2405,7 @@ int tscProcessQueryRsp(SSqlObj *pSql) {
pRes->data = NULL; pRes->data = NULL;
tscResetForNextRetrieve(pRes); tscResetForNextRetrieve(pRes);
tscDebug("%p query rsp received, qId:%"PRIu64, pSql, pRes->qId); tscDebug("0x%"PRIx64" query rsp received, qId:%"PRIu64, pSql->self, pRes->qId);
return 0; return 0;
} }
@ -2463,7 +2463,7 @@ int tscProcessRetrieveRspFromNode(SSqlObj *pSql) {
} }
pRes->row = 0; pRes->row = 0;
tscDebug("%p numOfRows:%d, offset:%" PRId64 ", complete:%d, qId:%"PRIu64, pSql, pRes->numOfRows, pRes->offset, tscDebug("0x%"PRIx64" numOfRows:%d, offset:%" PRId64 ", complete:%d, qId:%"PRIu64, pSql->self, pRes->numOfRows, pRes->offset,
pRes->completed, pRes->qId); pRes->completed, pRes->qId);
return 0; return 0;
@ -2507,14 +2507,14 @@ static int32_t getTableMetaFromMnode(SSqlObj *pSql, STableMetaInfo *pTableMetaIn
} }
} }
tscDebug("%p new pSqlObj:%p to get tableMeta, auto create:%d", pSql, pNew, pNew->cmd.autoCreated);
registerSqlObj(pNew); registerSqlObj(pNew);
tscDebug("0x%"PRIx64" new pSqlObj:0x%"PRIx64" to get tableMeta, auto create:%d", pSql->self, pNew->self,
pNew->cmd.autoCreated);
pNew->fp = tscTableMetaCallBack; pNew->fp = tscTableMetaCallBack;
pNew->param = (void *)pSql->self; pNew->param = (void *)pSql->self;
tscDebug("%p metaRid from %" PRId64 " to %" PRId64 , pSql, pSql->metaRid, pNew->self); tscDebug("0x%"PRIx64" metaRid from %" PRId64 " to %" PRId64 , pSql->self, pSql->metaRid, pNew->self);
pSql->metaRid = pNew->self; pSql->metaRid = pNew->self;
@ -2585,7 +2585,7 @@ int tscRenewTableMeta(SSqlObj *pSql, int32_t tableIndex) {
STableMeta* pTableMeta = pTableMetaInfo->pTableMeta; STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
if (pTableMeta) { if (pTableMeta) {
tscDebug("%p update table meta:%s, old meta numOfTags:%d, numOfCols:%d, uid:%" PRId64, pSql, name, tscDebug("0x%"PRIx64" update table meta:%s, old meta numOfTags:%d, numOfCols:%d, uid:%" PRId64, pSql->self, name,
tscGetNumOfTags(pTableMeta), tscGetNumOfColumns(pTableMeta), pTableMeta->id.uid); tscGetNumOfTags(pTableMeta), tscGetNumOfColumns(pTableMeta), pTableMeta->id.uid);
} }
@ -2645,12 +2645,12 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, int32_t clauseIndex) {
pNewQueryInfo->numOfTables = pQueryInfo->numOfTables; pNewQueryInfo->numOfTables = pQueryInfo->numOfTables;
registerSqlObj(pNew); registerSqlObj(pNew);
tscDebug("%p svgroupRid from %" PRId64 " to %" PRId64 , pSql, pSql->svgroupRid, pNew->self); tscDebug("0x%"PRIx64" svgroupRid from %" PRId64 " to %" PRId64 , pSql->self, pSql->svgroupRid, pNew->self);
pSql->svgroupRid = pNew->self; pSql->svgroupRid = pNew->self;
tscDebug("%p new sqlObj:%p to get vgroupInfo, numOfTables:%d", pSql, pNew, pNewQueryInfo->numOfTables); tscDebug("0x%"PRIx64" new sqlObj:%p to get vgroupInfo, numOfTables:%d", pSql->self, pNew, pNewQueryInfo->numOfTables);
pNew->fp = tscTableMetaCallBack; pNew->fp = tscTableMetaCallBack;
pNew->param = (void *)pSql->self; pNew->param = (void *)pSql->self;

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

@ -292,7 +292,7 @@ void taos_close(TAOS *taos) {
pHb->rpcRid = -1; pHb->rpcRid = -1;
} }
tscDebug("%p HB is freed", pHb); tscDebug("0x%"PRIx64" HB is freed", pHb->self);
taosReleaseRef(tscObjRef, pHb->self); taosReleaseRef(tscObjRef, pHb->self);
#ifdef __APPLE__ #ifdef __APPLE__
// to satisfy later tsem_destroy in taos_free_result // to satisfy later tsem_destroy in taos_free_result
@ -576,7 +576,7 @@ static bool tscKillQueryInDnode(SSqlObj* pSql) {
cmd == TSDB_SQL_FETCH)) { cmd == TSDB_SQL_FETCH)) {
pQueryInfo->type = TSDB_QUERY_TYPE_FREE_RESOURCE; pQueryInfo->type = TSDB_QUERY_TYPE_FREE_RESOURCE;
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;
tscDebug("%p send msg to dnode to free qhandle ASAP before free sqlObj, command:%s", pSql, sqlCmd[pCmd->command]); tscDebug("0x%"PRIx64" send msg to dnode to free qhandle ASAP before free sqlObj, command:%s", pSql->self, sqlCmd[pCmd->command]);
tscProcessSql(pSql); tscProcessSql(pSql);
return false; return false;
@ -594,7 +594,7 @@ void taos_free_result(TAOS_RES *res) {
bool freeNow = tscKillQueryInDnode(pSql); bool freeNow = tscKillQueryInDnode(pSql);
if (freeNow) { if (freeNow) {
tscDebug("%p free sqlObj in cache", pSql); tscDebug("0x%"PRIx64" free sqlObj in cache", pSql->self);
taosReleaseRef(tscObjRef, pSql->self); taosReleaseRef(tscObjRef, pSql->self);
} }
} }
@ -708,7 +708,7 @@ static void tscKillSTableQuery(SSqlObj *pSql) {
tscUnlockByThread(&pSql->squeryLock); tscUnlockByThread(&pSql->squeryLock);
tscDebug("%p super table query cancelled", pSql); tscDebug("0x%"PRIx64" super table query cancelled", pSql->self);
} }
void taos_stop_query(TAOS_RES *res) { void taos_stop_query(TAOS_RES *res) {
@ -717,7 +717,7 @@ void taos_stop_query(TAOS_RES *res) {
return; return;
} }
tscDebug("%p start to cancel query", res); tscDebug("0x%"PRIx64" start to cancel query", pSql->self);
SSqlCmd *pCmd = &pSql->cmd; SSqlCmd *pCmd = &pSql->cmd;
// set the error code for master pSqlObj firstly // set the error code for master pSqlObj firstly
@ -744,7 +744,7 @@ void taos_stop_query(TAOS_RES *res) {
} }
} }
tscDebug("%p query is cancelled", res); tscDebug("0x%"PRIx64" query is cancelled", pSql->self);
} }
bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) { bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
@ -877,7 +877,7 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
pRes->numOfClauseTotal = 0; pRes->numOfClauseTotal = 0;
tscDebug("%p Valid SQL: %s pObj:%p", pSql, sql, pObj); tscDebug("0x%"PRIx64" Valid SQL: %s pObj:%p", pSql->self, sql, pObj);
int32_t sqlLen = (int32_t)strlen(sql); int32_t sqlLen = (int32_t)strlen(sql);
if (sqlLen > tsMaxSQLStringLen) { if (sqlLen > tsMaxSQLStringLen) {
@ -889,7 +889,7 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
pSql->sqlstr = realloc(pSql->sqlstr, sqlLen + 1); pSql->sqlstr = realloc(pSql->sqlstr, sqlLen + 1);
if (pSql->sqlstr == NULL) { if (pSql->sqlstr == NULL) {
tscError("%p failed to malloc sql string buffer", pSql); tscError("%p failed to malloc sql string buffer", pSql);
tscDebug("%p Valid SQL result:%d, %s pObj:%p", pSql, pRes->code, taos_errstr(pSql), pObj); tscDebug("0x%"PRIx64" Valid SQL result:%d, %s pObj:%p", pSql->self, pRes->code, taos_errstr(pSql), pObj);
tfree(pSql); tfree(pSql);
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
@ -914,7 +914,7 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
} }
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tscDebug("%p Valid SQL result:%d, %s pObj:%p", pSql, code, taos_errstr(pSql), pObj); tscDebug("0x%"PRIx64" Valid SQL result:%d, %s pObj:%p", pSql->self, code, taos_errstr(pSql), pObj);
} }
taos_free_result(pSql); taos_free_result(pSql);
@ -1027,7 +1027,7 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
pRes->numOfClauseTotal = 0; pRes->numOfClauseTotal = 0;
assert(pSql->fp == NULL); assert(pSql->fp == NULL);
tscDebug("%p tableNameList: %s pObj:%p", pSql, tableNameList, pObj); tscDebug("0x%"PRIx64" tableNameList: %s pObj:%p", pSql->self, tableNameList, pObj);
int32_t tblListLen = (int32_t)strlen(tableNameList); int32_t tblListLen = (int32_t)strlen(tableNameList);
if (tblListLen > MAX_TABLE_NAME_LENGTH) { if (tblListLen > MAX_TABLE_NAME_LENGTH) {
@ -1061,7 +1061,7 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
tscDoQuery(pSql); tscDoQuery(pSql);
tscDebug("%p load multi table meta result:%d %s pObj:%p", pSql, pRes->code, taos_errstr(pSql), pObj); tscDebug("0x%"PRIx64" load multi table meta result:%d %s pObj:%p", pSql->self, pRes->code, taos_errstr(pSql), pObj);
if ((code = pRes->code) != TSDB_CODE_SUCCESS) { if ((code = pRes->code) != TSDB_CODE_SUCCESS) {
tscFreeSqlObj(pSql); tscFreeSqlObj(pSql);
} }

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

@ -70,7 +70,7 @@ static void setRetryInfo(SSqlStream* pStream, int32_t code) {
pSql->res.code = code; pSql->res.code = code;
int64_t retryDelayTime = tscGetRetryDelayTime(pStream, pStream->interval.sliding, pStream->precision); int64_t retryDelayTime = tscGetRetryDelayTime(pStream, pStream->interval.sliding, pStream->precision);
tscDebug("%p stream:%p, get table Meta failed, retry in %" PRId64 "ms", pSql, pStream, retryDelayTime); tscDebug("0x%"PRIx64" stream:%p, get table Meta failed, retry in %" PRId64 "ms", pSql->self, pStream, retryDelayTime);
tscSetRetryTimer(pStream, pSql, retryDelayTime); tscSetRetryTimer(pStream, pSql, retryDelayTime);
} }
@ -104,7 +104,7 @@ static void doLaunchQuery(void* param, TAOS_RES* tres, int32_t code) {
// failed to get table Meta or vgroup list, retry in 10sec. // failed to get table Meta or vgroup list, retry in 10sec.
if (code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS) {
tscTansformFuncForSTableQuery(pQueryInfo); tscTansformFuncForSTableQuery(pQueryInfo);
tscDebug("%p stream:%p, start stream query on:%s", pSql, pStream, tNameGetTableName(&pTableMetaInfo->name)); tscDebug("0x%"PRIx64" stream:%p, start stream query on:%s", pSql->self, pStream, tNameGetTableName(&pTableMetaInfo->name));
pSql->fp = tscProcessStreamQueryCallback; pSql->fp = tscProcessStreamQueryCallback;
pSql->fetchFp = tscProcessStreamQueryCallback; pSql->fetchFp = tscProcessStreamQueryCallback;
@ -131,7 +131,7 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
pStream->numOfRes = 0; // reset the numOfRes. pStream->numOfRes = 0; // reset the numOfRes.
SSqlObj *pSql = pStream->pSql; SSqlObj *pSql = pStream->pSql;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0); SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
tscDebug("%p add into timer", pSql); tscDebug("0x%"PRIx64" add into timer", pSql->self);
if (pStream->isProject) { if (pStream->isProject) {
/* /*
@ -237,7 +237,7 @@ static void tscStreamFillTimeGap(SSqlStream* pStream, TSKEY ts) {
} }
if (rowNum > 0) { if (rowNum > 0) {
tscDebug("%p stream:%p %d rows padded", pSql, pStream, rowNum); tscDebug("0x%"PRIx64" stream:%p %d rows padded", pSql, pStream, rowNum);
} }
pRes->numOfRows = 0; pRes->numOfRows = 0;
@ -263,7 +263,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
for(int32_t i = 0; i < numOfRows; ++i) { for(int32_t i = 0; i < numOfRows; ++i) {
TAOS_ROW row = taos_fetch_row(res); TAOS_ROW row = taos_fetch_row(res);
if (row != NULL) { if (row != NULL) {
tscDebug("%p stream:%p fetch result", pSql, pStream); tscDebug("0x%"PRIx64" stream:%p fetch result", pSql->self, pStream);
tscStreamFillTimeGap(pStream, *(TSKEY*)row[0]); tscStreamFillTimeGap(pStream, *(TSKEY*)row[0]);
pStream->stime = *(TSKEY *)row[0]; pStream->stime = *(TSKEY *)row[0];
// user callback function // user callback function
@ -293,7 +293,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
pStream->stime += 1; pStream->stime += 1;
} }
tscDebug("%p stream:%p, query on:%s, fetch result completed, fetched rows:%" PRId64, pSql, pStream, tNameGetTableName(&pTableMetaInfo->name), tscDebug("0x%"PRIx64" stream:%p, query on:%s, fetch result completed, fetched rows:%" PRId64, pSql->self, pStream, tNameGetTableName(&pTableMetaInfo->name),
pStream->numOfRes); pStream->numOfRes);
tfree(pTableMetaInfo->pTableMeta); tfree(pTableMetaInfo->pTableMeta);
@ -318,8 +318,8 @@ static void tscSetRetryTimer(SSqlStream *pStream, SSqlObj *pSql, int64_t timer)
/* /*
* current time window will be closed, since it too early to exceed the maxRetentWindow value * current time window will be closed, since it too early to exceed the maxRetentWindow value
*/ */
tscDebug("%p stream:%p, etime:%" PRId64 " is too old, exceeds the max retention time window:%" PRId64 ", stop the stream", tscDebug("0x%"PRIx64" stream:%p, etime:%" PRId64 " is too old, exceeds the max retention time window:%" PRId64 ", stop the stream",
pStream->pSql, pStream, pStream->stime, pStream->etime); pStream->pSql->self, pStream, pStream->stime, pStream->etime);
// TODO : How to terminate stream here // TODO : How to terminate stream here
if (pStream->callback) { if (pStream->callback) {
// Callback function from upper level // Callback function from upper level
@ -329,10 +329,10 @@ static void tscSetRetryTimer(SSqlStream *pStream, SSqlObj *pSql, int64_t timer)
return; return;
} }
tscDebug("%p stream:%p, next start at %" PRId64 ", in %" PRId64 "ms. delay:%" PRId64 "ms qrange %" PRId64 "-%" PRId64, pStream->pSql, pStream, tscDebug("0x%"PRIx64" stream:%p, next start at %" PRId64 ", in %" PRId64 "ms. delay:%" PRId64 "ms qrange %" PRId64 "-%" PRId64, pStream->pSql->self, pStream,
now + timer, timer, delay, pStream->stime, etime); now + timer, timer, delay, pStream->stime, etime);
} else { } else {
tscDebug("%p stream:%p, next start at %" PRId64 ", in %" PRId64 "ms. delay:%" PRId64 "ms qrange %" PRId64 "-%" PRId64, pStream->pSql, pStream, tscDebug("0x%"PRIx64" stream:%p, next start at %" PRId64 ", in %" PRId64 "ms. delay:%" PRId64 "ms qrange %" PRId64 "-%" PRId64, pStream->pSql->self, pStream,
pStream->stime, timer, delay, pStream->stime - pStream->interval.interval, pStream->stime - 1); pStream->stime, timer, delay, pStream->stime - pStream->interval.interval, pStream->stime - 1);
} }
@ -378,8 +378,8 @@ static void tscSetNextLaunchTimer(SSqlStream *pStream, SSqlObj *pSql) {
*/ */
timer = pStream->interval.sliding; timer = pStream->interval.sliding;
if (pStream->stime > pStream->etime) { if (pStream->stime > pStream->etime) {
tscDebug("%p stream:%p, stime:%" PRId64 " is larger than end time: %" PRId64 ", stop the stream", pStream->pSql, pStream, tscDebug("0x%"PRIx64" stream:%p, stime:%" PRId64 " is larger than end time: %" PRId64 ", stop the stream",
pStream->stime, pStream->etime); pStream->pSql->self, pStream, pStream->stime, pStream->etime);
// TODO : How to terminate stream here // TODO : How to terminate stream here
if (pStream->callback) { if (pStream->callback) {
// Callback function from upper level // Callback function from upper level
@ -392,7 +392,7 @@ static void tscSetNextLaunchTimer(SSqlStream *pStream, SSqlObj *pSql) {
int64_t stime = taosTimeTruncate(pStream->stime - 1, &pStream->interval, pStream->precision); int64_t stime = taosTimeTruncate(pStream->stime - 1, &pStream->interval, pStream->precision);
//int64_t stime = taosGetIntervalStartTimestamp(pStream->stime - 1, pStream->interval.interval, pStream->interval.interval, pStream->interval.intervalUnit, pStream->precision); //int64_t stime = taosGetIntervalStartTimestamp(pStream->stime - 1, pStream->interval.interval, pStream->interval.interval, pStream->interval.intervalUnit, pStream->precision);
if (stime >= pStream->etime) { if (stime >= pStream->etime) {
tscDebug("%p stream:%p, stime:%" PRId64 " is larger than end time: %" PRId64 ", stop the stream", pStream->pSql, pStream, tscDebug("0x%"PRIx64" stream:%p, stime:%" PRId64 " is larger than end time: %" PRId64 ", stop the stream", pStream->pSql->self, pStream,
pStream->stime, pStream->etime); pStream->stime, pStream->etime);
// TODO : How to terminate stream here // TODO : How to terminate stream here
if (pStream->callback) { if (pStream->callback) {
@ -558,7 +558,7 @@ static void tscCreateStream(void *param, TAOS_RES *res, int code) {
taosTmrReset(tscProcessStreamTimer, (int32_t)starttime, pStream, tscTmr, &pStream->pTimer); taosTmrReset(tscProcessStreamTimer, (int32_t)starttime, pStream, tscTmr, &pStream->pTimer);
tscDebug("%p stream:%p is opened, query on:%s, interval:%" PRId64 ", sliding:%" PRId64 ", first launched in:%" PRId64 ", sql:%s", pSql, tscDebug("0x%"PRIx64" stream:%p is opened, query on:%s, interval:%" PRId64 ", sliding:%" PRId64 ", first launched in:%" PRId64 ", sql:%s", pSql->self,
pStream, tNameGetTableName(&pTableMetaInfo->name), pStream->interval.interval, pStream->interval.sliding, starttime, pSql->sqlstr); pStream, tNameGetTableName(&pTableMetaInfo->name), pStream->interval.interval, pStream->interval.sliding, starttime, pSql->sqlstr);
} }
@ -645,7 +645,7 @@ void taos_close_stream(TAOS_STREAM *handle) {
taosTmrStopA(&(pStream->pTimer)); taosTmrStopA(&(pStream->pTimer));
tscDebug("%p stream:%p is closed", pSql, pStream); tscDebug("0x%"PRIx64" stream:%p is closed", pSql->self, pStream);
// notify CQ to release the pStream object // notify CQ to release the pStream object
pStream->fp(pStream->param, NULL, NULL); pStream->fp(pStream->param, NULL, NULL);
pStream->pSql = NULL; pStream->pSql = NULL;

151
src/client/src/tscSubquery.c
View File

@ -81,14 +81,14 @@ static bool allSubqueryDone(SSqlObj *pParentSql) {
SSubqueryState *subState = &pParentSql->subState; SSubqueryState *subState = &pParentSql->subState;
//lock in caller //lock in caller
tscDebug("%p total subqueries: %d", pParentSql, subState->numOfSub); tscDebug("0x%"PRIx64" total subqueries: %d", pParentSql->self, subState->numOfSub);
for (int i = 0; i < subState->numOfSub; i++) { for (int i = 0; i < subState->numOfSub; i++) {
if (0 == subState->states[i]) { if (0 == subState->states[i]) {
tscDebug("%p subquery:%p, index: %d NOT finished, abort query completion check", pParentSql, pParentSql->pSubs[i], i); tscDebug("0x%"PRIx64" subquery:%p, index: %d NOT finished, abort query completion check", pParentSql->self, pParentSql->pSubs[i], i);
done = false; done = false;
break; break;
} else { } else {
tscDebug("%p subquery:%p, index: %d finished", pParentSql, pParentSql->pSubs[i], i); tscDebug("0x%"PRIx64" subquery:%p, index: %d finished", pParentSql->self, pParentSql->pSubs[i], i);
} }
} }
@ -105,14 +105,14 @@ static bool subAndCheckDone(SSqlObj *pSql, SSqlObj *pParentSql, int idx) {
bool done = allSubqueryDone(pParentSql); bool done = allSubqueryDone(pParentSql);
if (done) { if (done) {
tscDebug("%p subquery:%p,%d all subs already done", pParentSql, pSql, idx); tscDebug("0x%"PRIx64" subquery:%p,%d all subs already done", pParentSql->self, pSql, idx);
pthread_mutex_unlock(&subState->mutex); pthread_mutex_unlock(&subState->mutex);
return false; return false;
} }
tscDebug("%p subquery:%p,%d state set to 1", pParentSql, pSql, idx); tscDebug("0x%"PRIx64" subquery:%p,%d state set to 1", pParentSql->self, pSql, idx);
subState->states[idx] = 1; subState->states[idx] = 1;
@ -158,18 +158,18 @@ static int64_t doTSBlockIntersect(SSqlObj* pSql, STimeWindow * win) {
SJoinSupporter* pSupporter = pSql->pSubs[i]->param; SJoinSupporter* pSupporter = pSql->pSubs[i]->param;
if (pSupporter->pTSBuf == NULL) { if (pSupporter->pTSBuf == NULL) {
tscDebug("%p at least one ts-comp is empty, 0 for secondary query after ts blocks intersecting", pSql); tscDebug("0x%"PRIx64" at least one ts-comp is empty, 0 for secondary query after ts blocks intersecting", pSql->self);
return 0; return 0;
} }
tsBufResetPos(pSupporter->pTSBuf); tsBufResetPos(pSupporter->pTSBuf);
if (!tsBufNextPos(pSupporter->pTSBuf)) { if (!tsBufNextPos(pSupporter->pTSBuf)) {
tscDebug("%p input1 is empty, 0 for secondary query after ts blocks intersecting", pSql); tscDebug("0x%"PRIx64" input1 is empty, 0 for secondary query after ts blocks intersecting", pSql->self);
return 0; return 0;
} }
tscDebug("%p sub:%p table idx:%d, input group number:%d", pSql, pSql->pSubs[i], i, pSupporter->pTSBuf->numOfGroups); tscDebug("0x%"PRIx64" sub:%p table idx:%d, input group number:%d", pSql->self, pSql->pSubs[i], i, pSupporter->pTSBuf->numOfGroups);
ctxlist[i].p = pSupporter; ctxlist[i].p = pSupporter;
ctxlist[i].res = output; ctxlist[i].res = output;
@ -375,9 +375,9 @@ static int64_t doTSBlockIntersect(SSqlObj* pSql, STimeWindow * win) {
TSKEY et = taosGetTimestampUs(); TSKEY et = taosGetTimestampUs();
for (int32_t i = 0; i < joinNum; ++i) { for (int32_t i = 0; i < joinNum; ++i) {
tscDebug("%p sub:%p tblidx:%d, input:%" PRId64 ", final:%" PRId64 " in %d vnodes for secondary query after ts blocks " tscDebug("0x%"PRIx64" sub:%p tblidx:%d, input:%" PRId64 ", final:%" PRId64 " in %d vnodes for secondary query after ts blocks "
"intersecting, skey:%" PRId64 ", ekey:%" PRId64 ", numOfVnode:%d, elapsed time:%" PRId64 " us", "intersecting, skey:%" PRId64 ", ekey:%" PRId64 ", numOfVnode:%d, elapsed time:%" PRId64 " us",
pSql, pSql->pSubs[i], i, ctxlist[i].numOfInput, ctxlist[i].res->numOfTotal, ctxlist[i].res->numOfGroups, win->skey, win->ekey, pSql->self, pSql->pSubs[i], i, ctxlist[i].numOfInput, ctxlist[i].res->numOfTotal, ctxlist[i].res->numOfGroups, win->skey, win->ekey,
tsBufGetNumOfGroup(ctxlist[i].res), et - st); tsBufGetNumOfGroup(ctxlist[i].res), et - st);
} }
@ -549,7 +549,7 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) {
assert(numOfSub > 0); assert(numOfSub > 0);
// scan all subquery, if one sub query has only ts, ignore it // scan all subquery, if one sub query has only ts, ignore it
tscDebug("%p start to launch secondary subqueries, %d out of %d needs to query", pSql, numOfSub, pSql->subState.numOfSub); tscDebug("0x%"PRIx64" start to launch secondary subqueries, %d out of %d needs to query", pSql->self, numOfSub, pSql->subState.numOfSub);
bool success = true; bool success = true;
@ -560,7 +560,7 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) {
pSupporter = pPrevSub->param; pSupporter = pPrevSub->param;
if (taosArrayGetSize(pSupporter->exprList) == 0) { if (taosArrayGetSize(pSupporter->exprList) == 0) {
tscDebug("%p subIndex: %d, no need to launch query, ignore it", pSql, i); tscDebug("0x%"PRIx64" subIndex: %d, no need to launch query, ignore it", pSql->self, i);
tscDestroyJoinSupporter(pSupporter); tscDestroyJoinSupporter(pSupporter);
taos_free_result(pPrevSub); taos_free_result(pPrevSub);
@ -665,8 +665,8 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) {
subquerySetState(pPrevSub, &pSql->subState, i, 0); subquerySetState(pPrevSub, &pSql->subState, i, 0);
size_t numOfCols = taosArrayGetSize(pQueryInfo->colList); size_t numOfCols = taosArrayGetSize(pQueryInfo->colList);
tscDebug("%p subquery:%p tableIndex:%d, vgroupIndex:%d, type:%d, exprInfo:%" PRIzu ", colList:%" PRIzu ", fieldsInfo:%d, name:%s", tscDebug("0x%"PRIx64" subquery:%p tableIndex:%d, vgroupIndex:%d, type:%d, exprInfo:%" PRIzu ", colList:%" PRIzu ", fieldsInfo:%d, name:%s",
pSql, pNew, 0, pTableMetaInfo->vgroupIndex, pQueryInfo->type, taosArrayGetSize(pQueryInfo->exprList), pSql->self, pNew, 0, pTableMetaInfo->vgroupIndex, pQueryInfo->type, taosArrayGetSize(pQueryInfo->exprList),
numOfCols, pQueryInfo->fieldsInfo.numOfOutput, tNameGetTableName(&pTableMetaInfo->name)); numOfCols, pQueryInfo->fieldsInfo.numOfOutput, tNameGetTableName(&pTableMetaInfo->name));
} }
@ -792,7 +792,7 @@ void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArr
if (taosArrayGetSize(result) > 0) { if (taosArrayGetSize(result) > 0) {
SVgroupTableInfo* prevGroup = taosArrayGet(result, taosArrayGetSize(result) - 1); SVgroupTableInfo* prevGroup = taosArrayGet(result, taosArrayGetSize(result) - 1);
tscDebug("%p vgId:%d, tables:%"PRIzu, pSql, prevGroup->vgInfo.vgId, taosArrayGetSize(prevGroup->itemList)); tscDebug("0x%"PRIx64" vgId:%d, tables:%"PRIzu, pSql->self, prevGroup->vgInfo.vgId, taosArrayGetSize(prevGroup->itemList));
} }
taosArrayPush(result, &info); taosArrayPush(result, &info);
@ -814,7 +814,7 @@ void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArr
pTableMetaInfo->pVgroupTables = result; pTableMetaInfo->pVgroupTables = result;
SVgroupTableInfo* g = taosArrayGet(result, taosArrayGetSize(result) - 1); SVgroupTableInfo* g = taosArrayGet(result, taosArrayGetSize(result) - 1);
tscDebug("%p vgId:%d, tables:%"PRIzu, pSql, g->vgInfo.vgId, taosArrayGetSize(g->itemList)); tscDebug("0x%"PRIx64" vgId:%d, tables:%"PRIzu, pSql->self, g->vgInfo.vgId, taosArrayGetSize(g->itemList));
} }
} }
@ -903,7 +903,7 @@ static int32_t getIntersectionOfTableTuple(SQueryInfo* pQueryInfo, SSqlObj* pPar
SSchema* pColSchema = tscGetColumnSchemaById(pTableMetaInfo->pTableMeta, tagColId); SSchema* pColSchema = tscGetColumnSchemaById(pTableMetaInfo->pTableMeta, tagColId);
tscDebug("%p all subquery retrieve <tid, tags> complete, do tags match", pParentSql); tscDebug("0x%"PRIx64" all subquery retrieve <tid, tags> complete, do tags match", pParentSql->self);
for (int32_t i = 0; i < joinNum; i++) { for (int32_t i = 0; i < joinNum; i++) {
SJoinSupporter* p = pParentSql->pSubs[i]->param; SJoinSupporter* p = pParentSql->pSubs[i]->param;
@ -1004,7 +1004,7 @@ static int32_t getIntersectionOfTableTuple(SQueryInfo* pQueryInfo, SSqlObj* pPar
continue; continue;
} }
tscDebug("%p tag matched, vgId:%d, val:%d, tid:%d, uid:%"PRIu64", tid:%d, uid:%"PRIu64, pParentSql, prev->vgId, tscDebug("0x%"PRIx64" tag matched, vgId:%d, val:%d, tid:%d, uid:%"PRIu64", tid:%d, uid:%"PRIu64, pParentSql->self, prev->vgId,
*(int*) prev->tag, prev->tid, prev->uid, cur->tid, cur->uid); *(int*) prev->tag, prev->tid, prev->uid, cur->tid, cur->uid);
assert(stackidx == tableNum); assert(stackidx == tableNum);
@ -1079,7 +1079,7 @@ static int32_t getIntersectionOfTableTuple(SQueryInfo* pQueryInfo, SSqlObj* pPar
taosArrayPush(resList, &ctxlist[i].res); taosArrayPush(resList, &ctxlist[i].res);
tscDebug("%p tags match complete, result num: %"PRIzu, pParentSql, num); tscDebug("0x%"PRIx64" tags match complete, result num: %"PRIzu, pParentSql->self, num);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1183,8 +1183,8 @@ static void tidTagRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
pTableMetaInfo->vgroupIndex += 1; pTableMetaInfo->vgroupIndex += 1;
assert(pTableMetaInfo->vgroupIndex < totalVgroups); assert(pTableMetaInfo->vgroupIndex < totalVgroups);
tscDebug("%p tid_tag from vgroup index:%d completed, try next vgroup:%d. total vgroups:%d. current numOfRes:%d", tscDebug("0x%"PRIx64" tid_tag from vgroup index:%d completed, try next vgroup:%d. total vgroups:%d. current numOfRes:%d",
pSql, pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, pSupporter->num); pSql->self, pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, pSupporter->num);
pCmd->command = TSDB_SQL_SELECT; pCmd->command = TSDB_SQL_SELECT;
tscResetForNextRetrieve(&pSql->res); tscResetForNextRetrieve(&pSql->res);
@ -1198,7 +1198,7 @@ static void tidTagRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
// no data exists in next vnode, mark the <tid, tags> query completed // no data exists in next vnode, mark the <tid, tags> query completed
// only when there is no subquery exits any more, proceeds to get the intersect of the <tid, tags> tuple sets. // only when there is no subquery exits any more, proceeds to get the intersect of the <tid, tags> tuple sets.
if (!subAndCheckDone(pSql, pParentSql, pSupporter->subqueryIndex)) { if (!subAndCheckDone(pSql, pParentSql, pSupporter->subqueryIndex)) {
tscDebug("%p tagRetrieve:%p,%d completed, total:%d", pParentSql, tres, pSupporter->subqueryIndex, pParentSql->subState.numOfSub); tscDebug("0x%"PRIx64" tagRetrieve:%p,%d completed, total:%d", pParentSql->self, tres, pSupporter->subqueryIndex, pParentSql->subState.numOfSub);
return; return;
} }
@ -1217,7 +1217,8 @@ static void tidTagRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
if (emptyTagList(resList, pParentSql->subState.numOfSub)) { // no results,return. if (emptyTagList(resList, pParentSql->subState.numOfSub)) { // no results,return.
assert(pParentSql->fp != tscJoinQueryCallback); assert(pParentSql->fp != tscJoinQueryCallback);
tscDebug("%p tag intersect does not generated qualified tables for join, free all sub SqlObj and quit", pParentSql); tscDebug("0x%"PRIx64" tag intersect does not generated qualified tables for join, free all sub SqlObj and quit",
pParentSql->self);
freeJoinSubqueryObj(pParentSql); freeJoinSubqueryObj(pParentSql);
// set no result command // set no result command
@ -1239,7 +1240,7 @@ static void tidTagRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
((SJoinSupporter*)psub->param)->pVgroupTables = tscVgroupTableInfoDup(pTableMetaInfo->pVgroupTables); ((SJoinSupporter*)psub->param)->pVgroupTables = tscVgroupTableInfoDup(pTableMetaInfo->pVgroupTables);
memset(pParentSql->subState.states, 0, sizeof(pParentSql->subState.states[0]) * pParentSql->subState.numOfSub); memset(pParentSql->subState.states, 0, sizeof(pParentSql->subState.states[0]) * pParentSql->subState.numOfSub);
tscDebug("%p reset all sub states to 0", pParentSql); tscDebug("0x%"PRIx64" reset all sub states to 0", pParentSql->self);
issueTsCompQuery(psub, psub->param, pParentSql); issueTsCompQuery(psub, psub->param, pParentSql);
} }
@ -1332,7 +1333,7 @@ static void tsCompRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
} }
if (pSupporter->pTSBuf == NULL) { if (pSupporter->pTSBuf == NULL) {
tscDebug("%p create tmp file for ts block:%s, size:%d bytes", pSql, pBuf->path, numOfRows); tscDebug("0x%"PRIx64" create tmp file for ts block:%s, size:%d bytes", pSql->self, pBuf->path, numOfRows);
pSupporter->pTSBuf = pBuf; pSupporter->pTSBuf = pBuf;
} else { } else {
assert(pQueryInfo->numOfTables == 1); // for subquery, only one assert(pQueryInfo->numOfTables == 1); // for subquery, only one
@ -1358,8 +1359,8 @@ static void tsCompRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
pTableMetaInfo->vgroupIndex += 1; pTableMetaInfo->vgroupIndex += 1;
assert(pTableMetaInfo->vgroupIndex < totalVgroups); assert(pTableMetaInfo->vgroupIndex < totalVgroups);
tscDebug("%p results from vgroup index:%d completed, try next vgroup:%d. total vgroups:%d. current numOfRes:%" PRId64, tscDebug("0x%"PRIx64" results from vgroup index:%d completed, try next vgroup:%d. total vgroups:%d. current numOfRes:%" PRId64,
pSql, pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, pSql->self, pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups,
pRes->numOfClauseTotal); pRes->numOfClauseTotal);
pCmd->command = TSDB_SQL_SELECT; pCmd->command = TSDB_SQL_SELECT;
@ -1382,12 +1383,12 @@ static void tsCompRetrieveCallback(void* param, TAOS_RES* tres, int32_t numOfRow
return; return;
} }
tscDebug("%p all subquery retrieve ts complete, do ts block intersect", pParentSql); tscDebug("0x%"PRIx64" all subquery retrieve ts complete, do ts block intersect", pParentSql->self);
STimeWindow win = TSWINDOW_INITIALIZER; STimeWindow win = TSWINDOW_INITIALIZER;
int64_t num = doTSBlockIntersect(pParentSql, &win); int64_t num = doTSBlockIntersect(pParentSql, &win);
if (num <= 0) { // no result during ts intersect if (num <= 0) { // no result during ts intersect
tscDebug("%p no results generated in ts intersection, free all sub SqlObj and quit", pParentSql); tscDebug("0x%"PRIx64" no results generated in ts intersection, free all sub SqlObj and quit", pParentSql->self);
freeJoinSubqueryObj(pParentSql); freeJoinSubqueryObj(pParentSql);
// set no result command // set no result command
@ -1455,23 +1456,23 @@ static void joinRetrieveFinalResCallback(void* param, TAOS_RES* tres, int numOfR
} }
if ((++pTableMetaInfo->vgroupIndex) < numOfVgroups) { if ((++pTableMetaInfo->vgroupIndex) < numOfVgroups) {
tscDebug("%p no result in current vnode anymore, try next vnode, vgIndex:%d", pSql, pTableMetaInfo->vgroupIndex); tscDebug("0x%"PRIx64" no result in current vnode anymore, try next vnode, vgIndex:%d", pSql->self, pTableMetaInfo->vgroupIndex);
pSql->cmd.command = TSDB_SQL_SELECT; pSql->cmd.command = TSDB_SQL_SELECT;
pSql->fp = tscJoinQueryCallback; pSql->fp = tscJoinQueryCallback;
tscProcessSql(pSql); tscProcessSql(pSql);
return; return;
} else { } else {
tscDebug("%p no result in current subquery anymore", pSql); tscDebug("0x%"PRIx64" no result in current subquery anymore", pSql->self);
} }
} }
if (!subAndCheckDone(pSql, pParentSql, pSupporter->subqueryIndex)) { if (!subAndCheckDone(pSql, pParentSql, pSupporter->subqueryIndex)) {
tscDebug("%p sub:%p,%d completed, total:%d", pParentSql, tres, pSupporter->subqueryIndex, pState->numOfSub); tscDebug("0x%"PRIx64" sub:%p,%d completed, total:%d", pParentSql->self, tres, pSupporter->subqueryIndex, pState->numOfSub);
return; return;
} }
tscDebug("%p all %d secondary subqueries retrieval completed, code:%d", tres, pState->numOfSub, pParentSql->res.code); tscDebug("0x%"PRIx64" all %d secondary subqueries retrieval completed, code:%d", pSql->self, pState->numOfSub, pParentSql->res.code);
if (pParentSql->res.code != TSDB_CODE_SUCCESS) { if (pParentSql->res.code != TSDB_CODE_SUCCESS) {
freeJoinSubqueryObj(pParentSql); freeJoinSubqueryObj(pParentSql);
@ -1482,14 +1483,14 @@ static void joinRetrieveFinalResCallback(void* param, TAOS_RES* tres, int numOfR
bool stableQuery = tscIsTwoStageSTableQuery(pQueryInfo, 0); bool stableQuery = tscIsTwoStageSTableQuery(pQueryInfo, 0);
for (int32_t i = 0; i < pState->numOfSub; ++i) { for (int32_t i = 0; i < pState->numOfSub; ++i) {
if (pParentSql->pSubs[i] == NULL) { if (pParentSql->pSubs[i] == NULL) {
tscDebug("%p %p sub:%d not retrieve data", pParentSql, NULL, i); tscDebug("0x%"PRIx64" %p sub:%d not retrieve data", pParentSql->self, NULL, i);
continue; continue;
} }
SSqlRes* pRes1 = &pParentSql->pSubs[i]->res; SSqlRes* pRes1 = &pParentSql->pSubs[i]->res;
if (pRes1->row > 0 && pRes1->numOfRows > 0) { if (pRes1->row > 0 && pRes1->numOfRows > 0) {
tscDebug("%p sub:%p index:%d numOfRows:%d total:%"PRId64 " (not retrieve)", pParentSql, pParentSql->pSubs[i], i, tscDebug("0x%"PRIx64" sub:%p index:%d numOfRows:%d total:%"PRId64 " (not retrieve)", pParentSql->self, pParentSql->pSubs[i], i,
pRes1->numOfRows, pRes1->numOfTotal); pRes1->numOfRows, pRes1->numOfTotal);
assert(pRes1->row < pRes1->numOfRows); assert(pRes1->row < pRes1->numOfRows);
} else { } else {
@ -1497,7 +1498,7 @@ static void joinRetrieveFinalResCallback(void* param, TAOS_RES* tres, int numOfR
pRes1->numOfClauseTotal += pRes1->numOfRows; pRes1->numOfClauseTotal += pRes1->numOfRows;
} }
tscDebug("%p sub:%p index:%d numOfRows:%d total:%"PRId64, pParentSql, pParentSql->pSubs[i], i, tscDebug("0x%"PRIx64" sub:%p index:%d numOfRows:%d total:%"PRId64, pParentSql->self, pParentSql->pSubs[i], i,
pRes1->numOfRows, pRes1->numOfTotal); pRes1->numOfRows, pRes1->numOfTotal);
} }
} }
@ -1615,7 +1616,7 @@ void tscFetchDatablockForSubquery(SSqlObj* pSql) {
} }
if ((++pTableMetaInfo->vgroupIndex) < numOfVgroups) { if ((++pTableMetaInfo->vgroupIndex) < numOfVgroups) {
tscDebug("%p no result in current vnode anymore, try next vnode, vgIndex:%d", pSub, tscDebug("0x%"PRIx64" no result in current vnode anymore, try next vnode, vgIndex:%d", pSub->self,
pTableMetaInfo->vgroupIndex); pTableMetaInfo->vgroupIndex);
pSub->cmd.command = TSDB_SQL_SELECT; pSub->cmd.command = TSDB_SQL_SELECT;
pSub->fp = tscJoinQueryCallback; pSub->fp = tscJoinQueryCallback;
@ -1623,7 +1624,7 @@ void tscFetchDatablockForSubquery(SSqlObj* pSql) {
tscProcessSql(pSub); tscProcessSql(pSub);
tryNextVnode = true; tryNextVnode = true;
} else { } else {
tscDebug("%p no result in current subquery anymore", pSub); tscDebug("0x%"PRIx64" no result in current subquery anymore", pSub->self);
} }
} }
} }
@ -1646,7 +1647,7 @@ void tscFetchDatablockForSubquery(SSqlObj* pSql) {
// TODO multi-vnode retrieve for projection query with limitation has bugs, since the global limiation is not handled // TODO multi-vnode retrieve for projection query with limitation has bugs, since the global limiation is not handled
// retrieve data from current vnode. // retrieve data from current vnode.
tscDebug("%p retrieve data from %d subqueries", pSql, numOfFetch); tscDebug("0x%"PRIx64" retrieve data from %d subqueries", pSql->self, numOfFetch);
SJoinSupporter* pSupporter = NULL; SJoinSupporter* pSupporter = NULL;
for (int32_t i = 0; i < pSql->subState.numOfSub; ++i) { for (int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
@ -1680,7 +1681,7 @@ void tscFetchDatablockForSubquery(SSqlObj* pSql) {
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
if (pRes1->row >= pRes1->numOfRows) { if (pRes1->row >= pRes1->numOfRows) {
tscDebug("%p subquery:%p retrieve data from vnode, subquery:%d, vgroupIndex:%d", pSql, pSql1, tscDebug("0x%"PRIx64" subquery:%p retrieve data from vnode, subquery:%d, vgroupIndex:%d", pSql->self, pSql1,
pSupporter->subqueryIndex, pTableMetaInfo->vgroupIndex); pSupporter->subqueryIndex, pTableMetaInfo->vgroupIndex);
tscResetForNextRetrieve(pRes1); tscResetForNextRetrieve(pRes1);
@ -2000,11 +2001,11 @@ void tscHandleMasterJoinQuery(SSqlObj* pSql) {
} }
memset(pSql->subState.states, 0, sizeof(*pSql->subState.states) * pSql->subState.numOfSub); memset(pSql->subState.states, 0, sizeof(*pSql->subState.states) * pSql->subState.numOfSub);
tscDebug("%p reset all sub states to 0", pSql); tscDebug("0x%"PRIx64" reset all sub states to 0", pSql->self);
bool hasEmptySub = false; bool hasEmptySub = false;
tscDebug("%p start subquery, total:%d", pSql, pQueryInfo->numOfTables); tscDebug("0x%"PRIx64" start subquery, total:%d", pSql->self, pQueryInfo->numOfTables);
for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) { for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) {
SJoinSupporter *pSupporter = tscCreateJoinSupporter(pSql, i); SJoinSupporter *pSupporter = tscCreateJoinSupporter(pSql, i);
@ -2465,7 +2466,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
return ret; return ret;
} }
tscDebug("%p retrieved query data from %d vnode(s)", pSql, pState->numOfSub); tscDebug("0x%"PRIx64" retrieved query data from %d vnode(s)", pSql->self, pState->numOfSub);
pSql->pSubs = calloc(pState->numOfSub, POINTER_BYTES); pSql->pSubs = calloc(pState->numOfSub, POINTER_BYTES);
if (pSql->pSubs == NULL) { if (pSql->pSubs == NULL) {
tfree(pSql->pSubs); tfree(pSql->pSubs);
@ -2489,7 +2490,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
} }
memset(pState->states, 0, sizeof(*pState->states) * pState->numOfSub); memset(pState->states, 0, sizeof(*pState->states) * pState->numOfSub);
tscDebug("%p reset all sub states to 0", pSql); tscDebug("0x%"PRIx64" reset all sub states to 0", pSql->self);
pRes->code = TSDB_CODE_SUCCESS; pRes->code = TSDB_CODE_SUCCESS;
@ -2531,7 +2532,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
assert(pNewQueryInfo->tsBuf != NULL); assert(pNewQueryInfo->tsBuf != NULL);
} }
tscDebug("%p sub:%p create subquery success. orderOfSub:%d", pSql, pNew, trs->subqueryIndex); tscDebug("0x%"PRIx64" sub:%p create subquery success. orderOfSub:%d", pSql->self, pNew, trs->subqueryIndex);
} }
if (i < pState->numOfSub) { if (i < pState->numOfSub) {
@ -2553,7 +2554,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
SSqlObj* pSub = pSql->pSubs[j]; SSqlObj* pSub = pSql->pSubs[j];
SRetrieveSupport* pSupport = pSub->param; SRetrieveSupport* pSupport = pSub->param;
tscDebug("%p sub:%p launch subquery, orderOfSub:%d.", pSql, pSub, pSupport->subqueryIndex); tscDebug("0x%"PRIx64" sub:%p launch subquery, orderOfSub:%d.", pSql->self, pSub, pSupport->subqueryIndex);
tscProcessSql(pSub); tscProcessSql(pSub);
} }
@ -2565,11 +2566,11 @@ static void tscFreeRetrieveSup(SSqlObj *pSql) {
void* p = atomic_val_compare_exchange_ptr(&pSql->param, trsupport, 0); void* p = atomic_val_compare_exchange_ptr(&pSql->param, trsupport, 0);
if (p == NULL) { if (p == NULL) {
tscDebug("%p retrieve supp already released", pSql); tscDebug("0x%"PRIx64" retrieve supp already released", pSql->self);
return; return;
} }
tscDebug("%p start to free subquery supp obj:%p", pSql, trsupport); tscDebug("0x%"PRIx64" start to free subquery supp obj:%p", pSql->self, trsupport);
tfree(trsupport->localBuffer); tfree(trsupport->localBuffer);
tfree(trsupport); tfree(trsupport);
} }
@ -2673,12 +2674,12 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO
*/ */
pSql->res.numOfRows = 0; pSql->res.numOfRows = 0;
trsupport->numOfRetry = MAX_NUM_OF_SUBQUERY_RETRY; // disable retry efforts trsupport->numOfRetry = MAX_NUM_OF_SUBQUERY_RETRY; // disable retry efforts
tscDebug("%p query is cancelled, sub:%p, orderOfSub:%d abort retrieve, code:%s", pParentSql, pSql, tscDebug("0x%"PRIx64" query is cancelled, sub:%p, orderOfSub:%d abort retrieve, code:%s", pParentSql->self, pSql,
subqueryIndex, tstrerror(pParentSql->res.code)); subqueryIndex, tstrerror(pParentSql->res.code));
} }
if (numOfRows >= 0) { // current query is successful, but other sub query failed, still abort current query. if (numOfRows >= 0) { // current query is successful, but other sub query failed, still abort current query.
tscDebug("%p sub:%p retrieve numOfRows:%d,orderOfSub:%d", pParentSql, pSql, numOfRows, subqueryIndex); tscDebug("0x%"PRIx64" sub:0x%"PRIx64" retrieve numOfRows:%d,orderOfSub:%d", pParentSql->self, pSql->self, numOfRows, subqueryIndex);
tscError("%p sub:%p abort further retrieval due to other queries failure,orderOfSub:%d,code:%s", pParentSql, pSql, tscError("%p sub:%p abort further retrieval due to other queries failure,orderOfSub:%d,code:%s", pParentSql, pSql,
subqueryIndex, tstrerror(pParentSql->res.code)); subqueryIndex, tstrerror(pParentSql->res.code));
} else { } else {
@ -2697,7 +2698,7 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO
} }
if (!subAndCheckDone(pSql, pParentSql, subqueryIndex)) { if (!subAndCheckDone(pSql, pParentSql, subqueryIndex)) {
tscDebug("%p sub:%p,%d freed, not finished, total:%d", pParentSql, pSql, trsupport->subqueryIndex, pState->numOfSub); tscDebug("0x%"PRIx64" sub:%p,%d freed, not finished, total:%d", pParentSql->self, pSql, trsupport->subqueryIndex, pState->numOfSub);
tscFreeRetrieveSup(pSql); tscFreeRetrieveSup(pSql);
return; return;
@ -2738,7 +2739,7 @@ static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* p
// data in from current vnode is stored in cache and disk // data in from current vnode is stored in cache and disk
uint32_t numOfRowsFromSubquery = (uint32_t)(trsupport->pExtMemBuffer[idx]->numOfTotalElems + trsupport->localBuffer->num); uint32_t numOfRowsFromSubquery = (uint32_t)(trsupport->pExtMemBuffer[idx]->numOfTotalElems + trsupport->localBuffer->num);
SVgroupsInfo* vgroupsInfo = pTableMetaInfo->vgroupList; SVgroupsInfo* vgroupsInfo = pTableMetaInfo->vgroupList;
tscDebug("%p sub:%p all data retrieved from ep:%s, vgId:%d, numOfRows:%d, orderOfSub:%d", pParentSql, pSql, tscDebug("0x%"PRIx64" sub:%p all data retrieved from ep:%s, vgId:%d, numOfRows:%d, orderOfSub:%d", pParentSql->self, pSql,
vgroupsInfo->vgroups[0].epAddr[0].fqdn, vgroupsInfo->vgroups[0].vgId, numOfRowsFromSubquery, idx); vgroupsInfo->vgroups[0].epAddr[0].fqdn, vgroupsInfo->vgroups[0].vgId, numOfRowsFromSubquery, idx);
tColModelCompact(pDesc->pColumnModel, trsupport->localBuffer, pDesc->pColumnModel->capacity); tColModelCompact(pDesc->pColumnModel, trsupport->localBuffer, pDesc->pColumnModel->capacity);
@ -2767,7 +2768,7 @@ static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* p
} }
if (!subAndCheckDone(pSql, pParentSql, idx)) { if (!subAndCheckDone(pSql, pParentSql, idx)) {
tscDebug("%p sub:%p orderOfSub:%d freed, not finished", pParentSql, pSql, trsupport->subqueryIndex); tscDebug("0x%"PRIx64" sub:%p orderOfSub:%d freed, not finished", pParentSql->self, pSql, trsupport->subqueryIndex);
tscFreeRetrieveSup(pSql); tscFreeRetrieveSup(pSql);
return; return;
@ -2776,14 +2777,14 @@ static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* p
// all sub-queries are returned, start to local merge process // all sub-queries are returned, start to local merge process
pDesc->pColumnModel->capacity = trsupport->pExtMemBuffer[idx]->numOfElemsPerPage; pDesc->pColumnModel->capacity = trsupport->pExtMemBuffer[idx]->numOfElemsPerPage;
tscDebug("%p retrieve from %d vnodes completed.final NumOfRows:%" PRId64 ",start to build loser tree", pParentSql, tscDebug("0x%"PRIx64" retrieve from %d vnodes completed.final NumOfRows:%" PRId64 ",start to build loser tree",
pState->numOfSub, pState->numOfRetrievedRows); pParentSql->self, pState->numOfSub, pState->numOfRetrievedRows);
SQueryInfo *pPQueryInfo = tscGetQueryInfoDetail(&pParentSql->cmd, 0); SQueryInfo *pPQueryInfo = tscGetQueryInfoDetail(&pParentSql->cmd, 0);
tscClearInterpInfo(pPQueryInfo); tscClearInterpInfo(pPQueryInfo);
tscCreateLocalMerger(trsupport->pExtMemBuffer, pState->numOfSub, pDesc, trsupport->pFinalColModel, trsupport->pFFColModel, pParentSql); tscCreateLocalMerger(trsupport->pExtMemBuffer, pState->numOfSub, pDesc, trsupport->pFinalColModel, trsupport->pFFColModel, pParentSql);
tscDebug("%p build loser tree completed", pParentSql); tscDebug("0x%"PRIx64" build loser tree completed", pParentSql->self);
pParentSql->res.precision = pSql->res.precision; pParentSql->res.precision = pSql->res.precision;
pParentSql->res.numOfRows = 0; pParentSql->res.numOfRows = 0;
@ -2807,7 +2808,7 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
// this query has been freed already // this query has been freed already
SRetrieveSupport *trsupport = (SRetrieveSupport *)param; SRetrieveSupport *trsupport = (SRetrieveSupport *)param;
if (pSql->param == NULL || param == NULL) { if (pSql->param == NULL || param == NULL) {
tscDebug("%p already freed in dnodecallback", pSql); tscDebug("0x%"PRIx64" already freed in dnodecallback", pSql->self);
return; return;
} }
@ -2822,8 +2823,8 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
if (pParentSql->res.code != TSDB_CODE_SUCCESS) { if (pParentSql->res.code != TSDB_CODE_SUCCESS) {
trsupport->numOfRetry = MAX_NUM_OF_SUBQUERY_RETRY; trsupport->numOfRetry = MAX_NUM_OF_SUBQUERY_RETRY;
tscDebug("%p query cancelled/failed, sub:%p, vgId:%d, orderOfSub:%d, code:%s, global code:%s", tscDebug("0x%"PRIx64" query cancelled/failed, sub:%p, vgId:%d, orderOfSub:%d, code:%s, global code:%s",
pParentSql, pSql, pVgroup->vgId, trsupport->subqueryIndex, tstrerror(numOfRows), tstrerror(pParentSql->res.code)); pParentSql->self, pSql, pVgroup->vgId, trsupport->subqueryIndex, tstrerror(numOfRows), tstrerror(pParentSql->res.code));
tscHandleSubqueryError(param, tres, numOfRows); tscHandleSubqueryError(param, tres, numOfRows);
return; return;
@ -2846,7 +2847,7 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
return; return;
} }
} else { } else {
tscDebug("%p sub:%p reach the max retry times, set global code:%s", pParentSql, pSql, tstrerror(numOfRows)); tscDebug("0x%"PRIx64" sub:%p reach the max retry times, set global code:%s", pParentSql->self, pSql, tstrerror(numOfRows));
atomic_val_compare_exchange_32(&pParentSql->res.code, TSDB_CODE_SUCCESS, numOfRows); // set global code and abort atomic_val_compare_exchange_32(&pParentSql->res.code, TSDB_CODE_SUCCESS, numOfRows); // set global code and abort
} }
@ -2861,8 +2862,8 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
assert(pRes->numOfRows == numOfRows); assert(pRes->numOfRows == numOfRows);
int64_t num = atomic_add_fetch_64(&pState->numOfRetrievedRows, numOfRows); int64_t num = atomic_add_fetch_64(&pState->numOfRetrievedRows, numOfRows);
tscDebug("%p sub:%p retrieve numOfRows:%d totalNumOfRows:%" PRIu64 " from ep:%s, orderOfSub:%d", pParentSql, pSql, tscDebug("0x%"PRIx64" sub:%p retrieve numOfRows:%d totalNumOfRows:%" PRIu64 " from ep:%s, orderOfSub:%d",
pRes->numOfRows, pState->numOfRetrievedRows, pSql->epSet.fqdn[pSql->epSet.inUse], idx); pParentSql->self, pSql, pRes->numOfRows, pState->numOfRetrievedRows, pSql->epSet.fqdn[pSql->epSet.inUse], idx);
if (num > tsMaxNumOfOrderedResults && tscIsProjectionQueryOnSTable(pQueryInfo, 0)) { if (num > tsMaxNumOfOrderedResults && tscIsProjectionQueryOnSTable(pQueryInfo, 0)) {
tscError("%p sub:%p num of OrderedRes is too many, max allowed:%" PRId32 " , current:%" PRId64, tscError("%p sub:%p num of OrderedRes is too many, max allowed:%" PRId32 " , current:%" PRId64,
@ -2985,8 +2986,8 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
return; return;
} }
tscDebug("%p sub:%p query complete, ep:%s, vgId:%d, orderOfSub:%d, retrieve data", trsupport->pParentSql, pSql, tscDebug("0x%"PRIx64" sub:0x%"PRIx64" query complete, ep:%s, vgId:%d, orderOfSub:%d, retrieve data", trsupport->pParentSql->self,
pVgroup->epAddr[pSql->epSet.inUse].fqdn, pVgroup->vgId, trsupport->subqueryIndex); pSql->self, pVgroup->epAddr[pSql->epSet.inUse].fqdn, pVgroup->vgId, trsupport->subqueryIndex);
if (pSql->res.qId == 0) { // qhandle is NULL, code is TSDB_CODE_SUCCESS means no results generated from this vnode if (pSql->res.qId == 0) { // qhandle is NULL, code is TSDB_CODE_SUCCESS means no results generated from this vnode
tscRetrieveFromDnodeCallBack(param, pSql, 0); tscRetrieveFromDnodeCallBack(param, pSql, 0);
@ -3049,7 +3050,7 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
} }
if (!subAndCheckDone(tres, pParentObj, pSupporter->index)) { if (!subAndCheckDone(tres, pParentObj, pSupporter->index)) {
tscDebug("%p insert:%p,%d completed, total:%d", pParentObj, tres, pSupporter->index, pParentObj->subState.numOfSub); tscDebug("0x%"PRIx64" insert:%p,%d completed, total:%d", pParentObj->self, tres, pSupporter->index, pParentObj->subState.numOfSub);
return; return;
} }
@ -3059,7 +3060,7 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
doFreeInsertSupporter(pParentObj); doFreeInsertSupporter(pParentObj);
if (pParentObj->res.code == TSDB_CODE_SUCCESS) { if (pParentObj->res.code == TSDB_CODE_SUCCESS) {
tscDebug("%p Async insertion completed, total inserted:%d", pParentObj, pParentObj->res.numOfRows); tscDebug("0x%"PRIx64" Async insertion completed, total inserted:%d", pParentObj->self, pParentObj->res.numOfRows);
// todo remove this parameter in async callback function definition. // todo remove this parameter in async callback function definition.
// all data has been sent to vnode, call user function // all data has been sent to vnode, call user function
@ -3085,14 +3086,14 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
subquerySetState(pSql, &pParentObj->subState, i, 0); subquerySetState(pSql, &pParentObj->subState, i, 0);
tscDebug("%p, failed sub:%d, %p", pParentObj, i, pSql); tscDebug("0x%"PRIx64", failed sub:%d, %p", pParentObj->self, i, pSql);
} }
} }
tscError("%p Async insertion completed, total inserted:%d rows, numOfFailed:%d, numOfTotal:%d", pParentObj, tscError("%p Async insertion completed, total inserted:%d rows, numOfFailed:%d, numOfTotal:%d", pParentObj,
pParentObj->res.numOfRows, numOfFailed, numOfSub); pParentObj->res.numOfRows, numOfFailed, numOfSub);
tscDebug("%p cleanup %d tableMeta in hashTable", pParentObj, pParentObj->cmd.numOfTables); tscDebug("0x%"PRIx64" cleanup %d tableMeta in hashTable", pParentObj->self, pParentObj->cmd.numOfTables);
for(int32_t i = 0; i < pParentObj->cmd.numOfTables; ++i) { for(int32_t i = 0; i < pParentObj->cmd.numOfTables; ++i) {
char name[TSDB_TABLE_FNAME_LEN] = {0}; char name[TSDB_TABLE_FNAME_LEN] = {0};
tNameExtractFullName(pParentObj->cmd.pTableNameList[i], name); tNameExtractFullName(pParentObj->cmd.pTableNameList[i], name);
@ -3106,7 +3107,7 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
// in case of insert, redo parsing the sql string and build new submit data block for two reasons: // in case of insert, redo parsing the sql string and build new submit data block for two reasons:
// 1. the table Id(tid & uid) may have been update, the submit block needs to be updated accordingly. // 1. the table Id(tid & uid) may have been update, the submit block needs to be updated accordingly.
// 2. vnode may need the schema information along with submit block to update its local table schema. // 2. vnode may need the schema information along with submit block to update its local table schema.
tscDebug("%p re-parse sql to generate submit data, retry:%d", pParentObj, pParentObj->retry); tscDebug("0x%"PRIx64" re-parse sql to generate submit data, retry:%d", pParentObj->self, pParentObj->retry);
pParentObj->retry++; pParentObj->retry++;
int32_t code = tsParseSql(pParentObj, true); int32_t code = tsParseSql(pParentObj, true);
@ -3158,7 +3159,7 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
pSup->pSql = pSql; pSup->pSql = pSql;
pSub->param = pSup; pSub->param = pSup;
tscDebug("%p sub:%p launch sub insert, orderOfSub:%d", pSql, pSub, i); tscDebug("0x%"PRIx64" sub:%p launch sub insert, orderOfSub:%d", pSql->self, pSub, i);
if (pSub->res.code != TSDB_CODE_SUCCESS) { if (pSub->res.code != TSDB_CODE_SUCCESS) {
tscHandleInsertRetry(pSql, pSub); tscHandleInsertRetry(pSql, pSub);
} }
@ -3186,14 +3187,14 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
} }
memset(pSql->subState.states, 0, sizeof(*pSql->subState.states) * pSql->subState.numOfSub); memset(pSql->subState.states, 0, sizeof(*pSql->subState.states) * pSql->subState.numOfSub);
tscDebug("%p reset all sub states to 0", pSql); tscDebug("0x%"PRIx64" reset all sub states to 0", pSql->self);
pSql->pSubs = calloc(pSql->subState.numOfSub, POINTER_BYTES); pSql->pSubs = calloc(pSql->subState.numOfSub, POINTER_BYTES);
if (pSql->pSubs == NULL) { if (pSql->pSubs == NULL) {
goto _error; goto _error;
} }
tscDebug("%p submit data to %d vnode(s)", pSql, pSql->subState.numOfSub); tscDebug("0x%"PRIx64" submit data to %d vnode(s)", pSql->self, pSql->subState.numOfSub);
while(numOfSub < pSql->subState.numOfSub) { while(numOfSub < pSql->subState.numOfSub) {
SInsertSupporter* pSupporter = calloc(1, sizeof(SInsertSupporter)); SInsertSupporter* pSupporter = calloc(1, sizeof(SInsertSupporter));
@ -3220,10 +3221,10 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
STableDataBlocks* pTableDataBlock = taosArrayGetP(pCmd->pDataBlocks, numOfSub); STableDataBlocks* pTableDataBlock = taosArrayGetP(pCmd->pDataBlocks, numOfSub);
pRes->code = tscCopyDataBlockToPayload(pNew, pTableDataBlock); pRes->code = tscCopyDataBlockToPayload(pNew, pTableDataBlock);
if (pRes->code == TSDB_CODE_SUCCESS) { if (pRes->code == TSDB_CODE_SUCCESS) {
tscDebug("%p sub:%p create subObj success. orderOfSub:%d", pSql, pNew, numOfSub); tscDebug("0x%"PRIx64" sub:%p create subObj success. orderOfSub:%d", pSql->self, pNew, numOfSub);
numOfSub++; numOfSub++;
} else { } else {
tscDebug("%p prepare submit data block failed in async insertion, vnodeIdx:%d, total:%d, code:%s", pSql, numOfSub, tscDebug("0x%"PRIx64" prepare submit data block failed in async insertion, vnodeIdx:%d, total:%d, code:%s", pSql->self, numOfSub,
pSql->subState.numOfSub, tstrerror(pRes->code)); pSql->subState.numOfSub, tstrerror(pRes->code));
goto _error; goto _error;
} }
@ -3240,7 +3241,7 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
// use the local variable // use the local variable
for (int32_t j = 0; j < numOfSub; ++j) { for (int32_t j = 0; j < numOfSub; ++j) {
SSqlObj *pSub = pSql->pSubs[j]; SSqlObj *pSub = pSql->pSubs[j];
tscDebug("%p sub:%p launch sub insert, orderOfSub:%d", pSql, pSub, j); tscDebug("0x%"PRIx64" sub:%p launch sub insert, orderOfSub:%d", pSql->self, pSub, j);
tscProcessSql(pSub); tscProcessSql(pSub);
} }

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

@ -496,10 +496,10 @@ void tscFreeSubobj(SSqlObj* pSql) {
return; return;
} }
tscDebug("%p start to free sub SqlObj, numOfSub:%d", pSql, pSql->subState.numOfSub); tscDebug("0x%"PRIx64" start to free sub SqlObj, numOfSub:%d", pSql->self, pSql->subState.numOfSub);
for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) { for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
tscDebug("%p free sub SqlObj:%p, index:%d", pSql, pSql->pSubs[i], i); tscDebug("0x%"PRIx64" free sub SqlObj:%p, index:%d", pSql->self, pSql->pSubs[i], i);
taos_free_result(pSql->pSubs[i]); taos_free_result(pSql->pSubs[i]);
pSql->pSubs[i] = NULL; pSql->pSubs[i] = NULL;
} }
@ -530,7 +530,7 @@ void tscFreeRegisteredSqlObj(void *pSql) {
int32_t num = atomic_sub_fetch_32(&pTscObj->numOfObj, 1); int32_t num = atomic_sub_fetch_32(&pTscObj->numOfObj, 1);
int32_t total = atomic_sub_fetch_32(&tscNumOfObj, 1); int32_t total = atomic_sub_fetch_32(&tscNumOfObj, 1);
tscDebug("%p free SqlObj, total in tscObj:%d, total:%d", pSql, num, total); tscDebug("0x%"PRIx64" free SqlObj, total in tscObj:%d, total:%d", p->self, num, total);
tscFreeSqlObj(p); tscFreeSqlObj(p);
taosReleaseRef(tscRefId, pTscObj->rid); taosReleaseRef(tscRefId, pTscObj->rid);
@ -553,7 +553,7 @@ void tscFreeSqlObj(SSqlObj* pSql) {
return; return;
} }
tscDebug("%p start to free sqlObj", pSql); tscDebug("0x%"PRIx64" start to free sqlObj", pSql->self);
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED; pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
@ -945,7 +945,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap) {
tscSortRemoveDataBlockDupRows(pOneTableBlock); tscSortRemoveDataBlockDupRows(pOneTableBlock);
char* ekey = (char*)pBlocks->data + pOneTableBlock->rowSize*(pBlocks->numOfRows-1); char* ekey = (char*)pBlocks->data + pOneTableBlock->rowSize*(pBlocks->numOfRows-1);
tscDebug("%p name:%s, name:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64, pSql, tNameGetTableName(&pOneTableBlock->tableName), tscDebug("0x%"PRIx64" name:%s, name:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64, pSql->self, tNameGetTableName(&pOneTableBlock->tableName),
pBlocks->tid, pBlocks->numOfRows, pBlocks->sversion, GET_INT64_VAL(pBlocks->data), GET_INT64_VAL(ekey)); pBlocks->tid, pBlocks->numOfRows, pBlocks->sversion, GET_INT64_VAL(pBlocks->data), GET_INT64_VAL(ekey));
int32_t len = pBlocks->numOfRows * (pOneTableBlock->rowSize + expandSize) + sizeof(STColumn) * tscGetNumOfColumns(pOneTableBlock->pTableMeta); int32_t len = pBlocks->numOfRows * (pOneTableBlock->rowSize + expandSize) + sizeof(STColumn) * tscGetNumOfColumns(pOneTableBlock->pTableMeta);
@ -2109,7 +2109,7 @@ void registerSqlObj(SSqlObj* pSql) {
int32_t num = atomic_add_fetch_32(&pSql->pTscObj->numOfObj, 1); int32_t num = atomic_add_fetch_32(&pSql->pTscObj->numOfObj, 1);
int32_t total = atomic_add_fetch_32(&tscNumOfObj, 1); int32_t total = atomic_add_fetch_32(&tscNumOfObj, 1);
tscDebug("%p new SqlObj from %p, total in tscObj:%d, total:%d", pSql, pSql->pTscObj, num, total); tscDebug("0x%"PRIx64" new SqlObj from %p, total in tscObj:%d, total:%d", pSql->self, pSql->pTscObj, num, total);
} }
SSqlObj* createSimpleSubObj(SSqlObj* pSql, __async_cb_func_t fp, void* param, int32_t cmd) { SSqlObj* createSimpleSubObj(SSqlObj* pSql, __async_cb_func_t fp, void* param, int32_t cmd) {
@ -2376,7 +2376,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, __async_cb_func_t
tscPrintSelectClause(pNew, 0); tscPrintSelectClause(pNew, 0);
} else { } else {
tscDebug("%p new sub insertion: %p, vnodeIdx:%d", pSql, pNew, pTableMetaInfo->vgroupIndex); tscDebug("0x%"PRIx64" new sub insertion: %p, vnodeIdx:%d", pSql->self, pNew, pTableMetaInfo->vgroupIndex);
} }
registerSqlObj(pNew); registerSqlObj(pNew);
@ -2627,7 +2627,7 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
int32_t totalVgroups = pTableMetaInfo->vgroupList->numOfVgroups; int32_t totalVgroups = pTableMetaInfo->vgroupList->numOfVgroups;
if (++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, tscDebug("0x%"PRIx64" results from vgroup index:%d completed, try next:%d. total vgroups:%d. current numOfRes:%" PRId64, pSql->self,
pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, pRes->numOfClauseTotal); pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, pRes->numOfClauseTotal);
/* /*
@ -2646,8 +2646,8 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
pQueryInfo->limit.offset = pRes->offset; pQueryInfo->limit.offset = pRes->offset;
assert((pRes->offset >= 0 && pRes->numOfRows == 0) || (pRes->offset == 0 && pRes->numOfRows >= 0)); assert((pRes->offset >= 0 && pRes->numOfRows == 0) || (pRes->offset == 0 && pRes->numOfRows >= 0));
tscDebug("%p new query to next vgroup, index:%d, limit:%" PRId64 ", offset:%" PRId64 ", glimit:%" PRId64, tscDebug("0x%"PRIx64" new query to next vgroup, index:%d, limit:%" PRId64 ", offset:%" PRId64 ", glimit:%" PRId64,
pSql, pTableMetaInfo->vgroupIndex, pQueryInfo->limit.limit, pQueryInfo->limit.offset, pQueryInfo->clauseLimit); pSql->self, pTableMetaInfo->vgroupIndex, pQueryInfo->limit.limit, pQueryInfo->limit.offset, pQueryInfo->clauseLimit);
/* /*
* For project query with super table join, the numOfSub is equalled to the number of all subqueries. * For project query with super table join, the numOfSub is equalled to the number of all subqueries.
@ -2664,7 +2664,7 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
pSql->fp = fp; pSql->fp = fp;
tscProcessSql(pSql); tscProcessSql(pSql);
} else { } else {
tscDebug("%p try all %d vnodes, query complete. current numOfRes:%" PRId64, pSql, totalVgroups, pRes->numOfClauseTotal); tscDebug("0x%"PRIx64" try all %d vnodes, query complete. current numOfRes:%" PRId64, pSql->self, totalVgroups, pRes->numOfClauseTotal);
} }
} }
@ -2690,7 +2690,7 @@ void tscTryQueryNextClause(SSqlObj* pSql, __async_cb_func_t fp) {
pSql->subState.numOfSub = 0; pSql->subState.numOfSub = 0;
pSql->fp = fp; pSql->fp = fp;
tscDebug("%p try data in the next subclause:%d, total subclause:%d", pSql, pCmd->clauseIndex, pCmd->numOfClause); tscDebug("0x%"PRIx64" try data in the next subclause:%d, total subclause:%d", pSql->self, pCmd->clauseIndex, pCmd->numOfClause);
if (pCmd->command > TSDB_SQL_LOCAL) { if (pCmd->command > TSDB_SQL_LOCAL) {
tscProcessLocalCmd(pSql); tscProcessLocalCmd(pSql);
} else { } else {