Skip to content

T
TDengine

爱划水de鲸鱼哥～ / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 f8cea1c5 编写于 作者: S slguan 提交者： GitHub
浏览文件
操作

Merge pull request #1507 from taosdata/feature/query 

Feature/query
上级 18da95ad 8a2d7d99
展开全部 隐藏空白更改
内联 并排
Showing with 1092 addition and 346 deletion
.gitignore
浏览文件 @ f8cea1c5
......@@ -12,6 +12,7 @@ rpms/
mac/
*.pyc
*.tmp
*.swp
src/connector/nodejs/node_modules/
src/connector/nodejs/out/
tests/test/
......
src/client/inc/tsclient.h
浏览文件 @ f8cea1c5
......@@ -357,7 +357,6 @@ typedef struct SSqlObj {
char freed : 4;
char listed : 4;
tsem_t rspSem;
tsem_t emptyRspSem;
SSqlCmd cmd;
SSqlRes res;
uint8_t numOfSubs;
......@@ -409,7 +408,7 @@ int tscProcessSql(SSqlObj *pSql);
int tscRenewMeterMeta(SSqlObj *pSql, char *tableId);
void tscQueueAsyncRes(SSqlObj *pSql);
void tscQueueAsyncError(void(*fp), void *param);
void tscQueueAsyncError(void(*fp), void *param, int32_t code);
int tscProcessLocalCmd(SSqlObj *pSql);
int tscCfgDynamicOptions(char *msg);
......@@ -450,7 +449,7 @@ void tscFreeSqlObj(SSqlObj *pObj);
void tscCloseTscObj(STscObj *pObj);
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, int32_t sqlLen);
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen);
void tscProcessMultiVnodesInsert(SSqlObj *pSql);
void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql);
......
src/client/src/tscAsync.c
浏览文件 @ f8cea1c5
......@@ -40,7 +40,7 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
static void tscAsyncFetchRowsProxy(void *param, TAOS_RES *tres, int numOfRows);
static void tscAsyncFetchSingleRowProxy(void *param, TAOS_RES *tres, int numOfRows);
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, int32_t sqlLen) {
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen) {
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
......@@ -51,17 +51,15 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) {
tscError("failed to malloc payload");
tfree(pSql);
tscQueueAsyncError(fp, param);
tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
return;
}
pSql->sqlstr = malloc(sqlLen + 1);
pSql->sqlstr = realloc(pSql->sqlstr, sqlLen + 1);
if (pSql->sqlstr == NULL) {
tscError("%p failed to malloc sql string buffer", pSql);
tscQueueAsyncError(fp, param);
tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
free(pCmd->payload);
free(pSql);
return;
}
......@@ -75,7 +73,7 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
if (code != TSDB_CODE_SUCCESS) {
pSql->res.code = (uint8_t)code;
pSql->res.code = code;
tscQueueAsyncRes(pSql);
return;
}
......@@ -88,15 +86,16 @@ void taos_query_a(TAOS *taos, const char *sqlstr, __async_cb_func_t fp, void *pa
STscObj *pObj = (STscObj *)taos;
if (pObj == NULL || pObj->signature != pObj) {
tscError("bug!!! pObj:%p", pObj);
globalCode = TSDB_CODE_DISCONNECTED;
tscQueueAsyncError(fp, param);
terrno = TSDB_CODE_DISCONNECTED;
tscQueueAsyncError(fp, param, TSDB_CODE_DISCONNECTED);
return;
}
int32_t sqlLen = strlen(sqlstr);
if (sqlLen > tsMaxSQLStringLen) {
tscError("sql string too long");
tscQueueAsyncError(fp, param);
terrno = TSDB_CODE_INVALID_SQL;
tscQueueAsyncError(fp, param, TSDB_CODE_INVALID_SQL);
return;
}
......@@ -105,7 +104,8 @@ void taos_query_a(TAOS *taos, const char *sqlstr, __async_cb_func_t fp, void *pa
SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj));
if (pSql == NULL) {
tscError("failed to malloc sqlObj");
tscQueueAsyncError(fp, param);
terrno = TSDB_CODE_CLI_OUT_OF_MEMORY;
tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
return;
}
......@@ -170,7 +170,7 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
pRes->code = numOfRows;
}
tscQueueAsyncError(pSql->fetchFp, param);
tscQueueAsyncError(pSql->fetchFp, param, pRes->code);
return;
}
......@@ -200,8 +200,8 @@ void taos_fetch_rows_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, int), voi
SSqlObj *pSql = (SSqlObj *)taosa;
if (pSql == NULL || pSql->signature != pSql) {
tscError("sql object is NULL");
globalCode = TSDB_CODE_DISCONNECTED;
tscQueueAsyncError(fp, param);
// globalCode = TSDB_CODE_DISCONNECTED;
tscQueueAsyncError(fp, param, TSDB_CODE_DISCONNECTED);
return;
}
......@@ -210,7 +210,7 @@ void taos_fetch_rows_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, int), voi
if (pRes->qhandle == 0) {
tscError("qhandle is NULL");
tscQueueAsyncError(fp, param);
tscQueueAsyncError(fp, param, TSDB_CODE_INVALID_QHANDLE);
return;
}
......@@ -232,8 +232,8 @@ void taos_fetch_row_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, TAOS_ROW),
SSqlObj *pSql = (SSqlObj *)taosa;
if (pSql == NULL || pSql->signature != pSql) {
tscError("sql object is NULL");
globalCode = TSDB_CODE_DISCONNECTED;
tscQueueAsyncError(fp, param);
// globalCode = TSDB_CODE_DISCONNECTED;
tscQueueAsyncError(fp, param, TSDB_CODE_DISCONNECTED);
return;
}
......@@ -242,7 +242,7 @@ void taos_fetch_row_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, TAOS_ROW),
if (pRes->qhandle == 0) {
tscError("qhandle is NULL");
tscQueueAsyncError(fp, param);
tscQueueAsyncError(fp, param, TSDB_CODE_INVALID_QHANDLE);
return;
}
......@@ -331,7 +331,7 @@ void tscProcessAsyncRes(SSchedMsg *pMsg) {
// pCmd may be released, so cache pCmd->command
int cmd = pCmd->command;
int code = pRes->code ? -pRes->code : pRes->numOfRows;
int code = pRes->code;// ? -pRes->code : pRes->numOfRows;
// in case of async insert, restore the user specified callback function
bool shouldFree = tscShouldFreeAsyncSqlObj(pSql);
......@@ -349,18 +349,20 @@ void tscProcessAsyncRes(SSchedMsg *pMsg) {
}
}
void tscProcessAsyncError(SSchedMsg *pMsg) {
static void tscProcessAsyncError(SSchedMsg *pMsg) {
void (*fp)() = pMsg->ahandle;
(*fp)(pMsg->thandle, NULL, -1);
(*fp)(pMsg->thandle, NULL, *(int32_t*)pMsg->msg);
}
void tscQueueAsyncError(void(*fp), void *param) {
void tscQueueAsyncError(void(*fp), void *param, int32_t code) {
int32_t* c = malloc(sizeof(int32_t));
*c = code;
SSchedMsg schedMsg;
schedMsg.fp = tscProcessAsyncError;
schedMsg.ahandle = fp;
schedMsg.thandle = param;
schedMsg.msg = NULL;
schedMsg.msg = c;
taosScheduleTask(tscQhandle, &schedMsg);
}
......@@ -412,7 +414,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
if (code != 0) {
pRes->code = code;
tscTrace("%p failed to renew tableMeta", pSql);
tsem_post(&pSql->rspSem);
// tsem_post(&pSql->rspSem);
} else {
tscTrace("%p renew tableMeta successfully, command:%d, code:%d, retry:%d",
pSql, pSql->cmd.command, pSql->res.code, pSql->retry);
......@@ -424,7 +426,7 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
code = tscSendMsgToServer(pSql);
if (code != 0) {
pRes->code = code;
tsem_post(&pSql->rspSem);
// tsem_post(&pSql->rspSem);
}
}
......
src/client/src/tscPrepare.c
浏览文件 @ f8cea1c5
......@@ -488,7 +488,6 @@ TAOS_STMT* taos_stmt_init(TAOS* taos) {
}
tsem_init(&pSql->rspSem, 0, 0);
tsem_init(&pSql->emptyRspSem, 0, 1);
pSql->signature = pSql;
pSql->pTscObj = pObj;
......
src/client/src/tscSQLParser.c
浏览文件 @ f8cea1c5
......@@ -117,7 +117,7 @@ static int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo);
static int32_t doCheckForStream(SSqlObj* pSql, SSqlInfo* pInfo);
static int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index);
static int32_t tSQLBinaryExprCreateFromSqlExpr(tSQLSyntaxNode **pExpr, tSQLExpr* pAst, int32_t* num,
static int32_t convertSyntaxTreeToExprTree(tExprNode **pExpr, tSQLExpr* pAst, int32_t* num,
SColIndexEx** pColIndex, SSqlExprInfo* pExprInfo);
/*
......@@ -215,7 +215,7 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
if (pQueryInfo->numOfTables == 0) {
pTableMetaInfo = tscAddEmptyMetaInfo(pQueryInfo);
} else {
pTableMetaInfo = &pQueryInfo->pTableMetaInfo[0];
pTableMetaInfo = pQueryInfo->pTableMetaInfo[0];
}
pCmd->command = pInfo->type;
......@@ -1208,12 +1208,12 @@ int32_t parseSelectClause(SSqlCmd* pCmd, int32_t clauseIndex, tSQLExprList* pSel
SSqlBinaryExprInfo* pBinExprInfo = &pFuncExpr->binExprInfo;
tSQLSyntaxNode* pNode = NULL;
tExprNode* pNode = NULL;
SColIndexEx* pColIndex = NULL;
int32_t ret = tSQLBinaryExprCreateFromSqlExpr(&pNode, pItem->pNode, &pBinExprInfo->numOfCols, &pColIndex, &pQueryInfo->exprsInfo);
int32_t ret = convertSyntaxTreeToExprTree(&pNode, pItem->pNode, &pBinExprInfo->numOfCols, &pColIndex, &pQueryInfo->exprsInfo);
if (ret != TSDB_CODE_SUCCESS) {
tSQLBinaryExprDestroy(&pNode, NULL);
tExprTreeDestroy(&pNode, NULL);
return invalidSqlErrMsg(pQueryInfo->msg, "invalid expression in select clause");
}
......@@ -5807,20 +5807,20 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) {
return TSDB_CODE_SUCCESS; // Does not build query message here
}
static int32_t tSQLBinaryExprCreateFromSqlExpr(tSQLSyntaxNode **pExpr, tSQLExpr* pAst, int32_t* num,
static int32_t convertSyntaxTreeToExprTree(tExprNode **pExpr, tSQLExpr* pAst, int32_t* num,
SColIndexEx** pColIndex, SSqlExprInfo* pExprInfo) {
tSQLSyntaxNode* pLeft = NULL;
tSQLSyntaxNode* pRight= NULL;
tExprNode* pLeft = NULL;
tExprNode* pRight= NULL;
if (pAst->pLeft != NULL) {
int32_t ret = tSQLBinaryExprCreateFromSqlExpr(&pLeft, pAst->pLeft, num, pColIndex, pExprInfo);
int32_t ret = convertSyntaxTreeToExprTree(&pLeft, pAst->pLeft, num, pColIndex, pExprInfo);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
}
}
if (pAst->pRight != NULL) {
int32_t ret = tSQLBinaryExprCreateFromSqlExpr(&pRight, pAst->pRight, num, pColIndex, pExprInfo);
int32_t ret = convertSyntaxTreeToExprTree(&pRight, pAst->pRight, num, pColIndex, pExprInfo);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
}
......@@ -5828,14 +5828,14 @@ static int32_t tSQLBinaryExprCreateFromSqlExpr(tSQLSyntaxNode **pExpr, tSQLExpr*
if (pAst->pLeft == NULL) {
if (pAst->nSQLOptr >= TK_TINYINT && pAst->nSQLOptr <= TK_DOUBLE) {
*pExpr = calloc(1, sizeof(tSQLSyntaxNode) + sizeof(tVariant));
*pExpr = calloc(1, sizeof(tExprNode) + sizeof(tVariant));
(*pExpr)->nodeType = TSQL_NODE_VALUE;
(*pExpr)->pVal = (tVariant*) ((char*)(*pExpr) + sizeof(tSQLSyntaxNode));
(*pExpr)->pVal = (tVariant*) ((char*)(*pExpr) + sizeof(tExprNode));
tVariantAssign((*pExpr)->pVal, &pAst->val);
} else if (pAst->nSQLOptr >= TK_COUNT && pAst->nSQLOptr <= TK_AVG_IRATE) {
*pExpr = calloc(1, sizeof(tSQLSyntaxNode) + sizeof(SSchemaEx));
*pExpr = calloc(1, sizeof(tExprNode) + sizeof(SSchemaEx));
(*pExpr)->nodeType = TSQL_NODE_COL;
(*pExpr)->pSchema = (SSchema*)((char*)(*pExpr) + sizeof(tSQLSyntaxNode));
(*pExpr)->pSchema = (SSchema*)((char*)(*pExpr) + sizeof(tExprNode));
strncpy((*pExpr)->pSchema->name, pAst->operand.z, pAst->operand.n);
// set the input column data byte and type.
......@@ -5855,7 +5855,7 @@ static int32_t tSQLBinaryExprCreateFromSqlExpr(tSQLSyntaxNode **pExpr, tSQLExpr*
strncpy((*pColIndex)[(*num) - 1].name, pAst->operand.z, pAst->operand.n);
} else {
*pExpr = (tSQLSyntaxNode *)calloc(1, sizeof(tSQLSyntaxNode));
*pExpr = (tExprNode *)calloc(1, sizeof(tExprNode));
(*pExpr)->_node.hasPK = false;
(*pExpr)->_node.pLeft = pLeft;
(*pExpr)->_node.pRight = pRight;
......
src/client/src/tscServer.c
浏览文件 @ f8cea1c5
......@@ -285,14 +285,16 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg) {
pRes->rspType = rpcMsg->msgType;
pRes->rspLen = rpcMsg->contLen;
char *tmp = (char *)realloc(pRes->pRsp, pRes->rspLen);
if (tmp == NULL) {
pRes->code = TSDB_CODE_CLI_OUT_OF_MEMORY;
} else {
pRes->pRsp = tmp;
if (pRes->rspLen) {
if (pRes->rspLen > 0) {
char *tmp = (char *)realloc(pRes->pRsp, pRes->rspLen);
if (tmp == NULL) {
pRes->code = TSDB_CODE_CLI_OUT_OF_MEMORY;
} else {
pRes->pRsp = tmp;
memcpy(pRes->pRsp, rpcMsg->pCont, pRes->rspLen);
}
} else {
pRes->pRsp = NULL;
}
// ignore the error information returned from mnode when set ignore flag in sql
......@@ -327,7 +329,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg) {
void *taosres = tscKeepConn[pCmd->command] ? pSql : NULL;
rpcMsg->code = pRes->code ? pRes->code : pRes->numOfRows;
tscTrace("%p Async SQL result:%s res:%p", pSql, tstrerror(pRes->code), taosres);
tscTrace("%p Async SQL result:%s res:%p", pSql, tstrerror(pRes->code), pSql);
/*
* Whether to free sqlObj or not should be decided before call the user defined function, since this SqlObj
......
src/client/src/tscSql.c
浏览文件 @ f8cea1c5
......@@ -155,6 +155,10 @@ static void syncConnCallback(void *param, TAOS_RES *tres, int code) {
STscObj *pObj = (STscObj *)param;
assert(pObj != NULL && pObj->pSql != NULL);
if (code < 0) {
pObj->pSql->res.code = code;
}
sem_post(&pObj->pSql->rspSem);
}
......@@ -177,17 +181,17 @@ TAOS *taos_connect(const char *ip, const char *user, const char *pass, const cha
sem_wait(&pSql->rspSem);
if (pSql->res.code != TSDB_CODE_SUCCESS) {
terrno = pSql->res.code;
taos_close(pObj);
return NULL;
}
tscTrace("%p DB connection is opening", pObj);
// version compare only requires the first 3 segments of the version string
int code = taosCheckVersion(version, taos_get_server_info(pObj), 3);
if (code != 0) {
pSql->res.code = code;
terrno = code;
taos_close(pObj);
return NULL;
} else {
......@@ -267,31 +271,29 @@ int taos_query_imp(STscObj *pObj, SSqlObj *pSql) {
return pRes->code;
}
static void syncQueryCallback(void *param, TAOS_RES *tres, int code) {
STscObj *pObj = (STscObj *)param;
assert(pObj != NULL && pObj->pSql != NULL);
static void waitForQueryRsp(void *param, TAOS_RES *tres, int code) {
assert(param != NULL);
SSqlObj *pSql = ((STscObj *)param)->pSql;
sem_post(&pObj->pSql->rspSem);
// valid error code is less than 0
if (code < 0) {
pSql->res.code = code;
}
sem_post(&pSql->rspSem);
}
int taos_query(TAOS *taos, const char *sqlstr) {
STscObj *pObj = (STscObj *)taos;
if (pObj == NULL || pObj->signature != pObj) {
globalCode = TSDB_CODE_DISCONNECTED;
terrno = TSDB_CODE_DISCONNECTED;
return TSDB_CODE_DISCONNECTED;
}
SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj));
if (pSql == NULL) {
tscError("failed to malloc sqlObj");
return TSDB_CODE_CLI_OUT_OF_MEMORY;
}
pObj->pSql = pSql;
tsem_init(&pSql->rspSem, 0, 0);
SSqlObj* pSql = pObj->pSql;
int32_t sqlLen = strlen(sqlstr);
doAsyncQuery(pObj, pObj->pSql, syncQueryCallback, taos, sqlstr, sqlLen);
size_t sqlLen = strlen(sqlstr);
doAsyncQuery(pObj, pObj->pSql, waitForQueryRsp, taos, sqlstr, sqlLen);
// wait for the callback function to post the semaphore
sem_wait(&pSql->rspSem);
......@@ -649,12 +651,12 @@ static void **tscBuildResFromSubqueries(SSqlObj *pSql) {
return pRes->tsrow;
}
static void asyncFetchCallback(void *param, TAOS_RES *tres, int numOfRows) {
static void waitForRetrieveRsp(void *param, TAOS_RES *tres, int numOfRows) {
SSqlObj* pSql = (SSqlObj*) tres;
if (numOfRows < 0) { // set the error code
pSql->res.code = -numOfRows;
}
sem_post(&pSql->rspSem);
}
......@@ -677,7 +679,7 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
// current data are exhausted, fetch more data
if (pRes->data == NULL || (pRes->data != NULL && pRes->row >= pRes->numOfRows && pRes->completed != true &&
(pCmd->command == TSDB_SQL_RETRIEVE || pCmd->command == TSDB_SQL_RETRIEVE_METRIC || pCmd->command == TSDB_SQL_FETCH))) {
taos_fetch_rows_a(res, asyncFetchCallback, pSql->pTscObj);
taos_fetch_rows_a(res, waitForRetrieveRsp, pSql->pTscObj);
sem_wait(&pSql->rspSem);
}
......@@ -754,20 +756,18 @@ void taos_free_result_imp(TAOS_RES *res, int keepCmd) {
if (pRes == NULL || pRes->qhandle == 0) {
/* Query rsp is not received from vnode, so the qhandle is NULL */
tscTrace("%p qhandle is null, abort free, fp:%p", pSql, pSql->fp);
if (pSql->fp != NULL) {
STscObj* pObj = pSql->pTscObj;
if (pSql == pObj->pSql) {
pObj->pSql = NULL;
tscFreeSqlObj(pSql);
}
if (tscShouldFreeAsyncSqlObj(pSql)) {
tscFreeSqlObj(pSql);
tscTrace("%p Async SqlObj is freed by app", pSql);
} else if (keepCmd) {
tscFreeSqlResult(pSql);
} else {
tscFreeSqlObjPartial(pSql);
if (keepCmd) {
tscFreeSqlResult(pSql);
} else {
tscFreeSqlObjPartial(pSql);
}
}
return;
}
......@@ -793,7 +793,7 @@ void taos_free_result_imp(TAOS_RES *res, int keepCmd) {
* be executed, and the retry efforts may result in double free the resources, e.g.,SRetrieveSupport
*/
if (pRes->code != TSDB_CODE_QUERY_CANCELLED &&
((pRes->numOfRows > 0 && pCmd->command < TSDB_SQL_LOCAL) ||
((pRes->numOfRows > 0 && pCmd->command < TSDB_SQL_LOCAL && pRes->completed == false) ||
(pRes->code == TSDB_CODE_SUCCESS && pRes->numOfRows == 0 && pCmd->command == TSDB_SQL_SELECT &&
pSql->pStream == NULL && pTableMetaInfo->pTableMeta != NULL))) {
pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
......@@ -836,39 +836,37 @@ void taos_free_result_imp(TAOS_RES *res, int keepCmd) {
}
} else {
// if no free resource msg is sent to vnode, we free this object immediately.
if (pSql->fp) {
bool free = tscShouldFreeAsyncSqlObj(pSql);
if (free) {
assert(pRes->numOfRows == 0 || (pCmd->command > TSDB_SQL_LOCAL));
tscFreeSqlObj(pSql);
tscTrace("%p Async sql result is freed by app", pSql);
} else if (keepCmd) {
tscFreeSqlResult(pSql);
tscTrace("%p sql result is freed while sql command is kept", pSql);
} else {
tscFreeSqlObjPartial(pSql);
tscTrace("%p sql result is freed", pSql);
if (keepCmd) {
tscFreeSqlResult(pSql);
tscTrace("%p sql result is freed while sql command is kept", pSql);
} else {
tscFreeSqlObjPartial(pSql);
tscTrace("%p sql result is freed by app", pSql);
}
}
}
}
void taos_free_result(TAOS_RES *res) { taos_free_result_imp(res, 0); }
// todo should not be used in async query
int taos_errno(TAOS *taos) {
STscObj *pObj = (STscObj *)taos;
int code;
if (pObj == NULL || pObj->signature != pObj) return globalCode;
if ((int8_t)(pObj->pSql->res.code) == -1)
code = TSDB_CODE_OTHERS;
else
code = pObj->pSql->res.code;
if (pObj == NULL || pObj->signature != pObj) {
return terrno;
}
return code;
return pObj->pSql->res.code;
}
//static bool validErrorCode(int32_t code) { return code >= TSDB_CODE_SUCCESS && code < TSDB_CODE_MAX_ERROR_CODE; }
/*
* In case of invalid sql error, additional information is attached to explain
* why the sql is invalid
......@@ -888,13 +886,15 @@ static bool hasAdditionalErrorInfo(int32_t code, SSqlCmd *pCmd) {
return z != NULL;
}
// todo should not be used in async model
char *taos_errstr(TAOS *taos) {
STscObj *pObj = (STscObj *)taos;
if (pObj == NULL || pObj->signature != pObj)
return (char*)tstrerror(globalCode);
return (char*)tstrerror(terrno);
SSqlObj *pSql = pObj->pSql;
SSqlObj* pSql = pObj->pSql;
if (hasAdditionalErrorInfo(pSql->res.code, &pSql->cmd)) {
return pSql->cmd.payload;
} else {
......
src/client/src/tscSub.c
浏览文件 @ f8cea1c5
......@@ -124,7 +124,6 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char*
pSql->sqlstr = sqlstr;
tsem_init(&pSql->rspSem, 0, 0);
tsem_init(&pSql->emptyRspSem, 0, 1);
SSqlRes *pRes = &pSql->res;
pRes->numOfRows = 1;
......
src/client/src/tscSubquery.c
浏览文件 @ f8cea1c5
......@@ -392,10 +392,10 @@ void freeSubqueryObj(SSqlObj* pSql) {
static void doQuitSubquery(SSqlObj* pParentSql) {
freeSubqueryObj(pParentSql);
tsem_wait(&pParentSql->emptyRspSem);
tsem_wait(&pParentSql->emptyRspSem);
// tsem_wait(&pParentSql->emptyRspSem);
// tsem_wait(&pParentSql->emptyRspSem);
tsem_post(&pParentSql->rspSem);
// tsem_post(&pParentSql->rspSem);
}
static void quitAllSubquery(SSqlObj* pSqlObj, SJoinSubquerySupporter* pSupporter) {
......@@ -567,7 +567,7 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) {
freeSubqueryObj(pParentSql);
}
tsem_post(&pParentSql->rspSem);
// tsem_post(&pParentSql->rspSem);
} else {
tscTrace("%p sub:%p completed, completed:%d, total:%d", pParentSql, tres, finished, numOfTotal);
}
......@@ -662,7 +662,7 @@ void tscFetchDatablockFromSubquery(SSqlObj* pSql) {
}
// wait for all subquery completed
tsem_wait(&pSql->rspSem);
// tsem_wait(&pSql->rspSem);
// update the records for each subquery
for(int32_t i = 0; i < pSql->numOfSubs; ++i) {
......@@ -797,10 +797,7 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) {
tscProcessSql(pSql);
} else { // first retrieve from vnode during the secondary stage sub-query
if (pParentSql->fp == NULL) {
tsem_wait(&pParentSql->emptyRspSem);
tsem_wait(&pParentSql->emptyRspSem);
tsem_post(&pParentSql->rspSem);
// tsem_post(&pParentSql->rspSem);
} else {
// set the command flag must be after the semaphore been correctly set.
// pPObj->cmd.command = TSDB_SQL_RETRIEVE_METRIC;
......@@ -954,10 +951,7 @@ int32_t tscHandleMasterJoinQuery(SSqlObj* pSql) {
}
}
tsem_post(&pSql->emptyRspSem);
tsem_wait(&pSql->rspSem);
tsem_post(&pSql->emptyRspSem);
// tsem_wait(&pSql->rspSem);
if (pSql->numOfSubs <= 0) {
pSql->cmd.command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
......
src/client/src/tscSystem.c
浏览文件 @ f8cea1c5
......@@ -40,15 +40,10 @@ void * tscQhandle;
void * tscCheckDiskUsageTmr;
int tsInsertHeadSize;
extern int tscEmbedded;
int tscNumOfThreads;
static pthread_once_t tscinit = PTHREAD_ONCE_INIT;
static pthread_mutex_t tscMutex;
int tscNumOfThreads;
extern int tsTscEnableRecordSql;
extern int tsNumOfLogLines;
static pthread_once_t tscinit = PTHREAD_ONCE_INIT;
void taosInitNote(int numOfNoteLines, int maxNotes, char* lable);
void deltaToUtcInitOnce();
void tscCheckDiskUsage(void *para, void *unused) {
taosGetDisk();
......@@ -60,7 +55,6 @@ int32_t tscInitRpc(const char *user, const char *secret) {
char secretEncrypt[32] = {0};
taosEncryptPass((uint8_t *)secret, strlen(secret), secretEncrypt);
pthread_mutex_lock(&tscMutex);
if (pVnodeConn == NULL) {
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localIp = tsLocalIp;
......@@ -78,7 +72,6 @@ int32_t tscInitRpc(const char *user, const char *secret) {
pVnodeConn = rpcOpen(&rpcInit);
if (pVnodeConn == NULL) {
tscError("failed to init connection to vnode");
pthread_mutex_unlock(&tscMutex);
return -1;
}
}
......@@ -100,12 +93,10 @@ int32_t tscInitRpc(const char *user, const char *secret) {
pTscMgmtConn = rpcOpen(&rpcInit);
if (pTscMgmtConn == NULL) {
tscError("failed to init connection to mgmt");
pthread_mutex_unlock(&tscMutex);
return -1;
}
}
pthread_mutex_unlock(&tscMutex);
return 0;
}
......@@ -113,7 +104,7 @@ void taos_init_imp() {
char temp[128];
struct stat dirstat;
pthread_mutex_init(&tscMutex, NULL);
errno = TSDB_CODE_SUCCESS;
srand(taosGetTimestampSec());
deltaToUtcInitOnce();
......
src/client/src/tscUtil.c
浏览文件 @ f8cea1c5
......@@ -469,7 +469,7 @@ void tscFreeSqlObjPartial(SSqlObj* pSql) {
pSql->freed = 0;
tscFreeSqlCmdData(pCmd);
tscTrace("%p free sqlObj partial completed", pSql);
tscTrace("%p partially free sqlObj completed", pSql);
}
void tscFreeSqlObj(SSqlObj* pSql) {
......@@ -487,8 +487,6 @@ void tscFreeSqlObj(SSqlObj* pSql) {
tfree(pCmd->payload);
pCmd->allocSize = 0;
tsem_destroy(&pSql->rspSem);
free(pSql);
}
......@@ -820,7 +818,9 @@ void tscCloseTscObj(STscObj* pObj) {
taosTmrStopA(&(pObj->pTimer));
tscFreeSqlObj(pSql);
sem_destroy(&pSql->rspSem);
pthread_mutex_destroy(&pObj->mutex);
tscTrace("%p DB connection is closed", pObj);
tfree(pObj);
}
......@@ -842,10 +842,9 @@ int tscAllocPayload(SSqlCmd* pCmd, int size) {
if (pCmd->payload == NULL) {
assert(pCmd->allocSize == 0);
pCmd->payload = (char*)malloc(size);
pCmd->payload = (char*)calloc(1, size);
if (pCmd->payload == NULL) return TSDB_CODE_CLI_OUT_OF_MEMORY;
pCmd->allocSize = size;
memset(pCmd->payload, 0, pCmd->allocSize);
} else {
if (pCmd->allocSize < size) {
char* b = realloc(pCmd->payload, size);
......@@ -853,6 +852,8 @@ int tscAllocPayload(SSqlCmd* pCmd, int size) {
pCmd->payload = b;
pCmd->allocSize = size;
}
memset(pCmd->payload, 0, pCmd->payloadLen);
}
//memset(pCmd->payload, 0, pCmd->allocSize);
......@@ -1105,7 +1106,7 @@ void tscClearFieldInfo(SFieldInfo* pFieldInfo) {
for(int32_t i = 0; i < pFieldInfo->numOfOutputCols; ++i) {
if (pFieldInfo->pExpr[i] != NULL) {
tSQLBinaryExprDestroy(&pFieldInfo->pExpr[i]->binExprInfo.pBinExpr, NULL);
tExprTreeDestroy(&pFieldInfo->pExpr[i]->binExprInfo.pBinExpr, NULL);
tfree(pFieldInfo->pExpr[i]->binExprInfo.pReqColumns);
tfree(pFieldInfo->pExpr[i]);
}
......@@ -1742,7 +1743,7 @@ bool tscShouldFreeAsyncSqlObj(SSqlObj* pSql) {
}
STscObj* pTscObj = pSql->pTscObj;
if (pSql->pStream != NULL || pTscObj->pHb == pSql) {
if (pSql->pStream != NULL || pTscObj->pHb == pSql || pTscObj->pSql == pSql) {
return false;
}
......@@ -1929,7 +1930,6 @@ STableMetaInfo* tscAddMeterMetaInfo(SQueryInfo* pQueryInfo, const char* name, ST
}
pTableMetaInfo->pTableMeta = pTableMeta;
// pTableMetaInfo->pMetricMeta = pMetricMeta;
pTableMetaInfo->numOfTags = numOfTags;
if (tags != NULL) {
......@@ -1963,7 +1963,7 @@ void doRemoveMeterMetaInfo(SQueryInfo* pQueryInfo, int32_t index, bool removeFro
}
void tscRemoveAllMeterMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool removeFromCache) {
tscTrace("%p deref the metric/meter meta in cache, numOfTables:%d", address, pQueryInfo->numOfTables);
tscTrace("%p deref the table meta in cache, numOfTables:%d", address, pQueryInfo->numOfTables);
int32_t index = pQueryInfo->numOfTables;
while (index >= 0) {
......
src/dnode/src/dnodeRead.c
浏览文件 @ f8cea1c5
......@@ -286,6 +286,7 @@ static void dnodeProcessRetrieveMsg(void *pVnode, SReadMsg *pMsg) {
dnodeContinueExecuteQuery(pVnode, pQInfo, pMsg);
} else { // no further execution invoked, release the ref to vnode
dnodeProcessReadResult(pVnode, pMsg);
dnodeReleaseVnode(pVnode);
}
}
......
src/os/linux/src/tlinux.c
浏览文件 @ f8cea1c5
......@@ -236,7 +236,7 @@ void *taosProcessAlarmSignal(void *tharg) {
void (*callback)(int) = tharg;
timer_t timerId;
struct sigevent sevent;
struct sigevent sevent = {0};
#ifdef _ALPINE
sevent.sigev_notify = SIGEV_THREAD;
......
src/query/inc/qast.h
浏览文件 @ f8cea1c5
......@@ -16,6 +16,7 @@
#ifndef TDENGINE_TAST_H
#define TDENGINE_TAST_H
#include <tbuffer.h>
#ifdef __cplusplus
extern "C" {
#endif
......@@ -27,14 +28,14 @@ extern "C" {
#include "taosdef.h"
#include "tvariant.h"
struct tSQLSyntaxNode;
struct tExprNode;
struct SSchema;
struct tSkipList;
struct tSkipListNode;
enum {
TSQL_NODE_EXPR = 0x1,
TSQL_NODE_COL = 0x2,
TSQL_NODE_EXPR = 0x1,
TSQL_NODE_COL = 0x2,
TSQL_NODE_VALUE = 0x4,
};
......@@ -60,44 +61,41 @@ typedef struct SBinaryFilterSupp {
void * pExtInfo;
} SBinaryFilterSupp;
typedef struct tSQLSyntaxNode {
typedef struct tExprNode {
uint8_t nodeType;
union {
struct {
uint8_t optr; // filter operator
uint8_t hasPK; // 0: do not contain primary filter, 1: contain
void * info; // support filter operation on this expression only available for leaf node
uint8_t optr; // filter operator
uint8_t hasPK; // 0: do not contain primary filter, 1: contain
void * info; // support filter operation on this expression only available for leaf node
struct tSQLSyntaxNode *pLeft; // left child pointer
struct tSQLSyntaxNode *pRight; // right child pointer
struct tExprNode *pLeft; // left child pointer
struct tExprNode *pRight; // right child pointer
} _node;
struct SSchema *pSchema;
tVariant * pVal;
};
} tSQLSyntaxNode;
} tExprNode;
void tSQLBinaryExprFromString(tExprNode **pExpr, SSchema *pSchema, int32_t numOfCols, char *src, int32_t len);
typedef struct tQueryResultset {
void ** pRes;
int64_t num;
} tQueryResultset;
void tSQLBinaryExprToString(tExprNode *pExpr, char *dst, int32_t *len);
void tSQLBinaryExprFromString(tSQLSyntaxNode **pExpr, SSchema *pSchema, int32_t numOfCols, char *src, int32_t len);
void tExprTreeDestroy(tExprNode **pExprs, void (*fp)(void*));
void tSQLBinaryExprToString(tSQLSyntaxNode *pExpr, char *dst, int32_t *len);
void tSQLBinaryExprTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SBinaryFilterSupp *param);
void tSQLBinaryExprDestroy(tSQLSyntaxNode **pExprs, void (*fp)(void*));
void tSQLBinaryExprTraverse(tSQLSyntaxNode *pExpr, SSkipList *pSkipList, SArray *result, SBinaryFilterSupp *param);
void tSQLBinaryExprCalcTraverse(tSQLSyntaxNode *pExprs, int32_t numOfRows, char *pOutput, void *param, int32_t order,
void tSQLBinaryExprCalcTraverse(tExprNode *pExprs, int32_t numOfRows, char *pOutput, void *param, int32_t order,
char *(*cb)(void *, char *, int32_t));
void tSQLBinaryExprTrv(tSQLSyntaxNode *pExprs, int32_t *val, int16_t *ids);
void tQueryResultClean(tQueryResultset *pRes);
void tSQLBinaryExprTrv(tExprNode *pExprs, int32_t *val, int16_t *ids);
uint8_t getBinaryExprOptr(SSQLToken *pToken);
SBuffer exprTreeToBinary(tExprNode* pExprTree);
tExprNode* exprTreeFromBinary(const void* pBuf, size_t size);
#ifdef __cplusplus
}
#endif
......
src/query/inc/qsqlparser.h
浏览文件 @ f8cea1c5
......@@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_QASTDEF_H
#define TDENGINE_QASTDEF_H
#ifndef TDENGINE_QSQLPARSER_H
#define TDENGINE_QSQLPARSER_H
#ifdef __cplusplus
extern "C" {
......@@ -329,4 +329,4 @@ int32_t tSQLParse(SSqlInfo *pSQLInfo, const char *pSql);
}
#endif
#endif // TDENGINE_QASTDEF_H
#endif // TDENGINE_QSQLPARSER_H
src/query/src/qast.c
浏览文件 @ f8cea1c5
此差异已折叠。
src/query/src/qtokenizer.c
浏览文件 @ f8cea1c5
......@@ -13,10 +13,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "hash.h"
#include "hashfunc.h"
#include "os.h"
#include "shash.h"
#include "taosdef.h"
#include "tstoken.h"
#include "ttokendef.h"
......
src/query/src/queryExecutor.c
浏览文件 @ f8cea1c5
......@@ -2609,7 +2609,6 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
GET_QINFO_ADDR(pRuntimeEnv), pQuery->window.skey, pQuery->window.ekey, pQuery->lastKey, pQuery->order.order);
tsdb_query_handle_t pQueryHandle = pRuntimeEnv->pQueryHandle;
while (tsdbNextDataBlock(pQueryHandle)) {
// check if query is killed or not set the status of query to pass the status check
if (isQueryKilled(GET_QINFO_ADDR(pRuntimeEnv))) {
......@@ -5149,7 +5148,7 @@ static void singleTableQueryImpl(SQInfo* pQInfo) {
int64_t st = taosGetTimestampUs();
// group by normal column, sliding window query, interval query are handled by interval query processor
if (pQuery->intervalTime != 0 || isGroupbyNormalCol(pQuery->pGroupbyExpr)) { // interval (down sampling operation)
if (isIntervalQuery(pQuery) || isGroupbyNormalCol(pQuery->pGroupbyExpr)) { // interval (down sampling operation)
tableIntervalProcessor(pQInfo);
} else {
if (isFixedOutputQuery(pQuery)) {
......@@ -5461,7 +5460,7 @@ static int32_t buildAirthmeticExprFromMsg(SSqlFunctionExpr *pExpr, SQueryTableMs
SSqlBinaryExprInfo *pBinaryExprInfo = &pExpr->binExprInfo;
SColumnInfo * pColMsg = pQueryMsg->colList;
#if 0
tSQLSyntaxNode* pBinExpr = NULL;
tExprNode* pBinExpr = NULL;
SSchema* pSchema = toSchema(pQueryMsg, pColMsg, pQueryMsg->numOfCols);
dTrace("qmsg:%p create binary expr from string:%s", pQueryMsg, pExpr->pBase.arg[0].argValue.pz);
......@@ -5962,7 +5961,7 @@ static void freeQInfo(SQInfo *pQInfo) {
if (pBinExprInfo->numOfCols > 0) {
tfree(pBinExprInfo->pReqColumns);
tSQLBinaryExprDestroy(&pBinExprInfo->pBinExpr, NULL);
tExprTreeDestroy(&pBinExprInfo->pBinExpr, NULL);
}
}
......
src/query/src/queryUtil.c
浏览文件 @ f8cea1c5
......@@ -24,6 +24,7 @@
#include "ttime.h"
#include "queryExecutor.h"
#include "queryUtil.h"
int32_t initWindowResInfo(SWindowResInfo *pWindowResInfo, SQueryRuntimeEnv *pRuntimeEnv, int32_t size,
int32_t threshold, int16_t type) {
......
src/query/tests/astTest.cpp 0 → 100644
浏览文件 @ f8cea1c5
此差异已折叠。
src/util/inc/tbuffer.h
浏览文件 @ f8cea1c5
......@@ -13,14 +13,15 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <stddef.h>
#include <stdint.h>
#include <stdbool.h>
#include <setjmp.h>
#ifndef TDENGINE_TBUFFER_H
#define TDENGINE_TBUFFER_H
#include "setjmp.h"
#include "os.h"
#ifdef __cplusplus
extern "C" {
#endif
/*
SBuffer can be used to read or write a buffer, but cannot be used for both
......@@ -80,37 +81,33 @@ int main(int argc, char** argv) {
*/
typedef struct {
jmp_buf jb;
char* data;
size_t pos;
size_t size;
char* data;
size_t pos;
size_t size;
} SBuffer;
// common functions can be used in both read & write
#define tbufThrowError(buf, code) longjmp((buf)->jb, (code))
size_t tbufTell(SBuffer* buf);
size_t tbufSeekTo(SBuffer* buf, size_t pos);
size_t tbufSkip(SBuffer* buf, size_t size);
void tbufClose(SBuffer* buf, bool keepData);
void tbufClose(SBuffer* buf, bool keepData);
// basic read functions
#define tbufBeginRead(buf, data, len) (((buf)->data = (char*)data), ((buf)->pos = 0), ((buf)->size = ((data) == NULL) ? 0 : (len)), setjmp((buf)->jb))
char* tbufRead(SBuffer* buf, size_t size);
void tbufReadToBuffer(SBuffer* buf, void* dst, size_t size);
#define tbufBeginRead(buf, _data, len) ((buf)->data = (char*)(_data), ((buf)->pos = 0), ((buf)->size = ((_data) == NULL) ? 0 : (len)), setjmp((buf)->jb))
char* tbufRead(SBuffer* buf, size_t size);
void tbufReadToBuffer(SBuffer* buf, void* dst, size_t size);
const char* tbufReadString(SBuffer* buf, size_t* len);
size_t tbufReadToString(SBuffer* buf, char* dst, size_t size);
size_t tbufReadToString(SBuffer* buf, char* dst, size_t size);
// basic write functions
#define tbufBeginWrite(buf) ((buf)->data = NULL, ((buf)->pos = 0), ((buf)->size = 0), setjmp((buf)->jb))
void tbufEnsureCapacity(SBuffer* buf, size_t size);
void tbufEnsureCapacity(SBuffer* buf, size_t size);
char* tbufGetData(SBuffer* buf, bool takeOver);
void tbufWrite(SBuffer* buf, const void* data, size_t size);
void tbufWriteAt(SBuffer* buf, size_t pos, const void* data, size_t size);
void tbufWriteStringLen(SBuffer* buf, const char* str, size_t len);
void tbufWriteString(SBuffer* buf, const char* str);
void tbufWrite(SBuffer* buf, const void* data, size_t size);
void tbufWriteAt(SBuffer* buf, size_t pos, const void* data, size_t size);
void tbufWriteStringLen(SBuffer* buf, const char* str, size_t len);
void tbufWriteString(SBuffer* buf, const char* str);
// read & write function for primitive types
#ifndef TBUFFER_DEFINE_FUNCTION
......@@ -120,17 +117,21 @@ void tbufWriteString(SBuffer* buf, const char* str);
void tbufWrite##name##At(SBuffer* buf, size_t pos, type data);
#endif
TBUFFER_DEFINE_FUNCTION( bool, Bool )
TBUFFER_DEFINE_FUNCTION( char, Char )
TBUFFER_DEFINE_FUNCTION( int8_t, Int8 )
TBUFFER_DEFINE_FUNCTION( uint8_t, Unt8 )
TBUFFER_DEFINE_FUNCTION( int16_t, Int16 )
TBUFFER_DEFINE_FUNCTION( uint16_t, Uint16 )
TBUFFER_DEFINE_FUNCTION( int32_t, Int32 )
TBUFFER_DEFINE_FUNCTION( uint32_t, Uint32 )
TBUFFER_DEFINE_FUNCTION( int64_t, Int64 )
TBUFFER_DEFINE_FUNCTION( uint64_t, Uint64 )
TBUFFER_DEFINE_FUNCTION( float, Float )
TBUFFER_DEFINE_FUNCTION( double, Double )
TBUFFER_DEFINE_FUNCTION(bool, Bool)
TBUFFER_DEFINE_FUNCTION(char, Char)
TBUFFER_DEFINE_FUNCTION(int8_t, Int8)
TBUFFER_DEFINE_FUNCTION(uint8_t, Unt8)
TBUFFER_DEFINE_FUNCTION(int16_t, Int16)
TBUFFER_DEFINE_FUNCTION(uint16_t, Uint16)
TBUFFER_DEFINE_FUNCTION(int32_t, Int32)
TBUFFER_DEFINE_FUNCTION(uint32_t, Uint32)
TBUFFER_DEFINE_FUNCTION(int64_t, Int64)
TBUFFER_DEFINE_FUNCTION(uint64_t, Uint64)
TBUFFER_DEFINE_FUNCTION(float, Float)
TBUFFER_DEFINE_FUNCTION(double, Double)
#ifdef __cplusplus
}
#endif
#endif
\ No newline at end of file
src/util/src/tbuffer.c
浏览文件 @ f8cea1c5
......@@ -30,7 +30,7 @@
tbufWriteAt(buf, pos, &data, sizeof(data));\
}
#include "../inc/tbuffer.h"
#include "tbuffer.h"
////////////////////////////////////////////////////////////////////////////////
......@@ -119,13 +119,14 @@ void tbufEnsureCapacity(SBuffer* buf, size_t size) {
}
char* tbufGetData(SBuffer* buf, bool takeOver) {
char* ret = buf->data;
if (takeOver) {
buf->pos = 0;
buf->size = 0;
buf->data = NULL;
}
return ret;
char* ret = buf->data;
if (takeOver) {
buf->pos = 0;
buf->size = 0;
buf->data = NULL;
}
return ret;
}
void tbufEndWrite(SBuffer* buf) {
......
src/vnode/tsdb/src/tsdbFile.c
浏览文件 @ f8cea1c5
......@@ -147,6 +147,11 @@ void tsdbInitFileGroupIter(STsdbFileH *pFileH, SFileGroupIter *pIter, int direct
}
void tsdbSeekFileGroupIter(SFileGroupIter *pIter, int fid) {
if (pIter->numOfFGroups == 0) {
assert(pIter->pFileGroup == NULL);
return;
}
int flags = (pIter->direction == TSDB_FGROUP_ITER_FORWARD) ? TD_GE : TD_LE;
void *ptr = taosbsearch(&fid, pIter->base, sizeof(SFileGroup), pIter->numOfFGroups, compFGroupKey, flags);
if (ptr == NULL) {
......
src/vnode/tsdb/src/tsdbRead.c
浏览文件 @ f8cea1c5
......@@ -424,39 +424,48 @@ static SDataBlockInfo getTrueDataBlockInfo(STsdbQueryHandle* pHandle, STableChec
SArray *getDefaultLoadColumns(STsdbQueryHandle *pQueryHandle, bool loadTS);
static void filterDataInDataBlock(STsdbQueryHandle *pQueryHandle, SDataCols* pCols, SArray *sa);
static bool loadQualifiedDataFromFileBlock(STsdbQueryHandle *pQueryHandle) {
static bool doLoadDataFromFileBlock(STsdbQueryHandle *pQueryHandle) {
SQueryFilePos *cur = &pQueryHandle->cur;
STableCheckInfo* pCheckInfo = taosArrayGet(pQueryHandle->pTableCheckInfo, pQueryHandle->activeIndex);
SCompBlock* pBlock = &pCheckInfo->pCompInfo->blocks[cur->slot];
SArray *sa = getDefaultLoadColumns(pQueryHandle, true);
if (QUERY_IS_ASC_QUERY(pQueryHandle->order)) {
if (pQueryHandle->window.ekey < pBlock->keyLast) {
SCompData* data = calloc(1, sizeof(SCompData)+ sizeof(SCompCol)*pBlock->numOfCols);
SCompData* data = calloc(1, sizeof(SCompData)+ sizeof(SCompCol)*pBlock->numOfCols);
data->numOfCols = pBlock->numOfCols;
data->uid = pCheckInfo->pTableObj->tableId.uid;
data->numOfCols = pBlock->numOfCols;
data->uid = pCheckInfo->pTableObj->tableId.uid;
pCheckInfo->pDataCols = tdNewDataCols(1000, 2, 4096);
tdInitDataCols(pCheckInfo->pDataCols, pCheckInfo->pTableObj->schema);
pCheckInfo->pDataCols = tdNewDataCols(1000, 2, 4096);
tdInitDataCols(pCheckInfo->pDataCols, pCheckInfo->pTableObj->schema);
SFile* pFile = &pCheckInfo->pFileGroup->files[TSDB_FILE_TYPE_DATA];
if (pFile->fd == FD_INITIALIZER) {
pFile->fd = open(pFile->fname, O_RDONLY);
}
SFile* pFile = &pCheckInfo->pFileGroup->files[TSDB_FILE_TYPE_DATA];
if (pFile->fd == FD_INITIALIZER) {
pFile->fd = open(pFile->fname, O_RDONLY);
}
if (tsdbLoadDataBlock(pFile, pBlock, 1, pCheckInfo->pDataCols, data) == 0) {
//do something
}
tsdbLoadDataBlock(pFile, pBlock, 1, pCheckInfo->pDataCols, data);
}
static bool loadQualifiedDataFromFileBlock(STsdbQueryHandle *pQueryHandle) {
SQueryFilePos *cur = &pQueryHandle->cur;
STableCheckInfo* pCheckInfo = taosArrayGet(pQueryHandle->pTableCheckInfo, pQueryHandle->activeIndex);
SCompBlock* pBlock = &pCheckInfo->pCompInfo->blocks[cur->slot];
SArray *sa = getDefaultLoadColumns(pQueryHandle, true);
if (QUERY_IS_ASC_QUERY(pQueryHandle->order)) {
// query ended in current block
if (pQueryHandle->window.ekey < pBlock->keyLast) {
doLoadDataFromFileBlock(pQueryHandle);
filterDataInDataBlock(pQueryHandle, pCheckInfo->pDataCols, sa);
}
} else {
} else {// todo desc query
if (pQueryHandle->window.ekey > pBlock->keyFirst) {
// loadDataBlockIntoMem_(pQueryHandle, pBlock, &pQueryHandle->pFields[cur->slot], cur->fileId, sa);
//
}
}
filterDataInDataBlock(pQueryHandle, pCheckInfo->pDataCols, sa);
return pQueryHandle->realNumOfRows > 0;
}
......@@ -508,7 +517,7 @@ bool moveToNextBlock(STsdbQueryHandle *pQueryHandle, int32_t step) {
// next block in the same file
cur->slot += step;
SCompBlock* pBlock = &pQueryHandle->pBlock[cur->slot];
SCompBlock* pBlock = &pCheckInfo->pCompInfo->blocks[cur->slot];
cur->pos = (step == QUERY_ASC_FORWARD_STEP) ? 0 : pBlock->numOfPoints - 1;
return loadQualifiedDataFromFileBlock(pQueryHandle);
}
......@@ -736,12 +745,23 @@ static bool getQualifiedDataBlock(STsdbQueryHandle *pQueryHandle, STableCheckInf
int32_t index = -1;
int32_t tid = pCheckInfo->tableId.tid;
SFile* pFile = &pCheckInfo->pFileGroup->files[TSDB_FILE_TYPE_DATA];
while (1) {
while (pCheckInfo->pFileGroup != NULL) {
if ((fid = getFileCompInfo(pCheckInfo, pCheckInfo->pFileGroup)) < 0) {
break;
}
SFile* pFile = &pCheckInfo->pFileGroup->files[TSDB_FILE_TYPE_DATA];
// no data block in current file, try next
if (pCheckInfo->compIndex[tid].numOfSuperBlocks == 0) {
dTrace("QInfo:%p no data block in file, fid:%d, tid:%d, try next", pQueryHandle->qinfo,
pCheckInfo->pFileGroup->fileId, tid);
pCheckInfo->pFileGroup = tsdbGetFileGroupNext(&pCheckInfo->fileIter);
continue;
}
index = binarySearchForBlockImpl(pCheckInfo->pCompInfo->blocks, pCheckInfo->compIndex[tid].numOfSuperBlocks, pQueryHandle->order, key);
......@@ -769,12 +789,11 @@ static bool getQualifiedDataBlock(STsdbQueryHandle *pQueryHandle, STableCheckInf
// load first data block into memory failed, caused by disk block error
bool blockLoaded = false;
SArray *sa = NULL;
SArray *sa = getDefaultLoadColumns(pQueryHandle, true);
// todo no need to loaded at all
cur->slot = index;
sa = getDefaultLoadColumns(pQueryHandle, true);
SCompBlock* pBlock = &pCheckInfo->pCompInfo->blocks[cur->slot];
SCompData* data = calloc(1, sizeof(SCompData)+ sizeof(SCompCol)*pBlock->numOfCols);
......@@ -784,6 +803,7 @@ static bool getQualifiedDataBlock(STsdbQueryHandle *pQueryHandle, STableCheckInf
pCheckInfo->pDataCols = tdNewDataCols(1000, 2, 4096);
tdInitDataCols(pCheckInfo->pDataCols, pCheckInfo->pTableObj->schema);
SFile* pFile = &pCheckInfo->pFileGroup->files[TSDB_FILE_TYPE_DATA];
if (pFile->fd == FD_INITIALIZER) {
pFile->fd = open(pFile->fname, O_RDONLY);
}
......@@ -973,11 +993,16 @@ SArray *tsdbRetrieveDataBlock(tsdb_query_handle_t *pQueryHandle, SArray *pIdList
STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex);
SDataBlockInfo binfo = getTrueDataBlockInfo(pHandle, pCheckInfo);
if (pHandle->realNumOfRows <= binfo.size) {
assert(pHandle->realNumOfRows <= binfo.size);
if (pHandle->realNumOfRows < binfo.size) {
return pHandle->pColumns;
} else {
// todo do load data block
assert(0);
SArray *sa = getDefaultLoadColumns(pHandle, true);
doLoadDataFromFileBlock(pHandle);
filterDataInDataBlock(pHandle, pCheckInfo->pDataCols, sa);
return pHandle->pColumns;
}
}
}
......@@ -1197,7 +1222,7 @@ static void getTagColumnInfo(SSyntaxTreeFilterSupporter* pSupporter, SSchema* pS
}
void filterPrepare(void* expr, void* param) {
tSQLSyntaxNode *pExpr = (tSQLSyntaxNode*) expr;
tExprNode *pExpr = (tExprNode*) expr;
if (pExpr->_node.info != NULL) {
return;
}
......@@ -1276,7 +1301,7 @@ bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
static int32_t doQueryTableList(STable* pSTable, SArray* pRes, const char* pCond) {
STColumn* stcol = schemaColAt(pSTable->tagSchema, 0);
tSQLSyntaxNode* pExpr = NULL;
tExprNode* pExpr = NULL;
tSQLBinaryExprFromString(&pExpr, stcol, schemaNCols(pSTable->tagSchema), pCond, strlen(pCond));
// failed to build expression, no result, return immediately
......@@ -1297,7 +1322,7 @@ static int32_t doQueryTableList(STable* pSTable, SArray* pRes, const char* pCond
};
tSQLBinaryExprTraverse(pExpr, pSTable->pIndex, pRes, &supp);
tSQLBinaryExprDestroy(&pExpr, tSQLListTraverseDestroyInfo);
tExprTreeDestroy(&pExpr, tSQLListTraverseDestroyInfo);
tansformQueryResult(pRes);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或