commit
5b654779ba
|
@ -114,21 +114,30 @@ typedef struct SWal {
|
|||
int64_t refId;
|
||||
TdThreadMutex mutex;
|
||||
// ref
|
||||
SHashObj *pRefHash; // ref -> SWalRef
|
||||
SHashObj *pRefHash; // refId -> SWalRef
|
||||
// path
|
||||
char path[WAL_PATH_LEN];
|
||||
// reusable write head
|
||||
SWalCkHead writeHead;
|
||||
} SWal; // WAL HANDLE
|
||||
} SWal;
|
||||
|
||||
typedef struct {
|
||||
int64_t refId;
|
||||
int64_t refVer;
|
||||
int64_t refFile;
|
||||
SWal *pWal;
|
||||
} SWalRef;
|
||||
|
||||
typedef struct {
|
||||
int8_t scanUncommited;
|
||||
int8_t scanNotApplied;
|
||||
int8_t scanMeta;
|
||||
int8_t enableRef;
|
||||
} SWalFilterCond;
|
||||
|
||||
typedef struct {
|
||||
SWal *pWal;
|
||||
int64_t readerId;
|
||||
TdFilePtr pLogFile;
|
||||
TdFilePtr pIdxFile;
|
||||
int64_t curFileFirstVer;
|
||||
|
@ -138,7 +147,8 @@ typedef struct {
|
|||
int8_t curStopped;
|
||||
TdThreadMutex mutex;
|
||||
SWalFilterCond cond;
|
||||
SWalCkHead *pHead;
|
||||
// TODO remove it
|
||||
SWalCkHead *pHead;
|
||||
} SWalReader;
|
||||
|
||||
// module initialization
|
||||
|
@ -157,11 +167,7 @@ int32_t walWrite(SWal *, int64_t index, tmsg_t msgType, const void *body, int32_
|
|||
int32_t walWriteWithSyncInfo(SWal *, int64_t index, tmsg_t msgType, SWalSyncInfo syncMeta, const void *body,
|
||||
int32_t bodyLen);
|
||||
|
||||
// This interface assign version automatically and return to caller.
|
||||
// When using this interface with concurrent writes,
|
||||
// wal will write all logs atomically,
|
||||
// but not sure which one will be actually write first,
|
||||
// and then the unique index of successful writen is returned.
|
||||
// Assign version automatically and return to caller,
|
||||
// -1 will be returned for failed writes
|
||||
int64_t walAppendLog(SWal *, tmsg_t msgType, SWalSyncInfo syncMeta, const void *body, int32_t bodyLen);
|
||||
|
||||
|
@ -191,17 +197,15 @@ void walSetReaderCapacity(SWalReader *pRead, int32_t capacity);
|
|||
int32_t walFetchHead(SWalReader *pRead, int64_t ver, SWalCkHead *pHead);
|
||||
int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead);
|
||||
int32_t walSkipFetchBody(SWalReader *pRead, const SWalCkHead *pHead);
|
||||
typedef struct {
|
||||
int64_t refId;
|
||||
int64_t ver;
|
||||
} SWalRef;
|
||||
|
||||
SWalRef *walRefCommittedVer(SWal *);
|
||||
|
||||
SWalRef *walOpenRef(SWal *);
|
||||
void walCloseRef(SWalRef *);
|
||||
void walCloseRef(SWal *pWal, int64_t refId);
|
||||
int32_t walRefVer(SWalRef *, int64_t ver);
|
||||
int32_t walUnrefVer(SWal *);
|
||||
void walUnrefVer(SWalRef *);
|
||||
|
||||
// help function for raft
|
||||
// helper function for raft
|
||||
bool walLogExist(SWal *, int64_t ver);
|
||||
bool walIsEmpty(SWal *);
|
||||
|
||||
|
|
|
@ -104,6 +104,8 @@ typedef struct {
|
|||
// TODO remove
|
||||
SWalReader* pWalReader;
|
||||
|
||||
SWalRef* pRef;
|
||||
|
||||
// push
|
||||
STqPushHandle pushHandle;
|
||||
|
||||
|
|
|
@ -212,6 +212,15 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen) {
|
|||
ASSERT(0);
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (offset.val.type == TMQ_OFFSET__LOG) {
|
||||
STqHandle* pHandle = taosHashGet(pTq->handles, offset.subKey, strlen(offset.subKey));
|
||||
if (walRefVer(pHandle->pRef, offset.val.version) < 0) {
|
||||
ASSERT(0);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
/*}*/
|
||||
/*}*/
|
||||
|
||||
|
@ -534,11 +543,14 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
|
|||
|
||||
pHandle->execHandle.subType = req.subType;
|
||||
pHandle->fetchMeta = req.withMeta;
|
||||
// TODO version should be assigned and refed during preprocess
|
||||
SWalRef* pRef = walRefCommittedVer(pTq->pVnode->pWal);
|
||||
if (pRef == NULL) {
|
||||
ASSERT(0);
|
||||
}
|
||||
int64_t ver = pRef->refVer;
|
||||
pHandle->pRef = pRef;
|
||||
|
||||
pHandle->pWalReader = walOpenReader(pTq->pVnode->pWal, NULL);
|
||||
|
||||
// TODO version should be assigned in preprocess
|
||||
int64_t ver = walGetCommittedVer(pTq->pVnode->pWal);
|
||||
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
|
||||
pHandle->execHandle.execCol.qmsg = req.qmsg;
|
||||
pHandle->snapshotVer = ver;
|
||||
|
@ -560,10 +572,14 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
|
|||
pHandle->execHandle.pExecReader = qExtractReaderFromStreamScanner(scanner);
|
||||
ASSERT(pHandle->execHandle.pExecReader);
|
||||
} else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
|
||||
pHandle->pWalReader = walOpenReader(pTq->pVnode->pWal, NULL);
|
||||
|
||||
pHandle->execHandle.pExecReader = tqOpenReader(pTq->pVnode);
|
||||
pHandle->execHandle.execDb.pFilterOutTbUid =
|
||||
taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
|
||||
} else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
|
||||
pHandle->pWalReader = walOpenReader(pTq->pVnode->pWal, NULL);
|
||||
|
||||
pHandle->execHandle.execTb.suid = req.suid;
|
||||
SArray* tbUidList = taosArrayInit(0, sizeof(int64_t));
|
||||
vnodeGetCtbIdList(pTq->pVnode, req.suid, tbUidList);
|
||||
|
|
|
@ -93,7 +93,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
|
|||
}
|
||||
|
||||
// init ref
|
||||
pWal->pRefHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
|
||||
pWal->pRefHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
|
||||
if (pWal->pRefHash == NULL) {
|
||||
taosMemoryFree(pWal);
|
||||
return NULL;
|
||||
|
|
|
@ -21,107 +21,112 @@ static int32_t walFetchBodyNew(SWalReader *pRead);
|
|||
static int32_t walSkipFetchBodyNew(SWalReader *pRead);
|
||||
|
||||
SWalReader *walOpenReader(SWal *pWal, SWalFilterCond *cond) {
|
||||
SWalReader *pRead = taosMemoryCalloc(1, sizeof(SWalReader));
|
||||
if (pRead == NULL) {
|
||||
SWalReader *pReader = taosMemoryCalloc(1, sizeof(SWalReader));
|
||||
if (pReader == NULL) {
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
return NULL;
|
||||
}
|
||||
|
||||
pRead->pWal = pWal;
|
||||
pRead->pIdxFile = NULL;
|
||||
pRead->pLogFile = NULL;
|
||||
pRead->curVersion = -1;
|
||||
pRead->curFileFirstVer = -1;
|
||||
pRead->curInvalid = 1;
|
||||
pRead->capacity = 0;
|
||||
pReader->pWal = pWal;
|
||||
pReader->readerId = tGenIdPI64();
|
||||
pReader->pIdxFile = NULL;
|
||||
pReader->pLogFile = NULL;
|
||||
pReader->curVersion = -1;
|
||||
pReader->curFileFirstVer = -1;
|
||||
pReader->curInvalid = 1;
|
||||
pReader->capacity = 0;
|
||||
if (cond) {
|
||||
pRead->cond = *cond;
|
||||
pReader->cond = *cond;
|
||||
} else {
|
||||
pRead->cond.scanMeta = 0;
|
||||
pRead->cond.scanUncommited = 0;
|
||||
pRead->cond.enableRef = 0;
|
||||
pReader->cond.scanUncommited = 0;
|
||||
pReader->cond.scanNotApplied = 0;
|
||||
pReader->cond.scanMeta = 0;
|
||||
pReader->cond.enableRef = 0;
|
||||
}
|
||||
|
||||
taosThreadMutexInit(&pRead->mutex, NULL);
|
||||
taosThreadMutexInit(&pReader->mutex, NULL);
|
||||
|
||||
/*if (pRead->cond.enableRef) {*/
|
||||
/*walOpenRef(pWal);*/
|
||||
/*}*/
|
||||
|
||||
pRead->pHead = taosMemoryMalloc(sizeof(SWalCkHead));
|
||||
if (pRead->pHead == NULL) {
|
||||
pReader->pHead = taosMemoryMalloc(sizeof(SWalCkHead));
|
||||
if (pReader->pHead == NULL) {
|
||||
terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
|
||||
taosMemoryFree(pRead);
|
||||
taosMemoryFree(pReader);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
return pRead;
|
||||
/*if (pReader->cond.enableRef) {*/
|
||||
/*taosHashPut(pWal->pRefHash, &pReader->readerId, sizeof(int64_t), &pReader, sizeof(void *));*/
|
||||
/*}*/
|
||||
|
||||
return pReader;
|
||||
}
|
||||
|
||||
void walCloseReader(SWalReader *pRead) {
|
||||
taosCloseFile(&pRead->pIdxFile);
|
||||
taosCloseFile(&pRead->pLogFile);
|
||||
taosMemoryFreeClear(pRead->pHead);
|
||||
taosMemoryFree(pRead);
|
||||
void walCloseReader(SWalReader *pReader) {
|
||||
taosCloseFile(&pReader->pIdxFile);
|
||||
taosCloseFile(&pReader->pLogFile);
|
||||
/*if (pReader->cond.enableRef) {*/
|
||||
/*taosHashRemove(pReader->pWal->pRefHash, &pReader->readerId, sizeof(int64_t));*/
|
||||
/*}*/
|
||||
taosMemoryFreeClear(pReader->pHead);
|
||||
taosMemoryFree(pReader);
|
||||
}
|
||||
|
||||
int32_t walNextValidMsg(SWalReader *pRead) {
|
||||
int64_t fetchVer = pRead->curVersion;
|
||||
int64_t lastVer = walGetLastVer(pRead->pWal);
|
||||
int64_t committedVer = walGetCommittedVer(pRead->pWal);
|
||||
int64_t appliedVer = walGetAppliedVer(pRead->pWal);
|
||||
int64_t endVer = pRead->cond.scanUncommited ? lastVer : committedVer;
|
||||
int32_t walNextValidMsg(SWalReader *pReader) {
|
||||
int64_t fetchVer = pReader->curVersion;
|
||||
int64_t lastVer = walGetLastVer(pReader->pWal);
|
||||
int64_t committedVer = walGetCommittedVer(pReader->pWal);
|
||||
int64_t appliedVer = walGetAppliedVer(pReader->pWal);
|
||||
int64_t endVer = pReader->cond.scanUncommited ? lastVer : committedVer;
|
||||
endVer = TMIN(appliedVer, endVer);
|
||||
|
||||
wDebug("vgId:%d wal start to fetch, ver %ld, last ver %ld commit ver %ld, applied ver %ld, end ver %ld",
|
||||
pRead->pWal->cfg.vgId, fetchVer, lastVer, committedVer, appliedVer, endVer);
|
||||
pRead->curStopped = 0;
|
||||
pReader->pWal->cfg.vgId, fetchVer, lastVer, committedVer, appliedVer, endVer);
|
||||
pReader->curStopped = 0;
|
||||
while (fetchVer <= endVer) {
|
||||
if (walFetchHeadNew(pRead, fetchVer) < 0) {
|
||||
if (walFetchHeadNew(pReader, fetchVer) < 0) {
|
||||
return -1;
|
||||
}
|
||||
if (pRead->pHead->head.msgType == TDMT_VND_SUBMIT ||
|
||||
(IS_META_MSG(pRead->pHead->head.msgType) && pRead->cond.scanMeta)) {
|
||||
if (walFetchBodyNew(pRead) < 0) {
|
||||
if (pReader->pHead->head.msgType == TDMT_VND_SUBMIT ||
|
||||
(IS_META_MSG(pReader->pHead->head.msgType) && pReader->cond.scanMeta)) {
|
||||
if (walFetchBodyNew(pReader) < 0) {
|
||||
return -1;
|
||||
}
|
||||
return 0;
|
||||
} else {
|
||||
if (walSkipFetchBodyNew(pRead) < 0) {
|
||||
if (walSkipFetchBodyNew(pReader) < 0) {
|
||||
return -1;
|
||||
}
|
||||
fetchVer++;
|
||||
ASSERT(fetchVer == pRead->curVersion);
|
||||
ASSERT(fetchVer == pReader->curVersion);
|
||||
}
|
||||
}
|
||||
pRead->curStopped = 1;
|
||||
pReader->curStopped = 1;
|
||||
return -1;
|
||||
}
|
||||
|
||||
static int64_t walReadSeekFilePos(SWalReader *pRead, int64_t fileFirstVer, int64_t ver) {
|
||||
static int64_t walReadSeekFilePos(SWalReader *pReader, int64_t fileFirstVer, int64_t ver) {
|
||||
int64_t ret = 0;
|
||||
|
||||
TdFilePtr pIdxTFile = pRead->pIdxFile;
|
||||
TdFilePtr pLogTFile = pRead->pLogFile;
|
||||
TdFilePtr pIdxTFile = pReader->pIdxFile;
|
||||
TdFilePtr pLogTFile = pReader->pLogFile;
|
||||
|
||||
// seek position
|
||||
int64_t offset = (ver - fileFirstVer) * sizeof(SWalIdxEntry);
|
||||
ret = taosLSeekFile(pIdxTFile, offset, SEEK_SET);
|
||||
if (ret < 0) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
wError("vgId:%d, failed to seek idx file, index:%" PRId64 ", pos:%" PRId64 ", since %s", pRead->pWal->cfg.vgId, ver,
|
||||
offset, terrstr());
|
||||
wError("vgId:%d, failed to seek idx file, index:%" PRId64 ", pos:%" PRId64 ", since %s", pReader->pWal->cfg.vgId,
|
||||
ver, offset, terrstr());
|
||||
return -1;
|
||||
}
|
||||
SWalIdxEntry entry = {0};
|
||||
if ((ret = taosReadFile(pIdxTFile, &entry, sizeof(SWalIdxEntry))) != sizeof(SWalIdxEntry)) {
|
||||
if (ret < 0) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
wError("vgId:%d, failed to read idx file, since %s", pRead->pWal->cfg.vgId, terrstr());
|
||||
wError("vgId:%d, failed to read idx file, since %s", pReader->pWal->cfg.vgId, terrstr());
|
||||
} else {
|
||||
terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
|
||||
wError("vgId:%d, read idx file incompletely, read bytes %" PRId64 ", bytes should be %" PRIu64,
|
||||
pRead->pWal->cfg.vgId, ret, sizeof(SWalIdxEntry));
|
||||
pReader->pWal->cfg.vgId, ret, sizeof(SWalIdxEntry));
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
|
@ -130,79 +135,79 @@ static int64_t walReadSeekFilePos(SWalReader *pRead, int64_t fileFirstVer, int64
|
|||
ret = taosLSeekFile(pLogTFile, entry.offset, SEEK_SET);
|
||||
if (ret < 0) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
wError("vgId:%d, failed to seek log file, index:%" PRId64 ", pos:%" PRId64 ", since %s", pRead->pWal->cfg.vgId, ver,
|
||||
entry.offset, terrstr());
|
||||
wError("vgId:%d, failed to seek log file, index:%" PRId64 ", pos:%" PRId64 ", since %s", pReader->pWal->cfg.vgId,
|
||||
ver, entry.offset, terrstr());
|
||||
return -1;
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
static int32_t walReadChangeFile(SWalReader *pRead, int64_t fileFirstVer) {
|
||||
static int32_t walReadChangeFile(SWalReader *pReader, int64_t fileFirstVer) {
|
||||
char fnameStr[WAL_FILE_LEN];
|
||||
|
||||
taosCloseFile(&pRead->pIdxFile);
|
||||
taosCloseFile(&pRead->pLogFile);
|
||||
taosCloseFile(&pReader->pIdxFile);
|
||||
taosCloseFile(&pReader->pLogFile);
|
||||
|
||||
walBuildLogName(pRead->pWal, fileFirstVer, fnameStr);
|
||||
TdFilePtr pLogTFile = taosOpenFile(fnameStr, TD_FILE_READ);
|
||||
if (pLogTFile == NULL) {
|
||||
walBuildLogName(pReader->pWal, fileFirstVer, fnameStr);
|
||||
TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_READ);
|
||||
if (pLogFile == NULL) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
wError("vgId:%d, cannot open file %s, since %s", pRead->pWal->cfg.vgId, fnameStr, terrstr());
|
||||
wError("vgId:%d, cannot open file %s, since %s", pReader->pWal->cfg.vgId, fnameStr, terrstr());
|
||||
return -1;
|
||||
}
|
||||
|
||||
pRead->pLogFile = pLogTFile;
|
||||
pReader->pLogFile = pLogFile;
|
||||
|
||||
walBuildIdxName(pRead->pWal, fileFirstVer, fnameStr);
|
||||
TdFilePtr pIdxTFile = taosOpenFile(fnameStr, TD_FILE_READ);
|
||||
if (pIdxTFile == NULL) {
|
||||
walBuildIdxName(pReader->pWal, fileFirstVer, fnameStr);
|
||||
TdFilePtr pIdxFile = taosOpenFile(fnameStr, TD_FILE_READ);
|
||||
if (pIdxFile == NULL) {
|
||||
terrno = TAOS_SYSTEM_ERROR(errno);
|
||||
wError("vgId:%d, cannot open file %s, since %s", pRead->pWal->cfg.vgId, fnameStr, terrstr());
|
||||
wError("vgId:%d, cannot open file %s, since %s", pReader->pWal->cfg.vgId, fnameStr, terrstr());
|
||||
return -1;
|
||||
}
|
||||
|
||||
pRead->pIdxFile = pIdxTFile;
|
||||
pReader->pIdxFile = pIdxFile;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int32_t walReadSeekVerImpl(SWalReader *pRead, int64_t ver) {
|
||||
SWal *pWal = pRead->pWal;
|
||||
int32_t walReadSeekVerImpl(SWalReader *pReader, int64_t ver) {
|
||||
SWal *pWal = pReader->pWal;
|
||||
|
||||
// bsearch in fileSet
|
||||
SWalFileInfo tmpInfo;
|
||||
tmpInfo.firstVer = ver;
|
||||
// bsearch in fileSet
|
||||
SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE);
|
||||
ASSERT(pRet != NULL);
|
||||
if (pRead->curFileFirstVer != pRet->firstVer) {
|
||||
if (pReader->curFileFirstVer != pRet->firstVer) {
|
||||
// error code was set inner
|
||||
if (walReadChangeFile(pRead, pRet->firstVer) < 0) {
|
||||
if (walReadChangeFile(pReader, pRet->firstVer) < 0) {
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
// error code was set inner
|
||||
if (walReadSeekFilePos(pRead, pRet->firstVer, ver) < 0) {
|
||||
if (walReadSeekFilePos(pReader, pRet->firstVer, ver) < 0) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
wDebug("wal version reset from %ld(invalid: %d) to %ld", pRead->curVersion, pRead->curInvalid, ver);
|
||||
wDebug("wal version reset from %ld(invalid: %d) to %ld", pReader->curVersion, pReader->curInvalid, ver);
|
||||
|
||||
pRead->curVersion = ver;
|
||||
pReader->curVersion = ver;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int32_t walReadSeekVer(SWalReader *pRead, int64_t ver) {
|
||||
SWal *pWal = pRead->pWal;
|
||||
if (!pRead->curInvalid && ver == pRead->curVersion) {
|
||||
int32_t walReadSeekVer(SWalReader *pReader, int64_t ver) {
|
||||
SWal *pWal = pReader->pWal;
|
||||
if (!pReader->curInvalid && ver == pReader->curVersion) {
|
||||
wDebug("wal version %ld match, no need to reset", ver);
|
||||
return 0;
|
||||
}
|
||||
|
||||
pRead->curInvalid = 1;
|
||||
pRead->curVersion = ver;
|
||||
pReader->curInvalid = 1;
|
||||
pReader->curVersion = ver;
|
||||
|
||||
if (ver > pWal->vers.lastVer || ver < pWal->vers.firstVer) {
|
||||
wDebug("vgId:%d, invalid index:%" PRId64 ", first index:%" PRId64 ", last index:%" PRId64, pRead->pWal->cfg.vgId,
|
||||
wDebug("vgId:%d, invalid index:%" PRId64 ", first index:%" PRId64 ", last index:%" PRId64, pReader->pWal->cfg.vgId,
|
||||
ver, pWal->vers.firstVer, pWal->vers.lastVer);
|
||||
terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
|
||||
return -1;
|
||||
|
@ -210,7 +215,7 @@ int32_t walReadSeekVer(SWalReader *pRead, int64_t ver) {
|
|||
if (ver < pWal->vers.snapshotVer) {
|
||||
}
|
||||
|
||||
if (walReadSeekVerImpl(pRead, ver) < 0) {
|
||||
if (walReadSeekVerImpl(pReader, ver) < 0) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
|
|
|
@ -0,0 +1,89 @@
|
|||
/*
|
||||
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
|
||||
*
|
||||
* This program is free software: you can use, redistribute, and/or modify
|
||||
* it under the terms of the GNU Affero General Public License, version 3
|
||||
* or later ("AGPL"), as published by the Free Software Foundation.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful, but WITHOUT
|
||||
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
||||
* FITNESS FOR A PARTICULAR PURPOSE.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include "cJSON.h"
|
||||
#include "os.h"
|
||||
#include "taoserror.h"
|
||||
#include "tutil.h"
|
||||
#include "walInt.h"
|
||||
|
||||
SWalRef *walOpenRef(SWal *pWal) {
|
||||
SWalRef *pRef = taosMemoryCalloc(1, sizeof(SWalRef));
|
||||
if (pRef == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
pRef->refId = tGenIdPI64();
|
||||
pRef->refVer = -1;
|
||||
pRef->refFile = -1;
|
||||
pRef->pWal = pWal;
|
||||
taosHashPut(pWal->pRefHash, &pRef->refId, sizeof(int64_t), &pRef, sizeof(void *));
|
||||
return pRef;
|
||||
}
|
||||
|
||||
void walCloseRef(SWal *pWal, int64_t refId) {
|
||||
SWalRef *pRef = *(SWalRef **)taosHashGet(pWal->pRefHash, &refId, sizeof(int64_t));
|
||||
taosHashRemove(pWal->pRefHash, &refId, sizeof(int64_t));
|
||||
taosMemoryFree(pRef);
|
||||
}
|
||||
|
||||
int32_t walRefVer(SWalRef *pRef, int64_t ver) {
|
||||
SWal *pWal = pRef->pWal;
|
||||
if (pRef->refVer != ver) {
|
||||
taosThreadMutexLock(&pWal->mutex);
|
||||
if (ver < pWal->vers.firstVer || ver > pWal->vers.lastVer) {
|
||||
taosThreadMutexUnlock(&pWal->mutex);
|
||||
terrno = TSDB_CODE_WAL_INVALID_VER;
|
||||
return -1;
|
||||
}
|
||||
|
||||
pRef->refVer = ver;
|
||||
// bsearch in fileSet
|
||||
SWalFileInfo tmpInfo;
|
||||
tmpInfo.firstVer = ver;
|
||||
SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE);
|
||||
ASSERT(pRet != NULL);
|
||||
pRef->refFile = pRet->firstVer;
|
||||
|
||||
taosThreadMutexUnlock(&pWal->mutex);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
void walUnrefVer(SWalRef *pRef) {
|
||||
pRef->refId = -1;
|
||||
pRef->refFile = -1;
|
||||
}
|
||||
|
||||
SWalRef *walRefCommittedVer(SWal *pWal) {
|
||||
SWalRef *pRef = walOpenRef(pWal);
|
||||
if (pRef == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
taosThreadMutexLock(&pWal->mutex);
|
||||
|
||||
int64_t ver = walGetCommittedVer(pWal);
|
||||
|
||||
pRef->refVer = ver;
|
||||
// bsearch in fileSet
|
||||
SWalFileInfo tmpInfo;
|
||||
tmpInfo.firstVer = ver;
|
||||
SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE);
|
||||
ASSERT(pRet != NULL);
|
||||
pRef->refFile = pRet->firstVer;
|
||||
|
||||
taosThreadMutexUnlock(&pWal->mutex);
|
||||
return pRef;
|
||||
}
|
|
@ -26,7 +26,7 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
|
|||
pIter = taosHashIterate(pWal->pRefHash, pIter);
|
||||
if (pIter == NULL) break;
|
||||
SWalRef *pRef = (SWalRef *)pIter;
|
||||
if (pRef->ver != -1) {
|
||||
if (pRef->refVer != -1) {
|
||||
taosHashCancelIterate(pWal->pRefHash, pIter);
|
||||
return -1;
|
||||
}
|
||||
|
@ -215,22 +215,23 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
|
|||
|
||||
static FORCE_INLINE int32_t walCheckAndRoll(SWal *pWal) {
|
||||
if (taosArrayGetSize(pWal->fileInfoSet) == 0) {
|
||||
/*pWal->vers.firstVer = index;*/
|
||||
if (walRollImpl(pWal) < 0) {
|
||||
return -1;
|
||||
}
|
||||
} else {
|
||||
int64_t passed = walGetSeq() - pWal->lastRollSeq;
|
||||
if (pWal->cfg.rollPeriod != -1 && pWal->cfg.rollPeriod != 0 && passed > pWal->cfg.rollPeriod) {
|
||||
if (walRollImpl(pWal) < 0) {
|
||||
return -1;
|
||||
}
|
||||
} else if (pWal->cfg.segSize != -1 && pWal->cfg.segSize != 0 && walGetLastFileSize(pWal) > pWal->cfg.segSize) {
|
||||
if (walRollImpl(pWal) < 0) {
|
||||
return -1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
int64_t passed = walGetSeq() - pWal->lastRollSeq;
|
||||
if (pWal->cfg.rollPeriod != -1 && pWal->cfg.rollPeriod != 0 && passed > pWal->cfg.rollPeriod) {
|
||||
if (walRollImpl(pWal) < 0) {
|
||||
return -1;
|
||||
}
|
||||
} else if (pWal->cfg.segSize != -1 && pWal->cfg.segSize != 0 && walGetLastFileSize(pWal) > pWal->cfg.segSize) {
|
||||
if (walRollImpl(pWal) < 0) {
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
@ -260,6 +261,16 @@ int32_t walEndSnapshot(SWal *pWal) {
|
|||
pWal->vers.snapshotVer = ver;
|
||||
int ts = taosGetTimestampSec();
|
||||
|
||||
int64_t minVerToDelete = ver;
|
||||
void *pIter = NULL;
|
||||
while (1) {
|
||||
pIter = taosHashIterate(pWal->pRefHash, pIter);
|
||||
if (pIter == NULL) break;
|
||||
SWalRef *pRef = *(SWalRef **)pIter;
|
||||
if (pRef->refVer == -1) continue;
|
||||
minVerToDelete = TMIN(minVerToDelete, pRef->refVer);
|
||||
}
|
||||
|
||||
int deleteCnt = 0;
|
||||
int64_t newTotSize = pWal->totSize;
|
||||
SWalFileInfo tmp;
|
||||
|
|
|
@ -101,7 +101,7 @@
|
|||
./test.sh -f tsim/parser/constCol.sim
|
||||
#./test.sh -f tsim/parser/create_db.sim
|
||||
./test.sh -f tsim/parser/create_mt.sim
|
||||
# TD-17653 ./test.sh -f tsim/parser/create_tb_with_tag_name.sim
|
||||
./test.sh -f tsim/parser/create_tb_with_tag_name.sim
|
||||
./test.sh -f tsim/parser/create_tb.sim
|
||||
./test.sh -f tsim/parser/dbtbnameValidate.sim
|
||||
./test.sh -f tsim/parser/distinct.sim
|
||||
|
|
|
@ -93,7 +93,7 @@ sql_error create table tb11 using st2 (id,t1,) tags (1,1,1);
|
|||
|
||||
sql create table tb12 using st2 (t1,id) tags (2,1);
|
||||
sql show tags from tb12;
|
||||
if $rows != 5 then
|
||||
if $rows != 4 then
|
||||
return -1
|
||||
endi
|
||||
if $data05 != 1 then
|
||||
|
@ -109,9 +109,9 @@ if $data35 != NULL then
|
|||
return -1
|
||||
endi
|
||||
|
||||
sql create table tb13 using st2 ("t1",'id') tags (2,1);
|
||||
sql create table tb13 using st2 (t1,id) tags (2,1);
|
||||
sql show tags from tb13;
|
||||
if $rows != 2 then
|
||||
if $rows != 4 then
|
||||
return -1
|
||||
endi
|
||||
if $data05 != 1 then
|
||||
|
|
Loading…
Reference in New Issue