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

Merge branch '3.0' into fix/TD-15257

This commit is contained in:
Ganlin Zhao 2022-07-02 11:20:04 +08:00
parent f44e5310da f98e4323c4
commit 7a588bd16f
30 changed files with 620 additions and 303 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/function/functionMgt.h
View File

@ -196,6 +196,7 @@ bool fmIsIntervalInterpoFunc(int32_t funcId);
bool fmIsInterpFunc(int32_t funcId); bool fmIsInterpFunc(int32_t funcId);
bool fmIsLastRowFunc(int32_t funcId); bool fmIsLastRowFunc(int32_t funcId);
bool fmIsSystemInfoFunc(int32_t funcId); bool fmIsSystemInfoFunc(int32_t funcId);
bool fmIsImplicitTsFunc(int32_t funcId);
int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc); int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);

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

@ -104,6 +104,7 @@ typedef struct SIndefRowsFuncLogicNode {
SNodeList* pFuncs; SNodeList* pFuncs;
bool isTailFunc; bool isTailFunc;
bool isUniqueFunc; bool isUniqueFunc;
bool isTimeLineFunc;
} SIndefRowsFuncLogicNode; } SIndefRowsFuncLogicNode;
typedef struct SInterpFuncLogicNode { typedef struct SInterpFuncLogicNode {

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

@ -251,7 +251,7 @@ typedef struct SSelectStmt {
char stmtName[TSDB_TABLE_NAME_LEN]; char stmtName[TSDB_TABLE_NAME_LEN];
uint8_t precision; uint8_t precision;
bool isEmptyResult; bool isEmptyResult;
bool isTimeOrderQuery; bool isTimeLineResult;
bool hasAggFuncs; bool hasAggFuncs;
bool hasRepeatScanFuncs; bool hasRepeatScanFuncs;
bool hasIndefiniteRowsFunc; bool hasIndefiniteRowsFunc;
@ -261,6 +261,7 @@ typedef struct SSelectStmt {
bool hasTailFunc; bool hasTailFunc;
bool hasInterpFunc; bool hasInterpFunc;
bool hasLastRowFunc; bool hasLastRowFunc;
bool hasTimeLineFunc;
bool groupSort; bool groupSort;
} SSelectStmt; } SSelectStmt;

5
include/libs/sync/syncTools.h
View File

@ -362,6 +362,11 @@ typedef struct SOffsetAndContLen {
} SOffsetAndContLen; } SOffsetAndContLen;
*/ */
// block1: SOffsetAndContLen
// block2: SOffsetAndContLen Array
// block3: SRpcMsg Array
// block4: SRpcMsg pCont Array
typedef struct SyncAppendEntriesBatch { typedef struct SyncAppendEntriesBatch {
uint32_t bytes; uint32_t bytes;
int32_t vgId; int32_t vgId;

1
include/util/taoserror.h
View File

@ -577,6 +577,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC TAOS_DEF_ERROR_CODE(0, 0x265B) #define TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC TAOS_DEF_ERROR_CODE(0, 0x265B)
#define TSDB_CODE_PAR_INVALID_TABLE_OPTION TAOS_DEF_ERROR_CODE(0, 0x265C) #define TSDB_CODE_PAR_INVALID_TABLE_OPTION TAOS_DEF_ERROR_CODE(0, 0x265C)
#define TSDB_CODE_PAR_INVALID_INTERP_CLAUSE TAOS_DEF_ERROR_CODE(0, 0x265D) #define TSDB_CODE_PAR_INVALID_INTERP_CLAUSE TAOS_DEF_ERROR_CODE(0, 0x265D)
#define TSDB_CODE_PAR_NO_VALID_FUNC_IN_WIN TAOS_DEF_ERROR_CODE(0, 0x265E)
//planner //planner
#define TSDB_CODE_PLAN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x2700) #define TSDB_CODE_PLAN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x2700)

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

@ -47,6 +47,7 @@ int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, S
void *mndBuildCreateVnodeReq(SMnode *, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *cntlen, bool standby); void *mndBuildCreateVnodeReq(SMnode *, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *cntlen, bool standby);
void *mndBuildDropVnodeReq(SMnode *, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen); void *mndBuildDropVnodeReq(SMnode *, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
void *mndBuildAlterVnodeReq(SMnode *, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen); void *mndBuildAlterVnodeReq(SMnode *, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
bool mndVgroupInDb(SVgObj *pVgroup, int64_t dbUid);
#ifdef __cplusplus #ifdef __cplusplus
} }

4
source/dnode/mnode/impl/src/mndDb.c
View File

@ -645,7 +645,7 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid == pNew->uid) { if (mndVgroupInDb(pVgroup, pNew->uid)) {
if (mndBuildAlterVgroupAction(pMnode, pTrans, pNew, pVgroup, pArray) != 0) { if (mndBuildAlterVgroupAction(pMnode, pTrans, pNew, pVgroup, pArray) != 0) {
sdbCancelFetch(pSdb, pIter); sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
@ -1006,7 +1006,7 @@ static int32_t mndGetDBTableNum(SDbObj *pDb, SMnode *pMnode) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid == pDb->uid) { if (mndVgroupInDb(pVgroup, pDb->uid)) {
numOfTables += pVgroup->numOfTables / TSDB_TABLE_NUM_UNIT; numOfTables += pVgroup->numOfTables / TSDB_TABLE_NUM_UNIT;
vindex++; vindex++;
} }

11
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -225,10 +225,11 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
SVgObj* pVgroup; SVgObj* pVgroup;
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (strcmp(pVgroup->dbName, pStream->targetDb) != 0) { if (!mndVgroupInDb(pVgroup, pStream->targetDbUid)) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
continue; continue;
} }
SStreamTask* pTask = tNewSStreamTask(pStream->uid); SStreamTask* pTask = tNewSStreamTask(pStream->uid);
if (pTask == NULL) { if (pTask == NULL) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
@ -420,10 +421,11 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
SVgObj* pVgroup; SVgObj* pVgroup;
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid != pStream->sourceDbUid) { if (!mndVgroupInDb(pVgroup, pStream->sourceDbUid)) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
continue; continue;
} }
SStreamTask* pTask = tNewSStreamTask(pStream->uid); SStreamTask* pTask = tNewSStreamTask(pStream->uid);
if (pInnerTask == NULL) { if (pInnerTask == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -483,10 +485,11 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
SVgObj* pVgroup; SVgObj* pVgroup;
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid != pStream->sourceDbUid) { if (!mndVgroupInDb(pVgroup, pStream->sourceDbUid)) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
continue; continue;
} }
SStreamTask* pTask = tNewSStreamTask(pStream->uid); SStreamTask* pTask = tNewSStreamTask(pStream->uid);
if (pTask == NULL) { if (pTask == NULL) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
@ -559,7 +562,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
while (1) { while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid != pTopic->dbUid) { if (!mndVgroupInDb(pVgroup, pTopic->dbUid)) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
continue; continue;
} }

11
source/dnode/mnode/impl/src/mndSma.c
View File

@ -822,6 +822,17 @@ int32_t mndDropSmasByStb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *p
if (pSma->stbUid == pStb->uid) { if (pSma->stbUid == pStb->uid) {
pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId); pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
if (pVgroup == NULL) goto _OVER; if (pVgroup == NULL) goto _OVER;
SStreamObj *pStream = mndAcquireStream(pMnode, pSma->name);
if (pStream != NULL && pStream->smaId == pSma->uid) {
if (mndDropStreamTasks(pMnode, pTrans, pStream) < 0) {
mError("stream:%s, failed to drop task since %s", pStream->name, terrstr());
goto _OVER;
}
if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
goto _OVER;
}
}
if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER; if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER; if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER; if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;

28
source/dnode/mnode/impl/src/mndStb.c
View File

@ -621,12 +621,7 @@ static int32_t mndSetCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
while (1) { while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid != pDb->uid) { if (!mndVgroupInDb(pVgroup, pDb->uid)) {
sdbRelease(pSdb, pVgroup);
continue;
}
if (pVgroup->isTsma) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
continue; continue;
} }
@ -664,12 +659,7 @@ static int32_t mndSetCreateStbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
while (1) { while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid != pDb->uid) { if (!mndVgroupInDb(pVgroup, pDb->uid)) {
sdbRelease(pSdb, pVgroup);
continue;
}
if (pVgroup->isTsma) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
continue; continue;
} }
@ -1297,12 +1287,7 @@ static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
while (1) { while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid != pDb->uid) { if (!mndVgroupInDb(pVgroup, pDb->uid)) {
sdbRelease(pSdb, pVgroup);
continue;
}
if (pVgroup->isTsma) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
continue; continue;
} }
@ -1688,12 +1673,7 @@ static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
while (1) { while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
if (pVgroup->dbUid != pDb->uid) { if (!mndVgroupInDb(pVgroup, pDb->uid)) {
sdbRelease(pSdb, pVgroup);
continue;
}
if (pVgroup->isTsma) {
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
continue; continue;
} }

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

@ -1146,13 +1146,13 @@ static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans)
} else { } else {
code = TSDB_CODE_ACTION_IN_PROGRESS; code = TSDB_CODE_ACTION_IN_PROGRESS;
} }
} } else if (pAction->rawWritten) {
if (pAction->rawWritten) {
if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) { if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
code = pAction->errCode; code = pAction->errCode;
} else { } else {
mDebug("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action); mDebug("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
} }
} else {
} }
} }

4
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -1759,4 +1759,6 @@ _OVER:
taosArrayDestroy(pArray); taosArrayDestroy(pArray);
return code; return code;
} }
bool mndVgroupInDb(SVgObj *pVgroup, int64_t dbUid) { return !pVgroup->isTsma && pVgroup->dbUid == dbUid; }

2
source/libs/function/inc/functionMgtInt.h
View File

@ -32,7 +32,7 @@ extern "C" {
#define FUNC_MGT_STRING_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(3) #define FUNC_MGT_STRING_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(3)
#define FUNC_MGT_DATETIME_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(4) #define FUNC_MGT_DATETIME_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(4)
#define FUNC_MGT_TIMELINE_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(5) #define FUNC_MGT_TIMELINE_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(5)
#define FUNC_MGT_TIMEORDER_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(6) #define FUNC_MGT_IMPLICIT_TS_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(6)
#define FUNC_MGT_PSEUDO_COLUMN_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(7) #define FUNC_MGT_PSEUDO_COLUMN_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(7)
#define FUNC_MGT_WINDOW_PC_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(8) #define FUNC_MGT_WINDOW_PC_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(8)
#define FUNC_MGT_SPECIAL_DATA_REQUIRED FUNC_MGT_FUNC_CLASSIFICATION_MASK(9) #define FUNC_MGT_SPECIAL_DATA_REQUIRED FUNC_MGT_FUNC_CLASSIFICATION_MASK(9)

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

@ -406,7 +406,7 @@ static int32_t translateWduration(SFunctionNode* pFunc, char* pErrBuf, int32_t l
static int32_t translateNowToday(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { static int32_t translateNowToday(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
// pseudo column do not need to check parameters // pseudo column do not need to check parameters
//add database precision as param // add database precision as param
uint8_t dbPrec = pFunc->node.resType.precision; uint8_t dbPrec = pFunc->node.resType.precision;
addDbPrecisonParam(&pFunc->pParameterList, dbPrec); addDbPrecisonParam(&pFunc->pParameterList, dbPrec);
@ -1214,7 +1214,7 @@ static int32_t translateIrate(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
} }
//add database precision as param // add database precision as param
uint8_t dbPrec = pFunc->node.resType.precision; uint8_t dbPrec = pFunc->node.resType.precision;
addDbPrecisonParam(&pFunc->pParameterList, dbPrec); addDbPrecisonParam(&pFunc->pParameterList, dbPrec);
@ -1228,13 +1228,7 @@ static int32_t translateFirstLast(SFunctionNode* pFunc, char* pErrBuf, int32_t l
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0); pFunc->node.resType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType;
if (QUERY_NODE_COLUMN != nodeType(pPara)) {
return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
"The parameters of first/last can only be columns");
}
pFunc->node.resType = ((SExprNode*)pPara)->resType;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1248,11 +1242,6 @@ static int32_t translateFirstLastImpl(SFunctionNode* pFunc, char* pErrBuf, int32
uint8_t paraType = ((SExprNode*)pPara)->resType.type; uint8_t paraType = ((SExprNode*)pPara)->resType.type;
int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes; int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes;
if (isPartial) { if (isPartial) {
if (QUERY_NODE_COLUMN != nodeType(pPara)) {
return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
"The parameters of first/last can only be columns");
}
pFunc->node.resType = pFunc->node.resType =
(SDataType){.bytes = getFirstLastInfoSize(paraBytes) + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY}; (SDataType){.bytes = getFirstLastInfoSize(paraBytes) + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY};
} else { } else {
@ -1528,7 +1517,7 @@ static int32_t translateToUnixtimestamp(SFunctionNode* pFunc, char* pErrBuf, int
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
} }
//add database precision as param // add database precision as param
uint8_t dbPrec = pFunc->node.resType.precision; uint8_t dbPrec = pFunc->node.resType.precision;
addDbPrecisonParam(&pFunc->pParameterList, dbPrec); addDbPrecisonParam(&pFunc->pParameterList, dbPrec);
@ -1548,7 +1537,7 @@ static int32_t translateTimeTruncate(SFunctionNode* pFunc, char* pErrBuf, int32_
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
} }
//add database precision as param // add database precision as param
uint8_t dbPrec = pFunc->node.resType.precision; uint8_t dbPrec = pFunc->node.resType.precision;
int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode *)nodesListGetNode(pFunc->pParameterList, 1)); int32_t ret = validateTimeUnitParam(dbPrec, (SValueNode *)nodesListGetNode(pFunc->pParameterList, 1));
@ -1586,7 +1575,7 @@ static int32_t translateTimeDiff(SFunctionNode* pFunc, char* pErrBuf, int32_t le
} }
} }
//add database precision as param // add database precision as param
uint8_t dbPrec = pFunc->node.resType.precision; uint8_t dbPrec = pFunc->node.resType.precision;
if (3 == numOfParams) { if (3 == numOfParams) {
@ -1997,7 +1986,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "interp", .name = "interp",
.type = FUNCTION_TYPE_INTERP, .type = FUNCTION_TYPE_INTERP,
.classification = FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_INTERVAL_INTERPO_FUNC, .classification = FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_INTERVAL_INTERPO_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLast, .translateFunc = translateFirstLast,
.getEnvFunc = getSelectivityFuncEnv, .getEnvFunc = getSelectivityFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2007,7 +1996,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "derivative", .name = "derivative",
.type = FUNCTION_TYPE_DERIVATIVE, .type = FUNCTION_TYPE_DERIVATIVE,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateDerivative, .translateFunc = translateDerivative,
.getEnvFunc = getDerivativeFuncEnv, .getEnvFunc = getDerivativeFuncEnv,
.initFunc = derivativeFuncSetup, .initFunc = derivativeFuncSetup,
@ -2017,7 +2006,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "irate", .name = "irate",
.type = FUNCTION_TYPE_IRATE, .type = FUNCTION_TYPE_IRATE,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateIrate, .translateFunc = translateIrate,
.getEnvFunc = getIrateFuncEnv, .getEnvFunc = getIrateFuncEnv,
.initFunc = irateFuncSetup, .initFunc = irateFuncSetup,
@ -2026,8 +2015,8 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
}, },
{ {
.name = "last_row", .name = "last_row",
.type = FUNCTION_TYPE_LAST_ROWT, .type = FUNCTION_TYPE_LAST_ROW,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLast, .translateFunc = translateFirstLast,
.getEnvFunc = getFirstLastFuncEnv, .getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2037,7 +2026,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "_cache_last_row", .name = "_cache_last_row",
.type = FUNCTION_TYPE_CACHE_LAST_ROW, .type = FUNCTION_TYPE_CACHE_LAST_ROW,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateLastRow, .translateFunc = translateLastRow,
.getEnvFunc = getMinmaxFuncEnv, .getEnvFunc = getMinmaxFuncEnv,
.initFunc = minmaxFunctionSetup, .initFunc = minmaxFunctionSetup,
@ -2047,7 +2036,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "first", .name = "first",
.type = FUNCTION_TYPE_FIRST, .type = FUNCTION_TYPE_FIRST,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLast, .translateFunc = translateFirstLast,
.getEnvFunc = getFirstLastFuncEnv, .getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2060,7 +2049,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "_first_partial", .name = "_first_partial",
.type = FUNCTION_TYPE_FIRST_PARTIAL, .type = FUNCTION_TYPE_FIRST_PARTIAL,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLastPartial, .translateFunc = translateFirstLastPartial,
.getEnvFunc = getFirstLastFuncEnv, .getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2071,7 +2060,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "_first_merge", .name = "_first_merge",
.type = FUNCTION_TYPE_FIRST_MERGE, .type = FUNCTION_TYPE_FIRST_MERGE,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLastMerge, .translateFunc = translateFirstLastMerge,
.getEnvFunc = getFirstLastFuncEnv, .getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2082,7 +2071,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "last", .name = "last",
.type = FUNCTION_TYPE_LAST, .type = FUNCTION_TYPE_LAST,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLast, .translateFunc = translateFirstLast,
.getEnvFunc = getFirstLastFuncEnv, .getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2095,7 +2084,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "_last_partial", .name = "_last_partial",
.type = FUNCTION_TYPE_LAST_PARTIAL, .type = FUNCTION_TYPE_LAST_PARTIAL,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLastPartial, .translateFunc = translateFirstLastPartial,
.getEnvFunc = getFirstLastFuncEnv, .getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2106,7 +2095,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "_last_merge", .name = "_last_merge",
.type = FUNCTION_TYPE_LAST_MERGE, .type = FUNCTION_TYPE_LAST_MERGE,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLastMerge, .translateFunc = translateFirstLastMerge,
.getEnvFunc = getFirstLastFuncEnv, .getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2117,7 +2106,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "twa", .name = "twa",
.type = FUNCTION_TYPE_TWA, .type = FUNCTION_TYPE_TWA,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_INTERVAL_INTERPO_FUNC | FUNC_MGT_FORBID_STREAM_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_INTERVAL_INTERPO_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateInNumOutDou, .translateFunc = translateInNumOutDou,
.dataRequiredFunc = statisDataRequired, .dataRequiredFunc = statisDataRequired,
.getEnvFunc = getTwaFuncEnv, .getEnvFunc = getTwaFuncEnv,
@ -2214,7 +2203,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "statecount", .name = "statecount",
.type = FUNCTION_TYPE_STATE_COUNT, .type = FUNCTION_TYPE_STATE_COUNT,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC, .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC,
.translateFunc = translateStateCount, .translateFunc = translateStateCount,
.getEnvFunc = getStateFuncEnv, .getEnvFunc = getStateFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2224,7 +2213,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "stateduration", .name = "stateduration",
.type = FUNCTION_TYPE_STATE_DURATION, .type = FUNCTION_TYPE_STATE_DURATION,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC, .classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC,
.translateFunc = translateStateDuration, .translateFunc = translateStateDuration,
.getEnvFunc = getStateFuncEnv, .getEnvFunc = getStateFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
@ -2264,7 +2253,8 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "tail", .name = "tail",
.type = FUNCTION_TYPE_TAIL, .type = FUNCTION_TYPE_TAIL,
.classification = FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC | FUNC_MGT_FORBID_GROUP_BY_FUNC, .classification = FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_FORBID_STREAM_FUNC |
FUNC_MGT_FORBID_WINDOW_FUNC | FUNC_MGT_FORBID_GROUP_BY_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateTail, .translateFunc = translateTail,
.getEnvFunc = getTailFuncEnv, .getEnvFunc = getTailFuncEnv,
.initFunc = tailFunctionSetup, .initFunc = tailFunctionSetup,
@ -2275,7 +2265,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.name = "unique", .name = "unique",
.type = FUNCTION_TYPE_UNIQUE, .type = FUNCTION_TYPE_UNIQUE,
.classification = FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC | .classification = FUNC_MGT_SELECT_FUNC | FUNC_MGT_INDEFINITE_ROWS_FUNC | FUNC_MGT_TIMELINE_FUNC |
FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC | FUNC_MGT_FORBID_GROUP_BY_FUNC, FUNC_MGT_FORBID_STREAM_FUNC | FUNC_MGT_FORBID_WINDOW_FUNC | FUNC_MGT_FORBID_GROUP_BY_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateUnique, .translateFunc = translateUnique,
.getEnvFunc = getUniqueFuncEnv, .getEnvFunc = getUniqueFuncEnv,
.initFunc = uniqueFunctionSetup, .initFunc = uniqueFunctionSetup,

190
source/libs/function/src/builtinsimpl.c
View File

@ -16,6 +16,7 @@
#include "builtinsimpl.h" #include "builtinsimpl.h"
#include "cJSON.h" #include "cJSON.h"
#include "function.h" #include "function.h"
#include "query.h"
#include "querynodes.h" #include "querynodes.h"
#include "taggfunction.h" #include "taggfunction.h"
#include "tcompare.h" #include "tcompare.h"
@ -24,7 +25,6 @@
#include "tglobal.h" #include "tglobal.h"
#include "thistogram.h" #include "thistogram.h"
#include "tpercentile.h" #include "tpercentile.h"
#include "query.h"
#define HISTOGRAM_MAX_BINS_NUM 1000 #define HISTOGRAM_MAX_BINS_NUM 1000
#define MAVG_MAX_POINTS_NUM 1000 #define MAVG_MAX_POINTS_NUM 1000
@ -81,7 +81,7 @@ typedef struct STopBotRes {
typedef struct SFirstLastRes { typedef struct SFirstLastRes {
bool hasResult; bool hasResult;
bool isNull; //used for last_row function only bool isNull; // used for last_row function only
int32_t bytes; int32_t bytes;
char buf[]; char buf[];
} SFirstLastRes; } SFirstLastRes;
@ -210,13 +210,13 @@ typedef struct SMavgInfo {
} SMavgInfo; } SMavgInfo;
typedef struct SSampleInfo { typedef struct SSampleInfo {
int32_t samples; int32_t samples;
int32_t totalPoints; int32_t totalPoints;
int32_t numSampled; int32_t numSampled;
uint8_t colType; uint8_t colType;
int16_t colBytes; int16_t colBytes;
char* data; char* data;
STuplePos* tuplePos; STuplePos* tuplePos;
} SSampleInfo; } SSampleInfo;
typedef struct STailItem { typedef struct STailItem {
@ -271,20 +271,19 @@ typedef struct SDerivInfo {
} SDerivInfo; } SDerivInfo;
typedef struct SRateInfo { typedef struct SRateInfo {
double firstValue; double firstValue;
TSKEY firstKey; TSKEY firstKey;
double lastValue; double lastValue;
TSKEY lastKey; TSKEY lastKey;
int8_t hasResult; // flag to denote has value int8_t hasResult; // flag to denote has value
} SRateInfo; } SRateInfo;
typedef struct SGroupKeyInfo{ typedef struct SGroupKeyInfo {
bool hasResult; bool hasResult;
bool isNull; bool isNull;
char data[]; char data[];
} SGroupKeyInfo; } SGroupKeyInfo;
#define SET_VAL(_info, numOfElem, res) \ #define SET_VAL(_info, numOfElem, res) \
do { \ do { \
if ((numOfElem) <= 0) { \ if ((numOfElem) <= 0) { \
@ -1474,7 +1473,7 @@ void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuple
int32_t offset = pTuplePos->offset; int32_t offset = pTuplePos->offset;
if (pTuplePos->pageId != -1 && pCtx->subsidiaries.num > 0) { if (pTuplePos->pageId != -1 && pCtx->subsidiaries.num > 0) {
int32_t numOfCols = pCtx->subsidiaries.num; int32_t numOfCols = pCtx->subsidiaries.num;
SFilePage* pPage = getBufPage(pCtx->pBuf, pageId); SFilePage* pPage = getBufPage(pCtx->pBuf, pageId);
bool* nullList = (bool*)((char*)pPage + offset); bool* nullList = (bool*)((char*)pPage + offset);
@ -2410,7 +2409,9 @@ int32_t apercentileCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx)
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t getFirstLastInfoSize(int32_t resBytes) { return sizeof(SFirstLastRes) + resBytes + sizeof(int64_t) + sizeof(STuplePos); } int32_t getFirstLastInfoSize(int32_t resBytes) {
return sizeof(SFirstLastRes) + resBytes + sizeof(int64_t) + sizeof(STuplePos);
}
bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) { bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
SColumnNode* pNode = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 0); SColumnNode* pNode = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 0);
@ -2492,7 +2493,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
} }
memcpy(pInfo->buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(pInfo->buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
//handle selectivity // handle selectivity
if (pCtx->subsidiaries.num > 0) { if (pCtx->subsidiaries.num > 0) {
STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
if (!pInfo->hasResult) { if (!pInfo->hasResult) {
@ -2502,7 +2503,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
} }
} }
pInfo->hasResult = true; pInfo->hasResult = true;
//DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
break; break;
} }
@ -2534,7 +2535,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
} }
memcpy(pInfo->buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(pInfo->buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
//handle selectivity // handle selectivity
if (pCtx->subsidiaries.num > 0) { if (pCtx->subsidiaries.num > 0) {
STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
if (!pInfo->hasResult) { if (!pInfo->hasResult) {
@ -2544,7 +2545,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
} }
} }
pInfo->hasResult = true; pInfo->hasResult = true;
//DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
break; break;
} }
@ -2598,7 +2599,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
} }
memcpy(pInfo->buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(pInfo->buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
//handle selectivity // handle selectivity
if (pCtx->subsidiaries.num > 0) { if (pCtx->subsidiaries.num > 0) {
STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
if (!pInfo->hasResult) { if (!pInfo->hasResult) {
@ -2608,7 +2609,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
} }
} }
pInfo->hasResult = true; pInfo->hasResult = true;
//DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
} }
break; break;
@ -2630,7 +2631,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
} }
memcpy(pInfo->buf, data, bytes); memcpy(pInfo->buf, data, bytes);
*(TSKEY*)(pInfo->buf + bytes) = cts; *(TSKEY*)(pInfo->buf + bytes) = cts;
//handle selectivity // handle selectivity
if (pCtx->subsidiaries.num > 0) { if (pCtx->subsidiaries.num > 0) {
STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
if (!pInfo->hasResult) { if (!pInfo->hasResult) {
@ -2641,7 +2642,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
} }
pInfo->hasResult = true; pInfo->hasResult = true;
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
//DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
} }
break; break;
} }
@ -2653,7 +2654,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
static void firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst) { static void firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst) {
SInputColumnInfoData* pColInfo = &pCtx->input; SInputColumnInfoData* pColInfo = &pCtx->input;
int32_t start = pColInfo->startRowIndex; int32_t start = pColInfo->startRowIndex;
pOutput->bytes = pInput->bytes; pOutput->bytes = pInput->bytes;
TSKEY* tsIn = (TSKEY*)(pInput->buf + pInput->bytes); TSKEY* tsIn = (TSKEY*)(pInput->buf + pInput->bytes);
@ -2671,7 +2672,7 @@ static void firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, S
} }
*tsOut = *tsIn; *tsOut = *tsIn;
memcpy(pOutput->buf, pInput->buf, pOutput->bytes); memcpy(pOutput->buf, pInput->buf, pOutput->bytes);
//handle selectivity // handle selectivity
STuplePos* pTuplePos = (STuplePos*)(pOutput->buf + pOutput->bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pOutput->buf + pOutput->bytes + sizeof(TSKEY));
if (pCtx->subsidiaries.num > 0) { if (pCtx->subsidiaries.num > 0) {
if (!pOutput->hasResult) { if (!pOutput->hasResult) {
@ -2718,7 +2719,7 @@ int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(pResInfo); SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(pResInfo);
colDataAppend(pCol, pBlock->info.rows, pRes->buf, pResInfo->isNullRes); colDataAppend(pCol, pBlock->info.rows, pRes->buf, pResInfo->isNullRes);
//handle selectivity // handle selectivity
STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY));
setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows); setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows);
@ -2729,8 +2730,8 @@ int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx); SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
int32_t resultBytes = getFirstLastInfoSize(pRes->bytes); int32_t resultBytes = getFirstLastInfoSize(pRes->bytes);
char* res = taosMemoryCalloc(resultBytes + VARSTR_HEADER_SIZE, sizeof(char)); char* res = taosMemoryCalloc(resultBytes + VARSTR_HEADER_SIZE, sizeof(char));
memcpy(varDataVal(res), pRes, resultBytes); memcpy(varDataVal(res), pRes, resultBytes);
varDataSetLen(res, resultBytes); varDataSetLen(res, resultBytes);
@ -2739,7 +2740,7 @@ int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId); SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
colDataAppend(pCol, pBlock->info.rows, res, false); colDataAppend(pCol, pBlock->info.rows, res, false);
//handle selectivity // handle selectivity
STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY));
setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows); setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows);
@ -2801,7 +2802,7 @@ int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
} }
*(TSKEY*)(pInfo->buf) = cts; *(TSKEY*)(pInfo->buf) = cts;
numOfElems++; numOfElems++;
//handle selectivity // handle selectivity
if (pCtx->subsidiaries.num > 0) { if (pCtx->subsidiaries.num > 0) {
STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
if (!pInfo->hasResult) { if (!pInfo->hasResult) {
@ -2811,7 +2812,7 @@ int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
} }
} }
pInfo->hasResult = true; pInfo->hasResult = true;
//DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
} }
break; break;
@ -2833,7 +2834,7 @@ int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
} }
*(TSKEY*)(pInfo->buf) = cts; *(TSKEY*)(pInfo->buf) = cts;
numOfElems++; numOfElems++;
//handle selectivity // handle selectivity
if (pCtx->subsidiaries.num > 0) { if (pCtx->subsidiaries.num > 0) {
STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pInfo->buf + bytes + sizeof(TSKEY));
if (!pInfo->hasResult) { if (!pInfo->hasResult) {
@ -2844,7 +2845,7 @@ int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
} }
pInfo->hasResult = true; pInfo->hasResult = true;
pResInfo->numOfRes = 1; pResInfo->numOfRes = 1;
//DO_UPDATE_TAG_COLUMNS(pCtx, ts); // DO_UPDATE_TAG_COLUMNS(pCtx, ts);
} }
break; break;
} }
@ -2854,7 +2855,6 @@ int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t lastRowFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { int32_t lastRowFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t slotId = pCtx->pExpr->base.resSchema.slotId; int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId); SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
@ -2863,14 +2863,13 @@ int32_t lastRowFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(pResInfo); SFirstLastRes* pRes = GET_ROWCELL_INTERBUF(pResInfo);
colDataAppend(pCol, pBlock->info.rows, pRes->buf + sizeof(TSKEY), pRes->isNull); colDataAppend(pCol, pBlock->info.rows, pRes->buf + sizeof(TSKEY), pRes->isNull);
//handle selectivity // handle selectivity
STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY)); STuplePos* pTuplePos = (STuplePos*)(pRes->buf + pRes->bytes + sizeof(TSKEY));
setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows); setSelectivityValue(pCtx, pBlock, pTuplePos, pBlock->info.rows);
return pResInfo->numOfRes; return pResInfo->numOfRes;
} }
bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) { bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(SDiffInfo); pEnv->calcMemSize = sizeof(SDiffInfo);
return true; return true;
@ -2884,7 +2883,11 @@ bool diffFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
SDiffInfo* pDiffInfo = GET_ROWCELL_INTERBUF(pResInfo); SDiffInfo* pDiffInfo = GET_ROWCELL_INTERBUF(pResInfo);
pDiffInfo->hasPrev = false; pDiffInfo->hasPrev = false;
pDiffInfo->prev.i64 = 0; pDiffInfo->prev.i64 = 0;
pDiffInfo->ignoreNegative = pCtx->param[1].param.i; // TODO set correct param if (pCtx->numOfParams > 1) {
pDiffInfo->ignoreNegative = pCtx->param[1].param.i; // TODO set correct param
} else {
pDiffInfo->ignoreNegative = false;
}
pDiffInfo->includeNull = false; pDiffInfo->includeNull = false;
pDiffInfo->firstOutput = false; pDiffInfo->firstOutput = false;
return true; return true;
@ -2999,10 +3002,8 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pInputCol = pInput->pData[0];
SColumnInfoData* pTsOutput = pCtx->pTsOutput;
int32_t numOfElems = 0; int32_t numOfElems = 0;
TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
int32_t startOffset = pCtx->offset; int32_t startOffset = pCtx->offset;
SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput; SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
@ -3014,9 +3015,6 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) { if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
if (pDiffInfo->includeNull) { if (pDiffInfo->includeNull) {
colDataSetNull_f(pOutput->nullbitmap, pos); colDataSetNull_f(pOutput->nullbitmap, pos);
if (tsList != NULL) {
colDataAppendInt64(pTsOutput, pos, &tsList[i]);
}
numOfElems += 1; numOfElems += 1;
} }
@ -3027,9 +3025,6 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
if (pDiffInfo->hasPrev) { if (pDiffInfo->hasPrev) {
doHandleDiff(pDiffInfo, pInputCol->info.type, pv, pOutput, pos, pCtx->order); doHandleDiff(pDiffInfo, pInputCol->info.type, pv, pOutput, pos, pCtx->order);
if (pTsOutput != NULL) {
colDataAppendInt64(pTsOutput, pos, &tsList[i]);
}
numOfElems++; numOfElems++;
} else { } else {
@ -3045,9 +3040,6 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
if (colDataIsNull_f(pInputCol->nullbitmap, i)) { if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
if (pDiffInfo->includeNull) { if (pDiffInfo->includeNull) {
colDataSetNull_f(pOutput->nullbitmap, pos); colDataSetNull_f(pOutput->nullbitmap, pos);
if (tsList != NULL) {
colDataAppendInt64(pTsOutput, pos, &tsList[i]);
}
numOfElems += 1; numOfElems += 1;
} }
@ -3059,9 +3051,6 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
// there is a row of previous data block to be handled in the first place. // there is a row of previous data block to be handled in the first place.
if (pDiffInfo->hasPrev) { if (pDiffInfo->hasPrev) {
doHandleDiff(pDiffInfo, pInputCol->info.type, pv, pOutput, pos, pCtx->order); doHandleDiff(pDiffInfo, pInputCol->info.type, pv, pOutput, pos, pCtx->order);
if (pTsOutput != NULL) {
colDataAppendInt64(pTsOutput, pos, &pDiffInfo->prevTs);
}
numOfElems++; numOfElems++;
} else { } else {
@ -3069,9 +3058,6 @@ int32_t diffFunction(SqlFunctionCtx* pCtx) {
} }
pDiffInfo->hasPrev = true; pDiffInfo->hasPrev = true;
if (pTsOutput != NULL) {
pDiffInfo->prevTs = tsList[i];
}
} }
} }
@ -3209,7 +3195,8 @@ void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSData
saveTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos); saveTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos);
} }
#ifdef BUF_PAGE_DEBUG #ifdef BUF_PAGE_DEBUG
qDebug("page_saveTuple i:%d, item:%p,pageId:%d, offset:%d\n", pEntryInfo->numOfRes, pItem, pItem->tuplePos.pageId, pItem->tuplePos.offset); qDebug("page_saveTuple i:%d, item:%p,pageId:%d, offset:%d\n", pEntryInfo->numOfRes, pItem, pItem->tuplePos.pageId,
pItem->tuplePos.offset);
#endif #endif
// allocate the buffer and keep the data of this row into the new allocated buffer // allocate the buffer and keep the data of this row into the new allocated buffer
pEntryInfo->numOfRes++; pEntryInfo->numOfRes++;
@ -3280,7 +3267,7 @@ void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pS
char* p = colDataGetData(pCol, rowIndex); char* p = colDataGetData(pCol, rowIndex);
if (IS_VAR_DATA_TYPE(pCol->info.type)) { if (IS_VAR_DATA_TYPE(pCol->info.type)) {
memcpy(pStart + offset, p, (pCol->info.type == TSDB_DATA_TYPE_JSON) ? getJsonValueLen(p): varDataTLen(p)); memcpy(pStart + offset, p, (pCol->info.type == TSDB_DATA_TYPE_JSON) ? getJsonValueLen(p) : varDataTLen(p));
} else { } else {
memcpy(pStart + offset, p, pCol->info.bytes); memcpy(pStart + offset, p, pCol->info.bytes);
} }
@ -3306,8 +3293,8 @@ void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pS
int32_t offset = 0; int32_t offset = 0;
for (int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
SqlFunctionCtx* pc = pCtx->subsidiaries.pCtx[i]; SqlFunctionCtx* pc = pCtx->subsidiaries.pCtx[i];
SFunctParam* pFuncParam = &pc->pExpr->base.pParam[0]; SFunctParam* pFuncParam = &pc->pExpr->base.pParam[0];
int32_t srcSlotId = pFuncParam->pCol->slotId; int32_t srcSlotId = pFuncParam->pCol->slotId;
SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, srcSlotId); SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, srcSlotId);
if ((nullList[i] = colDataIsNull_s(pCol, rowIndex)) == true) { if ((nullList[i] = colDataIsNull_s(pCol, rowIndex)) == true) {
@ -3317,7 +3304,7 @@ void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pS
char* p = colDataGetData(pCol, rowIndex); char* p = colDataGetData(pCol, rowIndex);
if (IS_VAR_DATA_TYPE(pCol->info.type)) { if (IS_VAR_DATA_TYPE(pCol->info.type)) {
memcpy(pStart + offset, p, (pCol->info.type == TSDB_DATA_TYPE_JSON) ? getJsonValueLen(p): varDataTLen(p)); memcpy(pStart + offset, p, (pCol->info.type == TSDB_DATA_TYPE_JSON) ? getJsonValueLen(p) : varDataTLen(p));
} else { } else {
memcpy(pStart + offset, p, pCol->info.bytes); memcpy(pStart + offset, p, pCol->info.bytes);
} }
@ -3349,7 +3336,8 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
colDataAppend(pCol, currentRow, (const char*)&pItem->v.i, false); colDataAppend(pCol, currentRow, (const char*)&pItem->v.i, false);
} }
#ifdef BUF_PAGE_DEBUG #ifdef BUF_PAGE_DEBUG
qDebug("page_finalize i:%d,item:%p,pageId:%d, offset:%d\n", i, pItem, pItem->tuplePos.pageId, pItem->tuplePos.offset); qDebug("page_finalize i:%d,item:%p,pageId:%d, offset:%d\n", i, pItem, pItem->tuplePos.pageId,
pItem->tuplePos.offset);
#endif #endif
setSelectivityValue(pCtx, pBlock, &pRes->pItems[i].tuplePos, currentRow); setSelectivityValue(pCtx, pBlock, &pRes->pItems[i].tuplePos, currentRow);
currentRow += 1; currentRow += 1;
@ -3589,7 +3577,7 @@ bool elapsedFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo
pInfo->min = MAX_TS_KEY; pInfo->min = MAX_TS_KEY;
pInfo->max = 0; pInfo->max = 0;
if (pCtx->numOfParams > 2) { if (pCtx->numOfParams > 1) {
pInfo->timeUnit = pCtx->param[1].param.i; pInfo->timeUnit = pCtx->param[1].param.i;
} else { } else {
pInfo->timeUnit = 1; pInfo->timeUnit = 1;
@ -4530,7 +4518,6 @@ int32_t mavgFunction(SqlFunctionCtx* pCtx) {
SMavgInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo); SMavgInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pInputCol = pInput->pData[0];
SColumnInfoData* pTsOutput = pCtx->pTsOutput; SColumnInfoData* pTsOutput = pCtx->pTsOutput;
@ -4570,10 +4557,6 @@ int32_t mavgFunction(SqlFunctionCtx* pCtx) {
colDataAppend(pOutput, pos, (char*)&result, false); colDataAppend(pOutput, pos, (char*)&result, false);
} }
// TODO: remove this after pTsOutput is handled
if (pTsOutput != NULL) {
colDataAppendInt64(pTsOutput, pos, &tsList[i]);
}
numOfElems++; numOfElems++;
} }
@ -4920,11 +4903,11 @@ static void doModeAdd(SModeInfo* pInfo, char* data, bool isNull) {
return; return;
} }
int32_t hashKeyBytes = IS_VAR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes; int32_t hashKeyBytes = IS_VAR_DATA_TYPE(pInfo->colType) ? varDataTLen(data) : pInfo->colBytes;
SModeItem** pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes); SModeItem** pHashItem = taosHashGet(pInfo->pHash, data, hashKeyBytes);
if (pHashItem == NULL) { if (pHashItem == NULL) {
int32_t size = sizeof(SModeItem) + pInfo->colBytes; int32_t size = sizeof(SModeItem) + pInfo->colBytes;
SModeItem* pItem = (SModeItem*)(pInfo->pItems + pInfo->numOfPoints * size); SModeItem* pItem = (SModeItem*)(pInfo->pItems + pInfo->numOfPoints * size);
memcpy(pItem->data, data, pInfo->colBytes); memcpy(pItem->data, data, pInfo->colBytes);
pItem->count += 1; pItem->count += 1;
@ -4937,7 +4920,7 @@ static void doModeAdd(SModeInfo* pInfo, char* data, bool isNull) {
int32_t modeFunction(SqlFunctionCtx* pCtx) { int32_t modeFunction(SqlFunctionCtx* pCtx) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
SModeInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo); SModeInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
@ -4985,7 +4968,6 @@ int32_t modeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
return pResInfo->numOfRes; return pResInfo->numOfRes;
} }
bool getTwaFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv) { bool getTwaFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(STwaInfo); pEnv->calcMemSize = sizeof(STwaInfo);
return true; return true;
@ -5237,7 +5219,7 @@ int32_t twaFinalize(struct SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
return functionFinalize(pCtx, pBlock); return functionFinalize(pCtx, pBlock);
} }
bool blockDistSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo) { bool blockDistSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
if (!functionSetup(pCtx, pResultInfo)) { if (!functionSetup(pCtx, pResultInfo)) {
return false; return false;
} }
@ -5269,7 +5251,7 @@ int32_t blockDistFunction(SqlFunctionCtx* pCtx) {
pDistInfo->defMinRows = p1.defMinRows; pDistInfo->defMinRows = p1.defMinRows;
pDistInfo->defMaxRows = p1.defMaxRows; pDistInfo->defMaxRows = p1.defMaxRows;
pDistInfo->rowSize = p1.rowSize; pDistInfo->rowSize = p1.rowSize;
pDistInfo->numOfSmallBlocks = p1.numOfSmallBlocks; pDistInfo->numOfSmallBlocks = p1.numOfSmallBlocks;
if (pDistInfo->minRows > p1.minRows) { if (pDistInfo->minRows > p1.minRows) {
@ -5355,16 +5337,18 @@ int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t row = 0; int32_t row = 0;
char st[256] = {0}; char st[256] = {0};
double totalRawSize = pData->totalRows * pData->rowSize; double totalRawSize = pData->totalRows * pData->rowSize;
int32_t len = int32_t len = sprintf(st + VARSTR_HEADER_SIZE,
sprintf(st + VARSTR_HEADER_SIZE, "Total_Blocks=[%d] Total_Size=[%.2f Kb] Average_size=[%.2f Kb] Compression_Ratio=[%.2f %c]", "Total_Blocks=[%d] Total_Size=[%.2f Kb] Average_size=[%.2f Kb] Compression_Ratio=[%.2f %c]",
pData->numOfBlocks, pData->totalSize / 1024.0, ((double)pData->totalSize) / pData->numOfBlocks, pData->numOfBlocks, pData->totalSize / 1024.0, ((double)pData->totalSize) / pData->numOfBlocks,
pData->totalSize * 100 / totalRawSize, '%'); pData->totalSize * 100 / totalRawSize, '%');
varDataSetLen(st, len); varDataSetLen(st, len);
colDataAppend(pColInfo, row++, st, false); colDataAppend(pColInfo, row++, st, false);
len = sprintf(st + VARSTR_HEADER_SIZE, "Total_Rows=[%"PRId64"] Inmem_Rows=[%d] MinRows=[%d] MaxRows=[%d] Average_Rows=[%"PRId64"]", len = sprintf(st + VARSTR_HEADER_SIZE,
pData->totalRows, pData->numOfInmemRows, pData->minRows, pData->maxRows, pData->totalRows / pData->numOfBlocks); "Total_Rows=[%" PRId64 "] Inmem_Rows=[%d] MinRows=[%d] MaxRows=[%d] Average_Rows=[%" PRId64 "]",
pData->totalRows, pData->numOfInmemRows, pData->minRows, pData->maxRows,
pData->totalRows / pData->numOfBlocks);
varDataSetLen(st, len); varDataSetLen(st, len);
colDataAppend(pColInfo, row++, st, false); colDataAppend(pColInfo, row++, st, false);
@ -5540,25 +5524,25 @@ bool irateFuncSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
SRateInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo); SRateInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
pInfo->firstKey = INT64_MIN; pInfo->firstKey = INT64_MIN;
pInfo->lastKey = INT64_MIN; pInfo->lastKey = INT64_MIN;
pInfo->firstValue = (double)INT64_MIN; pInfo->firstValue = (double)INT64_MIN;
pInfo->lastValue = (double)INT64_MIN; pInfo->lastValue = (double)INT64_MIN;
pInfo->hasResult = 0; pInfo->hasResult = 0;
return true; return true;
} }
int32_t irateFunction(SqlFunctionCtx* pCtx) { int32_t irateFunction(SqlFunctionCtx* pCtx) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
SRateInfo* pRateInfo = GET_ROWCELL_INTERBUF(pResInfo); SRateInfo* pRateInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pInputCol = pInput->pData[0];
SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput; SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
TSKEY* tsList = (int64_t*)pInput->pPTS->pData; TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
int32_t numOfElems = 0; int32_t numOfElems = 0;
int32_t type = pInputCol->info.type; int32_t type = pInputCol->info.type;
@ -5570,13 +5554,13 @@ int32_t irateFunction(SqlFunctionCtx* pCtx) {
numOfElems++; numOfElems++;
char* data = colDataGetData(pInputCol, i); char* data = colDataGetData(pInputCol, i);
double v = 0; double v = 0;
GET_TYPED_DATA(v, double, type, data); GET_TYPED_DATA(v, double, type, data);
if (INT64_MIN == pRateInfo->lastKey) { if (INT64_MIN == pRateInfo->lastKey) {
pRateInfo->lastValue = v; pRateInfo->lastValue = v;
pRateInfo->lastKey = tsList[i]; pRateInfo->lastKey = tsList[i];
continue; continue;
} }
@ -5587,7 +5571,7 @@ int32_t irateFunction(SqlFunctionCtx* pCtx) {
} }
pRateInfo->lastValue = v; pRateInfo->lastValue = v;
pRateInfo->lastKey = tsList[i]; pRateInfo->lastKey = tsList[i];
continue; continue;
} }
@ -5596,7 +5580,6 @@ int32_t irateFunction(SqlFunctionCtx* pCtx) {
pRateInfo->firstValue = v; pRateInfo->firstValue = v;
pRateInfo->firstKey = tsList[i]; pRateInfo->firstKey = tsList[i];
} }
} }
SET_VAL(pResInfo, numOfElems, 1); SET_VAL(pResInfo, numOfElems, 1);
@ -5622,7 +5605,7 @@ static double doCalcRate(const SRateInfo* pRateInfo, double tickPerSec) {
return 0; return 0;
} }
return (duration > 0)? ((double)diff) / (duration/tickPerSec):0.0; return (duration > 0) ? ((double)diff) / (duration / tickPerSec) : 0.0;
} }
int32_t irateFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { int32_t irateFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
@ -5633,7 +5616,7 @@ int32_t irateFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
pResInfo->isNullRes = (pResInfo->numOfRes == 0) ? 1 : 0; pResInfo->isNullRes = (pResInfo->numOfRes == 0) ? 1 : 0;
SRateInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo); SRateInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
double result = doCalcRate(pInfo, (double)TSDB_TICK_PER_SECOND(pCtx->param[1].param.i)); double result = doCalcRate(pInfo, (double)TSDB_TICK_PER_SECOND(pCtx->param[1].param.i));
colDataAppend(pCol, pBlock->info.rows, (const char*)&result, pResInfo->isNullRes); colDataAppend(pCol, pBlock->info.rows, (const char*)&result, pResInfo->isNullRes);
return pResInfo->numOfRes; return pResInfo->numOfRes;
@ -5641,14 +5624,14 @@ int32_t irateFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
int32_t groupKeyFunction(SqlFunctionCtx* pCtx) { int32_t groupKeyFunction(SqlFunctionCtx* pCtx) {
SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx); SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
SGroupKeyInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo); SGroupKeyInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
SInputColumnInfoData* pInput = &pCtx->input; SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pInputCol = pInput->pData[0]; SColumnInfoData* pInputCol = pInput->pData[0];
int32_t startIndex = pInput->startRowIndex; int32_t startIndex = pInput->startRowIndex;
//escape rest of data blocks to avoid first entry to be overwritten. // escape rest of data blocks to avoid first entry to be overwritten.
if (pInfo->hasResult) { if (pInfo->hasResult) {
goto _group_key_over; goto _group_key_over;
} }
@ -5661,7 +5644,8 @@ int32_t groupKeyFunction(SqlFunctionCtx* pCtx) {
char* data = colDataGetData(pInputCol, startIndex); char* data = colDataGetData(pInputCol, startIndex);
if (IS_VAR_DATA_TYPE(pInputCol->info.type)) { if (IS_VAR_DATA_TYPE(pInputCol->info.type)) {
memcpy(pInfo->data, data, (pInputCol->info.type == TSDB_DATA_TYPE_JSON) ? getJsonValueLen(data): varDataTLen(data)); memcpy(pInfo->data, data,
(pInputCol->info.type == TSDB_DATA_TYPE_JSON) ? getJsonValueLen(data) : varDataTLen(data));
} else { } else {
memcpy(pInfo->data, data, pInputCol->info.bytes); memcpy(pInfo->data, data, pInputCol->info.bytes);
} }

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

@ -181,6 +181,8 @@ bool fmIsForbidGroupByFunc(int32_t funcId) { return isSpecificClassifyFunc(funcI
bool fmIsSystemInfoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SYSTEM_INFO_FUNC); } bool fmIsSystemInfoFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_SYSTEM_INFO_FUNC); }
bool fmIsImplicitTsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_IMPLICIT_TS_FUNC); }
bool fmIsInterpFunc(int32_t funcId) { bool fmIsInterpFunc(int32_t funcId) {
if (funcId < 0 || funcId >= funcMgtBuiltinsNum) { if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
return false; return false;

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

@ -609,7 +609,7 @@ static int32_t selectStmtCopy(const SSelectStmt* pSrc, SSelectStmt* pDst) {
COPY_CHAR_ARRAY_FIELD(stmtName); COPY_CHAR_ARRAY_FIELD(stmtName);
COPY_SCALAR_FIELD(precision); COPY_SCALAR_FIELD(precision);
COPY_SCALAR_FIELD(isEmptyResult); COPY_SCALAR_FIELD(isEmptyResult);
COPY_SCALAR_FIELD(isTimeOrderQuery); COPY_SCALAR_FIELD(isTimeLineResult);
COPY_SCALAR_FIELD(hasAggFuncs); COPY_SCALAR_FIELD(hasAggFuncs);
COPY_SCALAR_FIELD(hasRepeatScanFuncs); COPY_SCALAR_FIELD(hasRepeatScanFuncs);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;

2
source/libs/parser/src/parAstCreater.c
View File

@ -717,7 +717,7 @@ SNode* createSelectStmt(SAstCreateContext* pCxt, bool isDistinct, SNodeList* pPr
select->pProjectionList = pProjectionList; select->pProjectionList = pProjectionList;
select->pFromTable = pTable; select->pFromTable = pTable;
sprintf(select->stmtName, "%p", select); sprintf(select->stmtName, "%p", select);
select->isTimeOrderQuery = true; select->isTimeLineResult = true;
return (SNode*)select; return (SNode*)select;
} }

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

@ -440,6 +440,10 @@ static bool isTimelineFunc(const SNode* pNode) {
return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsTimelineFunc(((SFunctionNode*)pNode)->funcId)); return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsTimelineFunc(((SFunctionNode*)pNode)->funcId));
} }
static bool isImplicitTsFunc(const SNode* pNode) {
return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsImplicitTsFunc(((SFunctionNode*)pNode)->funcId));
}
static bool isScanPseudoColumnFunc(const SNode* pNode) { static bool isScanPseudoColumnFunc(const SNode* pNode) {
return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsScanPseudoColumnFunc(((SFunctionNode*)pNode)->funcId)); return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsScanPseudoColumnFunc(((SFunctionNode*)pNode)->funcId));
} }
@ -479,6 +483,35 @@ static SNodeList* getProjectList(const SNode* pNode) {
return NULL; return NULL;
} }
static bool isTimeLineQuery(SNode* pStmt) {
if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) {
return ((SSelectStmt*)pStmt)->isTimeLineResult;
} else {
return false;
}
}
static bool isPrimaryKeyImpl(SNode* pExpr) {
if (QUERY_NODE_COLUMN == nodeType(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) {
return isPrimaryKeyImpl(nodesListGetNode(pFunc->pParameterList, 0));
} else if (FUNCTION_TYPE_WSTARTTS == pFunc->funcType || FUNCTION_TYPE_WENDTS == pFunc->funcType) {
return true;
}
}
return false;
}
static bool isPrimaryKey(STempTableNode* pTable, SNode* pExpr) {
if (!isTimeLineQuery(pTable->pSubquery)) {
return false;
}
return isPrimaryKeyImpl(pExpr);
}
static void setColumnInfoBySchema(const SRealTableNode* pTable, const SSchema* pColSchema, int32_t tagFlag, static void setColumnInfoBySchema(const SRealTableNode* pTable, const SSchema* pColSchema, int32_t tagFlag,
SColumnNode* pCol) { SColumnNode* pCol) {
strcpy(pCol->dbName, pTable->table.dbName); strcpy(pCol->dbName, pTable->table.dbName);
@ -500,7 +533,7 @@ static void setColumnInfoBySchema(const SRealTableNode* pTable, const SSchema* p
} }
} }
static void setColumnInfoByExpr(const STableNode* pTable, SExprNode* pExpr, SColumnNode** pColRef) { static void setColumnInfoByExpr(STempTableNode* pTable, SExprNode* pExpr, SColumnNode** pColRef) {
SColumnNode* pCol = *pColRef; SColumnNode* pCol = *pColRef;
// pCol->pProjectRef = (SNode*)pExpr; // pCol->pProjectRef = (SNode*)pExpr;
@ -508,15 +541,8 @@ static void setColumnInfoByExpr(const STableNode* pTable, SExprNode* pExpr, SCol
pExpr->pAssociation = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES); pExpr->pAssociation = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
} }
taosArrayPush(pExpr->pAssociation, &pColRef); taosArrayPush(pExpr->pAssociation, &pColRef);
if (NULL != pTable) { strcpy(pCol->tableAlias, pTable->table.tableAlias);
strcpy(pCol->tableAlias, pTable->tableAlias); pCol->colId = isPrimaryKey(pTable, (SNode*)pExpr) ? PRIMARYKEY_TIMESTAMP_COL_ID : 0;
} else if (QUERY_NODE_COLUMN == nodeType(pExpr)) {
SColumnNode* pProjCol = (SColumnNode*)pExpr;
strcpy(pCol->tableAlias, pProjCol->tableAlias);
pCol->tableId = pProjCol->tableId;
pCol->colId = pProjCol->colId;
pCol->colType = pProjCol->colType;
}
strcpy(pCol->colName, pExpr->aliasName); strcpy(pCol->colName, pExpr->aliasName);
if ('\0' == pCol->node.aliasName[0]) { if ('\0' == pCol->node.aliasName[0]) {
strcpy(pCol->node.aliasName, pCol->colName); strcpy(pCol->node.aliasName, pCol->colName);
@ -538,8 +564,9 @@ static int32_t createColumnsByTable(STranslateContext* pCxt, const STableNode* p
nodesListAppend(pList, (SNode*)pCol); nodesListAppend(pList, (SNode*)pCol);
} }
} else { } else {
SNodeList* pProjectList = getProjectList(((STempTableNode*)pTable)->pSubquery); STempTableNode* pTempTable = (STempTableNode*)pTable;
SNode* pNode; SNodeList* pProjectList = getProjectList(pTempTable->pSubquery);
SNode* pNode;
FOREACH(pNode, pProjectList) { FOREACH(pNode, pProjectList) {
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN); SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) { if (NULL == pCol) {
@ -547,7 +574,7 @@ static int32_t createColumnsByTable(STranslateContext* pCxt, const STableNode* p
} }
nodesListAppend(pList, (SNode*)pCol); nodesListAppend(pList, (SNode*)pCol);
SListCell* pCell = nodesListGetCell(pList, LIST_LENGTH(pList) - 1); SListCell* pCell = nodesListGetCell(pList, LIST_LENGTH(pList) - 1);
setColumnInfoByExpr(pTable, (SExprNode*)pNode, (SColumnNode**)&pCell->pNode); setColumnInfoByExpr(pTempTable, (SExprNode*)pNode, (SColumnNode**)&pCell->pNode);
} }
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -557,35 +584,6 @@ static bool isInternalPrimaryKey(const SColumnNode* pCol) {
return PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && 0 == strcmp(pCol->colName, PK_TS_COL_INTERNAL_NAME); return PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && 0 == strcmp(pCol->colName, PK_TS_COL_INTERNAL_NAME);
} }
static bool isTimeOrderQuery(SNode* pStmt) {
if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) {
return ((SSelectStmt*)pStmt)->isTimeOrderQuery;
} else {
return false;
}
}
static bool isPrimaryKeyImpl(STempTableNode* pTable, SNode* pExpr) {
if (QUERY_NODE_COLUMN == nodeType(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) {
return isPrimaryKeyImpl(pTable, nodesListGetNode(pFunc->pParameterList, 0));
} else if (FUNCTION_TYPE_WSTARTTS == pFunc->funcType || FUNCTION_TYPE_WENDTS == pFunc->funcType) {
return true;
}
}
return false;
}
static bool isPrimaryKey(STempTableNode* pTable, SNode* pExpr) {
if (!isTimeOrderQuery(pTable->pSubquery)) {
return false;
}
return isPrimaryKeyImpl(pTable, pExpr);
}
static int32_t findAndSetColumn(STranslateContext* pCxt, SColumnNode** pColRef, const STableNode* pTable, static int32_t findAndSetColumn(STranslateContext* pCxt, SColumnNode** pColRef, const STableNode* pTable,
bool* pFound) { bool* pFound) {
SColumnNode* pCol = *pColRef; SColumnNode* pCol = *pColRef;
@ -606,18 +604,19 @@ static int32_t findAndSetColumn(STranslateContext* pCxt, SColumnNode** pColRef,
} }
} }
} else { } else {
SNodeList* pProjectList = getProjectList(((STempTableNode*)pTable)->pSubquery); STempTableNode* pTempTable = (STempTableNode*)pTable;
SNode* pNode; SNodeList* pProjectList = getProjectList(pTempTable->pSubquery);
SNode* pNode;
FOREACH(pNode, pProjectList) { FOREACH(pNode, pProjectList) {
SExprNode* pExpr = (SExprNode*)pNode; SExprNode* pExpr = (SExprNode*)pNode;
if (0 == strcmp(pCol->colName, pExpr->aliasName)) { if (0 == strcmp(pCol->colName, pExpr->aliasName)) {
if (*pFound) { if (*pFound) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_AMBIGUOUS_COLUMN, pCol->colName); return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_AMBIGUOUS_COLUMN, pCol->colName);
} }
setColumnInfoByExpr(pTable, pExpr, pColRef); setColumnInfoByExpr(pTempTable, pExpr, pColRef);
*pFound = true; *pFound = true;
} else if (isPrimaryKey((STempTableNode*)pTable, pNode) && isInternalPrimaryKey(pCol)) { } else if (isPrimaryKey(pTempTable, pNode) && isInternalPrimaryKey(pCol)) {
setColumnInfoByExpr(pTable, pExpr, pColRef); setColumnInfoByExpr(pTempTable, pExpr, pColRef);
*pFound = true; *pFound = true;
} }
} }
@ -1259,6 +1258,7 @@ static void setFuncClassification(SNode* pCurrStmt, SFunctionNode* pFunc) {
pSelect->hasTailFunc = pSelect->hasTailFunc ? true : (FUNCTION_TYPE_TAIL == pFunc->funcType); pSelect->hasTailFunc = pSelect->hasTailFunc ? true : (FUNCTION_TYPE_TAIL == pFunc->funcType);
pSelect->hasInterpFunc = pSelect->hasInterpFunc ? true : (FUNCTION_TYPE_INTERP == pFunc->funcType); pSelect->hasInterpFunc = pSelect->hasInterpFunc ? true : (FUNCTION_TYPE_INTERP == pFunc->funcType);
pSelect->hasLastRowFunc = pSelect->hasLastRowFunc ? true : (FUNCTION_TYPE_LAST_ROW == pFunc->funcType); pSelect->hasLastRowFunc = pSelect->hasLastRowFunc ? true : (FUNCTION_TYPE_LAST_ROW == pFunc->funcType);
pSelect->hasTimeLineFunc = pSelect->hasLastRowFunc ? true : fmIsTimelineFunc(pFunc->funcId);
} }
} }
@ -1477,7 +1477,7 @@ static EDealRes rewriteColToSelectValFunc(STranslateContext* pCxt, SNode** pNode
strcpy(pFunc->node.aliasName, ((SExprNode*)*pNode)->aliasName); strcpy(pFunc->node.aliasName, ((SExprNode*)*pNode)->aliasName);
pCxt->errCode = nodesListMakeAppend(&pFunc->pParameterList, *pNode); pCxt->errCode = nodesListMakeAppend(&pFunc->pParameterList, *pNode);
if (TSDB_CODE_SUCCESS == pCxt->errCode) { if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pCxt->errCode == getFuncInfo(pCxt, pFunc); pCxt->errCode = getFuncInfo(pCxt, pFunc);
} }
if (TSDB_CODE_SUCCESS == pCxt->errCode) { if (TSDB_CODE_SUCCESS == pCxt->errCode) {
*pNode = (SNode*)pFunc; *pNode = (SNode*)pFunc;
@ -1633,6 +1633,16 @@ static int32_t checkAggColCoexist(STranslateContext* pCxt, SSelectStmt* pSelect)
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t checkWindowFuncCoexist(STranslateContext* pCxt, SSelectStmt* pSelect) {
if (NULL == pSelect->pWindow) {
return TSDB_CODE_SUCCESS;
}
if (NULL != pSelect->pWindow && !pSelect->hasAggFuncs) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_NO_VALID_FUNC_IN_WIN);
}
return TSDB_CODE_SUCCESS;
}
static int32_t toVgroupsInfo(SArray* pVgs, SVgroupsInfo** pVgsInfo) { static int32_t toVgroupsInfo(SArray* pVgs, SVgroupsInfo** pVgsInfo) {
size_t vgroupNum = taosArrayGetSize(pVgs); size_t vgroupNum = taosArrayGetSize(pVgs);
*pVgsInfo = taosMemoryCalloc(1, sizeof(SVgroupsInfo) + sizeof(SVgroupInfo) * vgroupNum); *pVgsInfo = taosMemoryCalloc(1, sizeof(SVgroupsInfo) + sizeof(SVgroupInfo) * vgroupNum);
@ -2137,7 +2147,7 @@ static int32_t translateGroupBy(STranslateContext* pCxt, SSelectStmt* pSelect) {
} }
if (NULL != pSelect->pGroupByList) { if (NULL != pSelect->pGroupByList) {
pCxt->currClause = SQL_CLAUSE_GROUP_BY; pCxt->currClause = SQL_CLAUSE_GROUP_BY;
pSelect->isTimeOrderQuery = false; pSelect->isTimeLineResult = false;
return translateExprList(pCxt, pSelect->pGroupByList); return translateExprList(pCxt, pSelect->pGroupByList);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -2471,9 +2481,9 @@ static int32_t createPrimaryKeyCol(STranslateContext* pCxt, SNode** pPrimaryKey)
return code; return code;
} }
static EDealRes rewriteTimelineFuncImpl(SNode* pNode, void* pContext) { static EDealRes appendTsForImplicitTsFuncImpl(SNode* pNode, void* pContext) {
STranslateContext* pCxt = pContext; STranslateContext* pCxt = pContext;
if (isTimelineFunc(pNode)) { if (isImplicitTsFunc(pNode)) {
SFunctionNode* pFunc = (SFunctionNode*)pNode; SFunctionNode* pFunc = (SFunctionNode*)pNode;
SNode* pPrimaryKey = NULL; SNode* pPrimaryKey = NULL;
pCxt->errCode = createPrimaryKeyCol(pCxt, &pPrimaryKey); pCxt->errCode = createPrimaryKeyCol(pCxt, &pPrimaryKey);
@ -2485,8 +2495,8 @@ static EDealRes rewriteTimelineFuncImpl(SNode* pNode, void* pContext) {
return DEAL_RES_CONTINUE; return DEAL_RES_CONTINUE;
} }
static int32_t rewriteTimelineFunc(STranslateContext* pCxt, SSelectStmt* pSelect) { static int32_t appendTsForImplicitTsFunc(STranslateContext* pCxt, SSelectStmt* pSelect) {
nodesWalkSelectStmt(pSelect, SQL_CLAUSE_FROM, rewriteTimelineFuncImpl, pCxt); nodesWalkSelectStmt(pSelect, SQL_CLAUSE_FROM, appendTsForImplicitTsFuncImpl, pCxt);
return pCxt->errCode; return pCxt->errCode;
} }
@ -2579,6 +2589,9 @@ static int32_t translateSelectFrom(STranslateContext* pCxt, SSelectStmt* pSelect
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = checkAggColCoexist(pCxt, pSelect); code = checkAggColCoexist(pCxt, pSelect);
} }
if (TSDB_CODE_SUCCESS == code) {
code = checkWindowFuncCoexist(pCxt, pSelect);
}
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = checkLimit(pCxt, pSelect); code = checkLimit(pCxt, pSelect);
} }
@ -2586,7 +2599,7 @@ static int32_t translateSelectFrom(STranslateContext* pCxt, SSelectStmt* pSelect
code = translateInterp(pCxt, pSelect); code = translateInterp(pCxt, pSelect);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = rewriteTimelineFunc(pCxt, pSelect); code = appendTsForImplicitTsFunc(pCxt, pSelect);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = replaceOrderByAlias(pCxt, pSelect->pProjectionList, pSelect->pOrderByList); code = replaceOrderByAlias(pCxt, pSelect->pProjectionList, pSelect->pOrderByList);
@ -3208,7 +3221,7 @@ static int32_t checkTableTagsSchema(STranslateContext* pCxt, SHashObj* pHash, SN
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
if ((TSDB_DATA_TYPE_VARCHAR == pTag->dataType.type && pTag->dataType.bytes > TSDB_MAX_BINARY_LEN) || if ((TSDB_DATA_TYPE_VARCHAR == pTag->dataType.type && pTag->dataType.bytes > TSDB_MAX_BINARY_LEN) ||
(TSDB_DATA_TYPE_NCHAR == pTag->dataType.type && pTag->dataType.bytes > TSDB_MAX_NCHAR_LEN)) { (TSDB_DATA_TYPE_NCHAR == pTag->dataType.type && pTag->dataType.bytes > TSDB_MAX_NCHAR_LEN)) {
code = code = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN); code = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN);
} }
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {

19
source/libs/parser/src/parUtil.c
View File

@ -185,19 +185,21 @@ static char* getSyntaxErrFormat(int32_t errCode) {
case TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG: case TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG:
return "The REDISTRIBUTE VGROUP statement only support 1 to 3 dnodes"; return "The REDISTRIBUTE VGROUP statement only support 1 to 3 dnodes";
case TSDB_CODE_PAR_FILL_NOT_ALLOWED_FUNC: case TSDB_CODE_PAR_FILL_NOT_ALLOWED_FUNC:
return "%s function does not supportted in fill query"; return "%s function is not supported in fill query";
case TSDB_CODE_PAR_INVALID_WINDOW_PC: case TSDB_CODE_PAR_INVALID_WINDOW_PC:
return "_WSTARTTS, _WENDTS and _WDURATION can only be used in window query"; return "_WSTARTTS, _WENDTS and _WDURATION can only be used in window query";
case TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC: case TSDB_CODE_PAR_WINDOW_NOT_ALLOWED_FUNC:
return "%s function does not supportted in time window query"; return "%s function is not supported in time window query";
case TSDB_CODE_PAR_STREAM_NOT_ALLOWED_FUNC: case TSDB_CODE_PAR_STREAM_NOT_ALLOWED_FUNC:
return "%s function does not supportted in stream query"; return "%s function is not supported in stream query";
case TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC: case TSDB_CODE_PAR_GROUP_BY_NOT_ALLOWED_FUNC:
return "%s function does not supportted in group query"; return "%s function is not supported in group query";
case TSDB_CODE_PAR_INVALID_TABLE_OPTION: case TSDB_CODE_PAR_INVALID_TABLE_OPTION:
return "Invalid option %s"; return "Invalid option %s";
case TSDB_CODE_PAR_INVALID_INTERP_CLAUSE: case TSDB_CODE_PAR_INVALID_INTERP_CLAUSE:
return "Invalid usage of RANGE clause, EVERY clause or FILL clause"; return "Invalid usage of RANGE clause, EVERY clause or FILL clause";
case TSDB_CODE_PAR_NO_VALID_FUNC_IN_WIN:
return "No valid function in window query";
case TSDB_CODE_OUT_OF_MEMORY: case TSDB_CODE_OUT_OF_MEMORY:
return "Out of memory"; return "Out of memory";
default: default:
@ -338,11 +340,11 @@ int32_t trimString(const char* src, int32_t len, char* dst, int32_t dlen) {
static bool isValidateTag(char* input) { static bool isValidateTag(char* input) {
if (!input) return false; if (!input) return false;
for (size_t i = 0; i < strlen(input); ++i) { for (size_t i = 0; i < strlen(input); ++i) {
#ifdef WINDOWS #ifdef WINDOWS
if (input[i] < 0x20 || input[i] > 0x7E) return false; if (input[i] < 0x20 || input[i] > 0x7E) return false;
#else #else
if (isprint(input[i]) == 0) return false; if (isprint(input[i]) == 0) return false;
#endif #endif
} }
return true; return true;
} }
@ -382,7 +384,8 @@ int32_t parseJsontoTagData(const char* json, SArray* pTagVals, STag** ppTag, voi
char* jsonKey = item->string; char* jsonKey = item->string;
if (!isValidateTag(jsonKey)) { if (!isValidateTag(jsonKey)) {
fprintf(stdout,"%s(%d) %s %08" PRId64 "\n", __FILE__, __LINE__,__func__,taosGetSelfPthreadId());fflush(stdout); fprintf(stdout, "%s(%d) %s %08" PRId64 "\n", __FILE__, __LINE__, __func__, taosGetSelfPthreadId());
fflush(stdout);
retCode = buildSyntaxErrMsg(pMsgBuf, "json key not validate", jsonKey); retCode = buildSyntaxErrMsg(pMsgBuf, "json key not validate", jsonKey);
goto end; goto end;
} }

4
source/libs/planner/src/planLogicCreater.c
View File

@ -542,6 +542,7 @@ static int32_t createIndefRowsFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt
pIdfRowsFunc->isTailFunc = pSelect->hasTailFunc; pIdfRowsFunc->isTailFunc = pSelect->hasTailFunc;
pIdfRowsFunc->isUniqueFunc = pSelect->hasUniqueFunc; pIdfRowsFunc->isUniqueFunc = pSelect->hasUniqueFunc;
pIdfRowsFunc->isTimeLineFunc = pSelect->hasTimeLineFunc;
// indefinite rows functions and _select_values functions // indefinite rows functions and _select_values functions
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsVectorFunc, &pIdfRowsFunc->pFuncs); int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsVectorFunc, &pIdfRowsFunc->pFuncs);
@ -727,6 +728,9 @@ static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
} }
SFillNode* pFillNode = (SFillNode*)(((SIntervalWindowNode*)pSelect->pWindow)->pFill); SFillNode* pFillNode = (SFillNode*)(((SIntervalWindowNode*)pSelect->pWindow)->pFill);
if (FILL_MODE_NONE == pFillNode->mode) {
return TSDB_CODE_SUCCESS;
}
SFillLogicNode* pFill = (SFillLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_FILL); SFillLogicNode* pFill = (SFillLogicNode*)nodesMakeNode(QUERY_NODE_LOGIC_PLAN_FILL);
if (NULL == pFill) { if (NULL == pFill) {

62
source/libs/planner/src/planOptimizer.c
View File

@ -594,7 +594,7 @@ static int32_t pushDownCondOptDealJoin(SOptimizeContext* pCxt, SJoinLogicNode* p
typedef struct SPartAggCondContext { typedef struct SPartAggCondContext {
SAggLogicNode* pAgg; SAggLogicNode* pAgg;
bool hasAggFunc; bool hasAggFunc;
} SPartAggCondContext; } SPartAggCondContext;
static EDealRes partAggCondHasAggFuncImpl(SNode* pNode, void* pContext) { static EDealRes partAggCondHasAggFuncImpl(SNode* pNode, void* pContext) {
@ -619,11 +619,11 @@ static int32_t partitionAggCondHasAggFunc(SAggLogicNode* pAgg, SNode* pCond) {
static int32_t partitionAggCondConj(SAggLogicNode* pAgg, SNode** ppAggFuncCond, SNode** ppGroupKeyCond) { static int32_t partitionAggCondConj(SAggLogicNode* pAgg, SNode** ppAggFuncCond, SNode** ppGroupKeyCond) {
SLogicConditionNode* pLogicCond = (SLogicConditionNode*)pAgg->node.pConditions; SLogicConditionNode* pLogicCond = (SLogicConditionNode*)pAgg->node.pConditions;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SNodeList* pAggFuncConds = NULL; SNodeList* pAggFuncConds = NULL;
SNodeList* pGroupKeyConds = NULL; SNodeList* pGroupKeyConds = NULL;
SNode* pCond = NULL; SNode* pCond = NULL;
FOREACH(pCond, pLogicCond->pParameterList) { FOREACH(pCond, pLogicCond->pParameterList) {
if (partitionAggCondHasAggFunc(pAgg, pCond)) { if (partitionAggCondHasAggFunc(pAgg, pCond)) {
code = nodesListMakeAppend(&pAggFuncConds, nodesCloneNode(pCond)); code = nodesListMakeAppend(&pAggFuncConds, nodesCloneNode(pCond));
@ -677,14 +677,14 @@ static int32_t pushCondToAggCond(SOptimizeContext* pCxt, SAggLogicNode* pAgg, SN
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
typedef struct SRewriteAggGroupKeyCondContext{ typedef struct SRewriteAggGroupKeyCondContext {
SAggLogicNode *pAgg; SAggLogicNode* pAgg;
int32_t errCode; int32_t errCode;
} SRewriteAggGroupKeyCondContext; } SRewriteAggGroupKeyCondContext;
static EDealRes rewriteAggGroupKeyCondForPushDownImpl(SNode** pNode, void* pContext) { static EDealRes rewriteAggGroupKeyCondForPushDownImpl(SNode** pNode, void* pContext) {
SRewriteAggGroupKeyCondContext* pCxt = pContext; SRewriteAggGroupKeyCondContext* pCxt = pContext;
SAggLogicNode* pAgg = pCxt->pAgg; SAggLogicNode* pAgg = pCxt->pAgg;
if (QUERY_NODE_COLUMN == nodeType(*pNode)) { if (QUERY_NODE_COLUMN == nodeType(*pNode)) {
SNode* pGroupKey = NULL; SNode* pGroupKey = NULL;
FOREACH(pGroupKey, pAgg->pGroupKeys) { FOREACH(pGroupKey, pAgg->pGroupKeys) {
@ -717,15 +717,15 @@ static int32_t pushDownCondOptDealAgg(SOptimizeContext* pCxt, SAggLogicNode* pAg
OPTIMIZE_FLAG_TEST_MASK(pAgg->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE)) { OPTIMIZE_FLAG_TEST_MASK(pAgg->node.optimizedFlag, OPTIMIZE_FLAG_PUSH_DOWN_CONDE)) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
//TODO: remove it after full implementation of pushing down to child // TODO: remove it after full implementation of pushing down to child
if (1 != LIST_LENGTH(pAgg->node.pChildren) || if (1 != LIST_LENGTH(pAgg->node.pChildren) ||
QUERY_NODE_LOGIC_PLAN_SCAN != nodeType(nodesListGetNode(pAgg->node.pChildren, 0)) && QUERY_NODE_LOGIC_PLAN_SCAN != nodeType(nodesListGetNode(pAgg->node.pChildren, 0)) &&
QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(nodesListGetNode(pAgg->node.pChildren, 0))) { QUERY_NODE_LOGIC_PLAN_PROJECT != nodeType(nodesListGetNode(pAgg->node.pChildren, 0))) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SNode* pAggFuncCond = NULL; SNode* pAggFuncCond = NULL;
SNode* pGroupKeyCond = NULL; SNode* pGroupKeyCond = NULL;
int32_t code = partitionAggCond(pAgg, &pAggFuncCond, &pGroupKeyCond); int32_t code = partitionAggCond(pAgg, &pAggFuncCond, &pGroupKeyCond);
if (TSDB_CODE_SUCCESS == code && NULL != pAggFuncCond) { if (TSDB_CODE_SUCCESS == code && NULL != pAggFuncCond) {
code = pushCondToAggCond(pCxt, pAgg, &pAggFuncCond); code = pushCondToAggCond(pCxt, pAgg, &pAggFuncCond);
@ -1462,9 +1462,17 @@ static int32_t rewriteTailOptCreateSort(SIndefRowsFuncLogicNode* pIndef, SLogicN
TSWAP(pSort->node.pChildren, pIndef->node.pChildren); TSWAP(pSort->node.pChildren, pIndef->node.pChildren);
pSort->node.precision = pIndef->node.precision; pSort->node.precision = pIndef->node.precision;
SFunctionNode* pTail = NULL;
SNode* pFunc = NULL;
FOREACH(pFunc, pIndef->pFuncs) {
if (FUNCTION_TYPE_TAIL == ((SFunctionNode*)pFunc)->funcType) {
pTail = (SFunctionNode*)pFunc;
break;
}
}
// tail(expr, [limit, offset,] _rowts) // tail(expr, [limit, offset,] _rowts)
SFunctionNode* pTail = (SFunctionNode*)nodesListGetNode(pIndef->pFuncs, 0); int32_t rowtsIndex = LIST_LENGTH(pTail->pParameterList) - 1;
int32_t rowtsIndex = LIST_LENGTH(pTail->pParameterList) - 1;
int32_t code = nodesListMakeStrictAppend( int32_t code = nodesListMakeStrictAppend(
&pSort->pSortKeys, rewriteTailOptCreateOrderByExpr(nodesListGetNode(pTail->pParameterList, rowtsIndex))); &pSort->pSortKeys, rewriteTailOptCreateOrderByExpr(nodesListGetNode(pTail->pParameterList, rowtsIndex)));
@ -1484,12 +1492,12 @@ static int32_t rewriteTailOptCreateSort(SIndefRowsFuncLogicNode* pIndef, SLogicN
return code; return code;
} }
static SNode* rewriteTailOptCreateProjectExpr(SFunctionNode* pTail) { static SNode* rewriteTailOptCreateProjectExpr(SFunctionNode* pFunc) {
SNode* pExpr = nodesCloneNode(nodesListGetNode(pTail->pParameterList, 0)); SNode* pExpr = nodesCloneNode(nodesListGetNode(pFunc->pParameterList, 0));
if (NULL == pExpr) { if (NULL == pExpr) {
return NULL; return NULL;
} }
strcpy(((SExprNode*)pExpr)->aliasName, pTail->node.aliasName); strcpy(((SExprNode*)pExpr)->aliasName, pFunc->node.aliasName);
return pExpr; return pExpr;
} }
@ -1502,12 +1510,22 @@ static int32_t rewriteTailOptCreateProject(SIndefRowsFuncLogicNode* pIndef, SLog
TSWAP(pProject->node.pTargets, pIndef->node.pTargets); TSWAP(pProject->node.pTargets, pIndef->node.pTargets);
pProject->node.precision = pIndef->node.precision; pProject->node.precision = pIndef->node.precision;
// tail(expr, [limit, offset,] _rowts) int32_t code = TSDB_CODE_SUCCESS;
SFunctionNode* pTail = (SFunctionNode*)nodesListGetNode(pIndef->pFuncs, 0); SFunctionNode* pTail = NULL;
int32_t limitIndex = LIST_LENGTH(pTail->pParameterList) > 2 ? 1 : -1; SNode* pFunc = NULL;
int32_t offsetIndex = LIST_LENGTH(pTail->pParameterList) > 3 ? 2 : -1; FOREACH(pFunc, pIndef->pFuncs) {
code = nodesListMakeStrictAppend(&pProject->pProjections, rewriteTailOptCreateProjectExpr((SFunctionNode*)pFunc));
if (TSDB_CODE_SUCCESS != code) {
break;
}
if (FUNCTION_TYPE_TAIL == ((SFunctionNode*)pFunc)->funcType) {
pTail = (SFunctionNode*)pFunc;
}
}
int32_t code = nodesListMakeStrictAppend(&pProject->pProjections, rewriteTailOptCreateProjectExpr(pTail)); // tail(expr, [limit, offset,] _rowts)
int32_t limitIndex = LIST_LENGTH(pTail->pParameterList) > 2 ? 1 : -1;
int32_t offsetIndex = LIST_LENGTH(pTail->pParameterList) > 3 ? 2 : -1;
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = rewriteTailOptCreateLimit(limitIndex < 0 ? NULL : nodesListGetNode(pTail->pParameterList, limitIndex), code = rewriteTailOptCreateLimit(limitIndex < 0 ? NULL : nodesListGetNode(pTail->pParameterList, limitIndex),
offsetIndex < 0 ? NULL : nodesListGetNode(pTail->pParameterList, offsetIndex), offsetIndex < 0 ? NULL : nodesListGetNode(pTail->pParameterList, offsetIndex),
@ -1862,7 +1880,7 @@ static EDealRes mergeProjectionsExpr(SNode** pNode, void* pContext) {
} }
static int32_t mergeProjectsOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan, SLogicNode* pSelfNode) { static int32_t mergeProjectsOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan, SLogicNode* pSelfNode) {
SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pSelfNode->pChildren, 0); SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pSelfNode->pChildren, 0);
SMergeProjectionsContext cxt = {.pChildProj = (SProjectLogicNode*)pChild, .errCode = TSDB_CODE_SUCCESS}; SMergeProjectionsContext cxt = {.pChildProj = (SProjectLogicNode*)pChild, .errCode = TSDB_CODE_SUCCESS};
nodesRewriteExprs(((SProjectLogicNode*)pSelfNode)->pProjections, mergeProjectionsExpr, &cxt); nodesRewriteExprs(((SProjectLogicNode*)pSelfNode)->pProjections, mergeProjectionsExpr, &cxt);

18
source/libs/planner/test/planBasicTest.cpp
View File

@ -35,6 +35,8 @@ TEST_F(PlanBasicTest, whereClause) {
run("SELECT * FROM t1 WHERE c1 > 10"); run("SELECT * FROM t1 WHERE c1 > 10");
run("SELECT * FROM t1 WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59'"); run("SELECT * FROM t1 WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59'");
run("SELECT ts, c1 FROM t1 WHERE ts > NOW AND ts IS NULL AND (c1 > 0 OR c3 < 20)");
} }
TEST_F(PlanBasicTest, func) { TEST_F(PlanBasicTest, func) {
@ -103,6 +105,22 @@ TEST_F(PlanBasicTest, lastRowFunc) {
run("SELECT LAST_ROW(c1), SUM(c3) FROM t1"); run("SELECT LAST_ROW(c1), SUM(c3) FROM t1");
} }
TEST_F(PlanBasicTest, timeLineFunc) {
useDb("root", "test");
run("SELECT CSUM(c1) FROM t1");
run("SELECT CSUM(c1) FROM st1");
}
TEST_F(PlanBasicTest, multiResFunc) {
useDb("root", "test");
run("SELECT LAST(*) FROM t1");
run("SELECT LAST(c1 + 10, c2) FROM st1");
}
TEST_F(PlanBasicTest, sampleFunc) { TEST_F(PlanBasicTest, sampleFunc) {
useDb("root", "test"); useDb("root", "test");

24
source/libs/sync/inc/syncSnapshot.h
View File

@ -40,8 +40,8 @@ typedef struct SSyncSnapshotSender {
bool start; bool start;
int32_t seq; int32_t seq;
int32_t ack; int32_t ack;
void * pReader; void *pReader;
void * pCurrentBlock; void *pCurrentBlock;
int32_t blockLen; int32_t blockLen;
SSnapshot snapshot; SSnapshot snapshot;
SSyncCfg lastConfig; SSyncCfg lastConfig;
@ -56,20 +56,20 @@ typedef struct SSyncSnapshotSender {
SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex); SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex);
void snapshotSenderDestroy(SSyncSnapshotSender *pSender); void snapshotSenderDestroy(SSyncSnapshotSender *pSender);
bool snapshotSenderIsStart(SSyncSnapshotSender *pSender); bool snapshotSenderIsStart(SSyncSnapshotSender *pSender);
void snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void *pReader); int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void *pReader);
void snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish); int32_t snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish);
int32_t snapshotSend(SSyncSnapshotSender *pSender); int32_t snapshotSend(SSyncSnapshotSender *pSender);
int32_t snapshotReSend(SSyncSnapshotSender *pSender); int32_t snapshotReSend(SSyncSnapshotSender *pSender);
cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender); cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender);
char * snapshotSender2Str(SSyncSnapshotSender *pSender); char *snapshotSender2Str(SSyncSnapshotSender *pSender);
char * snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event); char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event);
//--------------------------------------------------- //---------------------------------------------------
typedef struct SSyncSnapshotReceiver { typedef struct SSyncSnapshotReceiver {
bool start; bool start;
int32_t ack; int32_t ack;
void * pWriter; void *pWriter;
SyncTerm term; SyncTerm term;
SyncTerm privateTerm; SyncTerm privateTerm;
SSnapshot snapshot; SSnapshot snapshot;
@ -80,13 +80,13 @@ typedef struct SSyncSnapshotReceiver {
SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId); SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, SRaftId fromId);
void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver); void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver);
void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg); int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg);
bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver); int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver);
void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver); bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver);
cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver); cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
char * snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver); char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);
char * snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event); char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event);
//--------------------------------------------------- //---------------------------------------------------
// on message // on message

279
source/libs/sync/src/syncAppendEntries.c
View File

@ -628,6 +628,8 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
#endif #endif
static int32_t syncNodeMakeLogSame2(SSyncNode* ths, SyncAppendEntriesBatch* pMsg) { return 0; }
static int32_t syncNodeMakeLogSame(SSyncNode* ths, SyncAppendEntries* pMsg) { static int32_t syncNodeMakeLogSame(SSyncNode* ths, SyncAppendEntries* pMsg) {
int32_t code; int32_t code;
@ -719,7 +721,282 @@ static bool syncNodeOnAppendEntriesLogOK(SSyncNode* pSyncNode, SyncAppendEntries
return false; return false;
} }
int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatch* pMsg) { return 0; } int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatch* pMsg) {
int32_t ret = 0;
int32_t code = 0;
// if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId)) && !ths->pRaftCfg->isStandBy) {
syncNodeEventLog(ths, "recv sync-append-entries-batch, maybe replica already dropped");
return ret;
}
// maybe update term
if (pMsg->term > ths->pRaftStore->currentTerm) {
syncNodeUpdateTerm(ths, pMsg->term);
}
ASSERT(pMsg->term <= ths->pRaftStore->currentTerm);
// reset elect timer
if (pMsg->term == ths->pRaftStore->currentTerm) {
ths->leaderCache = pMsg->srcId;
syncNodeResetElectTimer(ths);
}
ASSERT(pMsg->dataLen >= 0);
// candidate to follower
//
// operation:
// to follower
do {
bool condition = pMsg->term == ths->pRaftStore->currentTerm && ths->state == TAOS_SYNC_STATE_CANDIDATE;
if (condition) {
syncNodeEventLog(ths, "recv sync-append-entries-batch, candidate to follower");
syncNodeBecomeFollower(ths, "from candidate by append entries");
// do not reply?
return ret;
}
} while (0);
// fake match2
//
// condition1:
// preIndex <= my commit index
//
// operation:
// if hasAppendEntries && pMsg->prevLogIndex == ths->commitIndex, append entry
// match my-commit-index or my-commit-index + 1
// no operation on log
do {
bool condition = (pMsg->term == ths->pRaftStore->currentTerm) && (ths->state == TAOS_SYNC_STATE_FOLLOWER) &&
(pMsg->prevLogIndex <= ths->commitIndex);
if (condition) {
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf),
"recv sync-append-entries-batch, fake match2, pre-index:%ld, pre-term:%lu, datalen:%d",
pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->dataLen);
syncNodeEventLog(ths, logBuf);
} while (0);
SyncIndex matchIndex = ths->commitIndex;
bool hasAppendEntries = pMsg->dataLen > 0;
if (hasAppendEntries && pMsg->prevLogIndex == ths->commitIndex) {
SRpcMsg rpcMsgArr[SYNC_MAX_BATCH_SIZE];
memset(rpcMsgArr, 0, sizeof(rpcMsgArr));
int32_t retArrSize = 0;
syncAppendEntriesBatch2RpcMsgArray(pMsg, rpcMsgArr, SYNC_MAX_BATCH_SIZE, &retArrSize);
// make log same
do {
SyncIndex logLastIndex = ths->pLogStore->syncLogLastIndex(ths->pLogStore);
bool hasExtraEntries = logLastIndex > pMsg->prevLogIndex;
if (hasExtraEntries) {
// make log same, rollback deleted entries
code = syncNodeMakeLogSame2(ths, pMsg);
ASSERT(code == 0);
}
} while (0);
// append entry batch
for (int32_t i = 0; i < retArrSize; ++i) {
SSyncRaftEntry* pAppendEntry = syncEntryBuild(1234);
code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pAppendEntry);
if (code != 0) {
return -1;
}
code = syncNodePreCommit(ths, pAppendEntry);
ASSERT(code == 0);
syncEntryDestory(pAppendEntry);
}
// fsync once
SSyncLogStoreData* pData = ths->pLogStore->data;
SWal* pWal = pData->pWal;
walFsync(pWal, true);
// update match index
matchIndex = pMsg->prevLogIndex + retArrSize;
}
// prepare response msg
SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild(ths->vgId);
pReply->srcId = ths->myRaftId;
pReply->destId = pMsg->srcId;
pReply->term = ths->pRaftStore->currentTerm;
pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
pReply->success = true;
pReply->matchIndex = matchIndex;
// send response
SRpcMsg rpcMsg;
syncAppendEntriesReply2RpcMsg(pReply, &rpcMsg);
syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
syncAppendEntriesReplyDestroy(pReply);
return ret;
}
} while (0);
// calculate logOK here, before will coredump, due to fake match
// bool logOK = syncNodeOnAppendEntriesLogOK(ths, pMsg);
bool logOK = true;
// not match
//
// condition1:
// term < myTerm
//
// condition2:
// !logOK
//
// operation:
// not match
// no operation on log
do {
bool condition1 = pMsg->term < ths->pRaftStore->currentTerm;
bool condition2 =
(pMsg->term == ths->pRaftStore->currentTerm) && (ths->state == TAOS_SYNC_STATE_FOLLOWER) && !logOK;
bool condition = condition1 || condition2;
if (condition) {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "recv sync-append-entries, not match, pre-index:%ld, pre-term:%lu, datalen:%d",
pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->dataLen);
syncNodeEventLog(ths, logBuf);
// prepare response msg
SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild(ths->vgId);
pReply->srcId = ths->myRaftId;
pReply->destId = pMsg->srcId;
pReply->term = ths->pRaftStore->currentTerm;
pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
pReply->success = false;
pReply->matchIndex = SYNC_INDEX_INVALID;
// send response
SRpcMsg rpcMsg;
syncAppendEntriesReply2RpcMsg(pReply, &rpcMsg);
syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
syncAppendEntriesReplyDestroy(pReply);
return ret;
}
} while (0);
// really match
//
// condition:
// logOK
//
// operation:
// match
// make log same
do {
bool condition = (pMsg->term == ths->pRaftStore->currentTerm) && (ths->state == TAOS_SYNC_STATE_FOLLOWER) && logOK;
if (condition) {
// has extra entries (> preIndex) in local log
SyncIndex myLastIndex = syncNodeGetLastIndex(ths);
bool hasExtraEntries = myLastIndex > pMsg->prevLogIndex;
// has entries in SyncAppendEntries msg
bool hasAppendEntries = pMsg->dataLen > 0;
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "recv sync-append-entries, match, pre-index:%ld, pre-term:%lu, datalen:%d",
pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->dataLen);
syncNodeEventLog(ths, logBuf);
if (hasExtraEntries) {
// make log same, rollback deleted entries
// code = syncNodeMakeLogSame(ths, pMsg);
ASSERT(code == 0);
}
int32_t retArrSize = 0;
if (hasAppendEntries) {
SRpcMsg rpcMsgArr[SYNC_MAX_BATCH_SIZE];
memset(rpcMsgArr, 0, sizeof(rpcMsgArr));
syncAppendEntriesBatch2RpcMsgArray(pMsg, rpcMsgArr, SYNC_MAX_BATCH_SIZE, &retArrSize);
// append entry batch
for (int32_t i = 0; i < retArrSize; ++i) {
SSyncRaftEntry* pAppendEntry = syncEntryBuild(1234);
code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pAppendEntry);
if (code != 0) {
return -1;
}
code = syncNodePreCommit(ths, pAppendEntry);
ASSERT(code == 0);
syncEntryDestory(pAppendEntry);
}
// fsync once
SSyncLogStoreData* pData = ths->pLogStore->data;
SWal* pWal = pData->pWal;
walFsync(pWal, true);
}
// prepare response msg
SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild(ths->vgId);
pReply->srcId = ths->myRaftId;
pReply->destId = pMsg->srcId;
pReply->term = ths->pRaftStore->currentTerm;
pReply->privateTerm = ths->pNewNodeReceiver->privateTerm;
pReply->success = true;
pReply->matchIndex = hasAppendEntries ? pMsg->prevLogIndex + retArrSize : pMsg->prevLogIndex;
// send response
SRpcMsg rpcMsg;
syncAppendEntriesReply2RpcMsg(pReply, &rpcMsg);
syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
syncAppendEntriesReplyDestroy(pReply);
// maybe update commit index, leader notice me
if (pMsg->commitIndex > ths->commitIndex) {
// has commit entry in local
if (pMsg->commitIndex <= ths->pLogStore->syncLogLastIndex(ths->pLogStore)) {
// advance commit index to sanpshot first
SSnapshot snapshot;
ths->pFsm->FpGetSnapshotInfo(ths->pFsm, &snapshot);
if (snapshot.lastApplyIndex >= 0 && snapshot.lastApplyIndex > ths->commitIndex) {
SyncIndex commitBegin = ths->commitIndex;
SyncIndex commitEnd = snapshot.lastApplyIndex;
ths->commitIndex = snapshot.lastApplyIndex;
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "commit by snapshot from index:%ld to index:%ld", commitBegin,
commitEnd);
syncNodeEventLog(ths, eventLog);
}
SyncIndex beginIndex = ths->commitIndex + 1;
SyncIndex endIndex = pMsg->commitIndex;
// update commit index
ths->commitIndex = pMsg->commitIndex;
// call back Wal
code = ths->pLogStore->updateCommitIndex(ths->pLogStore, ths->commitIndex);
ASSERT(code == 0);
code = syncNodeCommit(ths, beginIndex, endIndex, ths->state);
ASSERT(code == 0);
}
}
return ret;
}
} while (0);
return ret;
}
int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMsg) { int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
int32_t ret = 0; int32_t ret = 0;

26
source/libs/sync/src/syncSnapshot.c
View File

@ -80,7 +80,7 @@ void snapshotSenderDestroy(SSyncSnapshotSender *pSender) {
bool snapshotSenderIsStart(SSyncSnapshotSender *pSender) { return pSender->start; } bool snapshotSenderIsStart(SSyncSnapshotSender *pSender) { return pSender->start; }
// begin send snapshot by snapshot, pReader // begin send snapshot by snapshot, pReader
void snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void *pReader) { int32_t snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void *pReader) {
ASSERT(!snapshotSenderIsStart(pSender)); ASSERT(!snapshotSenderIsStart(pSender));
// init snapshot and reader // init snapshot and reader
@ -181,9 +181,11 @@ void snapshotSenderStart(SSyncSnapshotSender *pSender, SSnapshot snapshot, void
syncNodeEventLog(pSender->pSyncNode, eventLog); syncNodeEventLog(pSender->pSyncNode, eventLog);
taosMemoryFree(eventLog); taosMemoryFree(eventLog);
} while (0); } while (0);
return 0;
} }
void snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) { int32_t snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) {
// close reader // close reader
if (pSender->pReader != NULL) { if (pSender->pReader != NULL) {
int32_t ret = pSender->pSyncNode->pFsm->FpSnapshotStopRead(pSender->pSyncNode->pFsm, pSender->pReader); int32_t ret = pSender->pSyncNode->pFsm->FpSnapshotStopRead(pSender->pSyncNode->pFsm, pSender->pReader);
@ -208,6 +210,8 @@ void snapshotSenderStop(SSyncSnapshotSender *pSender, bool finish) {
syncNodeEventLog(pSender->pSyncNode, eventLog); syncNodeEventLog(pSender->pSyncNode, eventLog);
taosMemoryFree(eventLog); taosMemoryFree(eventLog);
} while (0); } while (0);
return 0;
} }
// when sender receive ack, call this function to send msg from seq // when sender receive ack, call this function to send msg from seq
@ -349,14 +353,14 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) {
char *snapshotSender2Str(SSyncSnapshotSender *pSender) { char *snapshotSender2Str(SSyncSnapshotSender *pSender) {
cJSON *pJson = snapshotSender2Json(pSender); cJSON *pJson = snapshotSender2Json(pSender);
char * serialized = cJSON_Print(pJson); char *serialized = cJSON_Print(pJson);
cJSON_Delete(pJson); cJSON_Delete(pJson);
return serialized; return serialized;
} }
char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) { char *snapshotSender2SimpleStr(SSyncSnapshotSender *pSender, char *event) {
int32_t len = 256; int32_t len = 256;
char * s = taosMemoryMalloc(len); char *s = taosMemoryMalloc(len);
SRaftId destId = pSender->pSyncNode->replicasId[pSender->replicaIndex]; SRaftId destId = pSender->pSyncNode->replicasId[pSender->replicaIndex];
char host[64]; char host[64];
@ -471,7 +475,7 @@ static void snapshotReceiverForceStop(SSyncSnapshotReceiver *pReceiver) {
// if receiver receive msg from seq = SYNC_SNAPSHOT_SEQ_BEGIN, start receiver // if receiver receive msg from seq = SYNC_SNAPSHOT_SEQ_BEGIN, start receiver
// if already start, force close, start again // if already start, force close, start again
void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg) { int32_t snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTerm, SyncSnapshotSend *pBeginMsg) {
if (!snapshotReceiverIsStart(pReceiver)) { if (!snapshotReceiverIsStart(pReceiver)) {
// first start // first start
snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg); snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg);
@ -486,9 +490,11 @@ void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncTerm privateTer
// start again // start again
snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg); snapshotReceiverDoStart(pReceiver, privateTerm, pBeginMsg);
} }
return 0;
} }
void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) { int32_t snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
if (pReceiver->pWriter != NULL) { if (pReceiver->pWriter != NULL) {
int32_t ret = int32_t ret =
pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false); pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, false);
@ -506,6 +512,8 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
syncNodeEventLog(pReceiver->pSyncNode, eventLog); syncNodeEventLog(pReceiver->pSyncNode, eventLog);
taosMemoryFree(eventLog); taosMemoryFree(eventLog);
} while (0); } while (0);
return 0;
} }
static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) { static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
@ -604,7 +612,7 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) {
cJSON_AddStringToObject(pFromId, "addr", u64buf); cJSON_AddStringToObject(pFromId, "addr", u64buf);
{ {
uint64_t u64 = pReceiver->fromId.addr; uint64_t u64 = pReceiver->fromId.addr;
cJSON * pTmp = pFromId; cJSON *pTmp = pFromId;
char host[128] = {0}; char host[128] = {0};
uint16_t port; uint16_t port;
syncUtilU642Addr(u64, host, sizeof(host), &port); syncUtilU642Addr(u64, host, sizeof(host), &port);
@ -637,14 +645,14 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) {
char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) { char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) {
cJSON *pJson = snapshotReceiver2Json(pReceiver); cJSON *pJson = snapshotReceiver2Json(pReceiver);
char * serialized = cJSON_Print(pJson); char *serialized = cJSON_Print(pJson);
cJSON_Delete(pJson); cJSON_Delete(pJson);
return serialized; return serialized;
} }
char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event) { char *snapshotReceiver2SimpleStr(SSyncSnapshotReceiver *pReceiver, char *event) {
int32_t len = 256; int32_t len = 256;
char * s = taosMemoryMalloc(len); char *s = taosMemoryMalloc(len);
SRaftId fromId = pReceiver->fromId; SRaftId fromId = pReceiver->fromId;
char host[128]; char host[128];

7
tests/system-test/2-query/elapsed.py
View File

@ -1397,9 +1397,9 @@ class TDTestCase:
tdSql.error(sql2) tdSql.error(sql2)
tdSql.error(sql3) tdSql.error(sql3)
tdSql.error("select elapsed(ts,10s) from (select ts,tbname from regular_table_1 order by ts ) where ts>=\"2015-01-01 00:00:00.000\" and ts < \"2015-01-01 00:10:00.000\" interval(1s) fill(prev);") tdSql.query("select elapsed(ts,10s) from (select ts,tbname from regular_table_1 order by ts ) where ts>=\"2015-01-01 00:00:00.000\" and ts < \"2015-01-01 00:10:00.000\" interval(1s) fill(prev);")
tdSql.error("select elapsed(ts,10s) from (select ts ,max(q_int),tbname from regular_table_1 order by ts ) where ts>=\"2015-01-01 00:00:00.000\" and ts < \"2015-01-01 00:10:00.000\" interval(1s) fill(prev);") tdSql.query("select elapsed(ts,10s) from (select ts ,max(q_int),tbname from regular_table_1 order by ts ) where ts>=\"2015-01-01 00:00:00.000\" and ts < \"2015-01-01 00:10:00.000\" interval(1s) fill(prev);")
# ===============================================inner nest============================================ # ===============================================inner nest============================================
@ -1486,10 +1486,9 @@ class TDTestCase:
tdSql.checkRows(1) tdSql.checkRows(1)
tdSql.checkData(0,0,9) tdSql.checkData(0,0,9)
tdSql.error('select elapsed(ts,10s) from ( select * from sub_table1_1 where ts>="2015-01-01 00:00:00.000" and ts < "2015-01-01 00:10:00.000") session(ts,1w) ; ') tdSql.query('select elapsed(ts,10s) from ( select * from sub_table1_1 where ts>="2015-01-01 00:00:00.000" and ts < "2015-01-01 00:10:00.000") session(ts,1w) ; ')
tdSql.error('select elapsed(ts,10s) from ( select ts ,q_int from sub_table1_1 where ts>="2015-01-01 00:00:00.000" and ts < "2015-01-01 00:10:00.000") session(ts,1w) ; ') tdSql.error('select elapsed(ts,10s) from ( select ts ,q_int from sub_table1_1 where ts>="2015-01-01 00:00:00.000" and ts < "2015-01-01 00:10:00.000") session(ts,1w) ; ')
tdSql.error('select elapsed(ts,10s) from sub_table1_1 where ts>="2015-01-01 00:00:00.000" and ts < "2015-01-01 00:10:00.000" interval(20s) fill (next) session(ts,1w) ; ') tdSql.error('select elapsed(ts,10s) from sub_table1_1 where ts>="2015-01-01 00:00:00.000" and ts < "2015-01-01 00:10:00.000" interval(20s) fill (next) session(ts,1w) ; ')
tdSql.query('select elapsed(ts,10s) from sub_empty_1 where ts>="2015-01-01 00:00:00.000" and ts < "2015-01-01 00:10:00.000" session(ts,1w) ; ') tdSql.query('select elapsed(ts,10s) from sub_empty_1 where ts>="2015-01-01 00:00:00.000" and ts < "2015-01-01 00:10:00.000" session(ts,1w) ; ')

10
tests/system-test/2-query/irate.py
View File

@ -58,8 +58,8 @@ class TDTestCase:
for coltype in coltypes: for coltype in coltypes:
colname = coltype[0] colname = coltype[0]
if coltype[1] in support_types and coltype[-1] != "TAG" : if coltype[1] in support_types and coltype[-1] != "TAG" :
irate_sql = "select irate({}) from (select * from {} order by tbname ) ".format(colname, tbname) irate_sql = "select irate({}) from {}".format(colname, tbname)
origin_sql = "select ts , {} , cast(ts as bigint) from (select ts , {} from {} order by ts desc limit 2 offset 0 ) order by ts".format(colname,colname, tbname) origin_sql = "select tail({}, 2), cast(ts as bigint) from {} order by ts".format(colname, tbname)
tdSql.query(irate_sql) tdSql.query(irate_sql)
irate_result = tdSql.queryResult irate_result = tdSql.queryResult
@ -68,10 +68,10 @@ class TDTestCase:
irate_value = irate_result[0][0] irate_value = irate_result[0][0]
if origin_result[1][-1] - origin_result[0][-1] == 0: if origin_result[1][-1] - origin_result[0][-1] == 0:
comput_irate_value = 0 comput_irate_value = 0
elif (origin_result[1][1] - origin_result[0][1])<0: elif (origin_result[1][0] - origin_result[0][0])<0:
comput_irate_value = origin_result[1][1]*1000/( origin_result[1][-1] - origin_result[0][-1]) comput_irate_value = origin_result[1][0]*1000/( origin_result[1][-1] - origin_result[0][-1])
else: else:
comput_irate_value = (origin_result[1][1] - origin_result[0][1])*1000/( origin_result[1][-1] - origin_result[0][-1]) comput_irate_value = (origin_result[1][0] - origin_result[0][0])*1000/( origin_result[1][-1] - origin_result[0][-1])
if comput_irate_value ==irate_value: if comput_irate_value ==irate_value:
tdLog.info(" irate work as expected , sql is %s "% irate_sql) tdLog.info(" irate work as expected , sql is %s "% irate_sql)
else: else:

19
tests/system-test/2-query/tail.py
View File

@ -92,8 +92,6 @@ class TDTestCase:
"select tail(c1,1) , min(c1) from t1", "select tail(c1,1) , min(c1) from t1",
"select tail(c1,1) , spread(c1) from t1", "select tail(c1,1) , spread(c1) from t1",
"select tail(c1,1) , diff(c1) from t1", "select tail(c1,1) , diff(c1) from t1",
"select tail(c1,1) , abs(c1) from t1",
"select tail(c1,1) , c1 from t1",
"select tail from stb1 partition by tbname", "select tail from stb1 partition by tbname",
"select tail(123--123)==1 from stb1 partition by tbname", "select tail(123--123)==1 from stb1 partition by tbname",
"select tail(123,123) from stb1 partition by tbname", "select tail(123,123) from stb1 partition by tbname",
@ -115,10 +113,7 @@ class TDTestCase:
"select tail(c1,1) , avg(c1) from stb1 partition by tbname", "select tail(c1,1) , avg(c1) from stb1 partition by tbname",
"select tail(c1,1) , min(c1) from stb1 partition by tbname", "select tail(c1,1) , min(c1) from stb1 partition by tbname",
"select tail(c1,1) , spread(c1) from stb1 partition by tbname", "select tail(c1,1) , spread(c1) from stb1 partition by tbname",
"select tail(c1,1) , diff(c1) from stb1 partition by tbname", "select tail(c1,1) , diff(c1) from stb1 partition by tbname",
"select tail(c1,1) , abs(c1) from stb1 partition by tbname",
"select tail(c1,1) , c1 from stb1 partition by tbname"
] ]
for error_sql in error_sql_lists: for error_sql in error_sql_lists:
tdSql.error(error_sql) tdSql.error(error_sql)
@ -266,17 +261,17 @@ class TDTestCase:
tdSql.query("select tail(c1,10,10) from ct1") tdSql.query("select tail(c1,10,10) from ct1")
tdSql.checkRows(3) tdSql.checkRows(3)
tdSql.error("select tail(c1,10,10),tbname from ct1") tdSql.query("select tail(c1,10,10),tbname from ct1")
tdSql.error("select tail(c1,10,10),t1 from ct1") tdSql.query("select tail(c1,10,10),t1 from ct1")
# tail with common col # tail with common col
tdSql.error("select tail(c1,10,10) ,ts from ct1") tdSql.query("select tail(c1,10,10) ,ts from ct1")
tdSql.error("select tail(c1,10,10) ,c1 from ct1") tdSql.query("select tail(c1,10,10) ,c1 from ct1")
# tail with scalar function # tail with scalar function
tdSql.error("select tail(c1,10,10) ,abs(c1) from ct1") tdSql.query("select tail(c1,10,10) ,abs(c1) from ct1")
tdSql.error("select tail(c1,10,10) , tail(c2,10,10) from ct1") tdSql.error("select tail(c1,10,10) , tail(c2,10,10) from ct1")
tdSql.error("select tail(c1,10,10) , abs(c2)+2 from ct1") tdSql.query("select tail(c1,10,10) , abs(c2)+2 from ct1")
# bug need fix for scalar value or compute again # bug need fix for scalar value or compute again
# tdSql.error(" select tail(c1,10,10) , 123 from ct1") # tdSql.error(" select tail(c1,10,10) , 123 from ct1")

2
tools/taos-tools

@ -1 +1 @@
Subproject commit 7105027650b51e701cfa1dac11b8fb42d447dd01 Subproject commit 5fdd694621fbb7bd2d6102ff4feaec92a7001038