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

Merge branch '3.0' into fix/liao_cov

This commit is contained in:
Haojun Liao 2022-11-14 19:02:30 +08:00 committed by GitHub
parent cf56c38e7a fa2e39a5fa
commit fc83468156
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
29 changed files with 350 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/common/ttypes.h
View File

@ -346,8 +346,8 @@ bool isValidDataType(int32_t type);
void assignVal(char *val, const char *src, int32_t len, int32_t type);
void operateVal(void *dst, void *s1, void *s2, int32_t optr, int32_t type);
void *getDataMin(int32_t type);
void *getDataMax(int32_t type);
void *getDataMin(int32_t type, void* value);
void *getDataMax(int32_t type, void* value);
#ifdef __cplusplus
}

2
include/libs/function/taosudf.h
View File

@ -104,7 +104,7 @@ typedef int32_t (*TUdfDestroyFunc)();
} while (0)
#define udfColDataSetNull_var(pColumn, row) ((pColumn->colData.varLenCol.varOffsets)[row] = -1)
typedef uint16_t VarDataLenT; // maxVarDataLen: 32767
typedef uint16_t VarDataLenT; // maxVarDataLen: 65535
#define VARSTR_HEADER_SIZE sizeof(VarDataLenT)
#define varDataLen(v) ((VarDataLenT *)(v))[0]
#define varDataVal(v) ((char *)(v) + VARSTR_HEADER_SIZE)

1
include/libs/nodes/querynodes.h
View File

@ -74,7 +74,6 @@ typedef struct SColumnNode {
char tableName[TSDB_TABLE_NAME_LEN];
char tableAlias[TSDB_TABLE_NAME_LEN];
char colName[TSDB_COL_NAME_LEN];
// SNode* pProjectRef;
int16_t dataBlockId;
int16_t slotId;
} SColumnNode;

2
include/util/types.h
View File

@ -78,7 +78,7 @@ static FORCE_INLINE double taos_align_get_double(const char *pBuf) {
{ (*(double *)(x)) = (*(double *)(y)); }
// #endif
typedef uint16_t VarDataLenT; // maxVarDataLen: 32767
typedef uint16_t VarDataLenT; // maxVarDataLen: 65535
#define VARSTR_HEADER_SIZE sizeof(VarDataLenT)
#define varDataLen(v) ((VarDataLenT *)(v))[0]

1
source/client/src/clientMain.c
View File

@ -677,6 +677,7 @@ static void destoryCatalogReq(SCatalogReq *pCatalogReq) {
taosArrayDestroy(pCatalogReq->pIndex);
taosArrayDestroy(pCatalogReq->pUser);
taosArrayDestroy(pCatalogReq->pTableIndex);
taosArrayDestroy(pCatalogReq->pTableCfg);
taosMemoryFree(pCatalogReq);
}

5
source/common/src/tdatablock.c
View File

@ -277,12 +277,9 @@ int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, int
pColumnInfoData->varmeta.allocLen = len + oldLen;
}
if (pSource->pData != NULL) {
if (pColumnInfoData->pData && pSource->pData) { // TD-20382
memcpy(pColumnInfoData->pData + oldLen, pSource->pData, len);
} else {
ASSERT(len == 0);
}
pColumnInfoData->varmeta.length = len + oldLen;
} else {
if (finalNumOfRows > (*capacity)) {

5
source/common/src/tmsg.c
View File

@ -2591,7 +2591,10 @@ int32_t tDeserializeSUseDbBatchRsp(void *buf, int32_t bufLen, SUseDbBatchRsp *pR
for (int32_t i = 0; i < numOfBatch; ++i) {
SUseDbRsp usedbRsp = {0};
if (tDeserializeSUseDbRspImp(&decoder, &usedbRsp) < 0) return -1;
if (tDeserializeSUseDbRspImp(&decoder, &usedbRsp) < 0) {
tDecoderClear(&decoder);
return -1;
}
taosArrayPush(pRsp->pArray, &usedbRsp);
}
tEndDecode(&decoder);

28
source/common/src/ttypes.c
View File

@ -61,26 +61,36 @@ tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX] = {
static float floatMin = -FLT_MAX, floatMax = FLT_MAX;
static double doubleMin = -DBL_MAX, doubleMax = DBL_MAX;
FORCE_INLINE void *getDataMin(int32_t type) {
FORCE_INLINE void *getDataMin(int32_t type, void* value) {
switch (type) {
case TSDB_DATA_TYPE_FLOAT:
return &floatMin;
*(float *)value = floatMin;
break;
case TSDB_DATA_TYPE_DOUBLE:
return &doubleMin;
*(double *)value = doubleMin;
break;
default:
return &tDataTypes[type].minValue;
}
*(int64_t *)value = tDataTypes[type].minValue;
break;
}
FORCE_INLINE void *getDataMax(int32_t type) {
return value;
}
FORCE_INLINE void *getDataMax(int32_t type, void* value) {
switch (type) {
case TSDB_DATA_TYPE_FLOAT:
return &floatMax;
*(float *)value = floatMax;
break;
case TSDB_DATA_TYPE_DOUBLE:
return &doubleMax;
*(double *)value = doubleMax;
break;
default:
return &tDataTypes[type].maxValue;
*(int64_t *)value = tDataTypes[type].maxValue;
break;
}
return value;
}
bool isValidDataType(int32_t type) { return type >= TSDB_DATA_TYPE_NULL && type < TSDB_DATA_TYPE_MAX; }

2
source/dnode/mnode/impl/src/mndQuery.c
View File

@ -178,8 +178,10 @@ int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
offset += sizeof(p->msgLen);
*(int32_t *)((char *)pRsp + offset) = htonl(p->rspCode);
offset += sizeof(p->rspCode);
if (p->msg != NULL) {
memcpy((char *)pRsp + offset, p->msg, p->msgLen);
offset += p->msgLen;
}
rpcFreeCont(p->msg);
}

2
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -388,8 +388,10 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
offset += sizeof(p->msgLen);
*(int32_t *)((char *)pRsp + offset) = htonl(p->rspCode);
offset += sizeof(p->rspCode);
if (p->msg) {
memcpy((char *)pRsp + offset, p->msg, p->msgLen);
offset += p->msgLen;
}
taosMemoryFreeClear(p->msg);
}

2
source/libs/command/src/command.c
View File

@ -528,7 +528,7 @@ static int32_t setCreateTBResultIntoDataBlock(SSDataBlock* pBlock, SDbCfgInfo* p
appendTableOptions(buf2, &len, pDbCfg, pCfg);
}
varDataLen(buf2) = len;
varDataLen(buf2) = (len > 65535) ? 65535 : len;
colDataAppend(pCol2, 0, buf2, false);

4
source/libs/function/src/builtins.c
View File

@ -2474,7 +2474,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "first",
.type = FUNCTION_TYPE_FIRST,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC | FUNC_MGT_KEEP_ORDER_FUNC,
.translateFunc = translateFirstLast,
.dynDataRequiredFunc = firstDynDataReq,
.getEnvFunc = getFirstLastFuncEnv,
@ -2512,7 +2512,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
.name = "last",
.type = FUNCTION_TYPE_LAST,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC | FUNC_MGT_KEEP_ORDER_FUNC,
.translateFunc = translateFirstLast,
.dynDataRequiredFunc = lastDynDataReq,
.getEnvFunc = getFirstLastFuncEnv,

2
source/libs/nodes/src/nodesUtilFuncs.c
View File

@ -622,7 +622,7 @@ void nodesDestroyNode(SNode* pNode) {
}
switch (nodeType(pNode)) {
case QUERY_NODE_COLUMN: // pProjectRef is weak reference, no need to release
case QUERY_NODE_COLUMN:
destroyExprNode((SExprNode*)pNode);
break;
case QUERY_NODE_VALUE: {

1
source/libs/parser/src/parInsertSql.c
View File

@ -1364,6 +1364,7 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, SVnodeModifOpStmt* pStmt,
break;
}
}
taosMemoryFree(pLine);
if (TSDB_CODE_SUCCESS == code && 0 == (*pNumOfRows) &&
(!TSDB_QUERY_HAS_TYPE(pStmt->insertType, TSDB_QUERY_TYPE_STMT_INSERT)) && !pStmt->fileProcessing) {

22
source/libs/parser/src/parTranslater.c
View File

@ -744,7 +744,8 @@ static bool isPrimaryKeyImpl(SNode* pExpr) {
return (PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)pExpr)->colId);
} else if (QUERY_NODE_FUNCTION == nodeType(pExpr)) {
SFunctionNode* pFunc = (SFunctionNode*)pExpr;
if (FUNCTION_TYPE_SELECT_VALUE == pFunc->funcType) {
if (FUNCTION_TYPE_SELECT_VALUE == pFunc->funcType || FUNCTION_TYPE_FIRST == pFunc->funcType ||
FUNCTION_TYPE_LAST == pFunc->funcType) {
return isPrimaryKeyImpl(nodesListGetNode(pFunc->pParameterList, 0));
} else if (FUNCTION_TYPE_WSTART == pFunc->funcType || FUNCTION_TYPE_WEND == pFunc->funcType) {
return true;
@ -787,7 +788,6 @@ static void setColumnInfoBySchema(const SRealTableNode* pTable, const SSchema* p
static void setColumnInfoByExpr(STempTableNode* pTable, SExprNode* pExpr, SColumnNode** pColRef) {
SColumnNode* pCol = *pColRef;
// pCol->pProjectRef = (SNode*)pExpr;
if (NULL == pExpr->pAssociation) {
pExpr->pAssociation = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
}
@ -2932,8 +2932,8 @@ static int32_t checkFill(STranslateContext* pCxt, SFillNode* pFill, SValueNode*
return TSDB_CODE_SUCCESS;
}
if (TSWINDOW_IS_EQUAL(pFill->timeRange, TSWINDOW_INITIALIZER) ||
TSWINDOW_IS_EQUAL(pFill->timeRange, TSWINDOW_DESC_INITIALIZER)) {
if (!pCxt->createStream && (TSWINDOW_IS_EQUAL(pFill->timeRange, TSWINDOW_INITIALIZER) ||
TSWINDOW_IS_EQUAL(pFill->timeRange, TSWINDOW_DESC_INITIALIZER))) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_FILL_TIME_RANGE);
}
@ -5268,9 +5268,7 @@ static int32_t checkTopicQuery(STranslateContext* pCxt, SSelectStmt* pSelect) {
}
static int32_t buildCreateTopicReq(STranslateContext* pCxt, SCreateTopicStmt* pStmt, SCMCreateTopicReq* pReq) {
SName name;
tNameSetDbName(&name, pCxt->pParseCxt->acctId, pStmt->topicName, strlen(pStmt->topicName));
tNameGetFullDbName(&name, pReq->name);
snprintf(pReq->name, sizeof(pReq->name), "%d.%s", pCxt->pParseCxt->acctId, pStmt->topicName);
pReq->igExists = pStmt->ignoreExists;
pReq->withMeta = pStmt->withMeta;
@ -5280,7 +5278,7 @@ static int32_t buildCreateTopicReq(STranslateContext* pCxt, SCreateTopicStmt* pS
}
int32_t code = TSDB_CODE_SUCCESS;
SName name;
if ('\0' != pStmt->subSTbName[0]) {
pReq->subType = TOPIC_SUB_TYPE__TABLE;
toName(pCxt->pParseCxt->acctId, pStmt->subDbName, pStmt->subSTbName, &name);
@ -5548,6 +5546,10 @@ static int32_t checkStreamQuery(STranslateContext* pCxt, SSelectStmt* pSelect) {
crossTableWithUdaf(pSelect)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "Unsupported stream query");
}
if (NULL != pSelect->pSubtable && TSDB_DATA_TYPE_VARCHAR != ((SExprNode*)pSelect->pSubtable)->resType.type) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY,
"SUBTABLE expression must be of VARCHAR type");
}
return TSDB_CODE_SUCCESS;
}
@ -6060,11 +6062,11 @@ static int32_t extractShowCreateTableResultSchema(int32_t* numOfCols, SSchema**
}
(*pSchema)[0].type = TSDB_DATA_TYPE_BINARY;
(*pSchema)[0].bytes = TSDB_TABLE_NAME_LEN;
(*pSchema)[0].bytes = SHOW_CREATE_TB_RESULT_FIELD1_LEN;
strcpy((*pSchema)[0].name, "Table");
(*pSchema)[1].type = TSDB_DATA_TYPE_BINARY;
(*pSchema)[1].bytes = TSDB_MAX_BINARY_LEN;
(*pSchema)[1].bytes = SHOW_CREATE_TB_RESULT_FIELD2_LEN;
strcpy((*pSchema)[1].name, "Create Table");
return TSDB_CODE_SUCCESS;

11
source/libs/scalar/src/filter.c
View File

@ -512,15 +512,17 @@ int32_t filterReuseRangeCtx(SFilterRangeCtx *ctx, int32_t type, int32_t options)
}
int32_t filterConvertRange(SFilterRangeCtx *cur, SFilterRange *ra, bool *notNull) {
int64_t tmp = 0;
if (!FILTER_GET_FLAG(ra->sflag, RANGE_FLG_NULL)) {
int32_t sr = cur->pCompareFunc(&ra->s, getDataMin(cur->type));
int32_t sr = cur->pCompareFunc(&ra->s, getDataMin(cur->type, &tmp));
if (sr == 0) {
FILTER_SET_FLAG(ra->sflag, RANGE_FLG_NULL);
}
}
if (!FILTER_GET_FLAG(ra->eflag, RANGE_FLG_NULL)) {
int32_t er = cur->pCompareFunc(&ra->e, getDataMax(cur->type));
int32_t er = cur->pCompareFunc(&ra->e, getDataMax(cur->type, &tmp));
if (er == 0) {
FILTER_SET_FLAG(ra->eflag, RANGE_FLG_NULL);
}
@ -696,14 +698,15 @@ int32_t filterAddRangeImpl(void *h, SFilterRange *ra, int32_t optr) {
int32_t filterAddRange(void *h, SFilterRange *ra, int32_t optr) {
SFilterRangeCtx *ctx = (SFilterRangeCtx *)h;
int64_t tmp = 0;
if (FILTER_GET_FLAG(ra->sflag, RANGE_FLG_NULL)) {
SIMPLE_COPY_VALUES(&ra->s, getDataMin(ctx->type));
SIMPLE_COPY_VALUES(&ra->s, getDataMin(ctx->type, &tmp));
// FILTER_CLR_FLAG(ra->sflag, RA_NULL);
}
if (FILTER_GET_FLAG(ra->eflag, RANGE_FLG_NULL)) {
SIMPLE_COPY_VALUES(&ra->e, getDataMax(ctx->type));
SIMPLE_COPY_VALUES(&ra->e, getDataMax(ctx->type, &tmp));
// FILTER_CLR_FLAG(ra->eflag, RA_NULL);
}

2
source/libs/scheduler/src/schRemote.c
View File

@ -287,8 +287,10 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t execId, SDa
SSubmitRsp *sum = pJob->execRes.res;
sum->affectedRows += rsp->affectedRows;
sum->nBlocks += rsp->nBlocks;
if (rsp->nBlocks > 0 && rsp->pBlocks) {
sum->pBlocks = taosMemoryRealloc(sum->pBlocks, sum->nBlocks * sizeof(*sum->pBlocks));
memcpy(sum->pBlocks + sum->nBlocks - rsp->nBlocks, rsp->pBlocks, rsp->nBlocks * sizeof(*sum->pBlocks));
}
taosMemoryFree(rsp->pBlocks);
taosMemoryFree(rsp);
} else {

6
source/libs/stream/src/streamState.c
View File

@ -114,12 +114,12 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath, int
return NULL;
}
char statePath[300];
char statePath[1024];
if (!specPath) {
sprintf(statePath, "%s/%d", path, pTask->taskId);
} else {
memset(statePath, 0, 300);
tstrncpy(statePath, path, 300);
memset(statePath, 0, 1024);
tstrncpy(statePath, path, 1024);
}
if (tdbOpen(statePath, szPage, pages, &pState->db, 0) < 0) {
goto _err;

37
source/libs/sync/src/syncAppendEntriesReply.c
View File

@ -37,43 +37,6 @@
// /\ UNCHANGED <<serverVars, candidateVars, logVars, elections>>
//
// only start once
static void syncNodeStartSnapshotOnce(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, SyncTerm lastApplyTerm,
SyncAppendEntriesReply* pMsg) {
if (beginIndex > endIndex) {
sNError(ths, "snapshot param error, start:%" PRId64 ", end:%" PRId64, beginIndex, endIndex);
return;
}
// get sender
SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(ths, &(pMsg->srcId));
ASSERT(pSender != NULL);
if (snapshotSenderIsStart(pSender)) {
sSError(pSender, "snapshot sender already start");
return;
}
SSnapshot snapshot = {
.data = NULL, .lastApplyIndex = endIndex, .lastApplyTerm = lastApplyTerm, .lastConfigIndex = SYNC_INDEX_INVALID};
void* pReader = NULL;
SSnapshotParam readerParam = {.start = beginIndex, .end = endIndex};
int32_t code = ths->pFsm->FpSnapshotStartRead(ths->pFsm, &readerParam, &pReader);
ASSERT(code == 0);
#if 0
if (pMsg->privateTerm < pSender->privateTerm) {
ASSERT(pReader != NULL);
snapshotSenderStart(pSender, readerParam, snapshot, pReader);
} else {
if (pReader != NULL) {
ths->pFsm->FpSnapshotStopRead(ths->pFsm, pReader);
}
}
#endif
}
int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
int32_t ret = 0;
SyncAppendEntriesReply* pMsg = pRpcMsg->pCont;

3
source/libs/sync/src/syncMain.c
View File

@ -1832,6 +1832,9 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) {
SElectTimer* pElectTimer = param;
SSyncNode* pNode = pElectTimer->pSyncNode;
if (pNode == NULL) return;
if (pNode->syncEqMsg == NULL) return;
SRpcMsg rpcMsg = {0};
int32_t code = syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pElectTimer->logicClock, pNode->electTimerMS, pNode);

13
source/libs/sync/src/syncRaftLog.c
View File

@ -197,7 +197,12 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
syncMeta.isWeek = pEntry->isWeak;
syncMeta.seqNum = pEntry->seqNum;
syncMeta.term = pEntry->term;
int64_t tsWriteBegin = taosGetTimestampMs();
index = walAppendLog(pWal, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen);
int64_t tsWriteEnd = taosGetTimestampMs();
int64_t tsElapsed = tsWriteEnd - tsWriteBegin;
if (index < 0) {
int32_t err = terrno;
const char* errStr = tstrerror(err);
@ -210,8 +215,8 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr
}
pEntry->index = index;
sNTrace(pData->pSyncNode, "write index:%" PRId64 ", type:%s, origin type:%s", pEntry->index,
TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType));
sNTrace(pData->pSyncNode, "write index:%" PRId64 ", type:%s, origin type:%s, elapsed:%" PRId64, pEntry->index,
TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType), tsElapsed);
return 0;
}
@ -236,7 +241,11 @@ int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncR
taosThreadMutexLock(&(pData->mutex));
int64_t tsBegin = taosGetTimestampMs();
code = walReadVer(pWalHandle, index);
int64_t tsEnd = taosGetTimestampMs();
int64_t tsElapsed = tsEnd - tsBegin;
// code = walReadVerCached(pWalHandle, index);
if (code != 0) {
int32_t err = terrno;

20
source/libs/sync/src/syncReplication.c
View File

@ -140,9 +140,10 @@ int32_t syncNodeReplicate(SSyncNode* pSyncNode) {
int32_t syncNodeSendAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, SRpcMsg* pRpcMsg) {
int32_t ret = 0;
SyncAppendEntries* pMsg = pRpcMsg->pCont;
syncLogSendAppendEntries(pSyncNode, pMsg, "");
syncNodeSendMsgById(destRaftId, pSyncNode, pRpcMsg);
if (pMsg == NULL) {
sError("vgId:%d, sync-append-entries msg is NULL", pSyncNode->vgId);
return 0;
}
SPeerState* pState = syncNodeGetPeerState(pSyncNode, destRaftId);
if (pState == NULL) {
@ -150,8 +151,19 @@ int32_t syncNodeSendAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftI
return 0;
}
// save index, otherwise pMsg will be free by rpc
SyncIndex saveLastSendIndex = pState->lastSendIndex;
bool update = false;
if (pMsg->dataLen > 0) {
pState->lastSendIndex = pMsg->prevLogIndex + 1;
saveLastSendIndex = pMsg->prevLogIndex + 1;
update = true;
}
syncLogSendAppendEntries(pSyncNode, pMsg, "");
syncNodeSendMsgById(destRaftId, pSyncNode, pRpcMsg);
if (update) {
pState->lastSendIndex = saveLastSendIndex;
pState->lastSendTime = taosGetTimestampMs();
}

11
source/libs/tdb/src/db/tdbPCache.c
View File

@ -169,7 +169,7 @@ int tdbPCacheAlter(SPCache *pCache, int32_t nPage) {
SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn) {
SPage *pPage;
i32 nRef;
i32 nRef = 0;
tdbPCacheLock(pCache);
@ -178,14 +178,17 @@ SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, TXN *pTxn) {
nRef = tdbRefPage(pPage);
}
ASSERT(pPage);
tdbPCacheUnlock(pCache);
// printf("thread %" PRId64 " fetch page %d pgno %d pPage %p nRef %d\n", taosGetSelfPthreadId(), pPage->id,
// TDB_PAGE_PGNO(pPage), pPage, nRef);
if (pPage) {
tdbDebug("pcache/fetch page %p/%d/%d/%d", pPage, TDB_PAGE_PGNO(pPage), pPage->id, nRef);
} else {
tdbDebug("pcache/fetch page %p", pPage);
}
return pPage;
}
@ -266,7 +269,7 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn)
}
// 4. Try a create new page
if (!pPage) {
if (!pPage && pTxn->xMalloc != NULL) {
ret = tdbPageCreate(pCache->szPage, &pPage, pTxn->xMalloc, pTxn->xArg);
if (ret < 0 || pPage == NULL) {
// TODO

222
source/libs/tdb/src/db/tdbPager.c
View File

@ -27,6 +27,116 @@ typedef struct {
TDB_STATIC_ASSERT(sizeof(SFileHdr) == 128, "Size of file header is not correct");
struct hashset_st {
size_t nbits;
size_t mask;
size_t capacity;
size_t *items;
size_t nitems;
double load_factor;
};
static const unsigned int prime = 39;
static const unsigned int prime2 = 5009;
hashset_t hashset_create(void) {
hashset_t set = tdbOsCalloc(1, sizeof(struct hashset_st));
if (!set) {
return NULL;
}
set->nbits = 4;
set->capacity = (size_t)(1 << set->nbits);
set->items = tdbOsCalloc(set->capacity, sizeof(size_t));
if (!set->items) {
tdbOsFree(set);
return NULL;
}
set->mask = set->capacity - 1;
set->nitems = 0;
set->load_factor = 0.75;
return set;
}
void hashset_destroy(hashset_t set) {
if (set) {
tdbOsFree(set->items);
tdbOsFree(set);
}
}
int hashset_add_member(hashset_t set, void *item) {
size_t value = (size_t) item;
size_t h;
if (value == 0) {
return -1;
}
for (h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
if (set->items[h] == value) {
return 0;
}
}
set->items[h] = value;
++set->nitems;
return 1;
}
int hashset_add(hashset_t set, void *item) {
int ret = hashset_add_member(set, item);
size_t old_capacity = set->capacity;
if (set->nitems >= (double)old_capacity * set->load_factor) {
size_t *old_items = set->items;
++set->nbits;
set->capacity = (size_t)(1 << set->nbits);
set->mask = set->capacity - 1;
set->items = tdbOsCalloc(set->capacity, sizeof(size_t));
if (!set->items) {
return -1;
}
set->nitems = 0;
for (size_t i = 0; i < old_capacity; ++i) {
hashset_add_member(set, (void*)old_items[i]);
}
tdbOsFree(old_items);
}
return ret;
}
int hashset_remove(hashset_t set, void *item) {
size_t value = (size_t) item;
for (size_t h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
if (set->items[h] == value) {
set->items[h] = 0;
--set->nitems;
return 1;
}
}
return 0;
}
int hashset_contains(hashset_t set, void *item) {
size_t value = (size_t) item;
for (size_t h = set->mask & (prime * value); set->items[h] != 0; h = set->mask & (h + prime2)) {
if (set->items[h] == value) {
return 1;
}
}
return 0;
}
#define TDB_PAGE_INITIALIZED(pPage) ((pPage)->pPager != NULL)
static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage *, void *, int), void *arg,
@ -209,12 +319,16 @@ int tdbPagerWrite(SPager *pPager, SPage *pPage) {
tRBTreePut(&pPager->rbt, (SRBTreeNode *)pPage);
// Write page to journal if neccessary
if (TDB_PAGE_PGNO(pPage) <= pPager->dbOrigSize) {
if (TDB_PAGE_PGNO(pPage) <= pPager->dbOrigSize && (pPager->jPageSet == NULL || !hashset_contains(pPager->jPageSet, (void*)((long)TDB_PAGE_PGNO(pPage))))) {
ret = tdbPagerWritePageToJournal(pPager, pPage);
if (ret < 0) {
tdbError("failed to write page to journal since %s", tstrerror(terrno));
return -1;
}
if (pPager->jPageSet) {
hashset_add(pPager->jPageSet, (void*)((long)TDB_PAGE_PGNO(pPage)));
}
}
return 0;
@ -233,6 +347,7 @@ int tdbPagerBegin(SPager *pPager, TXN *pTxn) {
return -1;
}
pPager->jPageSet = hashset_create();
// TODO: write the size of the file
pPager->inTran = 1;
@ -275,6 +390,9 @@ int tdbPagerCommit(SPager *pPager, TXN *pTxn) {
pPage->isDirty = 0;
tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
if (pPager->jPageSet) {
hashset_remove(pPager->jPageSet, (void*)((long)TDB_PAGE_PGNO(pPage)));
}
tdbPCacheRelease(pPager->pCache, pPage, pTxn);
}
@ -304,6 +422,9 @@ int tdbPagerPostCommit(SPager *pPager, TXN *pTxn) {
return -1;
}
if (pPager->jPageSet) {
hashset_destroy(pPager->jPageSet);
}
pPager->inTran = 0;
return 0;
@ -375,36 +496,61 @@ int tdbPagerAbort(SPager *pPager, TXN *pTxn) {
return -1;
}
tdb_fd_t jfd = tdbOsOpen(pPager->jFileName, TDB_O_RDWR, 0755);
if (jfd == NULL) {
return -1;
}
tdb_fd_t jfd = pPager->jfd;
ret = tdbGetFileSize(jfd, pPager->pageSize, &journalSize);
if (ret < 0) {
return -1;
}
// 1, read pages from jounal file
// 2, write original pages to buffered ones
u8 *pageBuf = tdbOsCalloc(1, pPager->pageSize);
if (pageBuf == NULL) {
return -1;
}
/* TODO: reset the buffered pages instead of releasing them
// loop to reset the dirty pages from file
for (pgIdx = 0, pPage = pPager->pDirty; pPage != NULL && pgIndex < journalSize; pPage = pPage->pDirtyNext, ++pgIdx) {
for (int pgIndex = 0; pgIndex < journalSize; ++pgIndex) {
// read pgno & the page from journal
SPgno pgno;
int ret = tdbOsRead(jfd, &pgno, sizeof(pgno));
if (ret < 0) {
tdbOsFree(pageBuf);
return -1;
}
ret = tdbOsRead(jfd, pageBuf, pPager->pageSize);
if (ret < 0) {
tdbOsFree(pageBuf);
return -1;
}
i64 offset = pPager->pageSize * (pgno - 1);
if (tdbOsLSeek(pPager->fd, offset, SEEK_SET) < 0) {
tdbError("failed to lseek fd due to %s. file:%s, offset:%" PRId64, strerror(errno), pPager->dbFileName, offset);
terrno = TAOS_SYSTEM_ERROR(errno);
tdbOsFree(pageBuf);
return -1;
}
ret = tdbOsWrite(pPager->fd, pageBuf, pPager->pageSize);
if (ret < 0) {
tdbError("failed to write buf due to %s. file: %s, bufsize:%d", strerror(errno), pPager->dbFileName,
pPager->pageSize);
terrno = TAOS_SYSTEM_ERROR(errno);
tdbOsFree(pageBuf);
return -1;
}
}
*/
if (tdbOsFSync(pPager->fd) < 0) {
tdbError("failed to fsync fd due to %s. dbfile:%s", strerror(errno), pPager->dbFileName);
terrno = TAOS_SYSTEM_ERROR(errno);
tdbOsFree(pageBuf);
return -1;
}
tdbOsFree(pageBuf);
// 3, release the dirty pages
SRBTreeIter iter = tRBTreeIterCreate(&pPager->rbt, 1);
SRBTreeNode *pNode = NULL;
@ -413,17 +559,55 @@ int tdbPagerAbort(SPager *pPager, TXN *pTxn) {
pPage->isDirty = 0;
tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
hashset_remove(pPager->jPageSet, (void*)((long)TDB_PAGE_PGNO(pPage)));
tdbPCacheRelease(pPager->pCache, pPage, pTxn);
}
tRBTreeCreate(&pPager->rbt, pageCmpFn);
// 4, remove the journal file
tdbOsClose(pPager->jfd);
(void)tdbOsRemove(pPager->jFileName);
hashset_destroy(pPager->jPageSet);
pPager->inTran = 0;
return 0;
}
int tdbPagerFlushPage(SPager *pPager, TXN *pTxn) {
SPage *pPage;
int ret;
// loop to write the dirty pages to file
SRBTreeIter iter = tRBTreeIterCreate(&pPager->rbt, 1);
SRBTreeNode *pNode = NULL;
while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
pPage = (SPage *)pNode;
ret = tdbPagerWritePageToDB(pPager, pPage);
if (ret < 0) {
tdbError("failed to write page to db since %s", tstrerror(terrno));
return -1;
}
}
tdbTrace("tdbttl commit:%p, %d/%d", pPager, pPager->dbOrigSize, pPager->dbFileSize);
pPager->dbOrigSize = pPager->dbFileSize;
// release the page
iter = tRBTreeIterCreate(&pPager->rbt, 1);
while ((pNode = tRBTreeIterNext(&iter)) != NULL) {
pPage = (SPage *)pNode;
pPage->isDirty = 0;
tRBTreeDrop(&pPager->rbt, (SRBTreeNode *)pPage);
tdbPCacheRelease(pPager->pCache, pPage, pTxn);
}
tRBTreeCreate(&pPager->rbt, pageCmpFn);
// 4, remove the journal file
tdbOsClose(pPager->jfd);
(void)tdbOsRemove(pPager->jFileName);
pPager->inTran = 0;
return 0;
}
@ -453,10 +637,8 @@ int tdbPagerFetchPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPa
// fetch a page container
memcpy(&pgid, pPager->fid, TDB_FILE_ID_LEN);
pgid.pgno = pgno;
pPage = tdbPCacheFetch(pPager->pCache, &pgid, pTxn);
if (pPage == NULL) {
ASSERT(0);
return -1;
while ((pPage = tdbPCacheFetch(pPager->pCache, &pgid, pTxn)) == NULL) {
tdbPagerFlushPage(pPager, pTxn);
}
tdbTrace("tdbttl fetch pager:%p", pPage->pPager);

5
source/libs/tdb/src/inc/tdbInt.h
View File

@ -384,6 +384,8 @@ struct STDB {
#endif
};
typedef struct hashset_st *hashset_t;
struct SPager {
char *dbFileName;
char *jFileName;
@ -394,7 +396,8 @@ struct SPager {
SPCache *pCache;
SPgno dbFileSize;
SPgno dbOrigSize;
SPage *pDirty;
//SPage *pDirty;
hashset_t jPageSet;
SRBTree rbt;
u8 inTran;
SPager *pNext; // used by TDB

2
source/libs/wal/src/walWrite.c
View File

@ -319,7 +319,7 @@ int32_t walEndSnapshot(SWal *pWal) {
SWalFileInfo *pInfo = taosArraySearch(pWal->fileInfoSet, &tmp, compareWalFileInfo, TD_LE);
if (pInfo) {
if (ver >= pInfo->lastVer) {
pInfo--;
pInfo++;
}
if (POINTER_DISTANCE(pInfo, pWal->fileInfoSet->pData) > 0) {
wDebug("vgId:%d, begin remove from %" PRId64, pWal->cfg.vgId, pInfo->firstVer);

5
source/util/src/tconfig.c
View File

@ -271,8 +271,11 @@ static int32_t cfgSetTimezone(SConfigItem *pItem, const char *value, ECfgSrcType
cfgStypeStr(stype), value, terrstr());
return -1;
}
pItem->stype = stype;
// apply new timezone
osSetTimezone(value);
return 0;
}

46
tests/parallel_test/cases.task
View File

@ -9,7 +9,7 @@
,,y,script,./test.sh -f tsim/user/basic.sim
,,y,script,./test.sh -f tsim/user/password.sim
,,y,script,./test.sh -f tsim/user/privilege_db.sim
,,,script,./test.sh -f tsim/user/privilege_sysinfo.sim
,,y,script,./test.sh -f tsim/user/privilege_sysinfo.sim
,,,script,./test.sh -f tsim/db/alter_option.sim
,,,script,./test.sh -f tsim/db/alter_replica_13.sim
,,,script,./test.sh -f tsim/db/alter_replica_31.sim
@ -23,16 +23,16 @@
,,,script,./test.sh -f tsim/db/create_all_options.sim
,,y,script,./test.sh -f tsim/db/delete_reuse1.sim
,,y,script,./test.sh -f tsim/db/delete_reuse2.sim
,,,script,./test.sh -f tsim/db/delete_reusevnode.sim
,,y,script,./test.sh -f tsim/db/delete_reusevnode.sim
,,y,script,./test.sh -f tsim/db/delete_reusevnode2.sim
,,,script,./test.sh -f tsim/db/delete_writing1.sim
,,,script,./test.sh -f tsim/db/delete_writing2.sim
,,y,script,./test.sh -f tsim/db/delete_writing1.sim
,,y,script,./test.sh -f tsim/db/delete_writing2.sim
,,y,script,./test.sh -f tsim/db/error1.sim
,,y,script,./test.sh -f tsim/db/keep.sim
,,y,script,./test.sh -f tsim/db/len.sim
,,y,script,./test.sh -f tsim/db/repeat.sim
,,y,script,./test.sh -f tsim/db/show_create_db.sim
,,,script,./test.sh -f tsim/db/show_create_table.sim
,,y,script,./test.sh -f tsim/db/show_create_table.sim
,,y,script,./test.sh -f tsim/db/tables.sim
,,y,script,./test.sh -f tsim/db/taosdlog.sim
,,,script,./test.sh -f tsim/dnode/balance_replica1.sim
@ -80,8 +80,8 @@
,,y,script,./test.sh -f tsim/insert/query_multi_file.sim
,,y,script,./test.sh -f tsim/insert/tcp.sim
,,y,script,./test.sh -f tsim/insert/update0.sim
,,,script,./test.sh -f tsim/insert/update1_sort_merge.sim
,,,script,./test.sh -f tsim/parser/alter__for_community_version.sim
,,y,script,./test.sh -f tsim/insert/update1_sort_merge.sim
,,y,script,./test.sh -f tsim/parser/alter__for_community_version.sim
,,y,script,./test.sh -f tsim/parser/alter_column.sim
,,y,script,./test.sh -f tsim/parser/alter_stable.sim
,,y,script,./test.sh -f tsim/parser/alter.sim
@ -92,21 +92,21 @@
,,y,script,./test.sh -f tsim/parser/binary_escapeCharacter.sim
,,,script,./test.sh -f tsim/parser/col_arithmetic_operation.sim
,,y,script,./test.sh -f tsim/parser/columnValue_bigint.sim
,,,script,./test.sh -f tsim/parser/columnValue_bool.sim
,,y,script,./test.sh -f tsim/parser/columnValue_bool.sim
,,y,script,./test.sh -f tsim/parser/columnValue_double.sim
,,y,script,./test.sh -f tsim/parser/columnValue_float.sim
,,y,script,./test.sh -f tsim/parser/columnValue_int.sim
,,y,script,./test.sh -f tsim/parser/columnValue_smallint.sim
,,y,script,./test.sh -f tsim/parser/columnValue_tinyint.sim
,,,script,./test.sh -f tsim/parser/columnValue_unsign.sim
,,,script,./test.sh -f tsim/parser/commit.sim
,,,script,./test.sh -f tsim/parser/condition.sim
,,y,script,./test.sh -f tsim/parser/commit.sim
,,y,script,./test.sh -f tsim/parser/condition.sim
,,y,script,./test.sh -f tsim/parser/constCol.sim
,,,script,./test.sh -f tsim/parser/create_db.sim
,,,script,./test.sh -f tsim/parser/create_mt.sim
,,y,script,./test.sh -f tsim/parser/create_db.sim
,,y,script,./test.sh -f tsim/parser/create_mt.sim
,,y,script,./test.sh -f tsim/parser/create_tb_with_tag_name.sim
,,y,script,./test.sh -f tsim/parser/create_tb.sim
,,,script,./test.sh -f tsim/parser/dbtbnameValidate.sim
,,y,script,./test.sh -f tsim/parser/dbtbnameValidate.sim
,,y,script,./test.sh -f tsim/parser/distinct.sim
,,y,script,./test.sh -f tsim/parser/fill_us.sim
,,,script,./test.sh -f tsim/parser/fill.sim
@ -120,9 +120,9 @@
,,y,script,./test.sh -f tsim/parser/import_commit1.sim
,,y,script,./test.sh -f tsim/parser/import_commit2.sim
,,y,script,./test.sh -f tsim/parser/import_commit3.sim
,,,script,./test.sh -f tsim/parser/import_file.sim
,,y,script,./test.sh -f tsim/parser/import_file.sim
,,y,script,./test.sh -f tsim/parser/import.sim
,,,script,./test.sh -f tsim/parser/insert_multiTbl.sim
,,y,script,./test.sh -f tsim/parser/insert_multiTbl.sim
,,y,script,./test.sh -f tsim/parser/insert_tb.sim
,,,script,./test.sh -f tsim/parser/join_manyblocks.sim
,,y,script,./test.sh -f tsim/parser/join_multitables.sim
@ -131,12 +131,12 @@
,,y,script,./test.sh -f tsim/parser/last_cache.sim
,,y,script,./test.sh -f tsim/parser/last_groupby.sim
,,y,script,./test.sh -f tsim/parser/lastrow.sim
,,,script,./test.sh -f tsim/parser/lastrow2.sim
,,y,script,./test.sh -f tsim/parser/lastrow2.sim
,,y,script,./test.sh -f tsim/parser/like.sim
,,,script,./test.sh -f tsim/parser/limit.sim
,,,script,./test.sh -f tsim/parser/limit1.sim
,,y,script,./test.sh -f tsim/parser/mixed_blocks.sim
,,,script,./test.sh -f tsim/parser/nchar.sim
,,y,script,./test.sh -f tsim/parser/nchar.sim
,,,script,./test.sh -f tsim/parser/nestquery.sim
,,,script,./test.sh -f tsim/parser/null_char.sim
,,y,script,./test.sh -f tsim/parser/precision_ns.sim
@ -146,7 +146,7 @@
,,y,script,./test.sh -f tsim/parser/select_distinct_tag.sim
,,y,script,./test.sh -f tsim/parser/select_from_cache_disk.sim
,,y,script,./test.sh -f tsim/parser/select_with_tags.sim
,,,script,./test.sh -f tsim/parser/selectResNum.sim
,,y,script,./test.sh -f tsim/parser/selectResNum.sim
,,,script,./test.sh -f tsim/parser/set_tag_vals.sim
,,y,script,./test.sh -f tsim/parser/single_row_in_tb.sim
,,,script,./test.sh -f tsim/parser/sliding.sim
@ -271,7 +271,7 @@
,,y,script,./test.sh -f tsim/stable/tag_filter.sim
,,y,script,./test.sh -f tsim/stable/tag_modify.sim
,,y,script,./test.sh -f tsim/stable/tag_rename.sim
,,,script,./test.sh -f tsim/stable/values.sim
,,y,script,./test.sh -f tsim/stable/values.sim
,,y,script,./test.sh -f tsim/stable/vnode3.sim
,,y,script,./test.sh -f tsim/stable/metrics_idx.sim
,,,script,./test.sh -f tsim/sma/drop_sma.sim
@ -325,7 +325,7 @@
,,y,script,./test.sh -f tsim/compute/bottom.sim
,,y,script,./test.sh -f tsim/compute/count.sim
,,y,script,./test.sh -f tsim/compute/diff.sim
,,,script,./test.sh -f tsim/compute/diff2.sim
,,y,script,./test.sh -f tsim/compute/diff2.sim
,,y,script,./test.sh -f tsim/compute/first.sim
,,y,script,./test.sh -f tsim/compute/interval.sim
,,y,script,./test.sh -f tsim/compute/last_row.sim
@ -358,7 +358,7 @@
,,y,script,./test.sh -f tsim/vector/metrics_query.sim
,,y,script,./test.sh -f tsim/vector/metrics_tag.sim
,,y,script,./test.sh -f tsim/vector/metrics_time.sim
,,,script,./test.sh -f tsim/vector/multi.sim
,,y,script,./test.sh -f tsim/vector/multi.sim
,,y,script,./test.sh -f tsim/vector/single.sim
,,y,script,./test.sh -f tsim/vector/table_field.sim
,,y,script,./test.sh -f tsim/vector/table_mix.sim
@ -380,7 +380,7 @@
,,y,script,./test.sh -f tsim/tag/column.sim
,,y,script,./test.sh -f tsim/tag/commit.sim
,,y,script,./test.sh -f tsim/tag/create.sim
,,,script,./test.sh -f tsim/tag/delete.sim
,,y,script,./test.sh -f tsim/tag/delete.sim
,,y,script,./test.sh -f tsim/tag/double.sim
,,y,script,./test.sh -f tsim/tag/filter.sim
,,y,script,./test.sh -f tsim/tag/float.sim
@ -392,7 +392,7 @@
,,y,script,./test.sh -f tsim/tag/tinyint.sim
,,y,script,./test.sh -f tsim/tag/drop_tag.sim
,,y,script,./test.sh -f tsim/tag/tbNameIn.sim
,,,script,./test.sh -f tmp/monitor.sim
,,y,script,./test.sh -f tmp/monitor.sim
#system test

2
tests/script/sh/checkAsan.sh
View File

@ -20,7 +20,7 @@ LOG_DIR=$TAOS_DIR/sim/tsim/asan
error_num=`cat ${LOG_DIR}/*.asan | grep "ERROR" | wc -l`
memory_leak=`cat ${LOG_DIR}/*.asan | grep "Direct leak" | wc -l`
indirect_leak=`cat ${LOG_DIR}/*.asan | grep "Indirect leak" | wc -l`
runtime_error=`cat ${LOG_DIR}/*.asan | grep "runtime error" | wc -l`
runtime_error=`cat ${LOG_DIR}/*.asan | grep "runtime error" | grep -v "trees.c:873" | wc -l`
echo -e "\033[44;32;1m"asan error_num: $error_num"\033[0m"
echo -e "\033[44;32;1m"asan memory_leak: $memory_leak"\033[0m"