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

Merge branch '3.0' into feature/TD-14481-3.0

This commit is contained in:
Cary Xu 2022-05-17 15:36:40 +08:00
parent 9b7e7fd408 b69fc6609e
commit a8fb62a13d
47 changed files with 952 additions and 620 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -41,6 +41,7 @@ typedef enum EFunctionType {
FUNCTION_TYPE_SUM, FUNCTION_TYPE_SUM,
FUNCTION_TYPE_TWA, FUNCTION_TYPE_TWA,
FUNCTION_TYPE_HISTOGRAM, FUNCTION_TYPE_HISTOGRAM,
FUNCTION_TYPE_HYPERLOGLOG,
// nonstandard SQL function // nonstandard SQL function
FUNCTION_TYPE_BOTTOM = 500, FUNCTION_TYPE_BOTTOM = 500,

4
include/libs/function/tudf.h
View File

@ -87,6 +87,7 @@ typedef struct SUdfInterBuf {
} SUdfInterBuf; } SUdfInterBuf;
typedef void *UdfcFuncHandle; typedef void *UdfcFuncHandle;
//low level APIs
/** /**
* setup udf * setup udf
* @param udf, in * @param udf, in
@ -115,6 +116,9 @@ int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t
*/ */
int32_t doTeardownUdf(UdfcFuncHandle handle); int32_t doTeardownUdf(UdfcFuncHandle handle);
void freeUdfInterBuf(SUdfInterBuf *buf);
//high level APIs
bool udfAggGetEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool udfAggGetEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo); bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
int32_t udfAggProcess(struct SqlFunctionCtx *pCtx); int32_t udfAggProcess(struct SqlFunctionCtx *pCtx);

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

@ -325,6 +325,7 @@ typedef struct SQuery {
bool showRewrite; bool showRewrite;
int32_t placeholderNum; int32_t placeholderNum;
SArray* pPlaceholderValues; SArray* pPlaceholderValues;
SNode* pContainPlaceholderRoot;
} SQuery; } SQuery;
void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext); void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);

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

@ -62,7 +62,7 @@ int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t
void qDestroyStmtDataBlock(void* pBlock); void qDestroyStmtDataBlock(void* pBlock);
STableMeta* qGetTableMetaInDataBlock(void* pDataBlock); STableMeta* qGetTableMetaInDataBlock(void* pDataBlock);
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId); int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx);
int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery); int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery);
int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen); int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen);
int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx, int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx,
@ -77,8 +77,8 @@ int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char*
void* smlInitHandle(SQuery* pQuery); void* smlInitHandle(SQuery* pQuery);
void smlDestroyHandle(void* pHandle); void smlDestroyHandle(void* pHandle);
int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta,
STableMeta* pTableMeta, char* tableName, char* msgBuf, int16_t msgBufLen); char* tableName, char* msgBuf, int16_t msgBufLen);
int32_t smlBuildOutput(void* handle, SHashObj* pVgHash); int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
#ifdef __cplusplus #ifdef __cplusplus

4
include/libs/planner/planner.h
View File

@ -34,7 +34,6 @@ typedef struct SPlanContext {
bool showRewrite; bool showRewrite;
int8_t triggerType; int8_t triggerType;
int64_t watermark; int64_t watermark;
int32_t placeholderNum;
char* pMsg; char* pMsg;
int32_t msgLen; int32_t msgLen;
} SPlanContext; } SPlanContext;
@ -48,9 +47,6 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
// @pSource one execution location of this group of datasource subplans // @pSource one execution location of this group of datasource subplans
int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource); int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource);
int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId,
bool* pEmptyResult);
// Convert to subplan to string for the scheduler to send to the executor // Convert to subplan to string for the scheduler to send to the executor
int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen); int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen);
int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan); int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan);

3
include/os/os.h
View File

@ -58,9 +58,6 @@ extern "C" {
#else #else
#include <winsock.h> #include <winsock.h>
#endif #endif
#define __typeof(a) auto
#endif #endif
#include <errno.h> #include <errno.h>

2
include/os/osFile.h
View File

@ -66,7 +66,7 @@ int32_t taosUnLockFile(TdFilePtr pFile);
int32_t taosUmaskFile(int32_t maskVal); int32_t taosUmaskFile(int32_t maskVal);
int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime); int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime);
int32_t taosDevInoFile(const char *path, int64_t *stDev, int64_t *stIno); int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno);
int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime); int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime);
bool taosCheckExistFile(const char *pathname); bool taosCheckExistFile(const char *pathname);

8
include/os/osMath.h
View File

@ -25,9 +25,11 @@ extern "C" {
#define TSWAP(a, b) \ #define TSWAP(a, b) \
do { \ do { \
__typeof(a) __tmp = (a); \ char *__tmp = taosMemoryMalloc(sizeof(a)); \
(a) = (b); \ memcpy(__tmp, &(a), sizeof(a)); \
(b) = __tmp; \ memcpy(&(a), &(b), sizeof(a)); \
memcpy(&(b), __tmp, sizeof(a)); \
taosMemoryFree(__tmp); \
} while (0) } while (0)
#ifdef WINDOWS #ifdef WINDOWS

2
include/util/taoserror.h
View File

@ -644,6 +644,8 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_INVALID_TIMELINE_FUNC TAOS_DEF_ERROR_CODE(0, 0x2647) #define TSDB_CODE_PAR_INVALID_TIMELINE_FUNC TAOS_DEF_ERROR_CODE(0, 0x2647)
#define TSDB_CODE_PAR_INVALID_PASSWD TAOS_DEF_ERROR_CODE(0, 0x2648) #define TSDB_CODE_PAR_INVALID_PASSWD TAOS_DEF_ERROR_CODE(0, 0x2648)
#define TSDB_CODE_PAR_INVALID_ALTER_TABLE TAOS_DEF_ERROR_CODE(0, 0x2649) #define TSDB_CODE_PAR_INVALID_ALTER_TABLE TAOS_DEF_ERROR_CODE(0, 0x2649)
#define TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY TAOS_DEF_ERROR_CODE(0, 0x264A)
#define TSDB_CODE_PAR_INVALID_MODIFY_COL TAOS_DEF_ERROR_CODE(0, 0x264B)
//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
include/util/thash.h
View File

@ -40,6 +40,7 @@ typedef void (*_hash_free_fn_t)(void *);
*/ */
uint32_t MurmurHash3_32(const char *key, uint32_t len); uint32_t MurmurHash3_32(const char *key, uint32_t len);
uint64_t MurmurHash3_64(const char *key, uint32_t len);
/** /**
* *
* @param key * @param key

2
include/util/tjson.h
View File

@ -25,7 +25,7 @@ extern "C" {
#define tjsonGetNumberValue(pJson, pName, val, code) \ #define tjsonGetNumberValue(pJson, pName, val, code) \
do { \ do { \
uint64_t _tmp = 0; \ uint64_t _tmp = 0; \
code = tjsonGetUBigIntValue(pJson, pName, &_tmp); \ code = tjsonGetBigIntValue(pJson, pName, &_tmp); \
val = _tmp; \ val = _tmp; \
} while (0) } while (0)

6
source/client/src/clientImpl.c
View File

@ -233,8 +233,7 @@ int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArra
.pAstRoot = pQuery->pRoot, .pAstRoot = pQuery->pRoot,
.showRewrite = pQuery->showRewrite, .showRewrite = pQuery->showRewrite,
.pMsg = pRequest->msgBuf, .pMsg = pRequest->msgBuf,
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE, .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE};
.placeholderNum = pQuery->placeholderNum};
SEpSet mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp); SEpSet mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
SCatalog* pCatalog = NULL; SCatalog* pCatalog = NULL;
int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog); int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
@ -949,7 +948,8 @@ int32_t setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableR
// TODO handle the compressed case // TODO handle the compressed case
pResultInfo->totalRows += pResultInfo->numOfRows; pResultInfo->totalRows += pResultInfo->numOfRows;
return setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows, convertUcs4); return setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows,
convertUcs4);
} }
TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* details, int maxlen) { TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* details, int maxlen) {

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

@ -146,7 +146,8 @@ int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockH
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, bool autoCreateTbl, SHashObj* pVgHash, SHashObj* pBlockHash) { int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, bool autoCreateTbl,
SHashObj* pVgHash, SHashObj* pBlockHash) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbFName)); STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbFName));
@ -157,7 +158,6 @@ int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t stmtGetExecInfo(TAOS_STMT* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) { int32_t stmtGetExecInfo(TAOS_STMT* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
@ -324,8 +324,10 @@ int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks *pDataBlock, STab
SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp); SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
SVgroupInfo vgInfo = {0}; SVgroupInfo vgInfo = {0};
STMT_ERR_RET(catalogGetTableHashVgroup(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &vgInfo)); STMT_ERR_RET(catalogGetTableHashVgroup(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname,
STMT_ERR_RET(taosHashPut(pStmt->exec.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo))); &vgInfo));
STMT_ERR_RET(
taosHashPut(pStmt->exec.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo)));
STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, vgInfo.vgId)); STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, vgInfo.vgId));
@ -336,7 +338,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
pStmt->bInfo.needParse = true; pStmt->bInfo.needParse = true;
pStmt->bInfo.inExecCache = false; pStmt->bInfo.inExecCache = false;
STableDataBlocks *pBlockInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)); STableDataBlocks* pBlockInExec =
taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (pBlockInExec) { if (pBlockInExec) {
pStmt->bInfo.needParse = false; pStmt->bInfo.needParse = false;
pStmt->bInfo.inExecCache = true; pStmt->bInfo.inExecCache = true;
@ -371,7 +374,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
STableDataBlocks* pNewBlock = NULL; STableDataBlocks* pNewBlock = NULL;
STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, 0)); STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, 0));
if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock, POINTER_BYTES)) { if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
POINTER_BYTES)) {
STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
@ -381,10 +385,10 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
STMT_RET(stmtCleanBindInfo(pStmt)); STMT_RET(stmtCleanBindInfo(pStmt));
} }
STableMeta* pTableMeta = NULL; STableMeta* pTableMeta = NULL;
SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp); SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
int32_t code = catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta); int32_t code =
catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta);
if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) { if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
STMT_ERR_RET(stmtCleanBindInfo(pStmt)); STMT_ERR_RET(stmtCleanBindInfo(pStmt));
@ -408,7 +412,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
if (pStmt->bInfo.inExecCache) { if (pStmt->bInfo.inExecCache) {
SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid)); SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
if (NULL == pCache) { if (NULL == pCache) {
tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash", pStmt->bInfo.tbFName, uid, cacheUid); tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
pStmt->bInfo.tbFName, uid, cacheUid);
STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR); STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
} }
@ -437,7 +442,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
STableDataBlocks* pNewBlock = NULL; STableDataBlocks* pNewBlock = NULL;
STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, uid)); STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, uid));
if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock, POINTER_BYTES)) { if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
POINTER_BYTES)) {
STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
} }
@ -522,7 +528,8 @@ int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest)); STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
} }
STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen)); STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName); tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName);
STMT_ERR_RET(stmtGetFromCache(pStmt)); STMT_ERR_RET(stmtGetFromCache(pStmt));
@ -548,7 +555,8 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)); STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) { if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName); tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR); STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
@ -566,7 +574,8 @@ int32_t stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD** fiel
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
} }
STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)); STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) { if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName); tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR); STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
@ -583,7 +592,8 @@ int32_t stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD** fiel
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
} }
STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)); STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) { if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName); tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR); STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
@ -618,7 +628,7 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
} }
if (STMT_TYPE_QUERY == pStmt->sql.type) { if (STMT_TYPE_QUERY == pStmt->sql.type) {
STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx, pStmt->exec.pRequest->requestId)); STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx));
SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId, SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
.acctId = pStmt->taos->acctId, .acctId = pStmt->taos->acctId,
@ -637,7 +647,8 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery)); STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery));
if (pStmt->sql.pQuery->haveResultSet) { if (pStmt->sql.pQuery->haveResultSet) {
setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema, pStmt->sql.pQuery->numOfResCols); setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
pStmt->sql.pQuery->numOfResCols);
setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision); setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
} }
@ -653,7 +664,8 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)); STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) { if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName); tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR); STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
@ -733,7 +745,8 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp *pRsp) {
} }
if (i < pRsp->nBlocks) { if (i < pRsp->nBlocks) {
tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid, blkRsp->uid); tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
blkRsp->uid);
pMeta->uid = blkRsp->uid; pMeta->uid = blkRsp->uid;
pStmt->bInfo.tbUid = blkRsp->uid; pStmt->bInfo.tbUid = blkRsp->uid;
@ -760,7 +773,8 @@ int stmtExec(TAOS_STMT *stmt) {
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true, NULL); launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true, NULL);
} else { } else {
STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash)); STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash));
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true, (autoCreateTbl ? (void**)&pRsp : NULL)); launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true,
(autoCreateTbl ? (void**)&pRsp : NULL));
} }
if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) { if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {

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

@ -386,7 +386,7 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
req.schema.sver = pStb->version; req.schema.sver = pStb->version;
req.schema.pSchema = pStb->pColumns; req.schema.pSchema = pStb->pColumns;
req.schemaTag.nCols = pStb->numOfTags; req.schemaTag.nCols = pStb->numOfTags;
req.schemaTag.nCols = 0; req.schemaTag.sver = 1;
req.schemaTag.pSchema = pStb->pTags; req.schemaTag.pSchema = pStb->pTags;
if (req.rollup) { if (req.rollup) {

40
source/dnode/vnode/src/meta/metaTable.c
View File

@ -420,7 +420,8 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
// get table entry // get table entry
SDecoder dc = {0}; SDecoder dc = {0};
tDecoderInit(&dc, pData, nData); tDecoderInit(&dc, pData, nData);
metaDecodeEntry(&dc, &entry); ret = metaDecodeEntry(&dc, &entry);
ASSERT(ret == 0);
if (entry.type != TSDB_NORMAL_TABLE) { if (entry.type != TSDB_NORMAL_TABLE) {
terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION; terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION;
@ -468,11 +469,11 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
goto _err; goto _err;
} }
pSchema->sver++; pSchema->sver++;
pSchema->nCols--;
tlen = (pSchema->nCols - iCol - 1) * sizeof(SSchema); tlen = (pSchema->nCols - iCol - 1) * sizeof(SSchema);
if (tlen) { if (tlen) {
memmove(pColumn, pColumn + 1, tlen); memmove(pColumn, pColumn + 1, tlen);
} }
pSchema->nCols--;
break; break;
case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
if (pColumn == NULL) { if (pColumn == NULL) {
@ -598,10 +599,41 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
goto _err; goto _err;
} }
{ if (iCol == 0) {
// TODO: // TODO : need to update tag index
} }
ctbEntry.version = version;
SKVRowBuilder kvrb = {0};
const SKVRow pOldTag = (const SKVRow)ctbEntry.ctbEntry.pTags;
SKVRow pNewTag = NULL;
tdInitKVRowBuilder(&kvrb);
for (int32_t i = 0; i < pTagSchema->nCols; i++) {
SSchema *pCol = &pTagSchema->pSchema[i];
if (iCol == i) {
tdAddColToKVRow(&kvrb, pCol->colId, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal);
} else {
void *p = tdGetKVRowValOfCol(pOldTag, pCol->colId);
if (p) {
if (IS_VAR_DATA_TYPE(pCol->type)) {
tdAddColToKVRow(&kvrb, pCol->colId, p, varDataTLen(p));
} else {
tdAddColToKVRow(&kvrb, pCol->colId, p, pCol->bytes);
}
}
}
}
ctbEntry.ctbEntry.pTags = tdGetKVRowFromBuilder(&kvrb);
tdDestroyKVRowBuilder(&kvrb);
// save to table.db
metaSaveToTbDb(pMeta, &ctbEntry);
// save to uid.idx
tdbDbUpsert(pMeta->pUidIdx, &ctbEntry.uid, sizeof(tb_uid_t), &version, sizeof(version), &pMeta->txn);
if (ctbEntry.pBuf) taosMemoryFree(ctbEntry.pBuf); if (ctbEntry.pBuf) taosMemoryFree(ctbEntry.pBuf);
if (stbEntry.pBuf) tdbFree(stbEntry.pBuf); if (stbEntry.pBuf) tdbFree(stbEntry.pBuf);
tdbDbcClose(pTbDbc); tdbDbcClose(pTbDbc);

3
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -3055,7 +3055,8 @@ static bool loadDataBlockFromTableSeq(STsdbReadHandle* pTsdbReadHandle) {
bool tsdbNextDataBlock(tsdbReaderT pHandle) { bool tsdbNextDataBlock(tsdbReaderT pHandle) {
STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle; STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
for (int32_t i = 0; i < taosArrayGetSize(pTsdbReadHandle->pColumns); ++i) { size_t numOfCols = taosArrayGetSize(pTsdbReadHandle->pColumns);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i); SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity); colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity);
} }

2
source/libs/executor/inc/executil.h
View File

@ -139,7 +139,7 @@ typedef struct {
int32_t colId; int32_t colId;
} SStddevInterResult; } SStddevInterResult;
void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, bool sortGroupResult); void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order);
void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList); void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList);
void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo); void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo);

13
source/libs/executor/src/executil.c
View File

@ -184,7 +184,7 @@ void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
pGroupResInfo->index = 0; pGroupResInfo->index = 0;
} }
static int32_t resultrowCompar1(const void* p1, const void* p2) { static int32_t resultrowComparAsc(const void* p1, const void* p2) {
SResKeyPos* pp1 = *(SResKeyPos**) p1; SResKeyPos* pp1 = *(SResKeyPos**) p1;
SResKeyPos* pp2 = *(SResKeyPos**) p2; SResKeyPos* pp2 = *(SResKeyPos**) p2;
@ -202,7 +202,11 @@ static int32_t resultrowCompar1(const void* p1, const void* p2) {
} }
} }
void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, bool sortGroupResult) { static int32_t resultrowComparDesc(const void* p1, const void* p2) {
return resultrowComparAsc(p2, p1);
}
void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order) {
if (pGroupResInfo->pRows != NULL) { if (pGroupResInfo->pRows != NULL) {
taosArrayDestroy(pGroupResInfo->pRows); taosArrayDestroy(pGroupResInfo->pRows);
} }
@ -224,8 +228,9 @@ void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, boo
taosArrayPush(pGroupResInfo->pRows, &p); taosArrayPush(pGroupResInfo->pRows, &p);
} }
if (sortGroupResult) { if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {
qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, resultrowCompar1); __compar_fn_t fn = (order == TSDB_ORDER_ASC)? resultrowComparAsc:resultrowComparDesc;
qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, fn);
} }
pGroupResInfo->index = 0; pGroupResInfo->index = 0;

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

@ -3707,7 +3707,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
finalizeMultiTupleQueryResult(pOperator->numOfExprs, pAggInfo->aggSup.pResultBuf, &pAggInfo->binfo.resultRowInfo, finalizeMultiTupleQueryResult(pOperator->numOfExprs, pAggInfo->aggSup.pResultBuf, &pAggInfo->binfo.resultRowInfo,
pAggInfo->binfo.rowCellInfoOffset); pAggInfo->binfo.rowCellInfoOffset);
initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, false); initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, 0);
OPTR_SET_OPENED(pOperator); OPTR_SET_OPENED(pOperator);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

2
source/libs/executor/src/groupoperator.c
View File

@ -314,7 +314,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
// } // }
blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity); blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);
initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, false); initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, 0);
while(1) { while(1) {
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);

61
source/libs/executor/src/scanoperator.c
View File

@ -102,7 +102,7 @@ static void getNextTimeWindow(SInterval* pInterval, STimeWindow* tw, int32_t ord
tw->ekey -= 1; tw->ekey -= 1;
} }
static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo) { static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockInfo, int32_t order) {
STimeWindow w = {0}; STimeWindow w = {0};
// 0 by default, which means it is not a interval operator of the upstream operator. // 0 by default, which means it is not a interval operator of the upstream operator.
@ -110,13 +110,7 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
return false; return false;
} }
// todo handle the time range case if (order == TSDB_ORDER_ASC) {
TSKEY sk = INT64_MIN;
TSKEY ek = INT64_MAX;
// TSKEY sk = MIN(pQueryAttr->window.skey, pQueryAttr->window.ekey);
// TSKEY ek = MAX(pQueryAttr->window.skey, pQueryAttr->window.ekey);
if (true) {
getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey, &w); getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.skey, &w);
assert(w.ekey >= pBlockInfo->window.skey); assert(w.ekey >= pBlockInfo->window.skey);
@ -124,8 +118,8 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
return true; return true;
} }
while (1) { // todo handle the desc order scan case while (1) {
getNextTimeWindow(pInterval, &w, TSDB_ORDER_ASC); getNextTimeWindow(pInterval, &w, order);
if (w.skey > pBlockInfo->window.ekey) { if (w.skey > pBlockInfo->window.ekey) {
break; break;
} }
@ -136,24 +130,24 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
} }
} }
} else { } else {
// getAlignQueryTimeWindow(pQueryAttr, pBlockInfo->window.ekey, sk, ek, &w); getAlignQueryTimeWindow(pInterval, pInterval->precision, pBlockInfo->window.ekey, &w);
// assert(w.skey <= pBlockInfo->window.ekey); assert(w.skey <= pBlockInfo->window.ekey);
//
// if (w.skey > pBlockInfo->window.skey) { if (w.skey > pBlockInfo->window.skey) {
// return true; return true;
// } }
//
// while(1) { while(1) {
// getNextTimeWindow(pQueryAttr, &w); getNextTimeWindow(pInterval, &w, order);
// if (w.ekey < pBlockInfo->window.skey) { if (w.ekey < pBlockInfo->window.skey) {
// break; break;
// } }
//
// assert(w.skey < pBlockInfo->window.skey); assert(w.skey < pBlockInfo->window.skey);
// if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) { if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) {
// return true; return true;
// } }
// } }
} }
return false; return false;
@ -172,7 +166,8 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
pCost->totalRows += pBlock->info.rows; pCost->totalRows += pBlock->info.rows;
*status = pInfo->dataBlockLoadFlag; *status = pInfo->dataBlockLoadFlag;
if (pTableScanInfo->pFilterNode != NULL || overlapWithTimeWindow(&pTableScanInfo->interval, &pBlock->info)) { if (pTableScanInfo->pFilterNode != NULL ||
overlapWithTimeWindow(&pTableScanInfo->interval, &pBlock->info, pTableScanInfo->cond.order)) {
(*status) = FUNC_DATA_REQUIRED_DATA_LOAD; (*status) = FUNC_DATA_REQUIRED_DATA_LOAD;
} }
@ -188,6 +183,13 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo), qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%d", GET_TASKID(pTaskInfo),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
pCost->skipBlocks += 1; pCost->skipBlocks += 1;
// clear all data in pBlock that are set when handing the previous block
for(int32_t i = 0; i < pBlockInfo->numOfCols; ++i) {
SColumnInfoData* pcol = taosArrayGet(pBlock->pDataBlock, i);
pcol->pData = NULL;
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) { } else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) {
pCost->loadBlockStatis += 1; pCost->loadBlockStatis += 1;
@ -466,6 +468,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
} }
pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]}; pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};
// pInfo->scanInfo = (SScanInfo){.numOfAsc = 0, .numOfDesc = 1}; // for debug purpose
pInfo->readHandle = *readHandle; pInfo->readHandle = *readHandle;
pInfo->interval = extractIntervalInfo(pTableScanNode); pInfo->interval = extractIntervalInfo(pTableScanNode);

83
source/libs/executor/src/timewindowoperator.c
View File

@ -54,8 +54,8 @@ static TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols, int32_t rows,
if (tsCols == NULL) { if (tsCols == NULL) {
ts = ascQuery ? win->skey : win->ekey; ts = ascQuery ? win->skey : win->ekey;
} else { } else {
int32_t offset = ascQuery ? 0 : rows - 1; // int32_t offset = ascQuery ? 0 : rows - 1;
ts = tsCols[offset]; ts = tsCols[0];
} }
return ts; return ts;
@ -172,14 +172,22 @@ static FORCE_INLINE int32_t getForwardStepsInBlock(int32_t numOfRows, __block_se
} }
} }
} else { } else {
int32_t end = searchFn((char*)pData, pos + 1, ekey, order); int32_t end = searchFn((char*)&pData[pos], numOfRows - pos, ekey, order);
if (end >= 0) { if (end >= 0) {
forwardStep = pos - end; forwardStep = end;
if (pData[end] == ekey) { if (pData[end + pos] == ekey) {
forwardStep += 1; forwardStep += 1;
} }
} }
// int32_t end = searchFn((char*)pData, pos + 1, ekey, order);
// if (end >= 0) {
// forwardStep = pos - end;
//
// if (pData[end] == ekey) {
// forwardStep += 1;
// }
// }
} }
assert(forwardStep >= 0); assert(forwardStep >= 0);
@ -203,17 +211,25 @@ int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
if (order == TSDB_ORDER_DESC) { if (order == TSDB_ORDER_DESC) {
// find the first position which is smaller than the key // find the first position which is smaller than the key
while (1) { while (1) {
if (key >= keyList[lastPos]) return lastPos; if (key >= keyList[firstPos]) return firstPos;
if (key == keyList[firstPos]) return firstPos; if (key == keyList[lastPos]) return lastPos;
if (key < keyList[firstPos]) return firstPos - 1;
if (key < keyList[lastPos]) {
lastPos += 1;
if (lastPos >= num) {
return -1;
} else {
return lastPos;
}
}
numOfRows = lastPos - firstPos + 1; numOfRows = lastPos - firstPos + 1;
midPos = (numOfRows >> 1) + firstPos; midPos = (numOfRows >> 1) + firstPos;
if (key < keyList[midPos]) { if (key < keyList[midPos]) {
lastPos = midPos - 1;
} else if (key > keyList[midPos]) {
firstPos = midPos + 1; firstPos = midPos + 1;
} else if (key > keyList[midPos]) {
lastPos = midPos - 1;
} else { } else {
break; break;
} }
@ -273,12 +289,12 @@ int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimary
if (ekey > pDataBlockInfo->window.skey && pPrimaryColumn) { if (ekey > pDataBlockInfo->window.skey && pPrimaryColumn) {
num = getForwardStepsInBlock(pDataBlockInfo->rows, searchFn, ekey, startPos, order, pPrimaryColumn); num = getForwardStepsInBlock(pDataBlockInfo->rows, searchFn, ekey, startPos, order, pPrimaryColumn);
if (item != NULL) { if (item != NULL) {
item->lastKey = pPrimaryColumn[startPos - (num - 1)] + step; item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
} }
} else { } else {
num = startPos + 1; num = pDataBlockInfo->rows - startPos;
if (item != NULL) { if (item != NULL) {
item->lastKey = pDataBlockInfo->window.skey + step; item->lastKey = pDataBlockInfo->window.ekey + step;
} }
} }
} }
@ -470,20 +486,17 @@ static int32_t getNextQualifiedWindow(SInterval* pInterval, STimeWindow* pNext,
return -1; return -1;
} }
TSKEY startKey = ascQuery ? pNext->skey : pNext->ekey; TSKEY skey = ascQuery ? pNext->skey : pNext->ekey;
int32_t startPos = 0; int32_t startPos = 0;
// tumbling time window query, a special case of sliding time window query // tumbling time window query, a special case of sliding time window query
if (pInterval->sliding == pInterval->interval && prevPosition != -1) { if (pInterval->sliding == pInterval->interval && prevPosition != -1) {
int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order); startPos = prevPosition + 1;
startPos = prevPosition + factor;
} else { } else {
if (startKey <= pDataBlockInfo->window.skey && ascQuery) { if ((skey <= pDataBlockInfo->window.skey && ascQuery) || (skey >= pDataBlockInfo->window.ekey && !ascQuery)) {
startPos = 0; startPos = 0;
} else if (startKey >= pDataBlockInfo->window.ekey && !ascQuery) {
startPos = pDataBlockInfo->rows - 1;
} else { } else {
startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, startKey, order); startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, skey, order);
} }
} }
@ -608,7 +621,7 @@ static void saveDataBlockLastRow(char** pRow, SArray* pDataBlock, int32_t rowInd
} }
static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pSDataBlock, static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pSDataBlock,
int32_t tableGroupId) { uint64_t tableGroupId) {
SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info; SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info;
SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
@ -620,7 +633,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
} }
int32_t step = 1; int32_t step = 1;
bool ascScan = true; bool ascScan = (pInfo->order == TSDB_ORDER_ASC);
// int32_t prevIndex = pResultRowInfo->curPos; // int32_t prevIndex = pResultRowInfo->curPos;
@ -630,7 +643,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
tsCols = (int64_t*)pColDataInfo->pData; tsCols = (int64_t*)pColDataInfo->pData;
} }
int32_t startPos = ascScan ? 0 : (pSDataBlock->info.rows - 1); int32_t startPos = 0;
TSKEY ts = getStartTsKey(&pSDataBlock->info.window, tsCols, pSDataBlock->info.rows, ascScan); TSKEY ts = getStartTsKey(&pSDataBlock->info.window, tsCols, pSDataBlock->info.rows, ascScan);
STimeWindow win = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, STimeWindow win = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval,
@ -654,9 +667,10 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
} }
int32_t forwardStep = 0; int32_t forwardStep = 0;
TSKEY ekey = win.ekey; TSKEY ekey = ascScan? win.ekey:win.skey;
forwardStep = forwardStep =
getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC); getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order);
ASSERT(forwardStep > 0);
// prev time window not interpolation yet. // prev time window not interpolation yet.
// int32_t curIndex = pResultRowInfo->curPos; // int32_t curIndex = pResultRowInfo->curPos;
@ -731,9 +745,9 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
taosArrayPush(pUpdated, &pos); taosArrayPush(pUpdated, &pos);
} }
ekey = nextWin.ekey; // reviseWindowEkey(pQueryAttr, &nextWin); ekey = ascScan? nextWin.ekey:nextWin.skey;
forwardStep = forwardStep =
getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC); getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order);
// window start(end) key interpolation // window start(end) key interpolation
doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos, forwardStep, doWindowBorderInterpolation(pOperatorInfo, pSDataBlock, pInfo->binfo.pCtx, pResult, &nextWin, startPos, forwardStep,
@ -761,7 +775,8 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SIntervalAggOperatorInfo* pInfo = pOperator->info; SIntervalAggOperatorInfo* pInfo = pOperator->info;
int32_t order = TSDB_ORDER_ASC; int32_t scanFlag = MAIN_SCAN;
SOperatorInfo* downstream = pOperator->pDownstream[0]; SOperatorInfo* downstream = pOperator->pDownstream[0];
while (1) { while (1) {
@ -773,8 +788,10 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
break; break;
} }
getTableScanInfo(pOperator, &pInfo->order, &scanFlag);
// the pDataBlock are always the same one, no need to call this again // the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true); setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, scanFlag, true);
STableQueryInfo* pTableQueryInfo = pInfo->pCurrent; STableQueryInfo* pTableQueryInfo = pInfo->pCurrent;
setIntervalQueryRange(pTableQueryInfo, pBlock->info.window.skey, &pTaskInfo->window); setIntervalQueryRange(pTableQueryInfo, pBlock->info.window.skey, &pTaskInfo->window);
@ -800,7 +817,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pInfo->binfo.resultRowInfo, finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pInfo->binfo.resultRowInfo,
pInfo->binfo.rowCellInfoOffset); pInfo->binfo.rowCellInfoOffset);
initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true); initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->order);
OPTR_SET_OPENED(pOperator); OPTR_SET_OPENED(pOperator);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -945,7 +962,7 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo, finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo,
pBInfo->rowCellInfoOffset); pBInfo->rowCellInfoOffset);
initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true); initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity); blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) { if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
@ -1070,6 +1087,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
doClearWindows(pInfo, pOperator->numOfExprs, pBlock); doClearWindows(pInfo, pOperator->numOfExprs, pBlock);
continue; continue;
} }
pInfo->order = TSDB_ORDER_ASC;
pUpdated = hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0); pUpdated = hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0);
} }
@ -1119,7 +1137,6 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
pInfo->order = TSDB_ORDER_ASC; pInfo->order = TSDB_ORDER_ASC;
pInfo->interval = *pInterval; pInfo->interval = *pInterval;
// pInfo->execModel = OPTR_EXEC_MODEL_STREAM;
pInfo->execModel = pTaskInfo->execModel; pInfo->execModel = pTaskInfo->execModel;
pInfo->win = pTaskInfo->window; pInfo->win = pTaskInfo->window;
pInfo->twAggSup = *pTwAggSupp; pInfo->twAggSup = *pTwAggSupp;
@ -1338,7 +1355,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo, finalizeMultiTupleQueryResult(pOperator->numOfExprs, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo,
pBInfo->rowCellInfoOffset); pBInfo->rowCellInfoOffset);
initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, true); initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity); blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf); doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) { if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {

4
source/libs/function/inc/builtinsimpl.h
View File

@ -90,6 +90,10 @@ bool histogramFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultIn
int32_t histogramFunction(SqlFunctionCtx* pCtx); int32_t histogramFunction(SqlFunctionCtx* pCtx);
int32_t histogramFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock); int32_t histogramFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getHLLFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t hllFunction(SqlFunctionCtx* pCtx);
int32_t hllFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getStateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv); bool getStateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool stateFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo); bool stateFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t stateCountFunction(SqlFunctionCtx* pCtx); int32_t stateCountFunction(SqlFunctionCtx* pCtx);

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

@ -263,6 +263,21 @@ static int32_t translateHistogram(SFunctionNode* pFunc, char* pErrBuf, int32_t l
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t translateHLL(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
if (1 != LIST_LENGTH(pFunc->pParameterList)) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
}
SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
if (QUERY_NODE_COLUMN != nodeType(pPara)) {
return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
"The input parameter of HYPERLOGLOG function can only be column");
}
pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_UBIGINT].bytes, .type = TSDB_DATA_TYPE_UBIGINT};
return TSDB_CODE_SUCCESS;
}
static int32_t translateStateCount(SFunctionNode* pFunc, char* pErrBuf, int32_t len) { static int32_t translateStateCount(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
if (3 != LIST_LENGTH(pFunc->pParameterList)) { if (3 != LIST_LENGTH(pFunc->pParameterList)) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName); return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
@ -829,6 +844,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.processFunc = histogramFunction, .processFunc = histogramFunction,
.finalizeFunc = histogramFinalize .finalizeFunc = histogramFinalize
}, },
{
.name = "hyperloglog",
.type = FUNCTION_TYPE_HYPERLOGLOG,
.classification = FUNC_MGT_AGG_FUNC,
.translateFunc = translateHLL,
.getEnvFunc = getHLLFuncEnv,
.initFunc = functionSetup,
.processFunc = hllFunction,
.finalizeFunc = hllFinalize
},
{ {
.name = "state_count", .name = "state_count",
.type = FUNCTION_TYPE_STATE_COUNT, .type = FUNCTION_TYPE_STATE_COUNT,

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

@ -28,6 +28,12 @@
#define TAIL_MAX_POINTS_NUM 100 #define TAIL_MAX_POINTS_NUM 100
#define TAIL_MAX_OFFSET 100 #define TAIL_MAX_OFFSET 100
#define HLL_BUCKET_BITS 14 // The bits of the bucket
#define HLL_DATA_BITS (64-HLL_BUCKET_BITS)
#define HLL_BUCKETS (1<<HLL_BUCKET_BITS)
#define HLL_BUCKET_MASK (HLL_BUCKETS-1)
#define HLL_ALPHA_INF 0.721347520444481703680 // constant for 0.5/ln(2)
typedef struct SSumRes { typedef struct SSumRes {
union { union {
int64_t isum; int64_t isum;
@ -129,6 +135,11 @@ typedef enum {
LOG_BIN LOG_BIN
} EHistoBinType; } EHistoBinType;
typedef struct SHLLFuncInfo {
uint64_t result;
uint8_t buckets[HLL_BUCKETS];
} SHLLInfo;
typedef struct SStateInfo { typedef struct SStateInfo {
union { union {
int64_t count; int64_t count;
@ -2729,6 +2740,140 @@ int32_t histogramFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
return pResInfo->numOfRes; return pResInfo->numOfRes;
} }
bool getHLLFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(SHLLInfo);
return true;
}
static uint8_t hllCountNum(void* data, int32_t bytes, int32_t *buk) {
uint64_t hash = MurmurHash3_64(data, bytes);
int32_t index = hash & HLL_BUCKET_MASK;
hash >>= HLL_BUCKET_BITS;
hash |= ((uint64_t)1 << HLL_DATA_BITS);
uint64_t bit = 1;
uint8_t count = 1;
while((hash & bit) == 0) {
count++;
bit <<= 1;
}
*buk = index;
return count;
}
static void hllBucketHisto(uint8_t *buckets, int32_t* bucketHisto) {
uint64_t *word = (uint64_t*) buckets;
uint8_t *bytes;
for (int32_t j = 0; j < HLL_BUCKETS>>3; j++) {
if (*word == 0) {
bucketHisto[0] += 8;
} else {
bytes = (uint8_t*) word;
bucketHisto[bytes[0]]++;
bucketHisto[bytes[1]]++;
bucketHisto[bytes[2]]++;
bucketHisto[bytes[3]]++;
bucketHisto[bytes[4]]++;
bucketHisto[bytes[5]]++;
bucketHisto[bytes[6]]++;
bucketHisto[bytes[7]]++;
}
word++;
}
}
static double hllTau(double x) {
if (x == 0. || x == 1.) return 0.;
double zPrime;
double y = 1.0;
double z = 1 - x;
do {
x = sqrt(x);
zPrime = z;
y *= 0.5;
z -= pow(1 - x, 2)*y;
} while(zPrime != z);
return z / 3;
}
static double hllSigma(double x) {
if (x == 1.0) return INFINITY;
double zPrime;
double y = 1;
double z = x;
do {
x *= x;
zPrime = z;
z += x * y;
y += y;
} while(zPrime != z);
return z;
}
// estimate the cardinality, the algorithm refer this paper: "New cardinality estimation algorithms for HyperLogLog sketches"
static uint64_t hllCountCnt(uint8_t *buckets) {
double m = HLL_BUCKETS;
int32_t buckethisto[64] = {0};
hllBucketHisto(buckets,buckethisto);
double z = m * hllTau((m-buckethisto[HLL_DATA_BITS+1])/(double)m);
for (int j = HLL_DATA_BITS; j >= 1; --j) {
z += buckethisto[j];
z *= 0.5;
}
z += m * hllSigma(buckethisto[0]/(double)m);
double E = (double)llroundl(HLL_ALPHA_INF*m*m/z);
return (uint64_t) E;
}
int32_t hllFunction(SqlFunctionCtx *pCtx) {
SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
SInputColumnInfoData* pInput = &pCtx->input;
SColumnInfoData* pCol = pInput->pData[0];
int32_t type = pCol->info.type;
int32_t bytes = pCol->info.bytes;
int32_t start = pInput->startRowIndex;
int32_t numOfRows = pInput->numOfRows;
int32_t numOfElems = 0;
for (int32_t i = start; i < numOfRows + start; ++i) {
if (pCol->hasNull && colDataIsNull_s(pCol, i)) {
continue;
}
numOfElems++;
char* data = colDataGetData(pCol, i);
if (IS_VAR_DATA_TYPE(type)) {
bytes = varDataLen(data);
data = varDataVal(data);
}
int32_t index = 0;
uint8_t count = hllCountNum(data, bytes, &index);
uint8_t oldcount = pInfo->buckets[index];
if (count > oldcount) {
pInfo->buckets[index] = count;
}
}
SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
return TSDB_CODE_SUCCESS;
}
int32_t hllFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
pInfo->result = hllCountCnt(pInfo->buckets);
return functionFinalize(pCtx, pBlock);
}
bool getStateFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) { bool getStateFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
pEnv->calcMemSize = sizeof(SStateInfo); pEnv->calcMemSize = sizeof(SStateInfo);
return true; return true;
@ -3243,7 +3388,7 @@ int32_t tailFunction(SqlFunctionCtx* pCtx) {
if (pInfo->offset >= pInput->numOfRows) { if (pInfo->offset >= pInput->numOfRows) {
return 0; return 0;
} else { } else {
pInfo->numOfPoints = MIN(pInfo->numOfPoints, pInput->numOfRows - pInfo->offset); pInfo->numOfPoints = TMIN(pInfo->numOfPoints, pInput->numOfRows - pInfo->offset);
} }
for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex - pInfo->offset; i += 1) { for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex - pInfo->offset; i += 1) {

24
source/libs/function/src/tudf.c
View File

@ -795,7 +795,6 @@ int32_t convertScalarParamToDataBlock(SScalarParam *input, int32_t numOfCols, SS
} }
output->info.hasVarCol = hasVarCol; output->info.hasVarCol = hasVarCol;
//TODO: free the array output->pDataBlock
output->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData)); output->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
for (int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
taosArrayPush(output->pDataBlock, (input + i)->columnData); taosArrayPush(output->pDataBlock, (input + i)->columnData);
@ -809,8 +808,12 @@ int32_t convertDataBlockToScalarParm(SSDataBlock *input, SScalarParam *output) {
return -1; return -1;
} }
output->numOfRows = input->info.rows; output->numOfRows = input->info.rows;
//TODO: memory
output->columnData = taosArrayGet(input->pDataBlock, 0); output->columnData = taosMemoryMalloc(sizeof(SColumnInfoData));
memcpy(output->columnData,
taosArrayGet(input->pDataBlock, 0),
sizeof(SColumnInfoData));
return 0; return 0;
} }
@ -833,7 +836,7 @@ int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *
fnDebug("udfc get uv task result. task: %p, uvTask: %p", task, uvTask); fnDebug("udfc get uv task result. task: %p, uvTask: %p", task, uvTask);
if (uvTask->type == UV_TASK_REQ_RSP) { if (uvTask->type == UV_TASK_REQ_RSP) {
if (uvTask->rspBuf.base != NULL) { if (uvTask->rspBuf.base != NULL) {
SUdfResponse rsp; SUdfResponse rsp = {0};
void* buf = decodeUdfResponse(uvTask->rspBuf.base, &rsp); void* buf = decodeUdfResponse(uvTask->rspBuf.base, &rsp);
assert(uvTask->rspBuf.len == POINTER_DISTANCE(buf, uvTask->rspBuf.base)); assert(uvTask->rspBuf.len == POINTER_DISTANCE(buf, uvTask->rspBuf.base));
task->errCode = rsp.code; task->errCode = rsp.code;
@ -1427,7 +1430,10 @@ int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t
int32_t err = callUdf(handle, callType, &inputBlock, NULL, NULL, &resultBlock, NULL); int32_t err = callUdf(handle, callType, &inputBlock, NULL, NULL, &resultBlock, NULL);
if (err == 0) { if (err == 0) {
convertDataBlockToScalarParm(&resultBlock, output); convertDataBlockToScalarParm(&resultBlock, output);
taosArrayDestroy(resultBlock.pDataBlock);
} }
taosArrayDestroy(inputBlock.pDataBlock);
return err; return err;
} }
@ -1508,16 +1514,15 @@ int32_t doTeardownUdf(UdfcFuncHandle handle) {
udfcRunUdfUvTask(task, UV_TASK_REQ_RSP); udfcRunUdfUvTask(task, UV_TASK_REQ_RSP);
SUdfTeardownResponse *rsp = &task->_teardown.rsp;
int32_t err = task->errCode; int32_t err = task->errCode;
udfcRunUdfUvTask(task, UV_TASK_DISCONNECT); udfcRunUdfUvTask(task, UV_TASK_DISCONNECT);
fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle);
taosMemoryFree(task->session); taosMemoryFree(task->session);
taosMemoryFree(task); taosMemoryFree(task);
fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle);
return err; return err;
} }
@ -1564,6 +1569,7 @@ bool udfAggInit(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResult
} }
udfRes->interResNum = buf.numOfResult; udfRes->interResNum = buf.numOfResult;
memcpy(udfRes->interResBuf, buf.buf, buf.bufLen); memcpy(udfRes->interResBuf, buf.buf, buf.bufLen);
freeUdfInterBuf(&buf);
return true; return true;
} }
@ -1621,7 +1627,7 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
blockDataDestroy(inputBlock); blockDataDestroy(inputBlock);
taosArrayDestroy(tempBlock.pDataBlock); taosArrayDestroy(tempBlock.pDataBlock);
taosMemoryFree(newState.buf); freeUdfInterBuf(&newState);
return udfCode; return udfCode;
} }
@ -1650,6 +1656,8 @@ int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock) {
GET_RES_INFO(pCtx)->numOfRes = udfRes->finalResNum; GET_RES_INFO(pCtx)->numOfRes = udfRes->finalResNum;
} }
freeUdfInterBuf(&resultBuf);
int32_t numOfResults = functionFinalizeWithResultBuf(pCtx, pBlock, udfRes->finalResBuf); int32_t numOfResults = functionFinalizeWithResultBuf(pCtx, pBlock, udfRes->finalResBuf);
releaseUdfFuncHandle(pCtx->udfName); releaseUdfFuncHandle(pCtx->udfName);
return udfCallCode == 0 ? numOfResults : udfCallCode; return udfCallCode == 0 ? numOfResults : udfCallCode;

28
source/libs/function/src/udfd.c
View File

@ -96,10 +96,14 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf);
int32_t udfdLoadUdf(char *udfName, SUdf *udf) { int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
strcpy(udf->name, udfName); strcpy(udf->name, udfName);
int32_t err = 0;
err = udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf);
if (err != 0) {
fnError("can not retrieve udf from mnode. udf name %s", udfName);
return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
}
udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf); err = uv_dlopen(udf->path, &udf->lib);
//strcpy(udf->path, "/home/slzhou/TDengine/debug/build/lib/libudf1.so");
int err = uv_dlopen(udf->path, &udf->lib);
if (err != 0) { if (err != 0) {
fnError("can not load library %s. error: %s", udf->path, uv_strerror(err)); fnError("can not load library %s. error: %s", udf->path, uv_strerror(err));
return TSDB_CODE_UDF_LOAD_UDF_FAILURE; return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
@ -142,7 +146,7 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
void udfdProcessSetupRequest(SUvUdfWork* uvUdf, SUdfRequest* request) { void udfdProcessSetupRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
// TODO: tracable id from client. connect, setup, call, teardown // TODO: tracable id from client. connect, setup, call, teardown
fnInfo("%" PRId64 " setup request. udf name: %s", request->seqNum, request->setup.udfName); fnInfo( "setup request. seq num: %" PRId64 ", udf name: %s", request->seqNum, request->setup.udfName);
SUdfSetupRequest *setup = &request->setup; SUdfSetupRequest *setup = &request->setup;
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SUdf *udf = NULL; SUdf *udf = NULL;
@ -276,7 +280,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
void udfdProcessTeardownRequest(SUvUdfWork* uvUdf, SUdfRequest* request) { void udfdProcessTeardownRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
SUdfTeardownRequest *teardown = &request->teardown; SUdfTeardownRequest *teardown = &request->teardown;
fnInfo("teardown. %" PRId64 "handle:%" PRIx64, request->seqNum, teardown->udfHandle); fnInfo("teardown. seq number: %" PRId64 ", handle:%" PRIx64, request->seqNum, teardown->udfHandle);
SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(teardown->udfHandle); SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(teardown->udfHandle);
SUdf *udf = handle->udf; SUdf *udf = handle->udf;
bool unloadUdf = false; bool unloadUdf = false;
@ -800,17 +804,11 @@ static int32_t udfdRun() {
global.udfsHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK); global.udfsHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
uv_mutex_init(&global.udfsMutex); uv_mutex_init(&global.udfsMutex);
if (udfdUvInit() != 0) {
fnError("uv init failure");
return -2;
}
fnInfo("start the udfd"); fnInfo("start the udfd");
int code = uv_run(global.loop, UV_RUN_DEFAULT); int code = uv_run(global.loop, UV_RUN_DEFAULT);
fnInfo("udfd stopped. result: %s, code: %d", uv_err_name(code), code); fnInfo("udfd stopped. result: %s, code: %d", uv_err_name(code), code);
int codeClose = uv_loop_close(global.loop); int codeClose = uv_loop_close(global.loop);
fnDebug("uv loop close. result: %s", uv_err_name(codeClose)); fnDebug("uv loop close. result: %s", uv_err_name(codeClose));
removeListeningPipe();
uv_mutex_destroy(&global.udfsMutex); uv_mutex_destroy(&global.udfsMutex);
taosHashCleanup(global.udfsHash); taosHashCleanup(global.udfsHash);
return 0; return 0;
@ -853,8 +851,14 @@ int main(int argc, char *argv[]) {
return -4; return -4;
} }
if (udfdUvInit() != 0) {
fnError("uv init failure");
return -5;
}
udfdRun(); udfdRun();
udfdCloseClientRpc(); removeListeningPipe();
udfdCloseClientRpc();
} }

86
source/libs/function/test/runUdf.c
View File

@ -34,20 +34,13 @@ static int32_t initLog() {
return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, NULL, 0); return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, NULL, 0);
} }
int main(int argc, char *argv[]) { int scalarFuncTest() {
parseArgs(argc, argv);
initLog();
if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) {
fnError("failed to start since read config error");
return -1;
}
udfcOpen();
uv_sleep(1000);
UdfcFuncHandle handle; UdfcFuncHandle handle;
doSetupUdf("udf1", &handle); if (doSetupUdf("udf1", &handle) != 0) {
fnError("setup udf failure");
return -1;
}
SSDataBlock block = {0}; SSDataBlock block = {0};
SSDataBlock *pBlock = &block; SSDataBlock *pBlock = &block;
@ -74,11 +67,78 @@ int main(int argc, char *argv[]) {
input.columnData = taosArrayGet(pBlock->pDataBlock, 0); input.columnData = taosArrayGet(pBlock->pDataBlock, 0);
SScalarParam output = {0}; SScalarParam output = {0};
doCallUdfScalarFunc(handle, &input, 1, &output); doCallUdfScalarFunc(handle, &input, 1, &output);
taosArrayDestroy(pBlock->pDataBlock);
SColumnInfoData *col = output.columnData; SColumnInfoData *col = output.columnData;
for (int32_t i = 0; i < output.numOfRows; ++i) { for (int32_t i = 0; i < output.numOfRows; ++i) {
fprintf(stderr, "%d\t%d\n", i, *(int32_t *)(col->pData + i * sizeof(int32_t))); fprintf(stderr, "%d\t%d\n", i, *(int32_t *)(col->pData + i * sizeof(int32_t)));
} }
colDataDestroy(output.columnData);
taosMemoryFree(output.columnData);
doTeardownUdf(handle); doTeardownUdf(handle);
return 0;
}
int aggregateFuncTest() {
UdfcFuncHandle handle;
if (doSetupUdf("udf2", &handle) != 0) {
fnError("setup udf failure");
return -1;
}
SSDataBlock block = {0};
SSDataBlock *pBlock = &block;
pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
pBlock->info.numOfCols = 1;
pBlock->info.rows = 4;
char data[16] = {0};
char bitmap[4] = {0};
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
SColumnInfoData colInfo = {0};
colInfo.info.type = TSDB_DATA_TYPE_INT;
colInfo.info.bytes = sizeof(int32_t);
colInfo.info.colId = 1;
colInfo.pData = data;
colInfo.nullbitmap = bitmap;
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
colDataAppendInt32(&colInfo, j, &j);
}
taosArrayPush(pBlock->pDataBlock, &colInfo);
}
SUdfInterBuf buf = {0};
SUdfInterBuf newBuf = {0};
SUdfInterBuf resultBuf = {0};
doCallUdfAggInit(handle, &buf);
doCallUdfAggProcess(handle, pBlock, &buf, &newBuf);
taosArrayDestroy(pBlock->pDataBlock);
doCallUdfAggFinalize(handle, &newBuf, &resultBuf);
fprintf(stderr, "agg result: %f\n", *(double*)resultBuf.buf);
freeUdfInterBuf(&buf);
freeUdfInterBuf(&newBuf);
freeUdfInterBuf(&resultBuf);
doTeardownUdf(handle);
return 0;
}
int main(int argc, char *argv[]) {
parseArgs(argc, argv);
initLog();
if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) {
fnError("failed to start since read config error");
return -1;
}
udfcOpen();
uv_sleep(1000);
scalarFuncTest();
aggregateFuncTest();
udfcClose(); udfcClose();
} }

12
source/libs/index/src/indexComm.c
View File

@ -266,7 +266,7 @@ int32_t indexConvertData(void* src, int8_t type, void** dst) {
TASSERT(0); TASSERT(0);
break; break;
} }
*dst = *dst - tlen; *dst = (char*)*dst - tlen;
// indexMayFillNumbericData(*dst, tlen); // indexMayFillNumbericData(*dst, tlen);
return tlen; return tlen;
} }
@ -306,7 +306,7 @@ int32_t indexConvertDataToStr(void* src, int8_t type, void** dst) {
tlen = taosEncodeBinary(NULL, src, sizeof(float)); tlen = taosEncodeBinary(NULL, src, sizeof(float));
*dst = taosMemoryCalloc(1, tlen + 1); *dst = taosMemoryCalloc(1, tlen + 1);
tlen = taosEncodeBinary(dst, src, sizeof(float)); tlen = taosEncodeBinary(dst, src, sizeof(float));
*dst = *dst - tlen; *dst = (char*) * dst - tlen;
break; break;
case TSDB_DATA_TYPE_UINT: case TSDB_DATA_TYPE_UINT:
*dst = taosMemoryCalloc(1, sizeof(int64_t) + 1); *dst = taosMemoryCalloc(1, sizeof(int64_t) + 1);
@ -320,7 +320,7 @@ int32_t indexConvertDataToStr(void* src, int8_t type, void** dst) {
tlen = taosEncodeBinary(NULL, src, sizeof(double)); tlen = taosEncodeBinary(NULL, src, sizeof(double));
*dst = taosMemoryCalloc(1, tlen + 1); *dst = taosMemoryCalloc(1, tlen + 1);
tlen = taosEncodeBinary(dst, src, sizeof(double)); tlen = taosEncodeBinary(dst, src, sizeof(double));
*dst = *dst - tlen; *dst = (char*) * dst - tlen;
break; break;
case TSDB_DATA_TYPE_UBIGINT: case TSDB_DATA_TYPE_UBIGINT:
assert(0); assert(0);
@ -331,7 +331,7 @@ int32_t indexConvertDataToStr(void* src, int8_t type, void** dst) {
tlen = taosEncodeBinary(NULL, varDataVal(src), varDataLen(src)); tlen = taosEncodeBinary(NULL, varDataVal(src), varDataLen(src));
*dst = taosMemoryCalloc(1, tlen + 1); *dst = taosMemoryCalloc(1, tlen + 1);
tlen = taosEncodeBinary(dst, varDataVal(src), varDataLen(src)); tlen = taosEncodeBinary(dst, varDataVal(src), varDataLen(src));
*dst = *dst - tlen; *dst = (char*) * dst - tlen;
break; break;
} }
@ -340,7 +340,7 @@ int32_t indexConvertDataToStr(void* src, int8_t type, void** dst) {
tlen = taosEncodeBinary(NULL, src, strlen(src)); tlen = taosEncodeBinary(NULL, src, strlen(src));
*dst = taosMemoryCalloc(1, tlen + 1); *dst = taosMemoryCalloc(1, tlen + 1);
tlen = taosEncodeBinary(dst, src, strlen(src)); tlen = taosEncodeBinary(dst, src, strlen(src));
*dst = *dst - tlen; *dst = (char*) * dst - tlen;
break; break;
#endif #endif
} }
@ -349,7 +349,7 @@ int32_t indexConvertDataToStr(void* src, int8_t type, void** dst) {
tlen = taosEncodeBinary(NULL, src, strlen(src)); tlen = taosEncodeBinary(NULL, src, strlen(src));
*dst = taosMemoryCalloc(1, tlen + 1); *dst = taosMemoryCalloc(1, tlen + 1);
tlen = taosEncodeBinary(dst, src, strlen(src)); tlen = taosEncodeBinary(dst, src, strlen(src));
*dst = *dst - tlen; *dst = (char*) * dst - tlen;
break; break;
#endif #endif
default: default:

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

@ -529,6 +529,18 @@ void nodesDestroyNode(SNodeptr pNode) {
nodesDestroyNode(pStmt->pTbNamePattern); nodesDestroyNode(pStmt->pTbNamePattern);
break; break;
} }
case QUERY_NODE_QUERY: {
SQuery* pQuery = (SQuery*)pNode;
nodesDestroyNode(pQuery->pRoot);
taosMemoryFreeClear(pQuery->pResSchema);
if (NULL != pQuery->pCmdMsg) {
taosMemoryFreeClear(pQuery->pCmdMsg->pMsg);
taosMemoryFreeClear(pQuery->pCmdMsg);
}
taosArrayDestroy(pQuery->pDbList);
taosArrayDestroy(pQuery->pTableList);
break;
}
case QUERY_NODE_LOGIC_PLAN_SCAN: { case QUERY_NODE_LOGIC_PLAN_SCAN: {
SScanLogicNode* pLogicNode = (SScanLogicNode*)pNode; SScanLogicNode* pLogicNode = (SScanLogicNode*)pNode;
destroyLogicNode((SLogicNode*)pLogicNode); destroyLogicNode((SLogicNode*)pLogicNode);

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

@ -4254,7 +4254,135 @@ static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
return rewriteToVnodeModifyOpStmt(pQuery, pBufArray); return rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
} }
static int32_t buildAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq) { static SSchema* getColSchema(STableMeta* pTableMeta, const char* pTagName) {
int32_t numOfFields = getNumOfTags(pTableMeta) + getNumOfColumns(pTableMeta);
for (int32_t i = 0; i < numOfFields; ++i) {
SSchema* pTagSchema = pTableMeta->schema + i;
if (0 == strcmp(pTagName, pTagSchema->name)) {
return pTagSchema;
}
}
return NULL;
}
static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
}
pReq->tagName = strdup(pStmt->colName);
if (NULL == pReq->tagName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
if (DEAL_RES_ERROR == translateValueImpl(pCxt, pStmt->pVal, schemaToDataType(pSchema))) {
return pCxt->errCode;
}
pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type);
pReq->nTagVal = pStmt->pVal->node.resType.bytes;
char* pVal = nodesGetValueFromNode(pStmt->pVal);
pReq->pTagVal = IS_VAR_DATA_TYPE(pStmt->pVal->node.resType.type) ? pVal + VARSTR_HEADER_SIZE : pVal;
return TSDB_CODE_SUCCESS;
}
static int32_t buildAddColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
if (NULL != getColSchema(pTableMeta, pStmt->colName)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_DUPLICATED_COLUMN);
}
pReq->colName = strdup(pStmt->colName);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pReq->type = pStmt->dataType.type;
pReq->flags = COL_SMA_ON;
pReq->bytes = pStmt->dataType.bytes;
return TSDB_CODE_SUCCESS;
}
static int32_t buildDropColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMN, pStmt->colName);
} else if (PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY);
}
pReq->colName = strdup(pStmt->colName);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static int32_t buildUpdateColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
pReq->colModBytes = calcTypeBytes(pStmt->dataType);
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMN, pStmt->colName);
} else if (!IS_VAR_DATA_TYPE(pSchema->type) || pSchema->bytes >= pReq->colModBytes) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_MODIFY_COL);
}
pReq->colName = strdup(pStmt->colName);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static int32_t buildRenameColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
if (NULL == getColSchema(pTableMeta, pStmt->colName)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMN, pStmt->colName);
}
if (NULL != getColSchema(pTableMeta, pStmt->newColName)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_DUPLICATED_COLUMN);
}
pReq->colName = strdup(pStmt->colName);
pReq->colNewName = strdup(pStmt->newColName);
if (NULL == pReq->colName || NULL == pReq->colNewName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static int32_t buildUpdateOptionsReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq) {
int32_t code = TSDB_CODE_SUCCESS;
if (-1 != pStmt->pOptions->ttl) {
code = checkRangeOption(pCxt, "ttl", pStmt->pOptions->ttl, TSDB_MIN_TABLE_TTL, INT32_MAX);
if (TSDB_CODE_SUCCESS == code) {
pReq->updateTTL = true;
pReq->newTTL = pStmt->pOptions->ttl;
}
}
if (TSDB_CODE_SUCCESS == code && '\0' != pStmt->pOptions->comment[0]) {
pReq->updateComment = true;
pReq->newComment = strdup(pStmt->pOptions->comment);
if (NULL == pReq->newComment) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
return code;
}
static int32_t buildAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
pReq->tbName = strdup(pStmt->tableName); pReq->tbName = strdup(pStmt->tableName);
if (NULL == pReq->tbName) { if (NULL == pReq->tbName) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
@ -4268,60 +4396,22 @@ static int32_t buildAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt,
case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES: case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE); return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: case TSDB_ALTER_TABLE_UPDATE_TAG_VAL:
pReq->tagName = strdup(pStmt->colName); return buildUpdateTagValReq(pCxt, pStmt, pTableMeta, pReq);
if (NULL == pReq->tagName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
if (DEAL_RES_ERROR == translateValue(pCxt, pStmt->pVal)) {
return pCxt->errCode;
}
pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type);
pReq->nTagVal = pStmt->pVal->node.resType.bytes;
char* pVal = nodesGetValueFromNode(pStmt->pVal);
pReq->pTagVal = IS_VAR_DATA_TYPE(pStmt->pVal->node.resType.type) ? pVal + VARSTR_HEADER_SIZE : pVal;
break;
case TSDB_ALTER_TABLE_ADD_COLUMN: case TSDB_ALTER_TABLE_ADD_COLUMN:
return buildAddColReq(pCxt, pStmt, pTableMeta, pReq);
case TSDB_ALTER_TABLE_DROP_COLUMN: case TSDB_ALTER_TABLE_DROP_COLUMN:
pReq->colName = strdup(pStmt->colName); return buildDropColReq(pCxt, pStmt, pTableMeta, pReq);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pReq->type = pStmt->dataType.type;
pReq->flags = COL_SMA_ON;
pReq->bytes = pStmt->dataType.bytes;
break;
case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
pReq->colName = strdup(pStmt->colName); return buildUpdateColReq(pCxt, pStmt, pTableMeta, pReq);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pReq->colModBytes = calcTypeBytes(pStmt->dataType);
break;
case TSDB_ALTER_TABLE_UPDATE_OPTIONS: case TSDB_ALTER_TABLE_UPDATE_OPTIONS:
if (-1 != pStmt->pOptions->ttl) { return buildUpdateOptionsReq(pCxt, pStmt, pReq);
pReq->updateTTL = true;
pReq->newTTL = pStmt->pOptions->ttl;
}
if ('\0' != pStmt->pOptions->comment[0]) {
pReq->updateComment = true;
pReq->newComment = strdup(pStmt->pOptions->comment);
if (NULL == pReq->newComment) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
break;
case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME:
pReq->colName = strdup(pStmt->colName); return buildRenameColReq(pCxt, pStmt, pTableMeta, pReq);
pReq->colNewName = strdup(pStmt->newColName);
if (NULL == pReq->colName || NULL == pReq->colNewName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
break;
default: default:
break; break;
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_FAILED;
} }
static int32_t serializeAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq, static int32_t serializeAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq,
@ -4394,7 +4484,7 @@ static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) {
} }
SVAlterTbReq req = {0}; SVAlterTbReq req = {0};
code = buildAlterTbReq(pCxt, pStmt, &req); code = buildAlterTbReq(pCxt, pStmt, pTableMeta, &req);
SArray* pArray = NULL; SArray* pArray = NULL;
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {

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

@ -154,6 +154,10 @@ static char* getSyntaxErrFormat(int32_t errCode) {
return "Invalid password"; return "Invalid password";
case TSDB_CODE_PAR_INVALID_ALTER_TABLE: case TSDB_CODE_PAR_INVALID_ALTER_TABLE:
return "Invalid alter table statement"; return "Invalid alter table statement";
case TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY:
return "Primary timestamp column cannot be dropped";
case TSDB_CODE_PAR_INVALID_MODIFY_COL:
return "Only binary/nchar column length could be modified";
case TSDB_CODE_OUT_OF_MEMORY: case TSDB_CODE_OUT_OF_MEMORY:
return "Out of memory"; return "Out of memory";
default: default:

27
source/libs/parser/src/parser.c
View File

@ -39,10 +39,16 @@ static int32_t parseSqlIntoAst(SParseContext* pCxt, SQuery** pQuery) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = authenticate(pCxt, *pQuery); code = authenticate(pCxt, *pQuery);
} }
if (TSDB_CODE_SUCCESS == code && 0 == (*pQuery)->placeholderNum) {
if (TSDB_CODE_SUCCESS == code && (*pQuery)->placeholderNum > 0) {
// TSWAP((*pQuery)->pContainPlaceholderRoot, (*pQuery)->pRoot);
return TSDB_CODE_SUCCESS;
}
if (TSDB_CODE_SUCCESS == code) {
code = translate(pCxt, *pQuery); code = translate(pCxt, *pQuery);
} }
if (TSDB_CODE_SUCCESS == code && 0 == (*pQuery)->placeholderNum) { if (TSDB_CODE_SUCCESS == code) {
code = calculateConstant(pCxt, *pQuery); code = calculateConstant(pCxt, *pQuery);
} }
return code; return code;
@ -142,26 +148,13 @@ int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery) {
return code; return code;
} }
void qDestroyQuery(SQuery* pQueryNode) { void qDestroyQuery(SQuery* pQueryNode) { nodesDestroyNode(pQueryNode); }
if (NULL == pQueryNode) {
return;
}
nodesDestroyNode(pQueryNode->pRoot);
taosMemoryFreeClear(pQueryNode->pResSchema);
if (NULL != pQueryNode->pCmdMsg) {
taosMemoryFreeClear(pQueryNode->pCmdMsg->pMsg);
taosMemoryFreeClear(pQueryNode->pCmdMsg);
}
taosArrayDestroy(pQueryNode->pDbList);
taosArrayDestroy(pQueryNode->pTableList);
taosMemoryFreeClear(pQueryNode);
}
int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) { int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) {
return extractResultSchema(pRoot, numOfCols, pSchema); return extractResultSchema(pRoot, numOfCols, pSchema);
} }
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId) { int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
if (colIdx < 0) { if (colIdx < 0) {

6
source/libs/parser/test/parInitialATest.cpp
View File

@ -283,13 +283,13 @@ TEST_F(ParserInitialATest, alterTable) {
setAlterColFunc("t1", TSDB_ALTER_TABLE_DROP_COLUMN, "c1"); setAlterColFunc("t1", TSDB_ALTER_TABLE_DROP_COLUMN, "c1");
run("ALTER TABLE t1 DROP COLUMN c1"); run("ALTER TABLE t1 DROP COLUMN c1");
setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, "c1", TSDB_DATA_TYPE_VARCHAR, 20 + VARSTR_HEADER_SIZE); setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, "c2", TSDB_DATA_TYPE_VARCHAR, 30 + VARSTR_HEADER_SIZE);
run("ALTER TABLE t1 MODIFY COLUMN c1 VARCHAR(20)"); run("ALTER TABLE t1 MODIFY COLUMN c2 VARCHAR(30)");
setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME, "c1", 0, 0, "cc1"); setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME, "c1", 0, 0, "cc1");
run("ALTER TABLE t1 RENAME COLUMN c1 cc1"); run("ALTER TABLE t1 RENAME COLUMN c1 cc1");
int64_t val = 10; int32_t val = 10;
setAlterTagFunc("st1s1", "tag1", (const uint8_t*)&val, sizeof(val)); setAlterTagFunc("st1s1", "tag1", (const uint8_t*)&val, sizeof(val));
run("ALTER TABLE st1s1 SET TAG tag1=10"); run("ALTER TABLE st1s1 SET TAG tag1=10");

268
source/libs/planner/src/planner.c
View File

@ -18,28 +18,6 @@
#include "planInt.h" #include "planInt.h"
#include "scalar.h" #include "scalar.h"
typedef struct SCollectPlaceholderValuesCxt {
int32_t errCode;
SArray* pValues;
} SCollectPlaceholderValuesCxt;
static EDealRes collectPlaceholderValuesImpl(SNode* pNode, void* pContext) {
if (QUERY_NODE_VALUE == nodeType(pNode) && ((SValueNode*)pNode)->placeholderNo > 0) {
SCollectPlaceholderValuesCxt* pCxt = pContext;
taosArrayInsert(pCxt->pValues, ((SValueNode*)pNode)->placeholderNo - 1, &pNode);
return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR;
}
return DEAL_RES_CONTINUE;
}
static int32_t collectPlaceholderValues(SPlanContext* pCxt, SQueryPlan* pPlan) {
pPlan->pPlaceholderValues = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
SCollectPlaceholderValuesCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pValues = pPlan->pPlaceholderValues};
nodesWalkPhysiPlan((SNode*)pPlan, collectPlaceholderValuesImpl, &cxt);
return cxt.errCode;
}
int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNodeList) { int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNodeList) {
SLogicNode* pLogicNode = NULL; SLogicNode* pLogicNode = NULL;
SLogicSubplan* pLogicSubplan = NULL; SLogicSubplan* pLogicSubplan = NULL;
@ -58,9 +36,6 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = createPhysiPlan(pCxt, pLogicPlan, pPlan, pExecNodeList); code = createPhysiPlan(pCxt, pLogicPlan, pPlan, pExecNodeList);
} }
if (TSDB_CODE_SUCCESS == code && pCxt->placeholderNum > 0) {
code = collectPlaceholderValues(pCxt, *pPlan);
}
nodesDestroyNode(pLogicNode); nodesDestroyNode(pLogicNode);
nodesDestroyNode(pLogicSubplan); nodesDestroyNode(pLogicSubplan);
@ -99,249 +74,6 @@ int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstream
return setSubplanExecutionNode(subplan->pNode, groupId, pSource); return setSubplanExecutionNode(subplan->pNode, groupId, pSource);
} }
static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) {
if (pParam->is_null && 1 == *(pParam->is_null)) {
pVal->node.resType.type = TSDB_DATA_TYPE_NULL;
pVal->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_NULL].bytes;
return TSDB_CODE_SUCCESS;
}
int32_t inputSize = (NULL != pParam->length ? *(pParam->length) : tDataTypes[pParam->buffer_type].bytes);
pVal->node.resType.type = pParam->buffer_type;
pVal->node.resType.bytes = inputSize;
switch (pParam->buffer_type) {
case TSDB_DATA_TYPE_BOOL:
pVal->datum.b = *((bool*)pParam->buffer);
break;
case TSDB_DATA_TYPE_TINYINT:
pVal->datum.i = *((int8_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_SMALLINT:
pVal->datum.i = *((int16_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_INT:
pVal->datum.i = *((int32_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_BIGINT:
pVal->datum.i = *((int64_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_FLOAT:
pVal->datum.d = *((float*)pParam->buffer);
break;
case TSDB_DATA_TYPE_DOUBLE:
pVal->datum.d = *((double*)pParam->buffer);
break;
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY:
pVal->datum.p = taosMemoryCalloc(1, pVal->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
if (NULL == pVal->datum.p) {
return TSDB_CODE_OUT_OF_MEMORY;
}
varDataSetLen(pVal->datum.p, pVal->node.resType.bytes);
strncpy(varDataVal(pVal->datum.p), (const char*)pParam->buffer, pVal->node.resType.bytes);
break;
case TSDB_DATA_TYPE_NCHAR: {
pVal->node.resType.bytes *= TSDB_NCHAR_SIZE;
pVal->datum.p = taosMemoryCalloc(1, pVal->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
if (NULL == pVal->datum.p) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t output = 0;
if (!taosMbsToUcs4(pParam->buffer, inputSize, (TdUcs4*)varDataVal(pVal->datum.p), pVal->node.resType.bytes,
&output)) {
return errno;
}
varDataSetLen(pVal->datum.p, output);
pVal->node.resType.bytes = output;
break;
}
case TSDB_DATA_TYPE_TIMESTAMP:
pVal->datum.i = *((int64_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_UTINYINT:
pVal->datum.u = *((uint8_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_USMALLINT:
pVal->datum.u = *((uint16_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_UINT:
pVal->datum.u = *((uint32_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_UBIGINT:
pVal->datum.u = *((uint64_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_DECIMAL:
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_MEDIUMBLOB:
// todo
default:
break;
}
pVal->translate = true;
return TSDB_CODE_SUCCESS;
}
static EDealRes updatePlanQueryId(SNode* pNode, void* pContext) {
int64_t queryId = *(uint64_t*)pContext;
if (QUERY_NODE_PHYSICAL_PLAN == nodeType(pNode)) {
SQueryPlan* planNode = (SQueryPlan*)pNode;
planNode->queryId = queryId;
} else if (QUERY_NODE_PHYSICAL_SUBPLAN == nodeType(pNode)) {
SSubplan* subplanNode = (SSubplan*)pNode;
subplanNode->id.queryId = queryId;
}
return DEAL_RES_CONTINUE;
}
static int32_t calcConstNode(SNode** pNode) {
if (NULL == *pNode) {
return TSDB_CODE_SUCCESS;
}
SNode* pNew = NULL;
int32_t code = scalarCalculateConstants(*pNode, &pNew);
if (TSDB_CODE_SUCCESS == code) {
*pNode = pNew;
}
return code;
}
static int32_t calcConstList(SNodeList* pList) {
SNode* pNode = NULL;
FOREACH(pNode, pList) {
SNode* pNew = NULL;
int32_t code = scalarCalculateConstants(pNode, &pNew);
if (TSDB_CODE_SUCCESS == code) {
REPLACE_NODE(pNew);
} else {
return code;
}
}
return TSDB_CODE_SUCCESS;
}
static bool isEmptyResultCond(SNode** pCond) {
if (NULL == *pCond || QUERY_NODE_VALUE != nodeType(*pCond)) {
return false;
}
if (((SValueNode*)*pCond)->datum.b) {
nodesDestroyNode(*pCond);
*pCond = NULL;
return false;
}
return true;
}
static int32_t calcConstSpecificPhysiNode(SPhysiNode* pPhyNode) {
switch (nodeType(pPhyNode)) {
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE:
case QUERY_NODE_PHYSICAL_PLAN_FILL:
return TSDB_CODE_SUCCESS;
case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
return calcConstList(((SProjectPhysiNode*)pPhyNode)->pProjections);
case QUERY_NODE_PHYSICAL_PLAN_JOIN:
return calcConstNode(&(((SJoinPhysiNode*)pPhyNode)->pOnConditions));
case QUERY_NODE_PHYSICAL_PLAN_AGG:
return calcConstList(((SAggPhysiNode*)pPhyNode)->pExprs);
case QUERY_NODE_PHYSICAL_PLAN_SORT:
return calcConstList(((SSortPhysiNode*)pPhyNode)->pExprs);
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
return calcConstList(((SWinodwPhysiNode*)pPhyNode)->pExprs);
case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
return calcConstList(((SPartitionPhysiNode*)pPhyNode)->pExprs);
default:
break;
}
return TSDB_CODE_SUCCESS;
}
static int32_t calcConstSubplan(SPhysiNode* pPhyNode, bool* pEmptyResult) {
int32_t code = calcConstNode(&pPhyNode->pConditions);
if (TSDB_CODE_SUCCESS == code) {
code = calcConstSpecificPhysiNode(pPhyNode);
}
if (TSDB_CODE_SUCCESS != code) {
return code;
}
*pEmptyResult = isEmptyResultCond(&pPhyNode->pConditions);
if (*pEmptyResult) {
return TSDB_CODE_SUCCESS;
}
*pEmptyResult = true;
bool subEmptyResult = false;
SNode* pChild = NULL;
FOREACH(pChild, pPhyNode->pChildren) {
code = calcConstSubplan((SPhysiNode*)pChild, &subEmptyResult);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
if (!subEmptyResult) {
*pEmptyResult = false;
}
}
return TSDB_CODE_SUCCESS;
}
static int32_t calcConstPhysiPlan(SQueryPlan* pPlan, bool* pEmptyResult) {
*pEmptyResult = true;
bool subEmptyResult = false;
SNodeListNode* pNode = nodesListGetNode(pPlan->pSubplans, 0);
SNode* pSubplan = NULL;
FOREACH(pSubplan, pNode->pNodeList) {
int32_t code = calcConstSubplan(((SSubplan*)pSubplan)->pNode, pEmptyResult);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
if (!subEmptyResult) {
*pEmptyResult = false;
}
}
return TSDB_CODE_SUCCESS;
}
int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId,
bool* pEmptyResult) {
int32_t size = taosArrayGetSize(pPlan->pPlaceholderValues);
int32_t code = 0;
if (colIdx < 0) {
for (int32_t i = 0; i < size; ++i) {
code = setValueByBindParam((SValueNode*)taosArrayGetP(pPlan->pPlaceholderValues, i), pParams + i);
if (code) {
return code;
}
}
} else {
code = setValueByBindParam((SValueNode*)taosArrayGetP(pPlan->pPlaceholderValues, colIdx), pParams);
if (code) {
return code;
}
}
if (colIdx < 0 || ((colIdx + 1) == size)) {
nodesWalkPhysiPlan((SNode*)pPlan, updatePlanQueryId, &queryId);
code = calcConstPhysiPlan(pPlan, pEmptyResult);
}
return code;
}
int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) { int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {
if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType) { if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType) {
SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink; SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink;

60
source/libs/planner/test/planStmtTest.cpp
View File

@ -20,35 +20,49 @@ using namespace std;
class PlanStmtTest : public PlannerTestBase { class PlanStmtTest : public PlannerTestBase {
public: public:
void prepare(const string& sql) { void buildParam(TAOS_MULTI_BIND* pBindParams, int32_t index, void* pVal, int32_t type, int32_t bytes = 0) {
run(sql); TAOS_MULTI_BIND* pBindParam = pBindParams + index;
// todo calloc pBindParams_ pBindParam->buffer_type = type;
} pBindParam->num = 1;
pBindParam->buffer_length = bytes > 0 ? bytes : tDataTypes[type].bytes;
pBindParam->buffer = taosMemoryCalloc(1, pBindParam->buffer_length);
pBindParam->length = (int32_t*)taosMemoryCalloc(1, sizeof(int32_t));
pBindParam->is_null = (char*)taosMemoryCalloc(1, sizeof(char));
*(pBindParam->length) = bytes > 0 ? bytes : tDataTypes[type].bytes;
*(pBindParam->is_null) = 0;
void bindParam(int32_t val) { switch (type) {
TAOS_MULTI_BIND* pBind = pBindParams_ + paramNo_++; case TSDB_DATA_TYPE_BOOL:
pBind->buffer_type = TSDB_DATA_TYPE_INT; *((bool*)pBindParam->buffer) = *(bool*)pVal;
pBind->num = 1; break;
pBind->buffer_length = sizeof(int32_t); case TSDB_DATA_TYPE_TINYINT:
pBind->buffer = taosMemoryCalloc(1, pBind->buffer_length); *((int8_t*)pBindParam->buffer) = *(int64_t*)pVal;
pBind->length = (int32_t*)taosMemoryCalloc(1, sizeof(int32_t)); break;
pBind->is_null = (char*)taosMemoryCalloc(1, sizeof(char)); case TSDB_DATA_TYPE_SMALLINT:
*((int32_t*)pBind->buffer) = val; case TSDB_DATA_TYPE_INT:
*(pBind->length) = sizeof(int32_t); case TSDB_DATA_TYPE_BIGINT:
*(pBind->is_null) = 0; case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_TIMESTAMP:
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_UTINYINT:
case TSDB_DATA_TYPE_USMALLINT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_UBIGINT:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_DECIMAL:
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_MEDIUMBLOB:
default:
break;
} }
void exec() {
// todo
} }
private:
TAOS_MULTI_BIND* pBindParams_;
int32_t paramNo_;
}; };
TEST_F(PlanStmtTest, stmt) { TEST_F(PlanStmtTest, stmt) {
useDb("root", "test"); useDb("root", "test");
// run("select * from t1 where c1 = ?"); prepare("SELECT * FROM t1 WHERE c1 = ?");
} }

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

@ -108,6 +108,57 @@ class PlannerTestBaseImpl {
} }
} }
void prepare(const string& sql) {
reset();
try {
doParseSql(sql, &stmtEnv_.pQuery_, true);
dump(g_dumpModule);
} catch (...) {
dump(DUMP_MODULE_ALL);
throw;
}
}
void bindParams(TAOS_MULTI_BIND* pParams, int32_t colIdx) {
try {
doBindParams(stmtEnv_.pQuery_, pParams, colIdx);
SPlanContext cxt = {0};
setPlanContext(stmtEnv_.pQuery_, &cxt);
SLogicNode* pLogicNode = nullptr;
doCreateLogicPlan(&cxt, &pLogicNode);
doOptimizeLogicPlan(&cxt, pLogicNode);
SLogicSubplan* pLogicSubplan = nullptr;
doSplitLogicPlan(&cxt, pLogicNode, &pLogicSubplan);
SQueryLogicPlan* pLogicPlan = nullptr;
doScaleOutLogicPlan(&cxt, pLogicSubplan, &pLogicPlan);
SQueryPlan* pPlan = nullptr;
doCreatePhysiPlan(&cxt, pLogicPlan, &pPlan);
dump(g_dumpModule);
} catch (...) {
dump(DUMP_MODULE_ALL);
throw;
}
}
void exec() {
try {
doParseBoundSql(stmtEnv_.pQuery_);
dump(g_dumpModule);
} catch (...) {
dump(DUMP_MODULE_ALL);
throw;
}
}
private: private:
struct caseEnv { struct caseEnv {
string acctId_; string acctId_;
@ -117,10 +168,15 @@ class PlannerTestBaseImpl {
struct stmtEnv { struct stmtEnv {
string sql_; string sql_;
array<char, 1024> msgBuf_; array<char, 1024> msgBuf_;
SQuery* pQuery_;
~stmtEnv() { qDestroyQuery(pQuery_); }
}; };
struct stmtRes { struct stmtRes {
string ast_; string ast_;
string prepareAst_;
string boundAst_;
string rawLogicPlan_; string rawLogicPlan_;
string optimizedLogicPlan_; string optimizedLogicPlan_;
string splitLogicPlan_; string splitLogicPlan_;
@ -132,8 +188,10 @@ class PlannerTestBaseImpl {
void reset() { void reset() {
stmtEnv_.sql_.clear(); stmtEnv_.sql_.clear();
stmtEnv_.msgBuf_.fill(0); stmtEnv_.msgBuf_.fill(0);
qDestroyQuery(stmtEnv_.pQuery_);
res_.ast_.clear(); res_.ast_.clear();
res_.boundAst_.clear();
res_.rawLogicPlan_.clear(); res_.rawLogicPlan_.clear();
res_.optimizedLogicPlan_.clear(); res_.optimizedLogicPlan_.clear();
res_.splitLogicPlan_.clear(); res_.splitLogicPlan_.clear();
@ -152,6 +210,9 @@ class PlannerTestBaseImpl {
if (DUMP_MODULE_ALL == module || DUMP_MODULE_PARSER == module) { if (DUMP_MODULE_ALL == module || DUMP_MODULE_PARSER == module) {
cout << "syntax tree : " << endl; cout << "syntax tree : " << endl;
cout << res_.ast_ << endl; cout << res_.ast_ << endl;
cout << "bound syntax tree : " << endl;
cout << res_.boundAst_ << endl;
} }
if (DUMP_MODULE_ALL == module || DUMP_MODULE_LOGIC == module) { if (DUMP_MODULE_ALL == module || DUMP_MODULE_LOGIC == module) {
@ -187,7 +248,7 @@ class PlannerTestBaseImpl {
} }
} }
void doParseSql(const string& sql, SQuery** pQuery) { void doParseSql(const string& sql, SQuery** pQuery, bool prepare = false) {
stmtEnv_.sql_ = sql; stmtEnv_.sql_ = sql;
transform(stmtEnv_.sql_.begin(), stmtEnv_.sql_.end(), stmtEnv_.sql_.begin(), ::tolower); transform(stmtEnv_.sql_.begin(), stmtEnv_.sql_.end(), stmtEnv_.sql_.begin(), ::tolower);
@ -200,8 +261,32 @@ class PlannerTestBaseImpl {
cxt.msgLen = stmtEnv_.msgBuf_.max_size(); cxt.msgLen = stmtEnv_.msgBuf_.max_size();
DO_WITH_THROW(qParseSql, &cxt, pQuery); DO_WITH_THROW(qParseSql, &cxt, pQuery);
if (prepare) {
res_.prepareAst_ = toString((*pQuery)->pRoot);
} else {
res_.ast_ = toString((*pQuery)->pRoot); res_.ast_ = toString((*pQuery)->pRoot);
} }
}
void doBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx) {
DO_WITH_THROW(qStmtBindParams, pQuery, pParams, colIdx);
if (colIdx < 0 || pQuery->placeholderNum == colIdx + 1) {
res_.boundAst_ = toString(pQuery->pRoot);
}
}
void doParseBoundSql(SQuery* pQuery) {
SParseContext cxt = {0};
cxt.acctId = atoi(caseEnv_.acctId_.c_str());
cxt.db = caseEnv_.db_.c_str();
cxt.pSql = stmtEnv_.sql_.c_str();
cxt.sqlLen = stmtEnv_.sql_.length();
cxt.pMsg = stmtEnv_.msgBuf_.data();
cxt.msgLen = stmtEnv_.msgBuf_.max_size();
DO_WITH_THROW(qStmtParseQuerySql, &cxt, pQuery);
res_.ast_ = toString(pQuery->pRoot);
}
void doCreateLogicPlan(SPlanContext* pCxt, SLogicNode** pLogicNode) { void doCreateLogicPlan(SPlanContext* pCxt, SLogicNode** pLogicNode) {
DO_WITH_THROW(createLogicPlan, pCxt, pLogicNode); DO_WITH_THROW(createLogicPlan, pCxt, pLogicNode);
@ -275,3 +360,11 @@ PlannerTestBase::~PlannerTestBase() {}
void PlannerTestBase::useDb(const std::string& acctId, const std::string& db) { impl_->useDb(acctId, db); } void PlannerTestBase::useDb(const std::string& acctId, const std::string& db) { impl_->useDb(acctId, db); }
void PlannerTestBase::run(const std::string& sql) { return impl_->run(sql); } void PlannerTestBase::run(const std::string& sql) { return impl_->run(sql); }
void PlannerTestBase::prepare(const std::string& sql) { return impl_->prepare(sql); }
void PlannerTestBase::bindParams(TAOS_MULTI_BIND* pParams, int32_t colIdx) {
return impl_->bindParams(pParams, colIdx);
}
void PlannerTestBase::exec() { return impl_->exec(); }

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

@ -19,6 +19,7 @@
#include <gtest/gtest.h> #include <gtest/gtest.h>
class PlannerTestBaseImpl; class PlannerTestBaseImpl;
struct TAOS_MULTI_BIND;
class PlannerTestBase : public testing::Test { class PlannerTestBase : public testing::Test {
public: public:
@ -27,6 +28,10 @@ class PlannerTestBase : public testing::Test {
void useDb(const std::string& acctId, const std::string& db); void useDb(const std::string& acctId, const std::string& db);
void run(const std::string& sql); void run(const std::string& sql);
// stmt mode APIs
void prepare(const std::string& sql);
void bindParams(TAOS_MULTI_BIND* pParams, int32_t colIdx);
void exec();
private: private:
std::unique_ptr<PlannerTestBaseImpl> impl_; std::unique_ptr<PlannerTestBaseImpl> impl_;

1
source/libs/scheduler/src/scheduler.c
View File

@ -1148,6 +1148,7 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
if (NULL == msg) { if (NULL == msg) {
SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT); SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
} }
SCH_ERR_RET(schProcessOnTaskSuccess(pJob, pTask));
break; break;
} }
case TDMT_VND_SUBMIT_RSP: { case TDMT_VND_SUBMIT_RSP: {

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

@ -28,9 +28,9 @@ struct SPCache {
SPage lru; SPage lru;
}; };
static inline int tdbPCachePageHash(const SPgid *pPgid) { static inline uint32_t tdbPCachePageHash(const SPgid *pPgid) {
u32 *t = (u32 *)((pPgid)->fileid); uint32_t *t = (uint32_t *)((pPgid)->fileid);
return t[0] + t[1] + t[2] + t[3] + t[4] + t[5] + (pPgid)->pgno; return (uint32_t)(t[0] + t[1] + t[2] + t[3] + t[4] + t[5] + (pPgid)->pgno);
} }
#define PAGE_IS_PINNED(pPage) ((pPage)->pLruNext == NULL) #define PAGE_IS_PINNED(pPage) ((pPage)->pLruNext == NULL)

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

@ -72,7 +72,7 @@ int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
return -1; return -1;
} }
ret = tdbGnrtFileID(pPager->dbFileName, pPager->fid, false); ret = tdbGnrtFileID(pPager->fd, pPager->fid, false);
if (ret < 0) { if (ret < 0) {
return -1; return -1;
} }

4
source/libs/tdb/src/db/tdbUtil.c
View File

@ -35,10 +35,10 @@ void tdbFree(void *p) {
} }
} }
int tdbGnrtFileID(const char *fname, uint8_t *fileid, bool unique) { int tdbGnrtFileID(tdb_fd_t fd, uint8_t *fileid, bool unique) {
int64_t stDev = 0, stIno = 0; int64_t stDev = 0, stIno = 0;
if (taosDevInoFile(fname, &stDev, &stIno) < 0) { if (taosDevInoFile(fd, &stDev, &stIno) < 0) {
return -1; return -1;
} }

2
source/libs/tdb/src/inc/tdbUtil.h
View File

@ -28,7 +28,7 @@ extern "C" {
#define TDB_ROUND8(x) (((x) + 7) & ~7) #define TDB_ROUND8(x) (((x) + 7) & ~7)
int tdbGnrtFileID(const char *fname, uint8_t *fileid, bool unique); int tdbGnrtFileID(tdb_fd_t fd, uint8_t *fileid, bool unique);
int tdbGetFileSize(tdb_fd_t fd, int szPage, SPgno *size); int tdbGetFileSize(tdb_fd_t fd, int szPage, SPgno *size);
void *tdbRealloc(void *ptr, size_t size); void *tdbRealloc(void *ptr, size_t size);

35
source/os/src/osFile.c
View File

@ -110,7 +110,7 @@ void taosGetTmpfilePath(const char *inputTmpDir, const char *fileNamePrefix, cha
int64_t taosCopyFile(const char *from, const char *to) { int64_t taosCopyFile(const char *from, const char *to) {
#ifdef WINDOWS #ifdef WINDOWS
assert(0); assert(0);
return 0; return -1;
#else #else
char buffer[4096]; char buffer[4096];
int64_t size = 0; int64_t size = 0;
@ -190,15 +190,35 @@ int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime) {
return 0; return 0;
} }
int32_t taosDevInoFile(const char *path, int64_t *stDev, int64_t *stIno) { int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
if (pFile == NULL) {
return 0;
}
assert(pFile->fd >= 0); // Please check if you have closed the file.
#ifdef WINDOWS
BY_HANDLE_FILE_INFORMATION bhfi;
HANDLE handle = (HANDLE)_get_osfhandle(pFile->fd);
if (GetFileInformationByHandle(handle, &bhfi) == FALSE) {
printf("taosFStatFile get file info fail.");
return -1;
}
if (stDev != NULL) {
*stDev = (int64_t)(bhfi.dwVolumeSerialNumber);
}
if (stIno != NULL) {
*stIno = (int64_t)((((uint64_t)bhfi.nFileIndexHigh) << 32) + bhfi.nFileIndexLow);
}
#else
struct stat fileStat; struct stat fileStat;
#ifdef WINDOWS int32_t code = fstat(pFile->fd, &fileStat);
int32_t code = _stat(path, &fileStat);
#else
int32_t code = stat(path, &fileStat);
#endif
if (code < 0) { if (code < 0) {
printf("taosFStatFile run fstat fail.");
return code; return code;
} }
@ -209,6 +229,7 @@ int32_t taosDevInoFile(const char *path, int64_t *stDev, int64_t *stIno) {
if (stIno != NULL) { if (stIno != NULL) {
*stIno = fileStat.st_ino; *stIno = fileStat.st_ino;
} }
#endif
return 0; return 0;
} }

3
source/os/src/osSysinfo.c
View File

@ -744,7 +744,8 @@ int32_t taosGetSystemUUID(char *uid, int32_t uidlen) {
#ifdef WINDOWS #ifdef WINDOWS
GUID guid; GUID guid;
CoCreateGuid(&guid); CoCreateGuid(&guid);
memcpy(uid, &guid, uidlen); snprintf(uid, uidlen, "%08X-%04X-%04X-%02X%02X-%02X%02X%02X%02X%02X%02X", guid.Data1, guid.Data2, guid.Data3, guid.Data4[0],
guid.Data4[1], guid.Data4[2], guid.Data4[3], guid.Data4[4], guid.Data4[5], guid.Data4[6], guid.Data4[7]);
return 0; return 0;
#elif defined(_TD_DARWIN_64) #elif defined(_TD_DARWIN_64)

36
source/util/src/thashutil.c
View File

@ -78,6 +78,42 @@ uint32_t MurmurHash3_32(const char *key, uint32_t len) {
return h1; return h1;
} }
uint64_t MurmurHash3_64(const char *key, uint32_t len) {
const uint64_t m = 0x87c37b91114253d5;
const int r = 47;
uint32_t seed = 0x12345678;
uint64_t h = seed ^ (len * m);
const uint8_t *data = (const uint8_t *)key;
const uint8_t *end = data + (len-(len&7));
while(data != end) {
uint64_t k = *((uint64_t*)data);
k *= m;
k ^= k >> r;
k *= m;
h ^= k;
h *= m;
data += 8;
}
switch(len & 7) {
case 7: h ^= (uint64_t)data[6] << 48; /* fall-thru */
case 6: h ^= (uint64_t)data[5] << 40; /* fall-thru */
case 5: h ^= (uint64_t)data[4] << 32; /* fall-thru */
case 4: h ^= (uint64_t)data[3] << 24; /* fall-thru */
case 3: h ^= (uint64_t)data[2] << 16; /* fall-thru */
case 2: h ^= (uint64_t)data[1] << 8; /* fall-thru */
case 1: h ^= (uint64_t)data[0];
h *= m; /* fall-thru */
};
h ^= h >> r;
h *= m;
h ^= h >> r;
return h;
}
uint32_t taosIntHash_32(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint32_t *)key; } uint32_t taosIntHash_32(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint32_t *)key; }
uint32_t taosIntHash_16(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint16_t *)key; } uint32_t taosIntHash_16(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint16_t *)key; }
uint32_t taosIntHash_8(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint8_t *)key; } uint32_t taosIntHash_8(const char *key, uint32_t UNUSED_PARAM(len)) { return *(uint8_t *)key; }

12
source/util/src/tjson.c
View File

@ -183,8 +183,12 @@ int32_t tjsonGetBigIntValue(const SJson* pJson, const char* pName, int64_t* pVal
if (NULL == p) { if (NULL == p) {
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
#ifdef WINDOWS
sscanf(p,"%lld",pVal);
#else
// sscanf(p,"%ld",pVal);
*pVal = strtol(p, NULL, 10); *pVal = strtol(p, NULL, 10);
#endif
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -214,8 +218,12 @@ int32_t tjsonGetUBigIntValue(const SJson* pJson, const char* pName, uint64_t* pV
if (NULL == p) { if (NULL == p) {
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
#ifdef WINDOWS
sscanf(p,"%llu",pVal);
#else
// sscanf(p,"%ld",pVal);
*pVal = strtoul(p, NULL, 10); *pVal = strtoul(p, NULL, 10);
#endif
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }