Merge branch 'develop' into feature/query
This commit is contained in:
commit
5b81134eef
|
@ -438,8 +438,9 @@ void tscKillSTableQuery(SSqlObj *pSql) {
|
|||
* here, we cannot set the command = TSDB_SQL_KILL_QUERY. Otherwise, it may cause
|
||||
* sub-queries not correctly released and master sql object of super table query reaches an abnormal state.
|
||||
*/
|
||||
pSql->pSubs[i]->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
|
||||
rpcCancelRequest(pSql->pSubs[i]->pRpcCtx);
|
||||
rpcCancelRequest(pSub->pRpcCtx);
|
||||
pSub->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
|
||||
tscQueueAsyncRes(pSub);
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
|
@ -617,19 +617,18 @@ void taos_stop_query(TAOS_RES *res) {
|
|||
if (pSql->signature != pSql) return;
|
||||
tscDebug("%p start to cancel query", res);
|
||||
|
||||
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
|
||||
|
||||
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
|
||||
if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) {
|
||||
tscKillSTableQuery(pSql);
|
||||
return;
|
||||
}
|
||||
|
||||
if (pSql->cmd.command >= TSDB_SQL_LOCAL) {
|
||||
return;
|
||||
if (pSql->cmd.command < TSDB_SQL_LOCAL) {
|
||||
rpcCancelRequest(pSql->pRpcCtx);
|
||||
}
|
||||
pSql->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
|
||||
tscQueueAsyncRes(pSql);
|
||||
|
||||
rpcCancelRequest(pSql->pRpcCtx);
|
||||
tscDebug("%p query is cancelled", res);
|
||||
}
|
||||
|
||||
|
|
|
@ -869,6 +869,11 @@ void *readTable(void *sarg) {
|
|||
int64_t sTime = rinfo->start_time;
|
||||
char *tb_prefix = rinfo->tb_prefix;
|
||||
FILE *fp = fopen(rinfo->fp, "a");
|
||||
if (NULL == fp) {
|
||||
printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
|
||||
return NULL;
|
||||
}
|
||||
|
||||
int num_of_DPT = rinfo->nrecords_per_table;
|
||||
int num_of_tables = rinfo->end_table_id - rinfo->start_table_id + 1;
|
||||
int totalData = num_of_DPT * num_of_tables;
|
||||
|
@ -924,6 +929,11 @@ void *readMetric(void *sarg) {
|
|||
TAOS *taos = rinfo->taos;
|
||||
char command[BUFFER_SIZE] = "\0";
|
||||
FILE *fp = fopen(rinfo->fp, "a");
|
||||
if (NULL == fp) {
|
||||
printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
|
||||
return NULL;
|
||||
}
|
||||
|
||||
int num_of_DPT = rinfo->nrecords_per_table;
|
||||
int num_of_tables = rinfo->end_table_id - rinfo->start_table_id + 1;
|
||||
int totalData = num_of_DPT * num_of_tables;
|
||||
|
|
File diff suppressed because it is too large
Load Diff
|
@ -51,6 +51,7 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
|
|||
break;
|
||||
case 'f':
|
||||
arguments->fqdn = arg;
|
||||
break;
|
||||
case 'g':
|
||||
arguments->dnodeGroups = arg;
|
||||
break;
|
||||
|
|
|
@ -96,6 +96,7 @@ void walModWalFile(char* walfile) {
|
|||
if (wfd < 0) {
|
||||
printf("wal:%s, failed to open(%s)\n", newWalFile, strerror(errno));
|
||||
free(buffer);
|
||||
close(rfd);
|
||||
return ;
|
||||
}
|
||||
|
||||
|
@ -116,6 +117,11 @@ void walModWalFile(char* walfile) {
|
|||
break;
|
||||
}
|
||||
|
||||
if (pHead->len >= 1024000 - sizeof(SWalHead)) {
|
||||
printf("wal:%s, SWalHead.len(%d) overflow, skip the rest of file\n", walfile, pHead->len);
|
||||
break;
|
||||
}
|
||||
|
||||
ret = read(rfd, pHead->cont, pHead->len);
|
||||
if ( ret != pHead->len) {
|
||||
printf("wal:%s, failed to read body, skip, len:%d ret:%d\n", walfile, pHead->len, ret);
|
||||
|
|
|
@ -99,6 +99,8 @@ static int32_t readVnodeCfg(SVnodeObj *pVnode, char* cfgFile)
|
|||
goto PARSE_OVER;
|
||||
}
|
||||
|
||||
content[maxLen] = (char)0;
|
||||
|
||||
root = cJSON_Parse(content);
|
||||
if (root == NULL) {
|
||||
printf("failed to json parse %s, invalid json format\n", cfgFile);
|
||||
|
|
|
@ -135,7 +135,10 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
|
|||
} else {
|
||||
assert(pQInfo == NULL);
|
||||
}
|
||||
|
||||
if (handle != NULL) {
|
||||
dnodePutItemIntoReadQueue(pVnode, handle);
|
||||
qReleaseQInfo(pVnode->qMgmt, (void**) &handle, false);
|
||||
}
|
||||
vDebug("vgId:%d, QInfo:%p, dnode query msg disposed", vgId, pQInfo);
|
||||
} else {
|
||||
assert(pCont != NULL);
|
||||
|
@ -146,14 +149,10 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
|
|||
} else {
|
||||
vDebug("vgId:%d, QInfo:%p, dnode query msg in progress", pVnode->vgId, *(void**) pCont);
|
||||
code = TSDB_CODE_VND_ACTION_IN_PROGRESS;
|
||||
qTableQuery(*handle); // do execute query
|
||||
}
|
||||
}
|
||||
|
||||
if (handle != NULL) {
|
||||
qTableQuery(*handle); // do execute query
|
||||
qReleaseQInfo(pVnode->qMgmt, (void**) &handle, false);
|
||||
}
|
||||
|
||||
return code;
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue