Merge remote-tracking branch 'origin/3.0' into fix/3.0_merge_main
This commit is contained in:
Xiaoyu Wang
commit
9578c4538a
|
|
@ -27,7 +27,7 @@ description: "TDengine 3.0 版本的语法变更说明"
|
|||
| - | :------- | :-------- | :------- |
|
||||
| 1 | ALTER ACCOUNT | 废除 | 2.x中为企业版功能，3.0不再支持。语法暂时保留了，执行报“This statement is no longer supported”错误。
|
||||
| 2 | ALTER ALL DNODES | 新增 | 修改所有DNODE的参数。
|
||||
| 3 | ALTER DATABASE | 调整 | 废除<ul><li>QUORUM：写入需要的副本确认数。3.0版本使用STRICT来指定强一致还是弱一致。3.0.0版本STRICT暂不支持修改。</li><li>BLOCKS：VNODE使用的内存块数。3.0版本使用BUFFER来表示VNODE写入内存池的大小。</li><li>UPDATE：更新操作的支持模式。3.0版本所有数据库都支持部分列更新。</li><li>CACHELAST：缓存最新一行数据的模式。3.0版本用CACHEMODEL代替。</li><li>COMP：3.0版本暂不支持修改。<br/>新增<li>CACHEMODEL：表示是否在内存中缓存子表的最近数据。</li><li>CACHESIZE：表示缓存子表最近数据的内存大小。</li><li>WAL_FSYNC_PERIOD：代替原FSYNC参数。</li><li>WAL_LEVEL：代替原WAL参数。</li><li>WAL_RETENTION_PERIOD：3.0.4.0版本新增，wal文件的额外保留策略，用于数据订阅。</li><li>WAL_RETENTION_SIZE：3.0.4.0版本新增，wal文件的额外保留策略，用于数据订阅。<br/>调整</li><li>REPLICA：3.0.0版本暂不支持修改。</li><li>KEEP：3.0版本新增支持带单位的设置方式。</li></ul>
|
||||
| 3 | ALTER DATABASE | 调整 | 废除<ul><li>QUORUM：写入需要的副本确认数。3.0版本使用STRICT来指定强一致还是弱一致。3.0.0版本STRICT暂不支持修改。</li><li>BLOCKS：VNODE使用的内存块数。3.0版本使用BUFFER来表示VNODE写入内存池的大小。</li><li>UPDATE：更新操作的支持模式。3.0版本所有数据库都支持部分列更新。</li><li>CACHELAST：缓存最新一行数据的模式。3.0版本用CACHEMODEL代替。</li><li>COMP：3.0版本暂不支持修改。</li><br/>新增<li>CACHEMODEL：表示是否在内存中缓存子表的最近数据。</li><li>CACHESIZE：表示缓存子表最近数据的内存大小。</li><li>WAL_FSYNC_PERIOD：代替原FSYNC参数。</li><li>WAL_LEVEL：代替原WAL参数。</li><li>WAL_RETENTION_PERIOD：3.0.4.0版本新增，wal文件的额外保留策略，用于数据订阅。</li><li>WAL_RETENTION_SIZE：3.0.4.0版本新增，wal文件的额外保留策略，用于数据订阅。<br/>调整</li><li>REPLICA：3.0.0版本暂不支持修改。</li><li>KEEP：3.0版本新增支持带单位的设置方式。</li></ul>
|
||||
| 4 | ALTER STABLE | 调整 | 废除<ul><li>CHANGE TAG：修改标签列的名称。3.0版本使用RENAME TAG代替。<br/>新增</li><li>RENAME TAG：代替原CHANGE TAG子句。</li><li>COMMENT：修改超级表的注释。</li></ul>
|
||||
| 5 | ALTER TABLE | 调整 | 废除<ul><li>CHANGE TAG：修改标签列的名称。3.0版本使用RENAME TAG代替。<br/>新增</li><li>RENAME TAG：代替原CHANGE TAG子句。</li><li>COMMENT：修改表的注释。</li><li>TTL：修改表的生命周期。</li></ul>
|
||||
| 6 | ALTER USER | 调整 | 废除<ul><li>PRIVILEGE：修改用户权限。3.0版本使用GRANT和REVOKE来授予和回收权限。<br/>新增</li><li>ENABLE：启用或停用此用户。</li><li>SYSINFO：修改用户是否可查看系统信息。</li></ul>
|
||||
|
|
|
|||
|
|
@ -1105,9 +1105,15 @@ typedef struct {
|
|||
char* pCode;
|
||||
} SFuncInfo;
|
||||
|
||||
typedef struct {
|
||||
int32_t funcVersion;
|
||||
int64_t funcCreatedTime;
|
||||
} SFuncExtraInfo;
|
||||
|
||||
typedef struct {
|
||||
int32_t numOfFuncs;
|
||||
SArray* pFuncInfos;
|
||||
SArray* pFuncExtraInfos;
|
||||
} SRetrieveFuncRsp;
|
||||
|
||||
int32_t tSerializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp);
|
||||
|
|
|
|||
|
|
@ -276,6 +276,8 @@ typedef enum EUdfFuncType { UDF_FUNC_TYPE_SCALAR = 1, UDF_FUNC_TYPE_AGG = 2 } EU
|
|||
|
||||
typedef struct SScriptUdfInfo {
|
||||
const char *name;
|
||||
int32_t version;
|
||||
int64_t createdTime;
|
||||
|
||||
EUdfFuncType funcType;
|
||||
int8_t scriptType;
|
||||
|
|
|
|||
|
|
@ -116,6 +116,7 @@ static const SSysDbTableSchema userFuncSchema[] = {
|
|||
{.name = "bufsize", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
|
||||
{.name = "func_language", .bytes = TSDB_TYPE_STR_MAX_LEN - 1 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
|
||||
{.name = "func_body", .bytes = TSDB_MAX_BINARY_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
|
||||
{.name = "func_version", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
|
||||
};
|
||||
|
||||
static const SSysDbTableSchema userIdxSchema[] = {
|
||||
|
|
|
|||
|
|
@ -1831,6 +1831,7 @@ int32_t tSerializeSCreateFuncReq(void *buf, int32_t bufLen, SCreateFuncReq *pReq
|
|||
if (tEncodeCStr(&encoder, pReq->pComment) < 0) return -1;
|
||||
}
|
||||
|
||||
|
||||
if (tEncodeI8(&encoder, pReq->orReplace) < 0) return -1;
|
||||
|
||||
tEndEncode(&encoder);
|
||||
|
|
@ -1875,6 +1876,7 @@ int32_t tDeserializeSCreateFuncReq(void *buf, int32_t bufLen, SCreateFuncReq *pR
|
|||
if (tDecodeCStrTo(&decoder, pReq->pComment) < 0) return -1;
|
||||
}
|
||||
|
||||
|
||||
if (!tDecodeIsEnd(&decoder)) {
|
||||
if (tDecodeI8(&decoder, &pReq->orReplace) < 0) return -1;
|
||||
} else {
|
||||
|
|
@ -1992,6 +1994,13 @@ int32_t tSerializeSRetrieveFuncRsp(void *buf, int32_t bufLen, SRetrieveFuncRsp *
|
|||
}
|
||||
}
|
||||
|
||||
if (pRsp->numOfFuncs != (int32_t)taosArrayGetSize(pRsp->pFuncExtraInfos)) return -1;
|
||||
for (int32_t i = 0; i < pRsp->numOfFuncs; ++i) {
|
||||
SFuncExtraInfo *extraInfo = taosArrayGet(pRsp->pFuncExtraInfos, i);
|
||||
if (tEncodeI32(&encoder, extraInfo->funcVersion) < 0) return -1;
|
||||
if (tEncodeI64(&encoder, extraInfo->funcCreatedTime) < 0) return -1;
|
||||
}
|
||||
|
||||
tEndEncode(&encoder);
|
||||
|
||||
int32_t tlen = encoder.pos;
|
||||
|
|
@ -2039,6 +2048,22 @@ int32_t tDeserializeSRetrieveFuncRsp(void *buf, int32_t bufLen, SRetrieveFuncRsp
|
|||
|
||||
taosArrayPush(pRsp->pFuncInfos, &fInfo);
|
||||
}
|
||||
|
||||
pRsp->pFuncExtraInfos = taosArrayInit(pRsp->numOfFuncs, sizeof(SFuncExtraInfo));
|
||||
if (pRsp->pFuncExtraInfos == NULL) return -1;
|
||||
if (tDecodeIsEnd(&decoder)) {
|
||||
for (int32_t i = 0; i < pRsp->numOfFuncs; ++i) {
|
||||
SFuncExtraInfo extraInfo = { 0 };
|
||||
taosArrayPush(pRsp->pFuncExtraInfos, &extraInfo);
|
||||
}
|
||||
} else {
|
||||
for (int32_t i = 0; i < pRsp->numOfFuncs; ++i) {
|
||||
SFuncExtraInfo extraInfo = { 0 };
|
||||
if (tDecodeI32(&decoder, &extraInfo.funcVersion) < 0) return -1;
|
||||
if (tDecodeI64(&decoder, &extraInfo.funcCreatedTime) < 0) return -1;
|
||||
taosArrayPush(pRsp->pFuncExtraInfos, &extraInfo);
|
||||
}
|
||||
}
|
||||
tEndDecode(&decoder);
|
||||
|
||||
tDecoderClear(&decoder);
|
||||
|
|
@ -2061,6 +2086,7 @@ void tFreeSRetrieveFuncRsp(SRetrieveFuncRsp *pRsp) {
|
|||
tFreeSFuncInfo(pInfo);
|
||||
}
|
||||
taosArrayDestroy(pRsp->pFuncInfos);
|
||||
taosArrayDestroy(pRsp->pFuncExtraInfos);
|
||||
}
|
||||
|
||||
int32_t tSerializeSTableCfgReq(void *buf, int32_t bufLen, STableCfgReq *pReq) {
|
||||
|
|
|
|||
|
|
@ -450,6 +450,8 @@ typedef struct {
|
|||
int32_t codeSize;
|
||||
char* pComment;
|
||||
char* pCode;
|
||||
int32_t funcVersion;
|
||||
SRWLatch lock;
|
||||
} SFuncObj;
|
||||
|
||||
typedef struct {
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@
|
|||
#include "mndTrans.h"
|
||||
#include "mndUser.h"
|
||||
|
||||
#define SDB_FUNC_VER 1
|
||||
#define SDB_FUNC_VER 2
|
||||
#define SDB_FUNC_RESERVE_SIZE 64
|
||||
|
||||
static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc);
|
||||
|
|
@ -83,6 +83,7 @@ static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc) {
|
|||
SDB_SET_BINARY(pRaw, dataPos, pFunc->pComment, pFunc->commentSize, _OVER)
|
||||
}
|
||||
SDB_SET_BINARY(pRaw, dataPos, pFunc->pCode, pFunc->codeSize, _OVER)
|
||||
SDB_SET_INT32(pRaw, dataPos, pFunc->funcVersion, _OVER)
|
||||
SDB_SET_RESERVE(pRaw, dataPos, SDB_FUNC_RESERVE_SIZE, _OVER)
|
||||
SDB_SET_DATALEN(pRaw, dataPos, _OVER);
|
||||
|
||||
|
|
@ -107,7 +108,7 @@ static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw) {
|
|||
int8_t sver = 0;
|
||||
if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
|
||||
|
||||
if (sver != SDB_FUNC_VER) {
|
||||
if (sver != 1 && sver != 2) {
|
||||
terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
|
||||
goto _OVER;
|
||||
}
|
||||
|
|
@ -144,8 +145,15 @@ static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw) {
|
|||
goto _OVER;
|
||||
}
|
||||
SDB_GET_BINARY(pRaw, dataPos, pFunc->pCode, pFunc->codeSize, _OVER)
|
||||
|
||||
if (sver >= 2) {
|
||||
SDB_GET_INT32(pRaw, dataPos, &pFunc->funcVersion, _OVER)
|
||||
}
|
||||
|
||||
SDB_GET_RESERVE(pRaw, dataPos, SDB_FUNC_RESERVE_SIZE, _OVER)
|
||||
|
||||
taosInitRWLatch(&pFunc->lock);
|
||||
|
||||
terrno = 0;
|
||||
|
||||
_OVER:
|
||||
|
|
@ -173,6 +181,44 @@ static int32_t mndFuncActionDelete(SSdb *pSdb, SFuncObj *pFunc) {
|
|||
|
||||
static int32_t mndFuncActionUpdate(SSdb *pSdb, SFuncObj *pOld, SFuncObj *pNew) {
|
||||
mTrace("func:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);
|
||||
|
||||
taosWLockLatch(&pOld->lock);
|
||||
|
||||
pOld->align = pNew->align;
|
||||
pOld->bufSize = pNew->bufSize;
|
||||
pOld->codeSize = pNew->codeSize;
|
||||
pOld->commentSize = pNew->commentSize;
|
||||
pOld->createdTime = pNew->createdTime;
|
||||
pOld->funcType = pNew->funcType;
|
||||
pOld->funcVersion = pNew->funcVersion;
|
||||
pOld->outputLen = pNew->outputLen;
|
||||
pOld->outputType = pNew->outputType;
|
||||
|
||||
if (pOld->pComment != NULL) {
|
||||
taosMemoryFree(pOld->pComment);
|
||||
pOld->pComment = NULL;
|
||||
}
|
||||
if (pNew->commentSize > 0 && pNew->pComment != NULL) {
|
||||
pOld->commentSize = pNew->commentSize;
|
||||
pOld->pComment = taosMemoryMalloc(pOld->commentSize);
|
||||
memcpy(pOld->pComment, pNew->pComment, pOld->commentSize);
|
||||
}
|
||||
|
||||
if (pOld->pCode != NULL) {
|
||||
taosMemoryFree(pOld->pCode);
|
||||
pOld->pCode = NULL;
|
||||
}
|
||||
if (pNew->codeSize > 0 && pNew->pCode != NULL) {
|
||||
pOld->codeSize = pNew->codeSize;
|
||||
pOld->pCode = taosMemoryMalloc(pOld->codeSize);
|
||||
memcpy(pOld->pCode, pNew->pCode, pOld->codeSize);
|
||||
}
|
||||
|
||||
pOld->scriptType = pNew->scriptType;
|
||||
pOld->signature = pNew->signature;
|
||||
|
||||
taosWUnLockLatch(&pOld->lock);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
@ -225,26 +271,47 @@ static int32_t mndCreateFunc(SMnode *pMnode, SRpcMsg *pReq, SCreateFuncReq *pCre
|
|||
|
||||
pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq, "create-func");
|
||||
if (pTrans == NULL) goto _OVER;
|
||||
|
||||
mInfo("trans:%d, used to create func:%s", pTrans->id, pCreate->name);
|
||||
|
||||
SSdbRaw *pRedoRaw = mndFuncActionEncode(&func);
|
||||
if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING) != 0) goto _OVER;
|
||||
SFuncObj *oldFunc = mndAcquireFunc(pMnode, pCreate->name);
|
||||
if (pCreate->orReplace == 1 && oldFunc != NULL) {
|
||||
func.funcVersion = oldFunc->funcVersion + 1;
|
||||
func.createdTime = oldFunc->createdTime;
|
||||
|
||||
SSdbRaw *pUndoRaw = mndFuncActionEncode(&func);
|
||||
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED) != 0) goto _OVER;
|
||||
SSdbRaw *pRedoRaw = mndFuncActionEncode(oldFunc);
|
||||
if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) goto _OVER;
|
||||
|
||||
SSdbRaw *pCommitRaw = mndFuncActionEncode(&func);
|
||||
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) goto _OVER;
|
||||
SSdbRaw *pUndoRaw = mndFuncActionEncode(oldFunc);
|
||||
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_READY) != 0) goto _OVER;
|
||||
|
||||
SSdbRaw *pCommitRaw = mndFuncActionEncode(&func);
|
||||
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) goto _OVER;
|
||||
} else {
|
||||
SSdbRaw *pRedoRaw = mndFuncActionEncode(&func);
|
||||
if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING) != 0) goto _OVER;
|
||||
|
||||
SSdbRaw *pUndoRaw = mndFuncActionEncode(&func);
|
||||
if (pUndoRaw == NULL || mndTransAppendUndolog(pTrans, pUndoRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED) != 0) goto _OVER;
|
||||
|
||||
SSdbRaw *pCommitRaw = mndFuncActionEncode(&func);
|
||||
if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER;
|
||||
if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) goto _OVER;
|
||||
}
|
||||
|
||||
if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
|
||||
|
||||
code = 0;
|
||||
|
||||
_OVER:
|
||||
if (oldFunc != NULL) {
|
||||
mndReleaseFunc(pMnode, oldFunc);
|
||||
}
|
||||
|
||||
taosMemoryFree(func.pCode);
|
||||
taosMemoryFree(func.pComment);
|
||||
mndTransDrop(pTrans);
|
||||
|
|
@ -304,6 +371,9 @@ static int32_t mndProcessCreateFuncReq(SRpcMsg *pReq) {
|
|||
mInfo("func:%s, already exist, ignore exist is set", createReq.name);
|
||||
code = 0;
|
||||
goto _OVER;
|
||||
} else if (createReq.orReplace) {
|
||||
mInfo("func:%s, replace function is set", createReq.name);
|
||||
code = 0;
|
||||
} else {
|
||||
terrno = TSDB_CODE_MND_FUNC_ALREADY_EXIST;
|
||||
goto _OVER;
|
||||
|
|
@ -413,6 +483,12 @@ static int32_t mndProcessRetrieveFuncReq(SRpcMsg *pReq) {
|
|||
goto RETRIEVE_FUNC_OVER;
|
||||
}
|
||||
|
||||
retrieveRsp.pFuncExtraInfos = taosArrayInit(retrieveReq.numOfFuncs, sizeof(SFuncExtraInfo));
|
||||
if (retrieveRsp.pFuncExtraInfos == NULL) {
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
goto RETRIEVE_FUNC_OVER;
|
||||
}
|
||||
|
||||
for (int32_t i = 0; i < retrieveReq.numOfFuncs; ++i) {
|
||||
char *funcName = taosArrayGet(retrieveReq.pFuncNames, i);
|
||||
|
||||
|
|
@ -451,6 +527,11 @@ static int32_t mndProcessRetrieveFuncReq(SRpcMsg *pReq) {
|
|||
}
|
||||
}
|
||||
taosArrayPush(retrieveRsp.pFuncInfos, &funcInfo);
|
||||
SFuncExtraInfo extraInfo = {0};
|
||||
extraInfo.funcVersion = pFunc->funcVersion;
|
||||
extraInfo.funcCreatedTime = pFunc->createdTime;
|
||||
taosArrayPush(retrieveRsp.pFuncExtraInfos, &extraInfo);
|
||||
|
||||
mndReleaseFunc(pMnode, pFunc);
|
||||
}
|
||||
|
||||
|
|
@ -547,7 +628,7 @@ static int32_t mndRetrieveFuncs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
|
|||
colDataSetVal(pColInfo, numOfRows, (const char *)&pFunc->bufSize, false);
|
||||
|
||||
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
|
||||
char* language = "";
|
||||
char *language = "";
|
||||
if (pFunc->scriptType == TSDB_FUNC_SCRIPT_BIN_LIB) {
|
||||
language = "C";
|
||||
} else if (pFunc->scriptType == TSDB_FUNC_SCRIPT_PYTHON) {
|
||||
|
|
@ -559,13 +640,18 @@ static int32_t mndRetrieveFuncs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
|
|||
colDataSetVal(pColInfo, numOfRows, (const char *)varLang, false);
|
||||
|
||||
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
|
||||
int32_t varCodeLen = (pFunc->codeSize + VARSTR_HEADER_SIZE) > TSDB_MAX_BINARY_LEN ? TSDB_MAX_BINARY_LEN : pFunc->codeSize + VARSTR_HEADER_SIZE;
|
||||
char *b4 = taosMemoryMalloc(varCodeLen);
|
||||
int32_t varCodeLen = (pFunc->codeSize + VARSTR_HEADER_SIZE) > TSDB_MAX_BINARY_LEN
|
||||
? TSDB_MAX_BINARY_LEN
|
||||
: pFunc->codeSize + VARSTR_HEADER_SIZE;
|
||||
char *b4 = taosMemoryMalloc(varCodeLen);
|
||||
memcpy(varDataVal(b4), pFunc->pCode, varCodeLen - VARSTR_HEADER_SIZE);
|
||||
varDataSetLen(b4, varCodeLen - VARSTR_HEADER_SIZE);
|
||||
colDataSetVal(pColInfo, numOfRows, (const char*)b4, false);
|
||||
colDataSetVal(pColInfo, numOfRows, (const char *)b4, false);
|
||||
taosMemoryFree(b4);
|
||||
|
||||
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
|
||||
colDataSetVal(pColInfo, numOfRows, (const char *)&pFunc->funcVersion, false);
|
||||
|
||||
numOfRows++;
|
||||
sdbRelease(pSdb, pFunc);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -671,12 +671,17 @@ void ctgTestRspUdfInfo(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pR
|
|||
SRetrieveFuncRsp funcRsp = {0};
|
||||
funcRsp.numOfFuncs = 1;
|
||||
funcRsp.pFuncInfos = taosArrayInit(1, sizeof(SFuncInfo));
|
||||
funcRsp.pFuncExtraInfos = taosArrayInit(1, sizeof(SFuncExtraInfo));
|
||||
SFuncInfo funcInfo = {0};
|
||||
strcpy(funcInfo.name, "func1");
|
||||
funcInfo.funcType = ctgTestFuncType;
|
||||
|
||||
(void)taosArrayPush(funcRsp.pFuncInfos, &funcInfo);
|
||||
|
||||
SFuncExtraInfo extraInfo = {0};
|
||||
extraInfo.funcVersion = 0;
|
||||
extraInfo.funcCreatedTime = taosGetTimestampMs();
|
||||
(void)taosArrayPush(funcRsp.pFuncExtraInfos, &extraInfo);
|
||||
|
||||
int32_t rspLen = tSerializeSRetrieveFuncRsp(NULL, 0, &funcRsp);
|
||||
void *pReq = rpcMallocCont(rspLen);
|
||||
tSerializeSRetrieveFuncRsp(pReq, rspLen, &funcRsp);
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ extern "C" {
|
|||
struct SSDataBlock;
|
||||
|
||||
typedef struct SFillColInfo {
|
||||
int32_t numOfFillExpr;
|
||||
SExprInfo* pExpr;
|
||||
bool notFillCol; // denote if this column needs fill operation
|
||||
SVariant fillVal;
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ static void setNotFillColumn(SFillInfo* pFillInfo, SColumnInfoData* pDstColInfo,
|
|||
} else {
|
||||
p = FILL_IS_ASC_FILL(pFillInfo) ? &pFillInfo->prev : &pFillInfo->next;
|
||||
}
|
||||
|
||||
|
||||
SGroupKeys* pKey = taosArrayGet(p->pRowVal, colIdx);
|
||||
doSetVal(pDstColInfo, rowIndex, pKey);
|
||||
}
|
||||
|
|
@ -578,7 +578,12 @@ int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint*
|
|||
GET_TYPED_DATA(v1, double, inputType, point1->val);
|
||||
GET_TYPED_DATA(v2, double, inputType, point2->val);
|
||||
|
||||
double r = DO_INTERPOLATION(v1, v2, point1->key, point2->key, point->key);
|
||||
double r = 0;
|
||||
if (!IS_BOOLEAN_TYPE(inputType)) {
|
||||
r = DO_INTERPOLATION(v1, v2, point1->key, point2->key, point->key);
|
||||
} else {
|
||||
r = (v1 < 1 || v2 < 1) ? 0 : 1;
|
||||
}
|
||||
SET_TYPED_DATA(point->val, outputType, r);
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
|
|
@ -630,6 +635,7 @@ SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprIn
|
|||
nodesValueNodeToVariant(pv, &pFillCol[i].fillVal);
|
||||
}
|
||||
}
|
||||
pFillCol->numOfFillExpr = numOfFillExpr;
|
||||
|
||||
for (int32_t i = 0; i < numOfNoFillExpr; ++i) {
|
||||
SExprInfo* pExprInfo = &pNotFillExpr[i];
|
||||
|
|
|
|||
|
|
@ -156,6 +156,16 @@ static FORCE_INLINE int32_t timeSliceEnsureBlockCapacity(STimeSliceOperatorInfo*
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static bool isIrowtsPseudoColumn(SExprInfo* pExprInfo) {
|
||||
char *name = pExprInfo->pExpr->_function.functionName;
|
||||
return (IS_TIMESTAMP_TYPE(pExprInfo->base.resSchema.type) && strcasecmp(name, "_irowts") == 0);
|
||||
}
|
||||
|
||||
static bool isIsfilledPseudoColumn(SExprInfo* pExprInfo) {
|
||||
char *name = pExprInfo->pExpr->_function.functionName;
|
||||
return (IS_BOOLEAN_TYPE(pExprInfo->base.resSchema.type) && strcasecmp(name, "_isfilled") == 0);
|
||||
}
|
||||
|
||||
static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp* pExprSup, SSDataBlock* pResBlock,
|
||||
bool beforeTs) {
|
||||
int32_t rows = pResBlock->info.rows;
|
||||
|
|
@ -170,10 +180,10 @@ static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
|
|||
int32_t dstSlot = pExprInfo->base.resSchema.slotId;
|
||||
SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
|
||||
|
||||
if (IS_TIMESTAMP_TYPE(pExprInfo->base.resSchema.type)) {
|
||||
if (isIrowtsPseudoColumn(pExprInfo)) {
|
||||
colDataSetVal(pDst, rows, (char*)&pSliceInfo->current, false);
|
||||
continue;
|
||||
} else if (IS_BOOLEAN_TYPE(pExprInfo->base.resSchema.type)) {
|
||||
} else if (isIsfilledPseudoColumn(pExprInfo)) {
|
||||
bool isFilled = true;
|
||||
colDataAppend(pDst, pResBlock->info.rows, (char*)&isFilled, false);
|
||||
continue;
|
||||
|
|
@ -203,6 +213,14 @@ static bool genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp
|
|||
int64_t v = 0;
|
||||
GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i);
|
||||
colDataSetVal(pDst, rows, (char*)&v, false);
|
||||
} else if (IS_BOOLEAN_TYPE(pDst->info.type)) {
|
||||
bool v = false;
|
||||
if (!IS_VAR_DATA_TYPE(pVar->nType)) {
|
||||
GET_TYPED_DATA(v, bool, pVar->nType, &pVar->i);
|
||||
} else {
|
||||
v = taosStr2Int8(varDataVal(pVar->pz), NULL, 10);
|
||||
}
|
||||
colDataSetVal(pDst, rows, (char*)&v, false);
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
|
@ -288,9 +306,9 @@ static void addCurrentRowToResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp*
|
|||
int32_t dstSlot = pExprInfo->base.resSchema.slotId;
|
||||
SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
|
||||
|
||||
if (IS_TIMESTAMP_TYPE(pExprInfo->base.resSchema.type)) {
|
||||
if (isIrowtsPseudoColumn(pExprInfo)) {
|
||||
colDataSetVal(pDst, pResBlock->info.rows, (char*)&pSliceInfo->current, false);
|
||||
} else if (IS_BOOLEAN_TYPE(pExprInfo->base.resSchema.type)) {
|
||||
} else if (isIsfilledPseudoColumn(pExprInfo)) {
|
||||
bool isFilled = false;
|
||||
colDataSetVal(pDst, pResBlock->info.rows, (char*)&isFilled, false);
|
||||
} else {
|
||||
|
|
@ -643,6 +661,9 @@ void destroyTimeSliceOperatorInfo(void* param) {
|
|||
taosArrayDestroy(pInfo->pLinearInfo);
|
||||
cleanupExprSupp(&pInfo->scalarSup);
|
||||
|
||||
for (int32_t i = 0; i < pInfo->pFillColInfo->numOfFillExpr; ++i) {
|
||||
taosVariantDestroy(&pInfo->pFillColInfo[i].fillVal);
|
||||
}
|
||||
taosMemoryFree(pInfo->pFillColInfo);
|
||||
taosMemoryFreeClear(param);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1575,7 +1575,7 @@ static int32_t translateInterp(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
|
|||
|
||||
uint8_t nodeType = nodeType(nodesListGetNode(pFunc->pParameterList, 0));
|
||||
uint8_t paraType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type;
|
||||
if (!IS_NUMERIC_TYPE(paraType) || QUERY_NODE_VALUE == nodeType) {
|
||||
if ((!IS_NUMERIC_TYPE(paraType) && !IS_BOOLEAN_TYPE(paraType))|| QUERY_NODE_VALUE == nodeType) {
|
||||
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1400,8 +1400,9 @@ void udfcUvHandleError(SClientUvConn *conn) {
|
|||
QUEUE_REMOVE(&task->procTaskQueue);
|
||||
uv_sem_post(&task->taskSem);
|
||||
}
|
||||
|
||||
uv_close((uv_handle_t *)conn->pipe, onUdfcPipeClose);
|
||||
if (!uv_is_closing((uv_handle_t *)conn->pipe)) {
|
||||
uv_close((uv_handle_t *)conn->pipe, onUdfcPipeClose);
|
||||
}
|
||||
}
|
||||
|
||||
void onUdfcPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
|
||||
|
|
@ -1553,7 +1554,9 @@ int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
|
|||
} else {
|
||||
SClientUvConn *conn = pipe->data;
|
||||
QUEUE_INSERT_TAIL(&conn->taskQueue, &uvTask->connTaskQueue);
|
||||
uv_close((uv_handle_t *)uvTask->pipe, onUdfcPipeClose);
|
||||
if (!uv_is_closing((uv_handle_t *)uvTask->pipe)) {
|
||||
uv_close((uv_handle_t *)uvTask->pipe, onUdfcPipeClose);
|
||||
}
|
||||
code = 0;
|
||||
}
|
||||
break;
|
||||
|
|
|
|||
|
|
@ -53,17 +53,9 @@ int32_t udfdCPluginOpen(SScriptUdfEnvItem *items, int numItems) { return 0; }
|
|||
|
||||
int32_t udfdCPluginClose() { return 0; }
|
||||
|
||||
int32_t udfdCPluginUdfInit(SScriptUdfInfo *udf, void **pUdfCtx) {
|
||||
int32_t err = 0;
|
||||
SUdfCPluginCtx *udfCtx = taosMemoryCalloc(1, sizeof(SUdfCPluginCtx));
|
||||
err = uv_dlopen(udf->path, &udfCtx->lib);
|
||||
if (err != 0) {
|
||||
fnError("can not load library %s. error: %s", udf->path, uv_strerror(err));
|
||||
return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
|
||||
}
|
||||
const char *udfName = udf->name;
|
||||
char initFuncName[TSDB_FUNC_NAME_LEN + 5] = {0};
|
||||
char *initSuffix = "_init";
|
||||
const char *udfdCPluginUdfInitLoadInitDestoryFuncs(SUdfCPluginCtx *udfCtx, const char *udfName) {
|
||||
char initFuncName[TSDB_FUNC_NAME_LEN + 5] = {0};
|
||||
char *initSuffix = "_init";
|
||||
strcpy(initFuncName, udfName);
|
||||
strncat(initFuncName, initSuffix, strlen(initSuffix));
|
||||
uv_dlsym(&udfCtx->lib, initFuncName, (void **)(&udfCtx->initFunc));
|
||||
|
|
@ -73,34 +65,55 @@ int32_t udfdCPluginUdfInit(SScriptUdfInfo *udf, void **pUdfCtx) {
|
|||
strcpy(destroyFuncName, udfName);
|
||||
strncat(destroyFuncName, destroySuffix, strlen(destroySuffix));
|
||||
uv_dlsym(&udfCtx->lib, destroyFuncName, (void **)(&udfCtx->destroyFunc));
|
||||
return udfName;
|
||||
}
|
||||
|
||||
void udfdCPluginUdfInitLoadAggFuncs(SUdfCPluginCtx *udfCtx, const char *udfName) {
|
||||
char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
|
||||
strcpy(processFuncName, udfName);
|
||||
uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->aggProcFunc));
|
||||
|
||||
char startFuncName[TSDB_FUNC_NAME_LEN + 6] = {0};
|
||||
char *startSuffix = "_start";
|
||||
strncpy(startFuncName, processFuncName, sizeof(startFuncName));
|
||||
strncat(startFuncName, startSuffix, strlen(startSuffix));
|
||||
uv_dlsym(&udfCtx->lib, startFuncName, (void **)(&udfCtx->aggStartFunc));
|
||||
|
||||
char finishFuncName[TSDB_FUNC_NAME_LEN + 7] = {0};
|
||||
char *finishSuffix = "_finish";
|
||||
strncpy(finishFuncName, processFuncName, sizeof(finishFuncName));
|
||||
strncat(finishFuncName, finishSuffix, strlen(finishSuffix));
|
||||
uv_dlsym(&udfCtx->lib, finishFuncName, (void **)(&udfCtx->aggFinishFunc));
|
||||
|
||||
char mergeFuncName[TSDB_FUNC_NAME_LEN + 6] = {0};
|
||||
char *mergeSuffix = "_merge";
|
||||
strncpy(mergeFuncName, processFuncName, sizeof(mergeFuncName));
|
||||
strncat(mergeFuncName, mergeSuffix, strlen(mergeSuffix));
|
||||
uv_dlsym(&udfCtx->lib, mergeFuncName, (void **)(&udfCtx->aggMergeFunc));
|
||||
}
|
||||
|
||||
int32_t udfdCPluginUdfInit(SScriptUdfInfo *udf, void **pUdfCtx) {
|
||||
int32_t err = 0;
|
||||
SUdfCPluginCtx *udfCtx = taosMemoryCalloc(1, sizeof(SUdfCPluginCtx));
|
||||
err = uv_dlopen(udf->path, &udfCtx->lib);
|
||||
if (err != 0) {
|
||||
fnError("can not load library %s. error: %s", udf->path, uv_strerror(err));
|
||||
return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
|
||||
}
|
||||
const char *udfName = udf->name;
|
||||
|
||||
udfdCPluginUdfInitLoadInitDestoryFuncs(udfCtx, udfName);
|
||||
|
||||
if (udf->funcType == UDF_FUNC_TYPE_SCALAR) {
|
||||
char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
|
||||
strcpy(processFuncName, udfName);
|
||||
uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->scalarProcFunc));
|
||||
} else if (udf->funcType == UDF_FUNC_TYPE_AGG) {
|
||||
char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
|
||||
strcpy(processFuncName, udfName);
|
||||
uv_dlsym(&udfCtx->lib, processFuncName, (void **)(&udfCtx->aggProcFunc));
|
||||
char startFuncName[TSDB_FUNC_NAME_LEN + 6] = {0};
|
||||
char *startSuffix = "_start";
|
||||
strncpy(startFuncName, processFuncName, sizeof(startFuncName));
|
||||
strncat(startFuncName, startSuffix, strlen(startSuffix));
|
||||
uv_dlsym(&udfCtx->lib, startFuncName, (void **)(&udfCtx->aggStartFunc));
|
||||
char finishFuncName[TSDB_FUNC_NAME_LEN + 7] = {0};
|
||||
char *finishSuffix = "_finish";
|
||||
strncpy(finishFuncName, processFuncName, sizeof(finishFuncName));
|
||||
strncat(finishFuncName, finishSuffix, strlen(finishSuffix));
|
||||
uv_dlsym(&udfCtx->lib, finishFuncName, (void **)(&udfCtx->aggFinishFunc));
|
||||
char mergeFuncName[TSDB_FUNC_NAME_LEN + 6] = {0};
|
||||
char *mergeSuffix = "_merge";
|
||||
strncpy(mergeFuncName, processFuncName, sizeof(mergeFuncName));
|
||||
strncat(mergeFuncName, mergeSuffix, strlen(mergeSuffix));
|
||||
uv_dlsym(&udfCtx->lib, mergeFuncName, (void **)(&udfCtx->aggMergeFunc));
|
||||
udfdCPluginUdfInitLoadAggFuncs(udfCtx, udfName);
|
||||
}
|
||||
|
||||
int32_t code = 0;
|
||||
if (udfCtx->initFunc) {
|
||||
// TODO: handle init call return error
|
||||
code = (udfCtx->initFunc)();
|
||||
if (code != 0) {
|
||||
uv_dlclose(&udfCtx->lib);
|
||||
|
|
@ -216,6 +229,7 @@ typedef struct SUdfdContext {
|
|||
|
||||
SArray *residentFuncs;
|
||||
|
||||
char udfDataDir[PATH_MAX];
|
||||
bool printVersion;
|
||||
} SUdfdContext;
|
||||
|
||||
|
|
@ -245,7 +259,9 @@ typedef struct SUvUdfWork {
|
|||
typedef enum { UDF_STATE_INIT = 0, UDF_STATE_LOADING, UDF_STATE_READY } EUdfState;
|
||||
|
||||
typedef struct SUdf {
|
||||
char name[TSDB_FUNC_NAME_LEN + 1];
|
||||
char name[TSDB_FUNC_NAME_LEN + 1];
|
||||
int32_t version;
|
||||
int64_t createdTime;
|
||||
|
||||
int8_t funcType;
|
||||
int8_t scriptType;
|
||||
|
|
@ -263,9 +279,11 @@ typedef struct SUdf {
|
|||
|
||||
SUdfScriptPlugin *scriptPlugin;
|
||||
void *scriptUdfCtx;
|
||||
|
||||
int64_t lastFetchTime; // last fetch time in milliseconds
|
||||
bool expired;
|
||||
} SUdf;
|
||||
|
||||
// TODO: add private udf structure.
|
||||
typedef struct SUdfcFuncHandle {
|
||||
SUdf *udf;
|
||||
} SUdfcFuncHandle;
|
||||
|
|
@ -318,6 +336,9 @@ static void udfdCloseWalkCb(uv_handle_t *handle, void *arg);
|
|||
static int32_t udfdRun();
|
||||
static void udfdConnectMnodeThreadFunc(void *args);
|
||||
|
||||
SUdf *udfdNewUdf(const char *udfName);
|
||||
void udfdGetFuncBodyPath(const SUdf *udf, char *path);
|
||||
|
||||
void udfdInitializeCPlugin(SUdfScriptPlugin *plugin) {
|
||||
plugin->scriptType = TSDB_FUNC_SCRIPT_BIN_LIB;
|
||||
plugin->openFunc = udfdCPluginOpen;
|
||||
|
|
@ -370,12 +391,13 @@ int32_t udfdInitializePythonPlugin(SUdfScriptPlugin *plugin) {
|
|||
}
|
||||
|
||||
if (plugin->openFunc) {
|
||||
int16_t lenPythonPath = strlen(tsUdfdLdLibPath) + strlen(tsTempDir) + 1 + 1; // tsTempDir:tsUdfdLdLibPath
|
||||
char *pythonPath = taosMemoryMalloc(lenPythonPath);
|
||||
int16_t lenPythonPath =
|
||||
strlen(tsUdfdLdLibPath) + strlen(global.udfDataDir) + 1 + 1; // global.udfDataDir:tsUdfdLdLibPath
|
||||
char *pythonPath = taosMemoryMalloc(lenPythonPath);
|
||||
#ifdef WINDOWS
|
||||
snprintf(pythonPath, lenPythonPath, "%s;%s", tsTempDir, tsUdfdLdLibPath);
|
||||
snprintf(pythonPath, lenPythonPath, "%s;%s", global.udfDataDir, tsUdfdLdLibPath);
|
||||
#else
|
||||
snprintf(pythonPath, lenPythonPath, "%s:%s", tsTempDir, tsUdfdLdLibPath);
|
||||
snprintf(pythonPath, lenPythonPath, "%s:%s", global.udfDataDir, tsUdfdLdLibPath);
|
||||
#endif
|
||||
SScriptUdfEnvItem items[] = {{"PYTHONPATH", pythonPath}, {"LOGDIR", tsLogDir}};
|
||||
err = plugin->openFunc(items, 2);
|
||||
|
|
@ -500,28 +522,14 @@ void convertUdf2UdfInfo(SUdf *udf, SScriptUdfInfo *udfInfo) {
|
|||
udfInfo->funcType = UDF_FUNC_TYPE_SCALAR;
|
||||
}
|
||||
udfInfo->name = udf->name;
|
||||
udfInfo->version = udf->version;
|
||||
udfInfo->createdTime = udf->createdTime;
|
||||
udfInfo->outputLen = udf->outputLen;
|
||||
udfInfo->outputType = udf->outputType;
|
||||
udfInfo->path = udf->path;
|
||||
udfInfo->scriptType = udf->scriptType;
|
||||
}
|
||||
|
||||
int32_t udfdRenameUdfFile(SUdf *udf) {
|
||||
char newPath[PATH_MAX];
|
||||
if (udf->scriptType == TSDB_FUNC_SCRIPT_BIN_LIB) {
|
||||
snprintf(newPath, PATH_MAX, "%s/lib%s.so", tsTempDir, udf->name);
|
||||
} else if (udf->scriptType == TSDB_FUNC_SCRIPT_PYTHON) {
|
||||
snprintf(newPath, PATH_MAX, "%s/%s.py", tsTempDir, udf->name);
|
||||
} else {
|
||||
return TSDB_CODE_UDF_SCRIPT_NOT_SUPPORTED;
|
||||
}
|
||||
int32_t code = taosRenameFile(udf->path, newPath);
|
||||
if (code == 0) {
|
||||
sprintf(udf->path, "%s", newPath);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
int32_t udfdInitUdf(char *udfName, SUdf *udf) {
|
||||
int32_t err = 0;
|
||||
err = udfdFillUdfInfoFromMNode(global.clientRpc, udfName, udf);
|
||||
|
|
@ -546,8 +554,6 @@ int32_t udfdInitUdf(char *udfName, SUdf *udf) {
|
|||
uv_mutex_unlock(&global.scriptPluginsMutex);
|
||||
udf->scriptPlugin = global.scriptPlugins[udf->scriptType];
|
||||
|
||||
udfdRenameUdfFile(udf);
|
||||
|
||||
SScriptUdfInfo info = {0};
|
||||
convertUdf2UdfInfo(udf, &info);
|
||||
err = udf->scriptPlugin->udfInitFunc(&info, &udf->scriptUdfCtx);
|
||||
|
|
@ -556,40 +562,60 @@ int32_t udfdInitUdf(char *udfName, SUdf *udf) {
|
|||
return err;
|
||||
}
|
||||
|
||||
fnInfo("udf init succeeded. name %s type %d context %p", udf->name, udf->scriptType, (void*)udf->scriptUdfCtx);
|
||||
fnInfo("udf init succeeded. name %s type %d context %p", udf->name, udf->scriptType, (void *)udf->scriptUdfCtx);
|
||||
return 0;
|
||||
}
|
||||
|
||||
SUdf *udfdGetOrCreateUdf(const char *udfName) {
|
||||
SUdf *udf = NULL;
|
||||
uv_mutex_lock(&global.udfsMutex);
|
||||
SUdf **udfInHash = taosHashGet(global.udfsHash, udfName, strlen(udfName));
|
||||
if (udfInHash) {
|
||||
++(*udfInHash)->refCount;
|
||||
udf = *udfInHash;
|
||||
uv_mutex_unlock(&global.udfsMutex);
|
||||
} else {
|
||||
SUdf *udfNew = taosMemoryCalloc(1, sizeof(SUdf));
|
||||
udfNew->refCount = 1;
|
||||
strncpy(udfNew->name, udfName, TSDB_FUNC_NAME_LEN);
|
||||
SUdf *udfdNewUdf(const char *udfName) {
|
||||
SUdf *udfNew = taosMemoryCalloc(1, sizeof(SUdf));
|
||||
udfNew->refCount = 1;
|
||||
udfNew->lastFetchTime = taosGetTimestampMs();
|
||||
strncpy(udfNew->name, udfName, TSDB_FUNC_NAME_LEN);
|
||||
|
||||
udfNew->state = UDF_STATE_INIT;
|
||||
uv_mutex_init(&udfNew->lock);
|
||||
uv_cond_init(&udfNew->condReady);
|
||||
udfNew->state = UDF_STATE_INIT;
|
||||
uv_mutex_init(&udfNew->lock);
|
||||
uv_cond_init(&udfNew->condReady);
|
||||
|
||||
udf = udfNew;
|
||||
udf->resident = false;
|
||||
for (int32_t i = 0; i < taosArrayGetSize(global.residentFuncs); ++i) {
|
||||
char *funcName = taosArrayGet(global.residentFuncs, i);
|
||||
if (strcmp(udfName, funcName) == 0) {
|
||||
udf->resident = true;
|
||||
break;
|
||||
}
|
||||
udfNew->resident = false;
|
||||
udfNew->expired = false;
|
||||
for (int32_t i = 0; i < taosArrayGetSize(global.residentFuncs); ++i) {
|
||||
char *funcName = taosArrayGet(global.residentFuncs, i);
|
||||
if (strcmp(udfName, funcName) == 0) {
|
||||
udfNew->resident = true;
|
||||
break;
|
||||
}
|
||||
SUdf **pUdf = &udf;
|
||||
taosHashPut(global.udfsHash, udfName, strlen(udfName), pUdf, POINTER_BYTES);
|
||||
uv_mutex_unlock(&global.udfsMutex);
|
||||
}
|
||||
return udfNew;
|
||||
}
|
||||
|
||||
SUdf *udfdGetOrCreateUdf(const char *udfName) {
|
||||
uv_mutex_lock(&global.udfsMutex);
|
||||
SUdf **pUdfHash = taosHashGet(global.udfsHash, udfName, strlen(udfName));
|
||||
int64_t currTime = taosGetTimestampSec();
|
||||
bool expired = false;
|
||||
if (pUdfHash) {
|
||||
expired = currTime - (*pUdfHash)->lastFetchTime > 10 * 1000; // 10s
|
||||
if (!expired) {
|
||||
++(*pUdfHash)->refCount;
|
||||
SUdf *udf = *pUdfHash;
|
||||
uv_mutex_unlock(&global.udfsMutex);
|
||||
fnInfo("udfd reuse existing udf. udf %s udf version %d, udf created time %" PRIx64, udf->name, udf->version,
|
||||
udf->createdTime);
|
||||
return udf;
|
||||
} else {
|
||||
(*pUdfHash)->expired = true;
|
||||
taosHashRemove(global.udfsHash, udfName, strlen(udfName));
|
||||
fnInfo("udfd expired, check for new version. existing udf %s udf version %d, udf created time %" PRIx64,
|
||||
(*pUdfHash)->name, (*pUdfHash)->version, (*pUdfHash)->createdTime);
|
||||
}
|
||||
}
|
||||
|
||||
SUdf *udf = udfdNewUdf(udfName);
|
||||
|
||||
SUdf **pUdf = &udf;
|
||||
taosHashPut(global.udfsHash, udfName, strlen(udfName), pUdf, POINTER_BYTES);
|
||||
uv_mutex_unlock(&global.udfsMutex);
|
||||
|
||||
return udf;
|
||||
}
|
||||
|
||||
|
|
@ -760,13 +786,13 @@ void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
|
|||
|
||||
uv_mutex_lock(&global.udfsMutex);
|
||||
udf->refCount--;
|
||||
if (udf->refCount == 0 && !udf->resident) {
|
||||
if (udf->refCount == 0 && (!udf->resident || udf->expired)) {
|
||||
unloadUdf = true;
|
||||
taosHashRemove(global.udfsHash, udf->name, strlen(udf->name));
|
||||
}
|
||||
uv_mutex_unlock(&global.udfsMutex);
|
||||
if (unloadUdf) {
|
||||
fnInfo("udf teardown. udf name: %s type %d: context %p", udf->name, udf->scriptType, (void*)(udf->scriptUdfCtx));
|
||||
fnInfo("udf teardown. udf name: %s type %d: context %p", udf->name, udf->scriptType, (void *)(udf->scriptUdfCtx));
|
||||
uv_cond_destroy(&udf->condReady);
|
||||
uv_mutex_destroy(&udf->lock);
|
||||
code = udf->scriptPlugin->udfDestroyFunc(udf->scriptUdfCtx);
|
||||
|
|
@ -791,6 +817,61 @@ void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
|
|||
return;
|
||||
}
|
||||
|
||||
void udfdGetFuncBodyPath(const SUdf *udf, char *path) {
|
||||
if (udf->scriptType == TSDB_FUNC_SCRIPT_BIN_LIB) {
|
||||
#ifdef WINDOWS
|
||||
snprintf(path, PATH_MAX, "%s%s_%d_%" PRIx64 ".dll", global.udfDataDir, udf->name, udf->version, udf->createdTime);
|
||||
#else
|
||||
snprintf(path, PATH_MAX, "%s/lib%s_%d_%" PRIx64 ".so", global.udfDataDir, udf->name, udf->version,
|
||||
udf->createdTime);
|
||||
#endif
|
||||
} else if (udf->scriptType == TSDB_FUNC_SCRIPT_PYTHON) {
|
||||
#ifdef WINDOWS
|
||||
snprintf(path, PATH_MAX, "%s%s_%d_%" PRIx64 ".py", global.udfDataDir, udf->name, udf->version, udf->createdTime);
|
||||
#else
|
||||
snprintf(path, PATH_MAX, "%s/%s_%d_%" PRIx64 ".py", global.udfDataDir, udf->name, udf->version, udf->createdTime);
|
||||
#endif
|
||||
} else {
|
||||
#ifdef WINDOWS
|
||||
snprintf(path, PATH_MAX, "%s%s_%d_%" PRIx64, global.udfDataDir, udf->name, udf->version, udf->createdTime);
|
||||
#else
|
||||
snprintf(path, PATH_MAX, "%s/lib%s_%d_%" PRIx64, global.udfDataDir, udf->name, udf->version, udf->createdTime);
|
||||
#endif
|
||||
}
|
||||
}
|
||||
|
||||
int32_t udfdSaveFuncBodyToFile(SFuncInfo *pFuncInfo, SUdf *udf) {
|
||||
if (!osDataSpaceAvailable()) {
|
||||
terrno = TSDB_CODE_NO_AVAIL_DISK;
|
||||
fnError("udfd create shared library failed since %s", terrstr(terrno));
|
||||
return terrno;
|
||||
}
|
||||
|
||||
char path[PATH_MAX] = {0};
|
||||
udfdGetFuncBodyPath(udf, path);
|
||||
bool fileExist = !(taosStatFile(path, NULL, NULL) < 0);
|
||||
if (fileExist) {
|
||||
strncpy(udf->path, path, PATH_MAX);
|
||||
fnInfo("udfd func body file. reuse existing file %s", path);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
|
||||
if (file == NULL) {
|
||||
fnError("udfd write udf shared library: %s failed, error: %d %s", path, errno, strerror(errno));
|
||||
return TSDB_CODE_FILE_CORRUPTED;
|
||||
}
|
||||
int64_t count = taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
|
||||
if (count != pFuncInfo->codeSize) {
|
||||
fnError("udfd write udf shared library failed");
|
||||
return TSDB_CODE_FILE_CORRUPTED;
|
||||
}
|
||||
taosCloseFile(&file);
|
||||
|
||||
strncpy(udf->path, path, PATH_MAX);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
|
||||
SUdfdRpcSendRecvInfo *msgInfo = (SUdfdRpcSendRecvInfo *)pMsg->info.ahandle;
|
||||
|
||||
|
|
@ -829,49 +910,25 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
|
|||
} else if (msgInfo->rpcType == UDFD_RPC_RETRIVE_FUNC) {
|
||||
SRetrieveFuncRsp retrieveRsp = {0};
|
||||
tDeserializeSRetrieveFuncRsp(pMsg->pCont, pMsg->contLen, &retrieveRsp);
|
||||
if (retrieveRsp.pFuncInfos == NULL) {
|
||||
goto _return;
|
||||
}
|
||||
|
||||
SFuncInfo *pFuncInfo = (SFuncInfo *)taosArrayGet(retrieveRsp.pFuncInfos, 0);
|
||||
// SUdf *udf = msgInfo->param;
|
||||
SUdf *udf = msgInfo->param;
|
||||
SUdf *udf = msgInfo->param;
|
||||
udf->funcType = pFuncInfo->funcType;
|
||||
udf->scriptType = pFuncInfo->scriptType;
|
||||
udf->outputType = pFuncInfo->outputType;
|
||||
udf->outputLen = pFuncInfo->outputLen;
|
||||
udf->bufSize = pFuncInfo->bufSize;
|
||||
|
||||
if (!osTempSpaceAvailable()) {
|
||||
terrno = TSDB_CODE_NO_AVAIL_DISK;
|
||||
msgInfo->code = terrno;
|
||||
fnError("udfd create shared library failed since %s", terrstr(terrno));
|
||||
goto _return;
|
||||
SFuncExtraInfo *pFuncExtraInfo = (SFuncExtraInfo *)taosArrayGet(retrieveRsp.pFuncExtraInfos, 0);
|
||||
udf->version = pFuncExtraInfo->funcVersion;
|
||||
udf->createdTime = pFuncExtraInfo->funcCreatedTime;
|
||||
msgInfo->code = udfdSaveFuncBodyToFile(pFuncInfo, udf);
|
||||
if (msgInfo->code != 0) {
|
||||
udf->lastFetchTime = 0;
|
||||
}
|
||||
|
||||
char path[PATH_MAX] = {0};
|
||||
#ifdef WINDOWS
|
||||
snprintf(path, sizeof(path), "%s%s", tsTempDir, pFuncInfo->name);
|
||||
#else
|
||||
snprintf(path, sizeof(path), "%s/%s", tsTempDir, pFuncInfo->name);
|
||||
#endif
|
||||
TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
|
||||
if (file == NULL) {
|
||||
fnError("udfd write udf shared library: %s failed, error: %d %s", path, errno, strerror(errno));
|
||||
msgInfo->code = TSDB_CODE_FILE_CORRUPTED;
|
||||
goto _return;
|
||||
}
|
||||
|
||||
int64_t count = taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
|
||||
if (count != pFuncInfo->codeSize) {
|
||||
fnError("udfd write udf shared library failed");
|
||||
msgInfo->code = TSDB_CODE_FILE_CORRUPTED;
|
||||
goto _return;
|
||||
}
|
||||
taosCloseFile(&file);
|
||||
strncpy(udf->path, path, PATH_MAX);
|
||||
tFreeSFuncInfo(pFuncInfo);
|
||||
taosArrayDestroy(retrieveRsp.pFuncInfos);
|
||||
msgInfo->code = 0;
|
||||
taosArrayDestroy(retrieveRsp.pFuncExtraInfos);
|
||||
}
|
||||
|
||||
_return:
|
||||
|
|
@ -1380,6 +1437,24 @@ int32_t udfdCleanup() {
|
|||
return 0;
|
||||
}
|
||||
|
||||
int32_t udfdCreateUdfSourceDir() {
|
||||
snprintf(global.udfDataDir, PATH_MAX, "%s/.udf", tsDataDir);
|
||||
int32_t code = taosMkDir(global.udfDataDir);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
snprintf(global.udfDataDir, PATH_MAX, "%s/.udf", tsTempDir);
|
||||
code = taosMkDir(global.udfDataDir);
|
||||
}
|
||||
fnInfo("udfd create udf source directory %s. result: %s", global.udfDataDir, tstrerror(code));
|
||||
|
||||
return code;
|
||||
}
|
||||
|
||||
int32_t udfdDestroyUdfSourceDir() {
|
||||
fnInfo("destory udf source directory %s", global.udfDataDir);
|
||||
taosRemoveDir(global.udfDataDir);
|
||||
return 0;
|
||||
}
|
||||
|
||||
int main(int argc, char *argv[]) {
|
||||
if (!taosCheckSystemIsLittleEnd()) {
|
||||
printf("failed to start since on non-little-end machines\n");
|
||||
|
|
@ -1408,10 +1483,15 @@ int main(int argc, char *argv[]) {
|
|||
|
||||
initEpSetFromCfg(tsFirst, tsSecond, &global.mgmtEp);
|
||||
if (udfdOpenClientRpc() != 0) {
|
||||
fnError("open rpc connection to mnode failure");
|
||||
fnError("open rpc connection to mnode failed");
|
||||
return -3;
|
||||
}
|
||||
|
||||
if (udfdCreateUdfSourceDir() != 0) {
|
||||
fnError("create udf source directory failed");
|
||||
return -4;
|
||||
}
|
||||
|
||||
if (udfdUvInit() != 0) {
|
||||
fnError("uv init failure");
|
||||
return -5;
|
||||
|
|
@ -1425,6 +1505,7 @@ int main(int argc, char *argv[]) {
|
|||
udfdRun();
|
||||
|
||||
removeListeningPipe();
|
||||
udfdDestroyUdfSourceDir();
|
||||
udfdCloseClientRpc();
|
||||
|
||||
udfdDeinitResidentFuncs();
|
||||
|
|
|
|||
|
|
@ -587,7 +587,8 @@ int32_t queryProcessRetrieveFuncRsp(void *output, char *msg, int32_t msgSize) {
|
|||
|
||||
memcpy(output, funcInfo, sizeof(*funcInfo));
|
||||
taosArrayDestroy(out.pFuncInfos);
|
||||
|
||||
taosArrayDestroy(out.pFuncExtraInfos);
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -280,7 +280,37 @@ if $data20 != 8.000000000 then
|
|||
return -1
|
||||
endi
|
||||
|
||||
sql create or replace function bit_and as '/tmp/udf/libbitand.so' outputtype int
|
||||
sql select func_version from information_schema.ins_functions where name='bit_and'
|
||||
if $data00 != 1 then
|
||||
return -1
|
||||
endi
|
||||
sql select bit_and(f1, f2) from t2;
|
||||
print $rows , $data00 , $data10 , $data20 , $data30 , $data40 , $data50
|
||||
if $rows != 6 then
|
||||
return -1
|
||||
endi
|
||||
if $data00 != 0 then
|
||||
return -1
|
||||
endi
|
||||
if $data10 != 1 then
|
||||
return -1
|
||||
endi
|
||||
|
||||
if $data20 != NULL then
|
||||
return -1
|
||||
endi
|
||||
|
||||
if $data30 != NULL then
|
||||
return -1
|
||||
endi
|
||||
|
||||
if $data40 != 0 then
|
||||
return -1
|
||||
endi
|
||||
if $data50 != 1 then
|
||||
return -1
|
||||
endi
|
||||
#sql drop function bit_and;
|
||||
#sql show functions;
|
||||
#if $rows != 1 then
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ class TDTestCase:
|
|||
dbname = "db"
|
||||
tbname = "tb"
|
||||
tbname1 = "tb1"
|
||||
tbname2 = "tb2"
|
||||
stbname = "stb"
|
||||
ctbname1 = "ctb1"
|
||||
ctbname2 = "ctb2"
|
||||
|
|
@ -1948,6 +1949,7 @@ class TDTestCase:
|
|||
tdSql.checkData(59, 1, 60)
|
||||
tdSql.checkData(60, 1, 60) #
|
||||
|
||||
|
||||
tdLog.printNoPrefix("==========step11:test multi-interp cases")
|
||||
tdSql.query(f"select interp(c0),interp(c1),interp(c2),interp(c3) from {dbname}.{tbname} range('2020-02-09 00:00:05', '2020-02-13 00:00:05') every(1d) fill(null)")
|
||||
tdSql.checkRows(5)
|
||||
|
|
@ -2002,7 +2004,362 @@ class TDTestCase:
|
|||
for i in range (tdSql.queryCols):
|
||||
tdSql.checkData(0, i, 13)
|
||||
|
||||
tdLog.printNoPrefix("==========step12:test error cases")
|
||||
tdLog.printNoPrefix("==========step12:test interp with boolean type")
|
||||
tdSql.execute(
|
||||
f'''create table if not exists {dbname}.{tbname2}
|
||||
(ts timestamp, c0 bool)
|
||||
'''
|
||||
)
|
||||
|
||||
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:01', false)")
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:03', true)")
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:05', false)")
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:07', true)")
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:09', true)")
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:11', false)")
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:13', false)")
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:15', NULL)")
|
||||
tdSql.execute(f"insert into {dbname}.{tbname2} values ('2020-02-02 00:00:17', NULL)")
|
||||
|
||||
# test fill null
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(NULL)")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, None)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, None)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, None)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, None)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, None)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, None)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, None)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, None)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, None)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, None)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
# test fill prev
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(prev)")
|
||||
tdSql.checkRows(18)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:01.000')
|
||||
|
||||
tdSql.checkData(0, 2, False)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, True)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, False)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, True)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, True)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, False)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, False)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, None)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, None)
|
||||
tdSql.checkData(17, 2, None)
|
||||
|
||||
tdSql.checkData(17, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
# test fill next
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(next)")
|
||||
tdSql.checkRows(18)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, False)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, True)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, False)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, True)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, True)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, False)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, False)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, None)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, None)
|
||||
tdSql.checkData(17, 2, None)
|
||||
|
||||
tdSql.checkData(17, 0, '2020-02-02 00:00:17.000')
|
||||
|
||||
# test fill value
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(value, 0)")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, False)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, False)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, False)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, False)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, False)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, False)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, False)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, False)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, False)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, False)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(value, 1234)")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, True)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, True)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, True)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, True)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, True)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, True)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, True)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, True)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, True)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, True)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(value, false)")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, False)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, False)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, False)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, False)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, False)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, False)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, False)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, False)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, False)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, False)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(value, true)")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, True)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, True)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, True)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, True)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, True)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, True)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, True)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, True)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, True)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, True)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(value, '0')")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, False)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, False)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, False)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, False)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, False)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, False)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, False)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, False)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, False)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, False)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(value, '123')")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, True)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, True)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, True)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, True)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, True)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, True)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, True)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, True)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, True)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, True)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(value, 'abc')")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, False)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, False)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, False)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, False)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, False)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, False)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, False)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, False)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, False)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, False)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(value, NULL)")
|
||||
tdSql.checkRows(19)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:00.000')
|
||||
|
||||
tdSql.checkData(0, 2, False)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, False)
|
||||
tdSql.checkData(3, 2, True)
|
||||
tdSql.checkData(4, 2, False)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, False)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, False)
|
||||
tdSql.checkData(9, 2, True)
|
||||
tdSql.checkData(10, 2, False)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, False)
|
||||
tdSql.checkData(13, 2, False)
|
||||
tdSql.checkData(14, 2, False)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, False)
|
||||
tdSql.checkData(17, 2, None)
|
||||
tdSql.checkData(18, 2, False)
|
||||
|
||||
tdSql.checkData(18, 0, '2020-02-02 00:00:18.000')
|
||||
|
||||
# test fill linear
|
||||
tdSql.query(f"select _irowts,_isfilled,interp(c0) from {dbname}.{tbname2} range('2020-02-02 00:00:00', '2020-02-02 00:00:18') every(1s) fill(linear)")
|
||||
tdSql.checkRows(17)
|
||||
tdSql.checkCols(3)
|
||||
|
||||
tdSql.checkData(0, 0, '2020-02-02 00:00:01.000')
|
||||
|
||||
tdSql.checkData(0, 2, False)
|
||||
tdSql.checkData(1, 2, False)
|
||||
tdSql.checkData(2, 2, True)
|
||||
tdSql.checkData(3, 2, False)
|
||||
tdSql.checkData(4, 2, False)
|
||||
tdSql.checkData(5, 2, False)
|
||||
tdSql.checkData(6, 2, True)
|
||||
tdSql.checkData(7, 2, True)
|
||||
tdSql.checkData(8, 2, True)
|
||||
tdSql.checkData(9, 2, False)
|
||||
tdSql.checkData(10, 2, False)
|
||||
tdSql.checkData(11, 2, False)
|
||||
tdSql.checkData(12, 2, False)
|
||||
tdSql.checkData(13, 2, None)
|
||||
tdSql.checkData(14, 2, None)
|
||||
tdSql.checkData(15, 2, None)
|
||||
tdSql.checkData(16, 2, None)
|
||||
|
||||
tdSql.checkData(16, 0, '2020-02-02 00:00:17.000')
|
||||
|
||||
tdLog.printNoPrefix("==========step13:test error cases")
|
||||
|
||||
tdSql.error(f"select interp(c0) from {dbname}.{tbname}")
|
||||
tdSql.error(f"select interp(c0) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05')")
|
||||
|
|
@ -2013,7 +2370,7 @@ class TDTestCase:
|
|||
|
||||
# input can only be numerical types
|
||||
tdSql.error(f"select interp(ts) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
|
||||
tdSql.error(f"select interp(c6) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
|
||||
#tdSql.error(f"select interp(c6) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
|
||||
tdSql.error(f"select interp(c7) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
|
||||
tdSql.error(f"select interp(c8) from {dbname}.{tbname} range('2020-02-10 00:00:05', '2020-02-15 00:00:05') every(1d) fill(null)")
|
||||
|
||||
|
|
|
|||
|
|
@ -22,7 +22,7 @@ class TDTestCase:
|
|||
tdSql.execute("insert into db.ctb using db.stb tags(1) (ts, c1) values (now, 1)")
|
||||
|
||||
tdSql.query("select count(*) from information_schema.ins_columns")
|
||||
tdSql.checkData(0, 0, 274)
|
||||
tdSql.checkData(0, 0, 275)
|
||||
|
||||
tdSql.query("select * from information_schema.ins_columns where table_name = 'ntb'")
|
||||
tdSql.checkRows(14)
|
||||
|
|
|
|||
Loading…
Reference in New Issue