Merge remote-tracking branch 'origin/develop' into feature/os
This commit is contained in:
commit
7256bd99cf
|
@ -1372,13 +1372,6 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) {
|
||||||
// retrieve actual query results from vnode during the second stage join subquery
|
// retrieve actual query results from vnode during the second stage join subquery
|
||||||
if (pParentSql->res.code != TSDB_CODE_SUCCESS) {
|
if (pParentSql->res.code != TSDB_CODE_SUCCESS) {
|
||||||
tscError("%p abort query due to other subquery failure. code:%d, global code:%d", pSql, code, pParentSql->res.code);
|
tscError("%p abort query due to other subquery failure. code:%d, global code:%d", pSql, code, pParentSql->res.code);
|
||||||
|
|
||||||
if (!(pTableMetaInfo->vgroupIndex > 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0))) {
|
|
||||||
if (atomic_sub_fetch_32(&pParentSql->subState.numOfRemain, 1) > 0) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
quitAllSubquery(pParentSql, pSupporter);
|
quitAllSubquery(pParentSql, pSupporter);
|
||||||
tscAsyncResultOnError(pParentSql);
|
tscAsyncResultOnError(pParentSql);
|
||||||
|
|
||||||
|
@ -1391,13 +1384,6 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) {
|
||||||
|
|
||||||
tscError("%p abort query, code:%s, global code:%s", pSql, tstrerror(code), tstrerror(pParentSql->res.code));
|
tscError("%p abort query, code:%s, global code:%s", pSql, tstrerror(code), tstrerror(pParentSql->res.code));
|
||||||
pParentSql->res.code = code;
|
pParentSql->res.code = code;
|
||||||
|
|
||||||
if (!(pTableMetaInfo->vgroupIndex > 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0))) {
|
|
||||||
if (atomic_sub_fetch_32(&pParentSql->subState.numOfRemain, 1) > 0) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
quitAllSubquery(pParentSql, pSupporter);
|
quitAllSubquery(pParentSql, pSupporter);
|
||||||
tscAsyncResultOnError(pParentSql);
|
tscAsyncResultOnError(pParentSql);
|
||||||
|
|
||||||
|
|
|
@ -444,7 +444,7 @@ int32_t tStrToInteger(const char* z, int16_t type, int32_t n, int64_t* value, bo
|
||||||
#define TSDB_PORT_HTTP 11
|
#define TSDB_PORT_HTTP 11
|
||||||
#define TSDB_PORT_ARBITRATOR 12
|
#define TSDB_PORT_ARBITRATOR 12
|
||||||
|
|
||||||
#define TSDB_MAX_WAL_SIZE (1024*1024*2)
|
#define TSDB_MAX_WAL_SIZE (1024*1024*3)
|
||||||
|
|
||||||
typedef enum {
|
typedef enum {
|
||||||
TAOS_QTYPE_RPC = 0,
|
TAOS_QTYPE_RPC = 0,
|
||||||
|
|
|
@ -6932,11 +6932,12 @@ static size_t getResultSize(SQInfo *pQInfo, int64_t *numOfRows) {
|
||||||
*/
|
*/
|
||||||
if (isTSCompQuery(pQuery) && (*numOfRows) > 0) {
|
if (isTSCompQuery(pQuery) && (*numOfRows) > 0) {
|
||||||
struct stat fStat;
|
struct stat fStat;
|
||||||
if (fstat(fileno(*(FILE **)pQuery->sdata[0]->data), &fStat) == 0) {
|
FILE *f = *(FILE **)pQuery->sdata[0]->data;
|
||||||
|
if ((f != NULL) && (fstat(fileno(f), &fStat) == 0)) {
|
||||||
*numOfRows = fStat.st_size;
|
*numOfRows = fStat.st_size;
|
||||||
return fStat.st_size;
|
return fStat.st_size;
|
||||||
} else {
|
} else {
|
||||||
qError("QInfo:%p failed to get file info, path:%s, reason:%s", pQInfo, pQuery->sdata[0]->data, strerror(errno));
|
qError("QInfo:%p failed to get file info, file:%p, reason:%s", pQInfo, f, strerror(errno));
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
|
@ -6951,7 +6952,7 @@ static int32_t doDumpQueryResult(SQInfo *pQInfo, char *data) {
|
||||||
// load data from file to msg buffer
|
// load data from file to msg buffer
|
||||||
if (isTSCompQuery(pQuery)) {
|
if (isTSCompQuery(pQuery)) {
|
||||||
|
|
||||||
FILE *f = *(FILE **)pQuery->sdata[0]->data;
|
FILE *f = *(FILE **)pQuery->sdata[0]->data; // TODO refactor
|
||||||
|
|
||||||
// make sure file exist
|
// make sure file exist
|
||||||
if (f) {
|
if (f) {
|
||||||
|
|
|
@ -474,7 +474,8 @@ void *syncRetrieveData(void *param) {
|
||||||
SSyncNode *pNode = pPeer->pSyncNode;
|
SSyncNode *pNode = pPeer->pSyncNode;
|
||||||
|
|
||||||
taosBlockSIGPIPE();
|
taosBlockSIGPIPE();
|
||||||
sInfo("%s, start to retrieve data, sstatus:%s", pPeer->id, syncStatus[pPeer->sstatus]);
|
sInfo("%s, start to retrieve data, sstatus:%s, numOfRetrieves:%d", pPeer->id, syncStatus[pPeer->sstatus],
|
||||||
|
pPeer->numOfRetrieves);
|
||||||
|
|
||||||
if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->vgId, pPeer->numOfRetrieves);
|
if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->vgId, pPeer->numOfRetrieves);
|
||||||
|
|
||||||
|
@ -496,9 +497,11 @@ void *syncRetrieveData(void *param) {
|
||||||
pPeer->numOfRetrieves++;
|
pPeer->numOfRetrieves++;
|
||||||
} else {
|
} else {
|
||||||
pPeer->numOfRetrieves = 0;
|
pPeer->numOfRetrieves = 0;
|
||||||
if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->vgId, 0);
|
// if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->vgId, 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (pNode->notifyFlowCtrl) (*pNode->notifyFlowCtrl)(pNode->vgId, 0);
|
||||||
|
|
||||||
pPeer->fileChanged = 0;
|
pPeer->fileChanged = 0;
|
||||||
taosCloseSocket(pPeer->syncFd);
|
taosCloseSocket(pPeer->syncFd);
|
||||||
|
|
||||||
|
|
|
@ -308,7 +308,7 @@ static void vnodeFlowCtrlMsgToWQueue(void *param, void *tmrId) {
|
||||||
if (pVnode->flowctrlLevel <= 0) code = TSDB_CODE_VND_IS_FLOWCTRL;
|
if (pVnode->flowctrlLevel <= 0) code = TSDB_CODE_VND_IS_FLOWCTRL;
|
||||||
|
|
||||||
pWrite->processedCount++;
|
pWrite->processedCount++;
|
||||||
if (pWrite->processedCount > 100) {
|
if (pWrite->processedCount >= 100) {
|
||||||
vError("vgId:%d, msg:%p, failed to process since %s, retry:%d", pVnode->vgId, pWrite, tstrerror(code),
|
vError("vgId:%d, msg:%p, failed to process since %s, retry:%d", pVnode->vgId, pWrite, tstrerror(code),
|
||||||
pWrite->processedCount);
|
pWrite->processedCount);
|
||||||
pWrite->processedCount = 1;
|
pWrite->processedCount = 1;
|
||||||
|
|
Loading…
Reference in New Issue