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 feature/vnode_refact1

This commit is contained in:
Hongze Cheng 2022-05-05 10:57:23 +00:00
parent 6818ee3a94 f49524daa1
commit 930ce6a0d5
16 changed files with 639 additions and 1911 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cmake/cmake.define
View File

@ -67,7 +67,13 @@ ELSE ()
IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64" OR ${CMAKE_SYSTEM_PROCESSOR} MATCHES "aarch64") IF (${CMAKE_SYSTEM_PROCESSOR} MATCHES "arm64" OR ${CMAKE_SYSTEM_PROCESSOR} MATCHES "aarch64")
ADD_DEFINITIONS("-D_TD_ARM_") ADD_DEFINITIONS("-D_TD_ARM_")
ELSE () ELSE ()
ADD_DEFINITIONS("-msse4.2 -mfma") ADD_DEFINITIONS("-msse4.2")
IF("${FMA_SUPPORT}" MATCHES "true")
MESSAGE(STATUS "turn fma function support on")
ADD_DEFINITIONS("-mfma")
ELSE ()
MESSAGE(STATUS "turn fma function support off")
ENDIF()
ENDIF () ENDIF ()
ENDIF () ENDIF ()

68
include/libs/parser/parser.h
View File

@ -20,8 +20,8 @@
extern "C" { extern "C" {
#endif #endif
#include "querynodes.h"
#include "query.h" #include "query.h"
#include "querynodes.h"
typedef struct SStmtCallback { typedef struct SStmtCallback {
TAOS_STMT* pStmt; TAOS_STMT* pStmt;
@ -34,24 +34,26 @@ typedef struct SStmtCallback {
typedef struct SParseContext { typedef struct SParseContext {
uint64_t requestId; uint64_t requestId;
int32_t acctId; int32_t acctId;
const char *db; const char* db;
bool topicQuery; bool topicQuery;
void *pTransporter; void* pTransporter;
SEpSet mgmtEpSet; SEpSet mgmtEpSet;
const char *pSql; // sql string const char* pSql; // sql string
size_t sqlLen; // length of the sql string size_t sqlLen; // length of the sql string
char *pMsg; // extended error message if exists to help identifying the problem in sql statement. char* pMsg; // extended error message if exists to help identifying the problem in sql statement.
int32_t msgLen; // max length of the msg int32_t msgLen; // max length of the msg
struct SCatalog *pCatalog; struct SCatalog* pCatalog;
SStmtCallback *pStmtCb; SStmtCallback* pStmtCb;
const char* pUser;
bool isSuperUser;
} SParseContext; } SParseContext;
typedef struct SCmdMsgInfo { typedef struct SCmdMsgInfo {
int16_t msgType; int16_t msgType;
SEpSet epSet; SEpSet epSet;
void* pMsg; void* pMsg;
int32_t msgLen; int32_t msgLen;
void* pExtension; // todo remove it soon void* pExtension; // todo remove it soon
} SCmdMsgInfo; } SCmdMsgInfo;
typedef enum EQueryExecMode { typedef enum EQueryExecMode {
@ -63,21 +65,21 @@ typedef enum EQueryExecMode {
typedef struct SQuery { typedef struct SQuery {
EQueryExecMode execMode; EQueryExecMode execMode;
bool haveResultSet; bool haveResultSet;
SNode* pRoot; SNode* pRoot;
int32_t numOfResCols; int32_t numOfResCols;
SSchema* pResSchema; SSchema* pResSchema;
int8_t precision; int8_t precision;
SCmdMsgInfo* pCmdMsg; SCmdMsgInfo* pCmdMsg;
int32_t msgType; int32_t msgType;
SArray* pDbList; SArray* pDbList;
SArray* pTableList; SArray* pTableList;
bool showRewrite; bool showRewrite;
int32_t placeholderNum; int32_t placeholderNum;
} SQuery; } SQuery;
int32_t qParseQuerySql(SParseContext* pCxt, SQuery** pQuery); int32_t qParseQuerySql(SParseContext* pCxt, SQuery** pQuery);
bool isInsertSql(const char* pStr, size_t length); bool isInsertSql(const char* pStr, size_t length);
void qDestroyQuery(SQuery* pQueryNode); void qDestroyQuery(SQuery* pQueryNode);
@ -89,14 +91,16 @@ int32_t qCloneStmtDataBlock(void** pDst, void* pSrc);
void qFreeStmtDataBlock(void* pDataBlock); void qFreeStmtDataBlock(void* pDataBlock);
int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc); int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc);
void qDestroyStmtDataBlock(void* pBlock); void qDestroyStmtDataBlock(void* pBlock);
int32_t qBindStmtColsValue(void *pBlock, TAOS_MULTI_BIND *bind, char *msgBuf, int32_t msgBufLen); int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen);
int32_t qBindStmtSingleColValue(void *pBlock, TAOS_MULTI_BIND *bind, char *msgBuf, int32_t msgBufLen, int32_t colIdx, int32_t rowNum); int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx,
int32_t qBuildStmtColFields(void *pDataBlock, int32_t *fieldNum, TAOS_FIELD** fields); int32_t rowNum);
int32_t qBuildStmtTagFields(void *pBlock, void *boundTags, int32_t *fieldNum, TAOS_FIELD** fields); int32_t qBuildStmtColFields(void* pDataBlock, int32_t* fieldNum, TAOS_FIELD** fields);
int32_t qBindStmtTagsValue(void *pBlock, void *boundTags, int64_t suid, SName *pName, TAOS_MULTI_BIND *bind, char *msgBuf, int32_t msgBufLen); int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD** fields);
void destroyBoundColumnInfo(void* pBoundInfo); int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, SName* pName, TAOS_MULTI_BIND* bind,
int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char *msgBuf, int32_t msgBufLen); char* msgBuf, int32_t msgBufLen);
void destroyBoundColumnInfo(void* pBoundInfo);
int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf,
int32_t msgBufLen);
#ifdef __cplusplus #ifdef __cplusplus
} }

18
source/client/src/clientStmt.c
View File

@ -649,6 +649,19 @@ int stmtGetParamNum(TAOS_STMT *stmt, int *nums) {
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS)); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 && STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
pStmt->bInfo.needParse = false;
}
if (pStmt->exec.pRequest && STMT_TYPE_QUERY == pStmt->sql.type && pStmt->sql.runTimes) {
taos_free_result(pStmt->exec.pRequest);
pStmt->exec.pRequest = NULL;
}
if (NULL == pStmt->exec.pRequest) {
STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
}
if (pStmt->bInfo.needParse) { if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt)); STMT_ERR_RET(stmtParseSql(pStmt));
} }
@ -658,8 +671,11 @@ int stmtGetParamNum(TAOS_STMT *stmt, int *nums) {
STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList)); STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList));
pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag; pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag;
pStmt->exec.pRequest->body.pDag = NULL; pStmt->exec.pRequest->body.pDag = NULL;
STMT_ERR_RET(stmtBackupQueryFields(pStmt));
} else {
STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
} }
*nums = taosArrayGetSize(pStmt->sql.pQueryPlan->pPlaceholderValues); *nums = taosArrayGetSize(pStmt->sql.pQueryPlan->pPlaceholderValues);
} else { } else {
STMT_ERR_RET(stmtFetchColFields(stmt, nums, NULL)); STMT_ERR_RET(stmtFetchColFields(stmt, nums, NULL));

47
source/common/src/tglobal.c
View File

@ -224,7 +224,8 @@ struct SConfig *taosGetCfg() {
return tsCfg; return tsCfg;
} }
static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *inputCfgDir, const char *envFile, char *apolloUrl) { static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *inputCfgDir, const char *envFile,
char *apolloUrl) {
char cfgDir[PATH_MAX] = {0}; char cfgDir[PATH_MAX] = {0};
char cfgFile[PATH_MAX + 100] = {0}; char cfgFile[PATH_MAX + 100] = {0};
@ -300,15 +301,10 @@ static int32_t taosAddServerLogCfg(SConfig *pCfg) {
static int32_t taosAddClientCfg(SConfig *pCfg) { static int32_t taosAddClientCfg(SConfig *pCfg) {
char defaultFqdn[TSDB_FQDN_LEN] = {0}; char defaultFqdn[TSDB_FQDN_LEN] = {0};
int32_t defaultServerPort = 6030; int32_t defaultServerPort = 6030;
char defaultFirstEp[TSDB_EP_LEN] = {0};
char defaultSecondEp[TSDB_EP_LEN] = {0};
if (taosGetFqdn(defaultFqdn) != 0) return -1; if (taosGetFqdn(defaultFqdn) != 0) return -1;
snprintf(defaultFirstEp, TSDB_EP_LEN, "%s:%d", defaultFqdn, defaultServerPort);
snprintf(defaultSecondEp, TSDB_EP_LEN, "%s:%d", defaultFqdn, defaultServerPort);
if (cfgAddString(pCfg, "firstEp", defaultFirstEp, 1) != 0) return -1; if (cfgAddString(pCfg, "firstEp", "", 1) != 0) return -1;
if (cfgAddString(pCfg, "secondEp", defaultSecondEp, 1) != 0) return -1; if (cfgAddString(pCfg, "secondEp", "", 1) != 0) return -1;
if (cfgAddString(pCfg, "fqdn", defaultFqdn, 1) != 0) return -1; if (cfgAddString(pCfg, "fqdn", defaultFqdn, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "serverPort", defaultServerPort, 1, 65056, 1) != 0) return -1; if (cfgAddInt32(pCfg, "serverPort", defaultServerPort, 1, 65056, 1) != 0) return -1;
if (cfgAddDir(pCfg, "tempDir", tsTempDir, 1) != 0) return -1; if (cfgAddDir(pCfg, "tempDir", tsTempDir, 1) != 0) return -1;
@ -478,15 +474,18 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32; tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort); snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
char defaultFirstEp[TSDB_EP_LEN] = {0};
snprintf(defaultFirstEp, TSDB_EP_LEN, "%s:%u", tsLocalFqdn, tsServerPort);
SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp"); SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp");
SEp firstEp = {0}; SEp firstEp = {0};
taosGetFqdnPortFromEp(pFirstEpItem->str, &firstEp); taosGetFqdnPortFromEp(strlen(pFirstEpItem->str) == 0 ? defaultFirstEp : pFirstEpItem->str, &firstEp);
snprintf(tsFirst, sizeof(tsFirst), "%s:%u", firstEp.fqdn, firstEp.port); snprintf(tsFirst, sizeof(tsFirst), "%s:%u", firstEp.fqdn, firstEp.port);
cfgSetItem(pCfg, "firstEp", tsFirst, pFirstEpItem->stype); cfgSetItem(pCfg, "firstEp", tsFirst, pFirstEpItem->stype);
SConfigItem *pSecondpItem = cfgGetItem(pCfg, "secondEp"); SConfigItem *pSecondpItem = cfgGetItem(pCfg, "secondEp");
SEp secondEp = {0}; SEp secondEp = {0};
taosGetFqdnPortFromEp(pSecondpItem->str, &secondEp); taosGetFqdnPortFromEp(strlen(pSecondpItem->str) == 0 ? defaultFirstEp : pSecondpItem->str, &secondEp);
snprintf(tsSecond, sizeof(tsSecond), "%s:%u", secondEp.fqdn, secondEp.port); snprintf(tsSecond, sizeof(tsSecond), "%s:%u", secondEp.fqdn, secondEp.port);
cfgSetItem(pCfg, "secondEp", tsSecond, pSecondpItem->stype); cfgSetItem(pCfg, "secondEp", tsSecond, pSecondpItem->stype);
@ -583,8 +582,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
return 0; return 0;
} }
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd, const char *envFile, int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd,
char *apolloUrl, SArray *pArgs, bool tsc) { const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc) {
osDefaultInit(); osDefaultInit();
SConfig *pCfg = cfgInit(); SConfig *pCfg = cfgInit();
@ -636,7 +635,24 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
return 0; return 0;
} }
int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc) { static int32_t taosCheckGlobalCfg() {
uint32_t ipv4 = taosGetIpv4FromFqdn(tsLocalFqdn);
if (ipv4 == 0xffffffff) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to get ip from fqdn:%s since %s, dnode can not be initialized", tsLocalFqdn, terrstr());
return -1;
}
if (tsServerPort <= 0) {
uError("invalid server port:%u, dnode can not be initialized", tsServerPort);
return -1;
}
return 0;
}
int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs,
bool tsc) {
if (tsCfg != NULL) return 0; if (tsCfg != NULL) return 0;
tsCfg = cfgInit(); tsCfg = cfgInit();
@ -674,6 +690,11 @@ int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile
taosSetSystemCfg(tsCfg); taosSetSystemCfg(tsCfg);
cfgDumpCfg(tsCfg, tsc, false); cfgDumpCfg(tsCfg, tsc, false);
if (taosCheckGlobalCfg() != 0) {
return -1;
}
return 0; return 0;
} }

1
source/dnode/mnode/impl/inc/mndTrans.h
View File

@ -61,6 +61,7 @@ void mndTransSetDbInfo(STrans *pTrans, SDbObj *pDb);
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans); int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans);
void mndTransProcessRsp(SNodeMsg *pRsp); void mndTransProcessRsp(SNodeMsg *pRsp);
void mndTransPullup(SMnode *pMnode); void mndTransPullup(SMnode *pMnode);
int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans);
#ifdef __cplusplus #ifdef __cplusplus
} }

13
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -997,7 +997,12 @@ static int32_t mndTransSendActionMsg(SMnode *pMnode, STrans *pTrans, SArray *pAr
pAction->msgReceived = 0; pAction->msgReceived = 0;
pAction->errCode = 0; pAction->errCode = 0;
} else { } else {
if (terrno == TSDB_CODE_INVALID_PTR) rpcFreeCont(rpcMsg.pCont); pAction->msgSent = 0;
pAction->msgReceived = 0;
pAction->errCode = terrno;
if (terrno == TSDB_CODE_INVALID_PTR || terrno == TSDB_CODE_NODE_OFFLINE) {
rpcFreeCont(rpcMsg.pCont);
}
mError("trans:%d, action:%d not send since %s", pTrans->id, action, terrstr()); mError("trans:%d, action:%d not send since %s", pTrans->id, action, terrstr());
return -1; return -1;
} }
@ -1275,7 +1280,7 @@ static int32_t mndProcessTransReq(SNodeMsg *pReq) {
return 0; return 0;
} }
static int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) { int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
SArray *pArray = NULL; SArray *pArray = NULL;
if (pTrans->stage == TRN_STAGE_REDO_ACTION) { if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
pArray = pTrans->redoActions; pArray = pTrans->redoActions;
@ -1293,14 +1298,14 @@ static int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
if (pAction == NULL) continue; if (pAction == NULL) continue;
if (pAction->msgReceived == 0) { if (pAction->msgReceived == 0) {
mInfo("trans:%d, action:%d set processed", pTrans->id, i); mInfo("trans:%d, action:%d set processed for kill msg received", pTrans->id, i);
pAction->msgSent = 1; pAction->msgSent = 1;
pAction->msgReceived = 1; pAction->msgReceived = 1;
pAction->errCode = 0; pAction->errCode = 0;
} }
if (pAction->errCode != 0) { if (pAction->errCode != 0) {
mInfo("trans:%d, action:%d set processed, errCode from %s to success", pTrans->id, i, mInfo("trans:%d, action:%d set processed for kill msg received, errCode from %s to success", pTrans->id, i,
tstrerror(pAction->errCode)); tstrerror(pAction->errCode));
pAction->msgSent = 1; pAction->msgSent = 1;
pAction->msgReceived = 1; pAction->msgReceived = 1;

279
source/dnode/mnode/impl/test/trans/trans2.cpp
View File

@ -86,7 +86,7 @@ class MndTestTrans2 : public ::testing::Test {
void SetUp() override {} void SetUp() override {}
void TearDown() override {} void TearDown() override {}
int32_t CreateUserLog(const char *acct, const char *user) { int32_t CreateUserLog(const char *acct, const char *user, ETrnType type, SDbObj *pDb) {
SUserObj userObj = {0}; SUserObj userObj = {0};
taosEncryptPass_c((uint8_t *)"taosdata", strlen("taosdata"), userObj.pass); taosEncryptPass_c((uint8_t *)"taosdata", strlen("taosdata"), userObj.pass);
tstrncpy(userObj.user, user, TSDB_USER_LEN); tstrncpy(userObj.user, user, TSDB_USER_LEN);
@ -96,7 +96,7 @@ class MndTestTrans2 : public ::testing::Test {
userObj.superUser = 1; userObj.superUser = 1;
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_USER, &rpcMsg); STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, type, &rpcMsg);
SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj); SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj);
mndTransAppendRedolog(pTrans, pRedoRaw); mndTransAppendRedolog(pTrans, pRedoRaw);
sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY); sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
@ -108,13 +108,18 @@ class MndTestTrans2 : public ::testing::Test {
char *param = strdup("====> test log <====="); char *param = strdup("====> test log <=====");
mndTransSetCb(pTrans, TEST_TRANS_START_FUNC, TEST_TRANS_STOP_FUNC, param, strlen(param) + 1); mndTransSetCb(pTrans, TEST_TRANS_START_FUNC, TEST_TRANS_STOP_FUNC, param, strlen(param) + 1);
if (pDb != NULL) {
mndTransSetDbInfo(pTrans, pDb);
}
int32_t code = mndTransPrepare(pMnode, pTrans); int32_t code = mndTransPrepare(pMnode, pTrans);
mndTransDrop(pTrans); mndTransDrop(pTrans);
return code; return code;
} }
int32_t CreateUserAction(const char *acct, const char *user, bool hasUndoAction, ETrnPolicy policy) { int32_t CreateUserAction(const char *acct, const char *user, bool hasUndoAction, ETrnPolicy policy, ETrnType type,
SDbObj *pDb) {
SUserObj userObj = {0}; SUserObj userObj = {0};
taosEncryptPass_c((uint8_t *)"taosdata", strlen("taosdata"), userObj.pass); taosEncryptPass_c((uint8_t *)"taosdata", strlen("taosdata"), userObj.pass);
tstrncpy(userObj.user, user, TSDB_USER_LEN); tstrncpy(userObj.user, user, TSDB_USER_LEN);
@ -124,7 +129,7 @@ class MndTestTrans2 : public ::testing::Test {
userObj.superUser = 1; userObj.superUser = 1;
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
STrans *pTrans = mndTransCreate(pMnode, policy, TRN_TYPE_CREATE_USER, &rpcMsg); STrans *pTrans = mndTransCreate(pMnode, policy, type, &rpcMsg);
SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj); SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj);
mndTransAppendRedolog(pTrans, pRedoRaw); mndTransAppendRedolog(pTrans, pRedoRaw);
sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY); sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
@ -170,6 +175,44 @@ class MndTestTrans2 : public ::testing::Test {
mndTransAppendUndoAction(pTrans, &action); mndTransAppendUndoAction(pTrans, &action);
} }
{
void *pRsp = taosMemoryCalloc(1, 256);
strcpy((char *)pRsp, "simple rsponse");
mndTransSetRpcRsp(pTrans, pRsp, 256);
}
if (pDb != NULL) {
mndTransSetDbInfo(pTrans, pDb);
}
int32_t code = mndTransPrepare(pMnode, pTrans);
mndTransDrop(pTrans);
return code;
}
int32_t CreateUserGlobal(const char *acct, const char *user) {
SUserObj userObj = {0};
taosEncryptPass_c((uint8_t *)"taosdata", strlen("taosdata"), userObj.pass);
tstrncpy(userObj.user, user, TSDB_USER_LEN);
tstrncpy(userObj.acct, acct, TSDB_USER_LEN);
userObj.createdTime = taosGetTimestampMs();
userObj.updateTime = userObj.createdTime;
userObj.superUser = 1;
SRpcMsg rpcMsg = {0};
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_USER, &rpcMsg);
SSdbRaw *pRedoRaw = mndUserActionEncode(&userObj);
mndTransAppendRedolog(pTrans, pRedoRaw);
sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
SSdbRaw *pUndoRaw = mndUserActionEncode(&userObj);
mndTransAppendUndolog(pTrans, pUndoRaw);
sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED);
char *param = strdup("====> test log <=====");
mndTransSetCb(pTrans, TEST_TRANS_START_FUNC, TEST_TRANS_STOP_FUNC, param, strlen(param) + 1);
int32_t code = mndTransPrepare(pMnode, pTrans); int32_t code = mndTransPrepare(pMnode, pTrans);
mndTransDrop(pTrans); mndTransDrop(pTrans);
@ -189,12 +232,12 @@ TEST_F(MndTestTrans2, 01_Log) {
ASSERT_NE(pMnode, nullptr); ASSERT_NE(pMnode, nullptr);
EXPECT_EQ(CreateUserLog(acct, user1), 0); EXPECT_EQ(CreateUserLog(acct, user1, TRN_TYPE_CREATE_USER, NULL), 0);
pUser1 = mndAcquireUser(pMnode, user1); pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_NE(pUser1, nullptr); ASSERT_NE(pUser1, nullptr);
// failed to create user and rollback // failed to create user and rollback
EXPECT_EQ(CreateUserLog(acct_invalid, user2), 0); EXPECT_EQ(CreateUserLog(acct_invalid, user2, TRN_TYPE_CREATE_USER, NULL), 0);
pUser2 = mndAcquireUser(pMnode, user2); pUser2 = mndAcquireUser(pMnode, user2);
ASSERT_EQ(pUser2, nullptr); ASSERT_EQ(pUser2, nullptr);
@ -214,44 +257,46 @@ TEST_F(MndTestTrans2, 02_Action) {
ASSERT_NE(pMnode, nullptr); ASSERT_NE(pMnode, nullptr);
// failed to create user and rollback
EXPECT_EQ(CreateUserAction(acct, user1, false, TRN_POLICY_ROLLBACK), 0);
pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_EQ(pUser1, nullptr);
mndReleaseUser(pMnode, pUser1);
// create user, and fake a response
{ {
EXPECT_EQ(CreateUserAction(acct, user1, true, TRN_POLICY_ROLLBACK), 0); // failed to create user and rollback
pUser1 = mndAcquireUser(pMnode, user1); EXPECT_EQ(CreateUserAction(acct, user1, false, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_USER, NULL), 0);
ASSERT_NE(pUser1, nullptr);
mndReleaseUser(pMnode, pUser1);
transId = 4;
pTrans = mndAcquireTrans(pMnode, transId);
EXPECT_EQ(pTrans->code, TSDB_CODE_INVALID_PTR);
EXPECT_EQ(pTrans->stage, TRN_STAGE_UNDO_ACTION);
EXPECT_EQ(pTrans->failedTimes, 1);
STransAction *pAction = (STransAction *)taosArrayGet(pTrans->undoActions, action);
pAction->msgSent = 1;
SNodeMsg rspMsg = {0};
rspMsg.pNode = pMnode;
int64_t signature = transId;
signature = (signature << 32);
signature += action;
rspMsg.rpcMsg.ahandle = (void *)signature;
mndTransProcessRsp(&rspMsg);
mndReleaseTrans(pMnode, pTrans);
pUser1 = mndAcquireUser(pMnode, user1); pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_EQ(pUser1, nullptr); ASSERT_EQ(pUser1, nullptr);
mndReleaseUser(pMnode, pUser1); mndReleaseUser(pMnode, pUser1);
// create user, and fake a response
{
EXPECT_EQ(CreateUserAction(acct, user1, true, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_USER, NULL), 0);
pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_NE(pUser1, nullptr);
mndReleaseUser(pMnode, pUser1);
transId = 4;
pTrans = mndAcquireTrans(pMnode, transId);
EXPECT_EQ(pTrans->code, TSDB_CODE_INVALID_PTR);
EXPECT_EQ(pTrans->stage, TRN_STAGE_UNDO_ACTION);
EXPECT_EQ(pTrans->failedTimes, 1);
STransAction *pAction = (STransAction *)taosArrayGet(pTrans->undoActions, action);
pAction->msgSent = 1;
SNodeMsg rspMsg = {0};
rspMsg.pNode = pMnode;
int64_t signature = transId;
signature = (signature << 32);
signature += action;
rspMsg.rpcMsg.ahandle = (void *)signature;
mndTransProcessRsp(&rspMsg);
mndReleaseTrans(pMnode, pTrans);
pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_EQ(pUser1, nullptr);
mndReleaseUser(pMnode, pUser1);
}
} }
{ {
EXPECT_EQ(CreateUserAction(acct, user1, false, TRN_POLICY_RETRY), 0); EXPECT_EQ(CreateUserAction(acct, user1, false, TRN_POLICY_RETRY, TRN_TYPE_CREATE_USER, NULL), 0);
pUser1 = mndAcquireUser(pMnode, user1); pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_NE(pUser1, nullptr); ASSERT_NE(pUser1, nullptr);
mndReleaseUser(pMnode, pUser1); mndReleaseUser(pMnode, pUser1);
@ -305,4 +350,164 @@ TEST_F(MndTestTrans2, 02_Action) {
mndReleaseUser(pMnode, pUser1); mndReleaseUser(pMnode, pUser1);
} }
} }
{
EXPECT_EQ(CreateUserAction(acct, user2, true, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_USER, NULL), 0);
SUserObj *pUser2 = (SUserObj *)sdbAcquire(pMnode->pSdb, SDB_USER, user2);
ASSERT_NE(pUser2, nullptr);
mndReleaseUser(pMnode, pUser2);
{
transId = 6;
pTrans = mndAcquireTrans(pMnode, transId);
EXPECT_EQ(pTrans->code, TSDB_CODE_INVALID_PTR);
EXPECT_EQ(pTrans->stage, TRN_STAGE_UNDO_ACTION);
EXPECT_EQ(pTrans->failedTimes, 1);
SNodeMsg rspMsg = {0};
rspMsg.pNode = pMnode;
int64_t signature = transId;
signature = (signature << 32);
signature += action;
rspMsg.rpcMsg.ahandle = (void *)signature;
rspMsg.rpcMsg.code = 0;
mndTransProcessRsp(&rspMsg);
mndReleaseTrans(pMnode, pTrans);
pUser2 = mndAcquireUser(pMnode, user2);
ASSERT_NE(pUser2, nullptr);
mndReleaseUser(pMnode, pUser2);
}
{
transId = 6;
pTrans = mndAcquireTrans(pMnode, transId);
EXPECT_EQ(pTrans->code, TSDB_CODE_INVALID_PTR);
EXPECT_EQ(pTrans->stage, TRN_STAGE_UNDO_ACTION);
EXPECT_EQ(pTrans->failedTimes, 2);
STransAction *pAction = (STransAction *)taosArrayGet(pTrans->undoActions, action);
pAction->msgSent = 1;
SNodeMsg rspMsg = {0};
rspMsg.pNode = pMnode;
int64_t signature = transId;
signature = (signature << 32);
signature += action;
rspMsg.rpcMsg.ahandle = (void *)signature;
mndTransProcessRsp(&rspMsg);
mndReleaseTrans(pMnode, pTrans);
pUser2 = mndAcquireUser(pMnode, user2);
ASSERT_EQ(pUser2, nullptr);
mndReleaseUser(pMnode, pUser2);
}
}
} }
TEST_F(MndTestTrans2, 03_Kill) {
const char *acct = "root";
const char *user1 = "kill1";
const char *user2 = "kill2";
SUserObj *pUser1 = NULL;
SUserObj *pUser2 = NULL;
STrans *pTrans = NULL;
int32_t transId = 0;
int32_t action = 0;
ASSERT_NE(pMnode, nullptr);
{
EXPECT_EQ(CreateUserAction(acct, user1, true, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_USER, NULL), 0);
pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_NE(pUser1, nullptr);
mndReleaseUser(pMnode, pUser1);
transId = 7;
pTrans = mndAcquireTrans(pMnode, transId);
EXPECT_EQ(pTrans->code, TSDB_CODE_INVALID_PTR);
EXPECT_EQ(pTrans->stage, TRN_STAGE_UNDO_ACTION);
EXPECT_EQ(pTrans->failedTimes, 1);
mndKillTrans(pMnode, pTrans);
mndReleaseTrans(pMnode, pTrans);
pUser1 = mndAcquireUser(pMnode, user1);
ASSERT_EQ(pUser1, nullptr);
mndReleaseUser(pMnode, pUser1);
}
}
TEST_F(MndTestTrans2, 04_Conflict) {
const char *acct = "root";
const char *user1 = "conflict1";
const char *user2 = "conflict2";
const char *user3 = "conflict3";
const char *user4 = "conflict4";
const char *user5 = "conflict5";
const char *user6 = "conflict6";
const char *user7 = "conflict7";
const char *user8 = "conflict8";
SUserObj *pUser = NULL;
STrans *pTrans = NULL;
int32_t transId = 0;
int32_t code = 0;
ASSERT_NE(pMnode, nullptr);
{
SDbObj dbObj1 = {0};
dbObj1.uid = 9521;
strcpy(dbObj1.name, "db");
SDbObj dbObj2 = {0};
dbObj2.uid = 9522;
strcpy(dbObj2.name, "conflict db");
EXPECT_EQ(CreateUserAction(acct, user1, true, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_STB, &dbObj1), 0);
pUser = mndAcquireUser(pMnode, user1);
ASSERT_NE(pUser, nullptr);
mndReleaseUser(pMnode, pUser);
transId = 8;
pTrans = mndAcquireTrans(pMnode, transId);
EXPECT_EQ(pTrans->code, TSDB_CODE_INVALID_PTR);
EXPECT_EQ(pTrans->stage, TRN_STAGE_UNDO_ACTION);
// stb scope
EXPECT_EQ(CreateUserLog(acct, user2, TRN_TYPE_CREATE_DNODE, NULL), -1);
code = terrno;
EXPECT_EQ(code, TSDB_CODE_MND_TRANS_CONFLICT);
EXPECT_EQ(CreateUserLog(acct, user2, TRN_TYPE_CREATE_DB, &dbObj1), -1);
EXPECT_EQ(CreateUserLog(acct, user2, TRN_TYPE_CREATE_DB, &dbObj2), 0);
EXPECT_EQ(CreateUserLog(acct, user3, TRN_TYPE_CREATE_STB, &dbObj1), 0);
// db scope
pTrans->type = TRN_TYPE_CREATE_DB;
EXPECT_EQ(CreateUserLog(acct, user4, TRN_TYPE_CREATE_DNODE, NULL), -1);
EXPECT_EQ(CreateUserLog(acct, user4, TRN_TYPE_CREATE_DB, &dbObj1), -1);
EXPECT_EQ(CreateUserLog(acct, user4, TRN_TYPE_CREATE_DB, &dbObj2), 0);
EXPECT_EQ(CreateUserLog(acct, user5, TRN_TYPE_CREATE_STB, &dbObj1), -1);
EXPECT_EQ(CreateUserLog(acct, user5, TRN_TYPE_CREATE_STB, &dbObj2), 0);
// global scope
pTrans->type = TRN_TYPE_CREATE_DNODE;
EXPECT_EQ(CreateUserLog(acct, user6, TRN_TYPE_CREATE_DNODE, NULL), 0);
EXPECT_EQ(CreateUserLog(acct, user7, TRN_TYPE_CREATE_DB, &dbObj1), -1);
EXPECT_EQ(CreateUserLog(acct, user7, TRN_TYPE_CREATE_DB, &dbObj2), -1);
EXPECT_EQ(CreateUserLog(acct, user7, TRN_TYPE_CREATE_STB, &dbObj1), -1);
EXPECT_EQ(CreateUserLog(acct, user7, TRN_TYPE_CREATE_STB, &dbObj2), -1);
// global scope
pTrans->type = TRN_TYPE_CREATE_USER;
EXPECT_EQ(CreateUserLog(acct, user7, TRN_TYPE_CREATE_DB, &dbObj1), 0);
EXPECT_EQ(CreateUserLog(acct, user8, TRN_TYPE_CREATE_DB, &dbObj2), 0);
mndKillTrans(pMnode, pTrans);
mndReleaseTrans(pMnode, pTrans);
pUser = mndAcquireUser(pMnode, user1);
ASSERT_EQ(pUser, nullptr);
mndReleaseUser(pMnode, pUser);
}
}

35
source/libs/executor/src/executorimpl.c
View File

@ -4830,6 +4830,7 @@ static SArray* createSortInfo(SNodeList* pNodeList, SNodeList* pNodeListTarget);
static SArray* createIndexMap(SNodeList* pNodeList); static SArray* createIndexMap(SNodeList* pNodeList);
static SArray* extractPartitionColInfo(SNodeList* pNodeList); static SArray* extractPartitionColInfo(SNodeList* pNodeList);
static int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode); static int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode);
static void setJoinColumnInfo(SColumnInfo* pInfo, const SColumnNode* pLeftNode);
static SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) { static SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {
SInterval interval = { SInterval interval = {
@ -5624,25 +5625,29 @@ SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOf
goto _error; goto _error;
} }
pOperator->resultInfo.capacity = 4096; initResultSizeInfo(pOperator, 4096);
pOperator->resultInfo.threshold = 4096 * 0.75;
// initResultRowInf pInfo->pRes = pResBlock;
// o(&pInfo->binfo.resultRowInfo, 8); pOperator->name = "MergeJoinOperator";
pInfo->pRes = pResBlock;
pOperator->name = "JoinOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_JOIN; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_JOIN;
pOperator->blocking = false; pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED; pOperator->status = OP_NOT_OPENED;
pOperator->pExpr = pExprInfo; pOperator->pExpr = pExprInfo;
pOperator->numOfOutput = numOfCols; pOperator->numOfOutput = numOfCols;
pOperator->info = pInfo; pOperator->info = pInfo;
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
SOperatorNode* pNode = (SOperatorNode*)pOnCondition;
setJoinColumnInfo(&pInfo->leftCol, (SColumnNode*)pNode->pLeft);
setJoinColumnInfo(&pInfo->rightCol, (SColumnNode*)pNode->pRight);
pOperator->fpSet = pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doMergeJoin, NULL, NULL, destroyBasicOperatorInfo, NULL, NULL, NULL); createOperatorFpSet(operatorDummyOpenFn, doMergeJoin, NULL, NULL, destroyBasicOperatorInfo, NULL, NULL, NULL);
int32_t code = appendDownstream(pOperator, pDownstream, numOfDownstream); int32_t code = appendDownstream(pOperator, pDownstream, numOfDownstream);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
return pOperator; return pOperator;
_error: _error:
@ -5651,3 +5656,11 @@ _error:
pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
return NULL; return NULL;
} }
void setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode) {
pColumn->slotId = pColumnNode->slotId;
pColumn->type = pColumnNode->node.resType.type;
pColumn->bytes = pColumnNode->node.resType.bytes;
pColumn->precision = pColumnNode->node.resType.precision;
pColumn->scale = pColumnNode->node.resType.scale;
}

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

@ -913,7 +913,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "tbname", .name = "tbname",
.type = FUNCTION_TYPE_TBNAME, .type = FUNCTION_TYPE_TBNAME,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC, .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_SCAN_PC_FUNC,
.translateFunc = translateTbnameColumn, .translateFunc = translateTbnameColumn,
.getEnvFunc = NULL, .getEnvFunc = NULL,
.initFunc = NULL, .initFunc = NULL,

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

@ -100,6 +100,17 @@ void generateInformationSchema(MockCatalogService* mcs) {
} }
} }
/*
* Table:t1
* Field | Type | DataType | Bytes |
* ==========================================================================
* ts | column | TIMESTAMP | 8 |
* c1 | column | INT | 4 |
* c2 | column | VARCHAR | 20 |
* c3 | column | BIGINT | 8 |
* c4 | column | DOUBLE | 8 |
* c5 | column | DOUBLE | 8 |
*/
void generateTestT1(MockCatalogService* mcs) { void generateTestT1(MockCatalogService* mcs) {
ITableBuilder& builder = mcs->createTableBuilder("test", "t1", TSDB_NORMAL_TABLE, 6) ITableBuilder& builder = mcs->createTableBuilder("test", "t1", TSDB_NORMAL_TABLE, 6)
.setPrecision(TSDB_TIME_PRECISION_MILLI) .setPrecision(TSDB_TIME_PRECISION_MILLI)
@ -113,6 +124,17 @@ void generateTestT1(MockCatalogService* mcs) {
builder.done(); builder.done();
} }
/*
* Super Table: st1
* Field | Type | DataType | Bytes |
* ==========================================================================
* ts | column | TIMESTAMP | 8 |
* c1 | column | INT | 4 |
* c2 | column | VARCHAR | 20 |
* tag1 | tag | INT | 4 |
* tag2 | tag | VARCHAR | 20 |
* Child Table: st1s1, st1s2
*/
void generateTestST1(MockCatalogService* mcs) { void generateTestST1(MockCatalogService* mcs) {
ITableBuilder& builder = mcs->createTableBuilder("test", "st1", TSDB_SUPER_TABLE, 3, 2) ITableBuilder& builder = mcs->createTableBuilder("test", "st1", TSDB_SUPER_TABLE, 3, 2)
.setPrecision(TSDB_TIME_PRECISION_MILLI) .setPrecision(TSDB_TIME_PRECISION_MILLI)

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

@ -23,4 +23,6 @@ TEST_F(PlanSuperTableTest, tbname) {
useDb("root", "test"); useDb("root", "test");
run("select tbname from st1"); run("select tbname from st1");
run("select tbname, tag1, tag2 from st1");
} }

4
source/libs/planner/test/planTestMain.cpp
View File

@ -36,11 +36,11 @@ class PlannerEnv : public testing::Environment {
static void parseArg(int argc, char* argv[]) { static void parseArg(int argc, char* argv[]) {
int opt = 0; int opt = 0;
const char* optstring = ""; const char* optstring = "";
static struct option long_options[] = {{"dump", no_argument, NULL, 'd'}, {0, 0, 0, 0}}; static struct option long_options[] = {{"dump", optional_argument, NULL, 'd'}, {0, 0, 0, 0}};
while ((opt = getopt_long(argc, argv, optstring, long_options, NULL)) != -1) { while ((opt = getopt_long(argc, argv, optstring, long_options, NULL)) != -1) {
switch (opt) { switch (opt) {
case 'd': case 'd':
g_isDump = true; setDumpModule(optarg);
break; break;
default: default:
break; break;

99
source/libs/planner/test/planTestUtil.cpp
View File

@ -34,7 +34,41 @@ using namespace testing;
} \ } \
} while (0); } while (0);
bool g_isDump = false; enum DumpModule {
DUMP_MODULE_NOTHING = 1,
DUMP_MODULE_PARSER,
DUMP_MODULE_LOGIC,
DUMP_MODULE_OPTIMIZED,
DUMP_MODULE_SPLIT,
DUMP_MODULE_SCALED,
DUMP_MODULE_PHYSICAL,
DUMP_MODULE_SUBPLAN,
DUMP_MODULE_ALL
};
DumpModule g_dumpModule = DUMP_MODULE_NOTHING;
void setDumpModule(const char* pModule) {
if (NULL == pModule) {
g_dumpModule = DUMP_MODULE_ALL;
} else if (0 == strncasecmp(pModule, "parser", strlen(pModule))) {
g_dumpModule = DUMP_MODULE_PARSER;
} else if (0 == strncasecmp(pModule, "logic", strlen(pModule))) {
g_dumpModule = DUMP_MODULE_LOGIC;
} else if (0 == strncasecmp(pModule, "optimized", strlen(pModule))) {
g_dumpModule = DUMP_MODULE_OPTIMIZED;
} else if (0 == strncasecmp(pModule, "split", strlen(pModule))) {
g_dumpModule = DUMP_MODULE_SPLIT;
} else if (0 == strncasecmp(pModule, "scaled", strlen(pModule))) {
g_dumpModule = DUMP_MODULE_SCALED;
} else if (0 == strncasecmp(pModule, "physical", strlen(pModule))) {
g_dumpModule = DUMP_MODULE_PHYSICAL;
} else if (0 == strncasecmp(pModule, "subplan", strlen(pModule))) {
g_dumpModule = DUMP_MODULE_SUBPLAN;
} else if (0 == strncasecmp(pModule, "all", strlen(pModule))) {
g_dumpModule = DUMP_MODULE_PHYSICAL;
}
}
class PlannerTestBaseImpl { class PlannerTestBaseImpl {
public: public:
@ -66,11 +100,9 @@ class PlannerTestBaseImpl {
SQueryPlan* pPlan = nullptr; SQueryPlan* pPlan = nullptr;
doCreatePhysiPlan(&cxt, pLogicPlan, &pPlan); doCreatePhysiPlan(&cxt, pLogicPlan, &pPlan);
if (g_isDump) { dump(g_dumpModule);
dump();
}
} catch (...) { } catch (...) {
dump(); dump(DUMP_MODULE_ALL);
throw; throw;
} }
} }
@ -109,23 +141,48 @@ class PlannerTestBaseImpl {
res_.physiSubplans_.clear(); res_.physiSubplans_.clear();
} }
void dump() { void dump(DumpModule module) {
if (DUMP_MODULE_NOTHING == module) {
return;
}
cout << "==========================================sql : [" << stmtEnv_.sql_ << "]" << endl; cout << "==========================================sql : [" << stmtEnv_.sql_ << "]" << endl;
cout << "syntax tree : " << endl;
cout << res_.ast_ << endl; if (DUMP_MODULE_ALL == module || DUMP_MODULE_PARSER == module) {
cout << "raw logic plan : " << endl; cout << "syntax tree : " << endl;
cout << res_.rawLogicPlan_ << endl; cout << res_.ast_ << endl;
cout << "optimized logic plan : " << endl; }
cout << res_.optimizedLogicPlan_ << endl;
cout << "split logic plan : " << endl; if (DUMP_MODULE_ALL == module || DUMP_MODULE_LOGIC == module) {
cout << res_.splitLogicPlan_ << endl; cout << "raw logic plan : " << endl;
cout << "scaled logic plan : " << endl; cout << res_.rawLogicPlan_ << endl;
cout << res_.scaledLogicPlan_ << endl; }
cout << "physical plan : " << endl;
cout << res_.physiPlan_ << endl; if (DUMP_MODULE_ALL == module || DUMP_MODULE_OPTIMIZED == module) {
cout << "physical subplan : " << endl; cout << "optimized logic plan : " << endl;
for (const auto& subplan : res_.physiSubplans_) { cout << res_.optimizedLogicPlan_ << endl;
cout << subplan << endl; }
if (DUMP_MODULE_ALL == module || DUMP_MODULE_SPLIT == module) {
cout << "split logic plan : " << endl;
cout << res_.splitLogicPlan_ << endl;
}
if (DUMP_MODULE_ALL == module || DUMP_MODULE_SCALED == module) {
cout << "scaled logic plan : " << endl;
cout << res_.scaledLogicPlan_ << endl;
}
if (DUMP_MODULE_ALL == module || DUMP_MODULE_PHYSICAL == module) {
cout << "physical plan : " << endl;
cout << res_.physiPlan_ << endl;
}
if (DUMP_MODULE_ALL == module || DUMP_MODULE_SUBPLAN == module) {
cout << "physical subplan : " << endl;
for (const auto& subplan : res_.physiSubplans_) {
cout << subplan << endl;
}
} }
} }

2
source/libs/planner/test/planTestUtil.h
View File

@ -32,6 +32,6 @@ class PlannerTestBase : public testing::Test {
std::unique_ptr<PlannerTestBaseImpl> impl_; std::unique_ptr<PlannerTestBaseImpl> impl_;
}; };
extern bool g_isDump; extern void setDumpModule(const char* pModule);
#endif // PLAN_TEST_UTIL_H #endif // PLAN_TEST_UTIL_H

22
source/libs/scalar/src/sclvector.c
View File

@ -297,6 +297,22 @@ static FORCE_INLINE void varToNchar(char* buf, SScalarParam* pOut, int32_t rowIn
taosMemoryFree(t); taosMemoryFree(t);
} }
static FORCE_INLINE void ncharToVar(char* buf, SScalarParam* pOut, int32_t rowIndex) {
int32_t inputLen = varDataLen(buf);
char* t = taosMemoryCalloc(1, inputLen + VARSTR_HEADER_SIZE);
int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(buf), varDataLen(buf), varDataVal(t));
if (len < 0) {
taosMemoryFree(t);
return;
}
varDataSetLen(t, len);
colDataAppend(pOut->columnData, rowIndex, t, false);
taosMemoryFree(t);
}
//TODO opt performance, tmp is not needed. //TODO opt performance, tmp is not needed.
int32_t vectorConvertFromVarData(const SScalarParam* pIn, SScalarParam* pOut, int32_t inType, int32_t outType) { int32_t vectorConvertFromVarData(const SScalarParam* pIn, SScalarParam* pOut, int32_t inType, int32_t outType) {
int32_t bufSize = pIn->columnData->info.bytes; int32_t bufSize = pIn->columnData->info.bytes;
@ -313,6 +329,10 @@ int32_t vectorConvertFromVarData(const SScalarParam* pIn, SScalarParam* pOut, in
func = varToUnsigned; func = varToUnsigned;
} else if (IS_FLOAT_TYPE(outType)) { } else if (IS_FLOAT_TYPE(outType)) {
func = varToFloat; func = varToFloat;
} else if (outType == TSDB_DATA_TYPE_BINARY) { // nchar -> binary
ASSERT(inType == TSDB_DATA_TYPE_NCHAR);
func = ncharToVar;
vton = true;
} else if (outType == TSDB_DATA_TYPE_NCHAR) { // binary -> nchar } else if (outType == TSDB_DATA_TYPE_NCHAR) { // binary -> nchar
ASSERT(inType == TSDB_DATA_TYPE_VARCHAR); ASSERT(inType == TSDB_DATA_TYPE_VARCHAR);
func = varToNchar; func = varToNchar;
@ -608,7 +628,7 @@ int8_t gConvertTypes[TSDB_DATA_TYPE_BLOB+1][TSDB_DATA_TYPE_BLOB+1] = {
/*BIGI*/ 0, 0, 0, 0, 0, 0, 6, 7, 7, 0, 7, 5, 5, 5, 7, 0, 7, 0, 0, /*BIGI*/ 0, 0, 0, 0, 0, 0, 6, 7, 7, 0, 7, 5, 5, 5, 7, 0, 7, 0, 0,
/*FLOA*/ 0, 0, 0, 0, 0, 0, 0, 7, 7, 6, 7, 6, 6, 6, 6, 0, 7, 0, 0, /*FLOA*/ 0, 0, 0, 0, 0, 0, 0, 7, 7, 6, 7, 6, 6, 6, 6, 0, 7, 0, 0,
/*DOUB*/ 0, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7, 7, 7, 7, 0, 7, 0, 0, /*DOUB*/ 0, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7, 7, 7, 7, 0, 7, 0, 0,
/*VARC*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 0, 7, 7, 7, 7, 0, 0, 0, 0, /*VARC*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 8, 7, 7, 7, 7, 0, 0, 0, 0,
/*TIME*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 9, 9, 9, 7, 0, 7, 0, 0, /*TIME*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 9, 9, 9, 7, 0, 7, 0, 0,
/*NCHA*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7, 0, 0, 0, 0, /*NCHA*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7, 0, 0, 0, 0,
/*UTIN*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 12, 13, 14, 0, 7, 0, 0, /*UTIN*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 12, 13, 14, 0, 7, 0, 0,

1928
tests/script/api/batchprepare.c
View File

File diff suppressed because it is too large Load Diff