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

Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/stream_compression

This commit is contained in:
Hongze Cheng 2022-09-29 00:02:42 +08:00
parent cf0aecd3dd 5d7bafad17
commit 90915f1da3
14 changed files with 157 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -872,7 +872,7 @@ SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) {
pSmaIdxKey = (SSmaIdxKey *)pCur->pKey;
if (taosArrayPush(pUids, &pSmaIdxKey->smaUid) < 0) {
if (!taosArrayPush(pUids, &pSmaIdxKey->smaUid)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
metaCloseSmaCursor(pCur);
taosArrayDestroy(pUids);
@ -915,7 +915,7 @@ SArray *metaGetSmaTbUids(SMeta *pMeta) {
}
}
if (taosArrayPush(pUids, &uid) < 0) {
if (!taosArrayPush(pUids, &uid)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
metaCloseSmaCursor(pCur);
taosArrayDestroy(pUids);

2
source/dnode/vnode/src/sma/smaCommit.c
View File

@ -213,7 +213,7 @@ static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pStat) {
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), committed, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName);
if (taosCheckExistFile(qTaskInfoFullName)) {
SQTaskFile qFile = {.nRef = 1, .padding = 0, .version = committed, .size = 0};
if (taosArrayPush(pFS->aQTaskInf, &qFile) < 0) {
if (!taosArrayPush(pFS->aQTaskInf, &qFile)) {
taosWUnLockLatch(RSMA_FS_LOCK(pStat));
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_FAILED;

2
source/dnode/vnode/src/sma/smaEnv.c
View File

@ -386,7 +386,7 @@ int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType) {
}
break;
default:
smaError("vgId:%d, undefined smaType:%", SMA_VID(pSma), smaType);
smaError("vgId:%d, undefined smaType:%" PRIi8, SMA_VID(pSma), smaType);
return TSDB_CODE_FAILED;
}

2
source/dnode/vnode/src/sma/smaFS.c
View File

@ -54,7 +54,7 @@ int32_t tdRSmaFSOpen(SSma *pSma, int64_t version) {
if ((terrno = tdRSmaFSUpsertQTaskFile(RSMA_FS(pStat), &qTaskFile)) < 0) {
goto _end;
}
smaInfo("vgId:%d, open fs, version:%" PRIi64 ", ref:%" PRIi64, TD_VID(pVnode), qTaskFile.version, qTaskFile.nRef);
smaInfo("vgId:%d, open fs, version:%" PRIi64 ", ref:%d", TD_VID(pVnode), qTaskFile.version, qTaskFile.nRef);
}
_end:

18
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -196,7 +196,8 @@ static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids,
if (!suid || !tbUids) {
terrno = TSDB_CODE_INVALID_PTR;
smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr());
smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64 " since %s", SMA_VID(pSma), suid ? *suid : -1,
terrstr());
return TSDB_CODE_FAILED;
}
@ -566,6 +567,7 @@ static int32_t tdUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid)
}
if (!taosArrayPush(pUidArray, uid)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosArrayDestroy(pUidArray);
return TSDB_CODE_FAILED;
}
if (taosHashPut(pStore->uidHash, &suid, sizeof(suid), &pUidArray, sizeof(pUidArray)) < 0) {
@ -1678,13 +1680,13 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
SRSmaInfoItem *pItem = NULL;
if (!(pRSmaRef = taosHashGet(smaMgmt.refHash, &param, POINTER_BYTES))) {
smaDebug("rsma fetch task not start since rsma info item:%p not exist in refHash:%p, rsetId:%" PRIi64, param,
*(int64_t *)&param, smaMgmt.refHash, smaMgmt.rsetId);
smaDebug("rsma fetch task not start since rsma info item:%p not exist in refHash:%p, rsetId:%d", param,
smaMgmt.refHash, smaMgmt.rsetId);
return;
}
if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) {
smaDebug("rsma fetch task not start since rsma stat already destroyed, rsetId:%" PRIi64 " refId:%d)",
smaDebug("rsma fetch task not start since rsma stat already destroyed, rsetId:%d refId:%" PRIi64 ")",
smaMgmt.rsetId, pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
return;
@ -1693,7 +1695,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
pSma = pStat->pSma;
if (!(pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, pRSmaRef->suid))) {
smaDebug("rsma fetch task not start since rsma info not exist, rsetId:%" PRIi64 " refId:%d)", smaMgmt.rsetId,
smaDebug("rsma fetch task not start since rsma info not exist, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
@ -1701,7 +1703,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
}
if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
smaDebug("rsma fetch task not start since rsma info already deleted, rsetId:%" PRIi64 " refId:%d)", smaMgmt.rsetId,
smaDebug("rsma fetch task not start since rsma info already deleted, rsetId:%d refId:%" PRIi64 ")", smaMgmt.rsetId,
pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
tdReleaseRSmaInfo(pSma, pRSmaInfo);
tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
@ -1717,7 +1719,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
case TASK_TRIGGER_STAT_PAUSED:
case TASK_TRIGGER_STAT_CANCELLED: {
smaDebug("vgId:%d, rsma fetch task not start for level %" PRIi8 " since stat is %" PRIi8
", rsetId rsetId:%" PRIi64 " refId:%d",
", rsetId:%d refId:%" PRIi64,
SMA_VID(pSma), pItem->level, rsmaTriggerStat, smaMgmt.rsetId, pRSmaRef->refId);
if (rsmaTriggerStat == TASK_TRIGGER_STAT_PAUSED) {
taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
@ -1845,7 +1847,7 @@ static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SA
void *msg = NULL;
taosGetQitem(qall, (void **)&msg);
if (msg) {
if (taosArrayPush(pSubmitArr, &msg) < 0) {
if (!taosArrayPush(pSubmitArr, &msg)) {
tdFreeRSmaSubmitItems(pSubmitArr);
goto _err;
}

10
source/dnode/vnode/src/sma/smaSnapshot.c
View File

@ -70,6 +70,8 @@ int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapRead
return TSDB_CODE_SUCCESS;
_err:
if (pReader) rsmaSnapReaderClose(&pReader);
*ppReader = NULL;
smaError("vgId:%d, vnode snapshot rsma reader open failed since %s", TD_VID(pVnode), tstrerror(code));
return TSDB_CODE_FAILED;
}
@ -101,8 +103,8 @@ static int32_t rsmaQTaskInfSnapReaderOpen(SRSmaSnapReader* pReader, int64_t vers
if (!taosCheckExistFile(qTaskInfoFullName)) {
tdRSmaFSUnRef(pSma, pStat, version);
smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as %s not exists",
TD_VID(pVnode), qTaskInfoFullName);
smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as %s not exist",
TD_VID(pVnode), version, qTaskInfoFullName);
return TSDB_CODE_SUCCESS;
}
@ -336,6 +338,7 @@ int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWrit
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), 0, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName);
TdFilePtr qTaskF = taosCreateFile(qTaskInfoFullName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (!qTaskF) {
taosMemoryFree(qWriter);
code = TAOS_SYSTEM_ERROR(errno);
smaError("vgId:%d, rsma snapshot writer open %s failed since %s", TD_VID(pSma->pVnode), qTaskInfoFullName,
tstrerror(code));
@ -356,6 +359,7 @@ int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWrit
_err:
smaError("vgId:%d, rsma snapshot writer open failed since %s", TD_VID(pSma->pVnode), tstrerror(code));
if (pWriter) rsmaSnapWriterClose(&pWriter, 0);
*ppWriter = NULL;
return code;
}
@ -449,11 +453,11 @@ static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData,
code = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
smaInfo("vgId:%d, vnode snapshot rsma write qtaskinfo %s succeed", SMA_VID(pWriter->pSma), qWriter->fname);
} else {
smaInfo("vgId:%d, vnode snapshot rsma write qtaskinfo is not needed", SMA_VID(pWriter->pSma));
}
smaInfo("vgId:%d, vnode snapshot rsma write qtaskinfo %s succeed", SMA_VID(pWriter->pSma), qWriter->fname);
_exit:
return code;

8
source/dnode/vnode/src/sma/smaUtil.c
View File

@ -290,19 +290,19 @@ int32_t tdRemoveTFile(STFile *pTFile) {
void *tdAcquireSmaRef(int32_t rsetId, int64_t refId) {
void *pResult = taosAcquireRef(rsetId, refId);
if (!pResult) {
smaWarn("rsma acquire ref for rsetId:%" PRIi64 " refId:%d failed since %s", rsetId, refId, terrstr());
smaWarn("rsma acquire ref for rsetId:%d refId:%" PRIi64 " failed since %s", rsetId, refId, terrstr());
} else {
smaDebug("rsma acquire ref for rsetId:%" PRIi64 " refId:%d success", rsetId, refId);
smaDebug("rsma acquire ref for rsetId:%d refId:%" PRIi64 " success", rsetId, refId);
}
return pResult;
}
int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId) {
if (taosReleaseRef(rsetId, refId) < 0) {
smaWarn("rsma release ref for rsetId:%" PRIi64 " refId:%d failed since %s", rsetId, refId, terrstr());
smaWarn("rsma release ref for rsetId:%d refId:%" PRIi64 " failed since %s", rsetId, refId, terrstr());
return TSDB_CODE_FAILED;
}
smaDebug("rsma release ref for rsetId:%" PRIi64 " refId:%d success", rsetId, refId);
smaDebug("rsma release ref for rsetId:%d refId:%" PRIi64 " success", rsetId, refId);
return TSDB_CODE_SUCCESS;
}

4
source/libs/parser/src/parAstParser.c
View File

@ -407,6 +407,10 @@ static int32_t collectMetaKeyFromShowTags(SCollectMetaKeyCxt* pCxt, SShowStmt* p
if (TSDB_CODE_SUCCESS == code) {
code = reserveDbVgInfoInCache(pCxt->pParseCxt->acctId, ((SValueNode*)pStmt->pDbName)->literal, pCxt->pMetaCache);
}
if (TSDB_CODE_SUCCESS == code && NULL != pStmt->pTbName) {
code = reserveTableVgroupInCache(pCxt->pParseCxt->acctId, ((SValueNode*)pStmt->pDbName)->literal,
((SValueNode*)pStmt->pTbName)->literal, pCxt->pMetaCache);
}
return code;
}

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

@ -2141,10 +2141,114 @@ static bool sysTableFromVnode(const char* pTable) {
static bool sysTableFromDnode(const char* pTable) { return 0 == strcmp(pTable, TSDB_INS_TABLE_DNODE_VARIABLES); }
static int32_t getTagsTableVgroupListImpl(STranslateContext* pCxt, SName* pTargetName, SName* pName,
SArray** pVgroupList) {
if (0 == pTargetName->type) {
return getDBVgInfoImpl(pCxt, pName, pVgroupList);
}
if (TSDB_DB_NAME_T == pTargetName->type) {
return getDBVgInfoImpl(pCxt, pTargetName, pVgroupList);
}
SVgroupInfo vgInfo = {0};
int32_t code = getTableHashVgroupImpl(pCxt, pTargetName, &vgInfo);
if (TSDB_CODE_SUCCESS == code) {
*pVgroupList = taosArrayInit(1, sizeof(SVgroupInfo));
if (NULL == *pVgroupList) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
if (TSDB_CODE_SUCCESS == code) {
taosArrayPush(*pVgroupList, &vgInfo);
}
return code;
}
static int32_t getTagsTableTargetNameFromOp(STranslateContext* pCxt, SOperatorNode* pOper, SName* pName) {
if (OP_TYPE_EQUAL != pOper->opType) {
return TSDB_CODE_SUCCESS;
}
SColumnNode* pCol = NULL;
SValueNode* pVal = NULL;
if (QUERY_NODE_COLUMN == nodeType(pOper->pLeft)) {
pCol = (SColumnNode*)pOper->pLeft;
} else if (QUERY_NODE_VALUE == nodeType(pOper->pLeft)) {
pVal = (SValueNode*)pOper->pLeft;
}
if (QUERY_NODE_COLUMN == nodeType(pOper->pRight)) {
pCol = (SColumnNode*)pOper->pRight;
} else if (QUERY_NODE_VALUE == nodeType(pOper->pRight)) {
pVal = (SValueNode*)pOper->pRight;
}
if (NULL == pCol || NULL == pVal) {
return TSDB_CODE_SUCCESS;
}
if (0 == strcmp(pCol->colName, "db_name")) {
return tNameSetDbName(pName, pCxt->pParseCxt->acctId, pVal->literal, strlen(pVal->literal));
} else if (0 == strcmp(pCol->colName, "table_name")) {
return tNameAddTbName(pName, pVal->literal, strlen(pVal->literal));
}
return TSDB_CODE_SUCCESS;
}
static void getTagsTableTargetObjName(STranslateContext* pCxt, SNode* pNode, SName* pName) {
if (QUERY_NODE_OPERATOR == nodeType(pNode)) {
getTagsTableTargetNameFromOp(pCxt, (SOperatorNode*)pNode, pName);
}
}
static int32_t getTagsTableTargetNameFromCond(STranslateContext* pCxt, SLogicConditionNode* pCond, SName* pName) {
if (LOGIC_COND_TYPE_AND != pCond->condType) {
return TSDB_CODE_SUCCESS;
}
SNode* pNode = NULL;
FOREACH(pNode, pCond->pParameterList) { getTagsTableTargetObjName(pCxt, pNode, pName); }
if ('\0' == pName->dbname[0]) {
pName->type = 0;
}
return TSDB_CODE_SUCCESS;
}
static int32_t getTagsTableTargetName(STranslateContext* pCxt, SNode* pWhere, SName* pName) {
if (NULL == pWhere) {
return TSDB_CODE_SUCCESS;
}
if (QUERY_NODE_OPERATOR == nodeType(pWhere)) {
return getTagsTableTargetNameFromOp(pCxt, (SOperatorNode*)pWhere, pName);
}
if (QUERY_NODE_LOGIC_CONDITION == nodeType(pWhere)) {
return getTagsTableTargetNameFromCond(pCxt, (SLogicConditionNode*)pWhere, pName);
}
return TSDB_CODE_SUCCESS;
}
static int32_t getTagsTableVgroupList(STranslateContext* pCxt, SName* pName, SArray** pVgroupList) {
if (!isSelectStmt(pCxt->pCurrStmt)) {
return TSDB_CODE_SUCCESS;
}
SSelectStmt* pSelect = (SSelectStmt*)pCxt->pCurrStmt;
SName targetName = {0};
int32_t code = getTagsTableTargetName(pCxt, pSelect->pWhere, &targetName);
if (TSDB_CODE_SUCCESS == code) {
code = getTagsTableVgroupListImpl(pCxt, &targetName, pName, pVgroupList);
}
return code;
}
static int32_t setVnodeSysTableVgroupList(STranslateContext* pCxt, SName* pName, SRealTableNode* pRealTable) {
int32_t code = TSDB_CODE_SUCCESS;
SArray* vgroupList = NULL;
if ('\0' != pRealTable->qualDbName[0]) {
if (0 == strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_TAGS)) {
code = getTagsTableVgroupList(pCxt, pName, &vgroupList);
} else if ('\0' != pRealTable->qualDbName[0]) {
if (0 != strcmp(pRealTable->qualDbName, TSDB_INFORMATION_SCHEMA_DB)) {
code = getDBVgInfo(pCxt, pRealTable->qualDbName, &vgroupList);
}
@ -2152,14 +2256,12 @@ static int32_t setVnodeSysTableVgroupList(STranslateContext* pCxt, SName* pName,
code = getDBVgInfoImpl(pCxt, pName, &vgroupList);
}
if (TSDB_CODE_SUCCESS == code && 0 == strcmp(pRealTable->table.dbName, TSDB_INFORMATION_SCHEMA_DB) &&
0 == strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_TAGS) && isSelectStmt(pCxt->pCurrStmt) &&
0 == taosArrayGetSize(vgroupList)) {
if (TSDB_CODE_SUCCESS == code && 0 == strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_TAGS) &&
isSelectStmt(pCxt->pCurrStmt) && 0 == taosArrayGetSize(vgroupList)) {
((SSelectStmt*)pCxt->pCurrStmt)->isEmptyResult = true;
}
if (TSDB_CODE_SUCCESS == code && 0 == strcmp(pRealTable->table.dbName, TSDB_INFORMATION_SCHEMA_DB) &&
0 == strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_TABLES)) {
if (TSDB_CODE_SUCCESS == code && 0 == strcmp(pRealTable->table.tableName, TSDB_INS_TABLE_TABLES)) {
code = addMnodeToVgroupList(&pCxt->pParseCxt->mgmtEpSet, &vgroupList);
}

4
source/libs/parser/test/mockCatalog.cpp
View File

@ -85,6 +85,10 @@ void generateInformationSchema(MockCatalogService* mcs) {
.addColumn("dnode_id", TSDB_DATA_TYPE_INT)
.addColumn("dnode_ep", TSDB_DATA_TYPE_BINARY, TSDB_EP_LEN)
.done();
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_TAGS, TSDB_SYSTEM_TABLE, 2)
.addColumn("table_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.done();
}
void generatePerformanceSchema(MockCatalogService* mcs) {

10
source/libs/parser/test/parShowToUse.cpp
View File

@ -196,6 +196,12 @@ TEST_F(ParserShowToUseTest, showTableDistributed) {
run("SHOW TABLE DISTRIBUTED st1");
}
TEST_F(ParserShowToUseTest, showTags) {
useDb("root", "test");
run("SHOW TAGS FROM st1s1");
}
// todo SHOW topics
TEST_F(ParserShowToUseTest, showUsers) {
@ -213,9 +219,9 @@ TEST_F(ParserShowToUseTest, showVariables) {
TEST_F(ParserShowToUseTest, showVgroups) {
useDb("root", "test");
run("SHOW vgroups");
run("SHOW VGROUPS");
run("SHOW test.vgroups");
run("SHOW test.VGROUPS");
}
TEST_F(ParserShowToUseTest, showVnodes) {

4
source/libs/parser/test/parTestMain.cpp
View File

@ -86,6 +86,7 @@ static void parseArg(int argc, char* argv[]) {
{"dump", no_argument, NULL, 'd'},
{"async", required_argument, NULL, 'a'},
{"skipSql", required_argument, NULL, 's'},
{"log", required_argument, NULL, 'l'},
{0, 0, 0, 0}
};
// clang-format on
@ -100,6 +101,9 @@ static void parseArg(int argc, char* argv[]) {
case 's':
setSkipSqlNum(optarg);
break;
case 'l':
setLogLevel(optarg);
break;
default:
break;
}

2
source/libs/planner/test/planOtherTest.cpp
View File

@ -84,6 +84,8 @@ TEST_F(PlanOtherTest, show) {
run("SHOW TABLE DISTRIBUTED st1");
run("SHOW DNODE 1 VARIABLES");
run("SHOW TAGS FROM st1s1");
}
TEST_F(PlanOtherTest, delete) {

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

@ -29,7 +29,7 @@ SStreamState* streamStateOpen(char* path, SStreamTask* pTask, bool specPath) {
if (!specPath) {
sprintf(statePath, "%s/%d", path, pTask->taskId);
} else {
memcpy(statePath, path, 300);
strncpy(statePath, path, 300);
}
if (tdbOpen(statePath, 4096, 256, &pState->db) < 0) {
goto _err;