Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1193
Star 22018
Fork 4786
T
TDengine
“8b8d7e08a46ce23cc77d295dc86a974184204592”上不存在“mobile/src/operators/feed_op.cpp”
提交 05d04dcd 编写于 作者: S Shengliang Guan
浏览文件
操作

Merge remote-tracking branch 'origin/3.0' into enh/TD-19461

上级 2953a886 12fd4bfc
隐藏空白更改
内联 并排
Showing with 6046 addition and 326 deletion
cmake/taosws_CMakeLists.txt.in
浏览文件 @ 05d04dcd
......@@ -2,7 +2,7 @@
# taosws-rs
ExternalProject_Add(taosws-rs
GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git
GIT_TAG 1bdfca3
GIT_TAG 0373a70
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
......
docs/en/12-taos-sql/05-insert.md
浏览文件 @ 05d04dcd
......@@ -108,11 +108,11 @@ INSERT INTO d21001 USING meters TAGS ('California.SanFrancisco', 2) VALUES ('202
## Insert Rows From A File
Besides using `VALUES` to insert one or multiple rows, the data to be inserted can also be prepared in a CSV file with comma as separator and each field value quoted by single quotes. Table definition is not required in the CSV file. For example, if file "/tmp/csvfile.csv" contains the below data:
Besides using `VALUES` to insert one or multiple rows, the data to be inserted can also be prepared in a CSV file with comma as separator and timestamp and string field value quoted by single quotes. Table definition is not required in the CSV file. For example, if file "/tmp/csvfile.csv" contains the below data:
```
'2021-07-13 14:07:34.630', '10.2', '219', '0.32'
'2021-07-13 14:07:35.779', '10.15', '217', '0.33'
'2021-07-13 14:07:34.630', 10.2, 219, 0.32
'2021-07-13 14:07:35.779', 10.15, 217, 0.33
```
Then data in this file can be inserted by the SQL statement below:
......
docs/en/12-taos-sql/21-node.md
浏览文件 @ 05d04dcd
......@@ -64,6 +64,12 @@ dnode_option: {
The parameters that you can modify through this statement are the same as those located in the dnode configuration file. Modifications that you make through this statement take effect immediately, while modifications to the configuration file take effect when the dnode restarts.
`value` is the value of the parameter, which needs to be in character format. For example, modify the log output level of dnode 1 to debug:
```sql
ALTER DNODE 1 'debugFlag' '143';
```
## Add an Mnode
```sql
......
docs/zh/12-taos-sql/05-insert.md
浏览文件 @ 05d04dcd
......@@ -109,11 +109,11 @@ INSERT INTO d21001 USING meters TAGS ('California.SanFrancisco', 2) VALUES ('202
## 插入来自文件的数据记录
除了使用 VALUES 关键字插入一行或多行数据外，也可以把要写入的数据放在 CSV 文件中（英文逗号分隔、英文单引号括住每个值）供 SQL 指令读取。其中 CSV 文件无需表头。例如，如果 /tmp/csvfile.csv 文件的内容为：
除了使用 VALUES 关键字插入一行或多行数据外，也可以把要写入的数据放在 CSV 文件中（英文逗号分隔、时间戳和字符串类型的值需要用英文单引号括住）供 SQL 指令读取。其中 CSV 文件无需表头。例如，如果 /tmp/csvfile.csv 文件的内容为：
```
'2021-07-13 14:07:34.630', '10.2', '219', '0.32'
'2021-07-13 14:07:35.779', '10.15', '217', '0.33'
'2021-07-13 14:07:34.630', 10.2, 219, 0.32
'2021-07-13 14:07:35.779', 10.15, 217, 0.33
```
那么通过如下指令可以把这个文件中的数据写入子表中：
......
docs/zh/12-taos-sql/21-node.md
浏览文件 @ 05d04dcd
......@@ -65,6 +65,12 @@ dnode_option: {
上面语法中的这些可修改配置项其配置方式与 dnode 配置文件中的配置方式相同，区别是修改是动态的立即生效，且不需要重启 dnode。
value 是参数的值，需要是字符格式。如修改 dnode 1 的日志输出级别为 debug：
```sql
ALTER DNODE 1 'debugFlag' '143';
```
## 添加管理节点
```sql
......
include/client/taos.h
浏览文件 @ 05d04dcd
......@@ -126,6 +126,22 @@ typedef struct setConfRet {
char retMsg[RET_MSG_LENGTH];
} setConfRet;
typedef struct TAOS_VGROUP_HASH_INFO {
int32_t vgId;
uint32_t hashBegin;
uint32_t hashEnd;
} TAOS_VGROUP_HASH_INFO ;
typedef struct TAOS_DB_ROUTE_INFO {
int32_t routeVersion;
int16_t hashPrefix;
int16_t hashSuffix;
int8_t hashMethod;
int32_t vgNum;
TAOS_VGROUP_HASH_INFO *vgHash;
} TAOS_DB_ROUTE_INFO ;
DLL_EXPORT void taos_cleanup(void);
DLL_EXPORT int taos_options(TSDB_OPTION option, const void *arg, ...);
DLL_EXPORT setConfRet taos_set_config(const char *config);
......@@ -196,6 +212,9 @@ DLL_EXPORT void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, vo
DLL_EXPORT void taos_fetch_raw_block_a(TAOS_RES *res, __taos_async_fn_t fp, void *param);
DLL_EXPORT const void *taos_get_raw_block(TAOS_RES *res);
DLL_EXPORT int taos_get_db_route_info(TAOS* taos, const char* db, TAOS_DB_ROUTE_INFO* dbInfo);
DLL_EXPORT int taos_get_table_vgId(TAOS* taos, const char* db, const char* table, int* vgId);
DLL_EXPORT int taos_load_table_info(TAOS *taos, const char *tableNameList);
DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision);
DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw(TAOS* taos, char* lines, int len, int32_t *totalRows, int protocol, int precision);
......
include/common/tcommon.h
浏览文件 @ 05d04dcd
......@@ -83,6 +83,27 @@ typedef struct {
int32_t exprIdx;
} STupleKey;
typedef struct STuplePos {
union {
struct {
int32_t pageId;
int32_t offset;
};
STupleKey streamTupleKey;
};
} STuplePos;
typedef struct SFirstLastRes {
bool hasResult;
// used for last_row function only, isNullRes in SResultRowEntry can not be passed to downstream.So,
// this attribute is required
bool isNull;
int32_t bytes;
int64_t ts;
STuplePos pos;
char buf[];
} SFirstLastRes;
static inline int STupleKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
STupleKey* pTuple1 = (STupleKey*)pKey1;
STupleKey* pTuple2 = (STupleKey*)pKey2;
......
include/libs/catalog/catalog.h
浏览文件 @ 05d04dcd
......@@ -163,7 +163,9 @@ int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* vers
* @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
* @return error code
*/
int32_t catalogGetDBVgInfo(SCatalog* pCatalog, SRequestConnInfo* pConn, const char* pDBName, SArray** pVgroupList);
int32_t catalogGetDBVgList(SCatalog* pCatalog, SRequestConnInfo* pConn, const char* pDBName, SArray** pVgroupList);
int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, TAOS_DB_ROUTE_INFO* pInfo);
int32_t catalogUpdateDBVgInfo(SCatalog* pCatalog, const char* dbName, uint64_t dbId, SDBVgInfo* dbInfo);
......
include/libs/function/functionMgt.h
浏览文件 @ 05d04dcd
......@@ -129,6 +129,7 @@ typedef enum EFunctionType {
FUNCTION_TYPE_TO_COLUMN,
FUNCTION_TYPE_GROUP_KEY,
FUNCTION_TYPE_CACHE_LAST_ROW,
FUNCTION_TYPE_CACHE_LAST,
// distributed splitting functions
FUNCTION_TYPE_APERCENTILE_PARTIAL = 4000,
......@@ -216,6 +217,8 @@ bool fmIsKeepOrderFunc(int32_t funcId);
bool fmIsCumulativeFunc(int32_t funcId);
bool fmIsInterpPseudoColumnFunc(int32_t funcId);
void getLastCacheDataType(SDataType* pType);
int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);
typedef enum EFuncDataRequired {
......
source/client/src/clientImpl.c
浏览文件 @ 05d04dcd
......@@ -643,7 +643,7 @@ int32_t buildSyncExecNodeList(SRequestObj* pRequest, SArray** pNodeList, SArray*
.requestObjRefId = pRequest->self,
.mgmtEps = getEpSet_s(&pInst->mgmtEp)};
code = catalogGetDBVgInfo(pCtg, &conn, dbFName, &pVgList);
code = catalogGetDBVgList(pCtg, &conn, dbFName, &pVgList);
if (code) {
goto _return;
}
......
source/client/src/clientMain.c
浏览文件 @ 05d04dcd
......@@ -989,6 +989,106 @@ const void *taos_get_raw_block(TAOS_RES *res) {
return pRequest->body.resInfo.pData;
}
int taos_get_db_route_info(TAOS* taos, const char* db, TAOS_DB_ROUTE_INFO* dbInfo) {
if (NULL == taos) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
return terrno;
}
if (NULL == db || NULL == dbInfo) {
tscError("invalid input param, db:%p, dbInfo:%p", db, dbInfo);
terrno = TSDB_CODE_TSC_INVALID_INPUT;
return terrno;
}
int64_t connId = *(int64_t *)taos;
SRequestObj *pRequest = NULL;
char *sql = "taos_get_db_route_info";
int32_t code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
return terrno;
}
STscObj *pTscObj = pRequest->pTscObj;
SCatalog *pCtg = NULL;
code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
if (code != TSDB_CODE_SUCCESS) {
goto _return;
}
SRequestConnInfo conn = {
.pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
char dbFName[TSDB_DB_FNAME_LEN] = {0};
snprintf(dbFName, sizeof(dbFName), "%d.%s", pTscObj->acctId, db);
code = catalogGetDBVgInfo(pCtg, &conn, dbFName, dbInfo);
if (code) {
goto _return;
}
_return:
terrno = code;
destroyRequest(pRequest);
return code;
}
int taos_get_table_vgId(TAOS* taos, const char* db, const char* table, int* vgId) {
if (NULL == taos) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
return terrno;
}
if (NULL == db || NULL == table || NULL == vgId) {
tscError("invalid input param, db:%p, table:%p, vgId:%p", db, table, vgId);
terrno = TSDB_CODE_TSC_INVALID_INPUT;
return terrno;
}
int64_t connId = *(int64_t *)taos;
SRequestObj *pRequest = NULL;
char *sql = "taos_get_table_vgId";
int32_t code = buildRequest(connId, sql, strlen(sql), NULL, false, &pRequest);
if (code != TSDB_CODE_SUCCESS) {
return terrno;
}
STscObj *pTscObj = pRequest->pTscObj;
SCatalog *pCtg = NULL;
code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCtg);
if (code != TSDB_CODE_SUCCESS) {
goto _return;
}
SRequestConnInfo conn = {
.pTrans = pTscObj->pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
SName tableName;
toName(pTscObj->acctId, db, table, &tableName);
SVgroupInfo vgInfo;
code = catalogGetTableHashVgroup(pCtg, &conn, &tableName, &vgInfo);
if (code) {
goto _return;
}
*vgId = vgInfo.vgId;
_return:
terrno = code;
destroyRequest(pRequest);
return code;
}
int taos_load_table_info(TAOS *taos, const char *tableNameList) {
if (NULL == taos) {
terrno = TSDB_CODE_TSC_DISCONNECTED;
......
source/client/src/clientTmq.c
浏览文件 @ 05d04dcd
......@@ -979,6 +979,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
const SArray* container = &topic_list->container;
int32_t sz = taosArrayGetSize(container);
void* buf = NULL;
SMsgSendInfo* sendInfo = NULL;
SCMSubscribeReq req = {0};
int32_t code = -1;
......@@ -1016,7 +1017,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
void* abuf = buf;
tSerializeSCMSubscribeReq(&abuf, &req);
SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
if (sendInfo == NULL) goto FAIL;
SMqSubscribeCbParam param = {
......@@ -1046,6 +1047,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
// avoid double free if msg is sent
buf = NULL;
sendInfo = NULL;
tsem_wait(&param.rspSem);
tsem_destroy(&param.rspSem);
......@@ -1078,8 +1080,9 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
code = 0;
FAIL:
if (req.topicNames != NULL) taosArrayDestroyP(req.topicNames, taosMemoryFree);
taosArrayDestroyP(req.topicNames, taosMemoryFree);
taosMemoryFree(buf);
taosMemoryFree(sendInfo);
return code;
}
......@@ -1613,7 +1616,11 @@ void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
if (rspWrapper == NULL) {
taosReadAllQitems(tmq->mqueue, tmq->qall);
taosGetQitem(tmq->qall, (void**)&rspWrapper);
if (rspWrapper == NULL) return NULL;
if (rspWrapper == NULL) {
tscDebug("consumer %" PRId64 " mqueue empty", tmq->consumerId);
return NULL;
}
}
if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__END_RSP) {
......@@ -1711,6 +1718,8 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
void* rspObj;
int64_t startTime = taosGetTimestampMs();
tscDebug("consumer:%" PRId64 ", start poll at %" PRId64, tmq->consumerId, startTime);
#if 0
tmqHandleAllDelayedTask(tmq);
tmqPollImpl(tmq, timeout);
......@@ -1745,15 +1754,18 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
rspObj = tmqHandleAllRsp(tmq, timeout, false);
if (rspObj) {
tscDebug("consumer:%" PRId64 ", return rsp", tmq->consumerId);
return (TAOS_RES*)rspObj;
} else if (terrno == TSDB_CODE_TQ_NO_COMMITTED_OFFSET) {
tscDebug("consumer:%" PRId64 ", return null since no committed offset", tmq->consumerId);
return NULL;
}
if (timeout != -1) {
int64_t endTime = taosGetTimestampMs();
int64_t leftTime = endTime - startTime;
if (leftTime > timeout) {
tscDebug("consumer:%" PRId64 ", (epoch %d) timeout, no rsp", tmq->consumerId, tmq->epoch);
tscDebug("consumer:%" PRId64 ", (epoch %d) timeout, no rsp, start time %" PRId64 ", end time %" PRId64,
tmq->consumerId, tmq->epoch, startTime, endTime);
return NULL;
}
tsem_timewait(&tmq->rspSem, leftTime * 1000);
......
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ 05d04dcd
......@@ -299,6 +299,11 @@ int32_t tsdbMerge(STsdb *pTsdb);
#define TSDB_CACHE_LAST(c) (((c).cacheLast & 2) > 0)
// tsdbCache ==============================================================================================
typedef struct {
TSKEY ts;
SColVal colVal;
} SLastCol;
int32_t tsdbOpenCache(STsdb *pTsdb);
void tsdbCloseCache(STsdb *pTsdb);
int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb *pTsdb);
......
source/dnode/vnode/src/tsdb/tsdbCache.c
浏览文件 @ 05d04dcd
......@@ -15,11 +15,6 @@
#include "tsdb.h"
typedef struct {
TSKEY ts;
SColVal colVal;
} SLastCol;
int32_t tsdbOpenCache(STsdb *pTsdb) {
int32_t code = 0;
SLRUCache *pCache = NULL;
......@@ -61,9 +56,18 @@ static void getTableCacheKey(tb_uid_t uid, int cacheType, char *key, int *len) {
*len = sizeof(uint64_t);
}
static void deleteTableCacheLastrow(const void *key, size_t keyLen, void *value) { taosMemoryFree(value); }
static void deleteTableCacheLast(const void *key, size_t keyLen, void *value) {
SArray *pLastArray = (SArray *)value;
int16_t nCol = taosArrayGetSize(pLastArray);
for (int16_t iCol = 0; iCol < nCol; ++iCol) {
SLastCol *pLastCol = (SLastCol *)taosArrayGet(pLastArray, iCol);
if (IS_VAR_DATA_TYPE(pLastCol->colVal.type) && pLastCol->colVal.value.nData > 0) {
taosMemoryFree(pLastCol->colVal.value.pData);
}
}
static void deleteTableCacheLast(const void *key, size_t keyLen, void *value) { taosArrayDestroy(value); }
taosArrayDestroy(value);
}
int32_t tsdbCacheDeleteLastrow(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey) {
int32_t code = 0;
......@@ -75,13 +79,23 @@ int32_t tsdbCacheDeleteLastrow(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey) {
getTableCacheKey(uid, 0, key, &keyLen);
LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen);
if (h) {
STSRow *pRow = (STSRow *)taosLRUCacheValue(pCache, h);
if (pRow->ts <= eKey) {
SArray *pLast = (SArray *)taosLRUCacheValue(pCache, h);
bool invalidate = false;
int16_t nCol = taosArrayGetSize(pLast);
for (int16_t iCol = 0; iCol < nCol; ++iCol) {
SLastCol *tTsVal = (SLastCol *)taosArrayGet(pLast, iCol);
if (eKey >= tTsVal->ts) {
invalidate = true;
break;
}
}
if (invalidate) {
taosLRUCacheRelease(pCache, h, true);
} else {
taosLRUCacheRelease(pCache, h, false);
}
// void taosLRUCacheErase(SLRUCache * cache, const void *key, size_t keyLen);
}
......@@ -130,14 +144,23 @@ int32_t tsdbCacheDelete(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey) {
getTableCacheKey(uid, 0, key, &keyLen);
LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen);
if (h) {
STSRow *pRow = (STSRow *)taosLRUCacheValue(pCache, h);
if (pRow->ts <= eKey) {
SArray *pLast = (SArray *)taosLRUCacheValue(pCache, h);
bool invalidate = false;
int16_t nCol = taosArrayGetSize(pLast);
for (int16_t iCol = 0; iCol < nCol; ++iCol) {
SLastCol *tTsVal = (SLastCol *)taosArrayGet(pLast, iCol);
if (eKey >= tTsVal->ts) {
invalidate = true;
break;
}
}
if (invalidate) {
taosLRUCacheRelease(pCache, h, true);
} else {
taosLRUCacheRelease(pCache, h, false);
}
// void taosLRUCacheErase(SLRUCache * cache, const void *key, size_t keyLen);
}
// getTableCacheKey(uid, "l", key, &keyLen);
......@@ -177,6 +200,46 @@ int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, ST
getTableCacheKey(uid, 0, key, &keyLen);
LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen);
if (h) {
STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, uid, -1, 1);
TSKEY keyTs = row->ts;
bool invalidate = false;
SArray *pLast = (SArray *)taosLRUCacheValue(pCache, h);
int16_t nCol = taosArrayGetSize(pLast);
int16_t iCol = 0;
SLastCol *tTsVal = (SLastCol *)taosArrayGet(pLast, iCol);
if (keyTs > tTsVal->ts) {
STColumn *pTColumn = &pTSchema->columns[0];
SColVal tColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, (SValue){.val = keyTs});
taosArraySet(pLast, iCol, &(SLastCol){.ts = keyTs, .colVal = tColVal});
}
for (++iCol; iCol < nCol; ++iCol) {
SLastCol *tTsVal1 = (SLastCol *)taosArrayGet(pLast, iCol);
if (keyTs >= tTsVal1->ts) {
SColVal *tColVal = &tTsVal1->colVal;
SColVal colVal = {0};
tTSRowGetVal(row, pTSchema, iCol, &colVal);
if (!COL_VAL_IS_NONE(&colVal)) {
if (keyTs == tTsVal1->ts && !COL_VAL_IS_NONE(tColVal)) {
invalidate = true;
break;
}
} else {
taosArraySet(pLast, iCol, &(SLastCol){.ts = keyTs, .colVal = colVal});
}
}
}
_invalidate:
taosMemoryFreeClear(pTSchema);
taosLRUCacheRelease(pCache, h, invalidate);
/*
cacheRow = (STSRow *)taosLRUCacheValue(pCache, h);
if (row->ts >= cacheRow->ts) {
if (row->ts == cacheRow->ts) {
......@@ -218,9 +281,9 @@ int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, ST
if (status != TAOS_LRU_STATUS_OK) {
code = -1;
}
/* tsdbCacheInsertLastrow(pCache, uid, row, dup); */
// tsdbCacheInsertLastrow(pCache, uid, row, dup);
}
}
}*/
} /*else {
if (dup) {
cacheRow = tdRowDup(row);
......@@ -456,6 +519,11 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) {
return code;
}
if (state->pDataFReader != NULL) {
tsdbDataFReaderClose(&state->pDataFReader);
state->pDataFReader = NULL;
}
code = tsdbDataFReaderOpen(&state->pDataFReader, state->pTsdb, pFileSet);
if (code) goto _err;
......@@ -599,6 +667,8 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) {
*/
state->pBlockIdx = taosArraySearch(state->aBlockIdx, state->pBlockIdxExp, tCmprBlockIdx, TD_EQ);
if (!state->pBlockIdx) {
tsdbDataFReaderClose(&state->pDataFReader);
state->pDataFReader = NULL;
goto _next_fileset;
}
......@@ -1049,7 +1119,7 @@ _err:
return code;
}
static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRow) {
static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, SArray **ppColArray) {
int32_t code = 0;
STSchema *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, uid, -1, 1);
......@@ -1057,7 +1127,7 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRo
int16_t iCol = 0;
int16_t noneCol = 0;
bool setNoneCol = false;
SArray *pColArray = taosArrayInit(nCol, sizeof(SColVal));
SArray *pColArray = taosArrayInit(nCol, sizeof(SLastCol));
SColVal *pColVal = &(SColVal){0};
TSKEY lastRowTs = TSKEY_MAX;
......@@ -1073,12 +1143,15 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRo
break;
}
TSKEY rowTs = TSDBROW_TS(pRow);
if (lastRowTs == TSKEY_MAX) {
lastRowTs = TSDBROW_TS(pRow);
lastRowTs = rowTs;
STColumn *pTColumn = &pTSchema->columns[0];
*pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, (SValue){.val = lastRowTs});
if (taosArrayPush(pColArray, pColVal) == NULL) {
if (taosArrayPush(pColArray, &(SLastCol){.ts = lastRowTs, .colVal = *pColVal}) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
......@@ -1086,7 +1159,18 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRo
for (iCol = 1; iCol < nCol; ++iCol) {
tsdbRowGetColVal(pRow, pTSchema, iCol, pColVal);
if (taosArrayPush(pColArray, pColVal) == NULL) {
SLastCol lastCol = {.ts = lastRowTs, .colVal = *pColVal};
if (IS_VAR_DATA_TYPE(pColVal->type) && pColVal->value.nData > 0) {
lastCol.colVal.value.pData = taosMemoryMalloc(lastCol.colVal.value.nData);
if (lastCol.colVal.value.pData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
memcpy(lastCol.colVal.value.pData, pColVal->value.pData, pColVal->value.nData);
}
if (taosArrayPush(pColArray, &lastCol) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
......@@ -1097,15 +1181,15 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRo
}
}
if (!setNoneCol) {
// goto build the result ts row
// done, goto return pColArray
break;
} else {
continue;
}
}
if ((TSDBROW_TS(pRow) < lastRowTs)) {
// goto build the result ts row
if ((rowTs < lastRowTs)) {
// done, goto return pColArray
break;
}
......@@ -1117,7 +1201,21 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRo
tsdbRowGetColVal(pRow, pTSchema, iCol, pColVal);
if (COL_VAL_IS_NONE(tColVal) && !COL_VAL_IS_NONE(pColVal)) {
taosArraySet(pColArray, iCol, pColVal);
SLastCol lastCol = {.ts = rowTs, .colVal = *pColVal};
if (IS_VAR_DATA_TYPE(pColVal->type) && pColVal->value.nData > 0) {
SLastCol *pLastCol = (SLastCol *)taosArrayGet(pColArray, iCol);
taosMemoryFree(pLastCol->colVal.value.pData);
lastCol.colVal.value.pData = taosMemoryMalloc(lastCol.colVal.value.nData);
if (lastCol.colVal.value.pData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
memcpy(lastCol.colVal.value.pData, pColVal->value.pData, pColVal->value.nData);
}
taosArraySet(pColArray, iCol, &lastCol);
} else if (COL_VAL_IS_NONE(tColVal) && COL_VAL_IS_NONE(pColVal) && !setNoneCol) {
noneCol = iCol;
setNoneCol = true;
......@@ -1127,15 +1225,14 @@ static int32_t mergeLastRow(tb_uid_t uid, STsdb *pTsdb, bool *dup, STSRow **ppRo
// build the result ts row here
*dup = false;
if (taosArrayGetSize(pColArray) == nCol) {
code = tdSTSRowNew(pColArray, pTSchema, ppRow);
if (code) goto _err;
if (taosArrayGetSize(pColArray) != nCol) {
*ppColArray = NULL;
taosArrayDestroy(pColArray);
} else {
*ppRow = NULL;
*ppColArray = pColArray;
}
nextRowIterClose(&iter);
taosArrayDestroy(pColArray);
taosMemoryFreeClear(pTSchema);
return code;
......@@ -1185,7 +1282,18 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) {
for (iCol = 1; iCol < nCol; ++iCol) {
tsdbRowGetColVal(pRow, pTSchema, iCol, pColVal);
if (taosArrayPush(pColArray, &(SLastCol){.ts = lastRowTs, .colVal = *pColVal}) == NULL) {
SLastCol lastCol = {.ts = lastRowTs, .colVal = *pColVal};
if (IS_VAR_DATA_TYPE(pColVal->type) && pColVal->value.nData > 0) {
lastCol.colVal.value.pData = taosMemoryMalloc(lastCol.colVal.value.nData);
if (lastCol.colVal.value.pData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
memcpy(lastCol.colVal.value.pData, pColVal->value.pData, pColVal->value.nData);
}
if (taosArrayPush(pColArray, &lastCol) == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
......@@ -1196,7 +1304,7 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) {
}
}
if (!setNoneCol) {
// goto build the result ts row
// done, goto return pColArray
break;
} else {
continue;
......@@ -1207,11 +1315,26 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) {
setNoneCol = false;
for (iCol = noneCol; iCol < nCol; ++iCol) {
// high version's column value
SColVal *tColVal = (SColVal *)taosArrayGet(pColArray, iCol);
SLastCol *lastColVal = (SLastCol *)taosArrayGet(pColArray, iCol);
SColVal *tColVal = &lastColVal->colVal;
tsdbRowGetColVal(pRow, pTSchema, iCol, pColVal);
if (!COL_VAL_IS_VALUE(tColVal) && COL_VAL_IS_VALUE(pColVal)) {
taosArraySet(pColArray, iCol, &(SLastCol){.ts = rowTs, .colVal = *pColVal});
SLastCol lastCol = {.ts = rowTs, .colVal = *pColVal};
if (IS_VAR_DATA_TYPE(pColVal->type) && pColVal->value.nData > 0) {
SLastCol *pLastCol = (SLastCol *)taosArrayGet(pColArray, iCol);
taosMemoryFree(pLastCol->colVal.value.pData);
lastCol.colVal.value.pData = taosMemoryMalloc(lastCol.colVal.value.nData);
if (lastCol.colVal.value.pData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
memcpy(lastCol.colVal.value.pData, pColVal->value.pData, pColVal->value.nData);
}
taosArraySet(pColArray, iCol, &lastCol);
} else if (!COL_VAL_IS_VALUE(tColVal) && !COL_VAL_IS_VALUE(pColVal) && !setNoneCol) {
noneCol = iCol;
setNoneCol = true;
......@@ -1219,7 +1342,6 @@ static int32_t mergeLast(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray) {
}
} while (setNoneCol);
// build the result ts row here
if (taosArrayGetSize(pColArray) <= 0) {
*ppLastArray = NULL;
taosArrayDestroy(pColArray);
......@@ -1252,13 +1374,13 @@ int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUH
h = taosLRUCacheLookup(pCache, key, keyLen);
if (!h) {
STSRow *pRow = NULL;
SArray *pArray = NULL;
bool dup = false; // which is always false for now
code = mergeLastRow(uid, pTsdb, &dup, &pRow);
code = mergeLastRow(uid, pTsdb, &dup, &pArray);
// if table's empty or error, return code of -1
if (code < 0 || pRow == NULL) {
if (!dup && pRow) {
taosMemoryFree(pRow);
if (code < 0 || pArray == NULL) {
if (!dup && pArray) {
taosArrayDestroy(pArray);
}
taosThreadMutexUnlock(&pTsdb->lruMutex);
......@@ -1268,9 +1390,9 @@ int32_t tsdbCacheGetLastrowH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUH
return 0;
}
_taos_lru_deleter_t deleter = deleteTableCacheLastrow;
LRUStatus status =
taosLRUCacheInsert(pCache, key, keyLen, pRow, TD_ROW_LEN(pRow), deleter, NULL, TAOS_LRU_PRIORITY_LOW);
size_t charge = pArray->capacity * pArray->elemSize + sizeof(*pArray);
_taos_lru_deleter_t deleter = deleteTableCacheLast;
LRUStatus status = taosLRUCacheInsert(pCache, key, keyLen, pArray, charge, deleter, NULL, TAOS_LRU_PRIORITY_LOW);
if (status != TAOS_LRU_STATUS_OK) {
code = -1;
}
......@@ -1328,15 +1450,17 @@ int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUHand
SArray *pLastArray = NULL;
code = mergeLast(uid, pTsdb, &pLastArray);
// if table's empty or error, return code of -1
// if (code < 0 || pRow == NULL) {
if (code < 0 || pLastArray == NULL) {
taosThreadMutexUnlock(&pTsdb->lruMutex);
*handle = NULL;
return 0;
}
size_t charge = pLastArray->capacity * pLastArray->elemSize + sizeof(*pLastArray);
_taos_lru_deleter_t deleter = deleteTableCacheLast;
LRUStatus status = taosLRUCacheInsert(pCache, key, keyLen, pLastArray, pLastArray->capacity, deleter, NULL,
TAOS_LRU_PRIORITY_LOW);
LRUStatus status =
taosLRUCacheInsert(pCache, key, keyLen, pLastArray, charge, deleter, NULL, TAOS_LRU_PRIORITY_LOW);
if (status != TAOS_LRU_STATUS_OK) {
code = -1;
}
......
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
浏览文件 @ 05d04dcd
......@@ -29,31 +29,71 @@ typedef struct SCacheRowsReader {
SArray* pTableList; // table id list
} SCacheRowsReader;
static void saveOneRow(STSRow* pRow, SSDataBlock* pBlock, SCacheRowsReader* pReader, const int32_t* slotIds) {
#define HASTYPE(_type, _t) (((_type) & (_t)) == (_t))
static void saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* pReader, const int32_t* slotIds,
void** pRes) {
ASSERT(pReader->numOfCols <= taosArrayGetSize(pBlock->pDataBlock));
int32_t numOfRows = pBlock->info.rows;
SColVal colVal = {0};
for (int32_t i = 0; i < pReader->numOfCols; ++i) {
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST)) {
for (int32_t i = 0; i < pReader->numOfCols; ++i) {
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
SFirstLastRes* p = (SFirstLastRes*)varDataVal(pRes[i]);
if (slotIds[i] == -1) {
colDataAppend(pColInfoData, numOfRows, (const char*)&pRow->ts, false);
} else {
int32_t slotId = slotIds[i];
if (slotIds[i] == -1) { // the primary timestamp
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, 0);
p->ts = pColVal->ts;
p->bytes = TSDB_KEYSIZE;
*(int64_t*)p->buf = pColVal->ts;
} else {
int32_t slotId = slotIds[i];
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, slotId);
p->ts = pColVal->ts;
p->isNull = !COL_VAL_IS_VALUE(&pColVal->colVal);
if (!p->isNull) {
if (IS_VAR_DATA_TYPE(pColVal->colVal.type)) {
varDataSetLen(p->buf, pColVal->colVal.value.nData);
memcpy(varDataVal(p->buf), pColVal->colVal.value.pData, pColVal->colVal.value.nData);
p->bytes = pColVal->colVal.value.nData + VARSTR_HEADER_SIZE; // binary needs to plus the header size
} else {
memcpy(p->buf, &pColVal->colVal.value, pReader->pSchema->columns[slotId].bytes);
p->bytes = pReader->pSchema->columns[slotId].bytes;
}
}
}
tTSRowGetVal(pRow, pReader->pSchema, slotId, &colVal);
// pColInfoData->info.bytes includes the VARSTR_HEADER_SIZE, need to substruct it
p->hasResult = true;
varDataSetLen(pRes[i], pColInfoData->info.bytes - VARSTR_HEADER_SIZE);
colDataAppend(pColInfoData, numOfRows, (const char*)pRes[i], false);
}
} else {
ASSERT(HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST_ROW));
if (IS_VAR_DATA_TYPE(colVal.type)) {
if (!COL_VAL_IS_VALUE(&colVal)) {
colDataAppendNULL(pColInfoData, numOfRows);
for (int32_t i = 0; i < pReader->numOfCols; ++i) {
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
if (slotIds[i] == -1) {
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, 0);
colDataAppend(pColInfoData, numOfRows, (const char*)&pColVal->ts, false);
} else {
int32_t slotId = slotIds[i];
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, slotId);
SColVal* pVal = &pColVal->colVal;
if (IS_VAR_DATA_TYPE(pColVal->colVal.type)) {
if (!COL_VAL_IS_VALUE(&pColVal->colVal)) {
colDataAppendNULL(pColInfoData, numOfRows);
} else {
varDataSetLen(pReader->transferBuf[slotId], pVal->value.nData);
memcpy(varDataVal(pReader->transferBuf[slotId]), pVal->value.pData, pVal->value.nData);
colDataAppend(pColInfoData, numOfRows, pReader->transferBuf[slotId], false);
}
} else {
varDataSetLen(pReader->transferBuf[slotId], colVal.value.nData);
memcpy(varDataVal(pReader->transferBuf[slotId]), colVal.value.pData, colVal.value.nData);
colDataAppend(pColInfoData, numOfRows, pReader->transferBuf[slotId], false);
colDataAppend(pColInfoData, numOfRows, (const char*)&pVal->value.val, !COL_VAL_IS_VALUE(pVal));
}
} else {
colDataAppend(pColInfoData, numOfRows, (const char*)&colVal.value, !COL_VAL_IS_VALUE(&colVal));
}
}
}
......@@ -118,35 +158,36 @@ void* tsdbCacherowsReaderClose(void* pReader) {
return NULL;
}
static int32_t doExtractCacheRow(SCacheRowsReader* pr, SLRUCache* lruCache, uint64_t uid, STSRow** pRow,
static int32_t doExtractCacheRow(SCacheRowsReader* pr, SLRUCache* lruCache, uint64_t uid, SArray** pRow,
LRUHandle** h) {
int32_t code = TSDB_CODE_SUCCESS;
if ((pr->type & CACHESCAN_RETRIEVE_LAST_ROW) == CACHESCAN_RETRIEVE_LAST_ROW) {
code = tsdbCacheGetLastrowH(lruCache, uid, pr->pVnode->pTsdb, h);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
*pRow = NULL;
// no data in the table of Uid
if (*h != NULL) {
*pRow = (STSRow*)taosLRUCacheValue(lruCache, *h);
}
if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST_ROW)) {
code = tsdbCacheGetLastrowH(lruCache, uid, pr->pVnode->pTsdb, h);
} else {
code = tsdbCacheGetLastH(lruCache, uid, pr->pVnode->pTsdb, h);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
}
// no data in the table of Uid
if (*h != NULL) {
SArray* pLast = (SArray*)taosLRUCacheValue(lruCache, *h);
tsdbCacheLastArray2Row(pLast, pRow, pr->pSchema);
}
if (code != TSDB_CODE_SUCCESS) {
return code;
}
// no data in the table of Uid
if (*h != NULL) {
*pRow = (SArray*)taosLRUCacheValue(lruCache, *h);
}
return code;
}
static void freeItem(void* pItem) {
SLastCol* pCol = (SLastCol*) pItem;
if (IS_VAR_DATA_TYPE(pCol->colVal.type)) {
taosMemoryFree(pCol->colVal.value.pData);
}
}
int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32_t* slotIds, SArray* pTableUidList) {
if (pReader == NULL || pResBlock == NULL) {
return TSDB_CODE_INVALID_PARA;
......@@ -157,13 +198,41 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
int32_t code = TSDB_CODE_SUCCESS;
SLRUCache* lruCache = pr->pVnode->pTsdb->lruCache;
LRUHandle* h = NULL;
STSRow* pRow = NULL;
SArray* pRow = NULL;
size_t numOfTables = taosArrayGetSize(pr->pTableList);
bool hasRes = false;
SArray* pLastCols = NULL;
void** pRes = taosMemoryCalloc(pr->numOfCols, POINTER_BYTES);
if (pRes == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _end;
}
for (int32_t j = 0; j < pr->numOfCols; ++j) {
pRes[j] = taosMemoryCalloc(1, sizeof(SFirstLastRes) + pr->pSchema->columns[slotIds[j]].bytes + VARSTR_HEADER_SIZE);
SFirstLastRes* p = (SFirstLastRes*)varDataVal(pRes[j]);
p->ts = INT64_MIN;
}
pLastCols = taosArrayInit(pr->pSchema->numOfCols, sizeof(SLastCol));
if (pLastCols == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _end;
}
for (int32_t i = 0; i < pr->pSchema->numOfCols; ++i) {
struct STColumn* pCol = &pr->pSchema->columns[i];
SLastCol p = {.ts = INT64_MIN, .colVal.type = pCol->type};
if (IS_VAR_DATA_TYPE(pCol->type)) {
p.colVal.value.pData = taosMemoryCalloc(pCol->bytes, sizeof(char));
}
taosArrayPush(pLastCols, &p);
}
// retrieve the only one last row of all tables in the uid list.
if ((pr->type & CACHESCAN_RETRIEVE_TYPE_SINGLE) == CACHESCAN_RETRIEVE_TYPE_SINGLE) {
int64_t lastKey = INT64_MIN;
bool internalResult = false;
if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) {
for (int32_t i = 0; i < numOfTables; ++i) {
STableKeyInfo* pKeyInfo = taosArrayGet(pr->pTableList, i);
......@@ -176,23 +245,59 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
continue;
}
if (pRow->ts > lastKey) {
// Set result row into the same rowIndex repeatly, so we need to check if the internal result row has already
// appended or not.
if (internalResult) {
pResBlock->info.rows -= 1;
taosArrayClear(pTableUidList);
{
for (int32_t k = 0; k < pr->numOfCols; ++k) {
int32_t slotId = slotIds[k];
if (slotId == -1) { // the primary timestamp
SLastCol* p = taosArrayGet(pLastCols, 0);
SLastCol* pCol = (SLastCol*)taosArrayGet(pRow, 0);
if (pCol->ts > p->ts) {
hasRes = true;
p->ts = pCol->ts;
p->colVal = pCol->colVal;
// only set value for last row query
if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST_ROW)) {
if (taosArrayGetSize(pTableUidList) == 0) {
taosArrayPush(pTableUidList, &pKeyInfo->uid);
} else {
taosArraySet(pTableUidList, 0, &pKeyInfo->uid);
}
}
}
} else {
SLastCol* p = taosArrayGet(pLastCols, slotId);
SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, slotId);
if (pColVal->ts > p->ts) {
if (!COL_VAL_IS_VALUE(&pColVal->colVal) && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
continue;
}
hasRes = true;
p->ts = pColVal->ts;
uint8_t* px = p->colVal.value.pData;
p->colVal = pColVal->colVal;
if (COL_VAL_IS_VALUE(&pColVal->colVal) && IS_VAR_DATA_TYPE(pColVal->colVal.type)) {
p->colVal.value.pData = px;
memcpy(px, pColVal->colVal.value.pData, pColVal->colVal.value.nData);
}
}
}
}
saveOneRow(pRow, pResBlock, pr, slotIds);
taosArrayPush(pTableUidList, &pKeyInfo->uid);
internalResult = true;
lastKey = pRow->ts;
}
tsdbCacheRelease(lruCache, h);
}
} else if ((pr->type & CACHESCAN_RETRIEVE_TYPE_ALL) == CACHESCAN_RETRIEVE_TYPE_ALL) {
if (hasRes) {
saveOneRow(pLastCols, pResBlock, pr, slotIds, pRes);
}
} else if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_ALL)) {
for (int32_t i = pr->tableIndex; i < numOfTables; ++i) {
STableKeyInfo* pKeyInfo = (STableKeyInfo*) taosArrayGet(pr->pTableList, i);
code = doExtractCacheRow(pr, lruCache, pKeyInfo->uid, &pRow, &h);
......@@ -204,19 +309,27 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
continue;
}
saveOneRow(pRow, pResBlock, pr, slotIds);
taosArrayPush(pTableUidList, &pKeyInfo->uid);
saveOneRow(pRow, pResBlock, pr, slotIds, pRes);
// TODO reset the pRes
taosArrayPush(pTableUidList, &pKeyInfo->uid);
tsdbCacheRelease(lruCache, h);
pr->tableIndex += 1;
if (pResBlock->info.rows >= pResBlock->info.capacity) {
return TSDB_CODE_SUCCESS;
goto _end;
}
}
} else {
return TSDB_CODE_INVALID_PARA;
code = TSDB_CODE_INVALID_PARA;
}
return TSDB_CODE_SUCCESS;
_end:
for (int32_t j = 0; j < pr->numOfCols; ++j) {
taosMemoryFree(pRes[j]);
}
taosMemoryFree(pRes);
taosArrayDestroyEx(pLastCols, freeItem);
return code;
}
source/libs/catalog/src/catalog.c
浏览文件 @ 05d04dcd
......@@ -739,7 +739,7 @@ _return:
CTG_API_LEAVE(code);
}
int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, SArray** vgroupList) {
int32_t catalogGetDBVgList(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, SArray** vgroupList) {
CTG_API_ENTER();
if (NULL == pCtg || NULL == dbFName || NULL == pConn || NULL == vgroupList) {
......@@ -778,6 +778,64 @@ _return:
CTG_API_LEAVE(code);
}
int32_t catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, TAOS_DB_ROUTE_INFO* pInfo) {
CTG_API_ENTER();
if (NULL == pCtg || NULL == dbFName || NULL == pConn || NULL == pInfo) {
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
SCtgDBCache* dbCache = NULL;
int32_t code = 0;
SDBVgInfo* dbInfo = NULL;
CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pConn, dbFName, &dbCache, &dbInfo, NULL));
if (dbCache) {
dbInfo = dbCache->vgCache.vgInfo;
}
pInfo->routeVersion = dbInfo->vgVersion;
pInfo->hashPrefix = dbInfo->hashPrefix;
pInfo->hashSuffix = dbInfo->hashSuffix;
pInfo->hashMethod = dbInfo->hashMethod;
pInfo->vgNum = taosHashGetSize(dbInfo->vgHash);
if (pInfo->vgNum <= 0) {
ctgError("invalid vgNum %d in db %s's vgHash", pInfo->vgNum, dbFName);
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
pInfo->vgHash = taosMemoryCalloc(pInfo->vgNum, sizeof(TAOS_VGROUP_HASH_INFO));
if (NULL == pInfo->vgHash) {
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
SVgroupInfo* vgInfo = NULL;
int32_t i = 0;
void* pIter = taosHashIterate(dbInfo->vgHash, NULL);
while (pIter) {
vgInfo = pIter;
pInfo->vgHash[i].vgId = vgInfo->vgId;
pInfo->vgHash[i].hashBegin = vgInfo->hashBegin;
pInfo->vgHash[i].hashEnd = vgInfo->hashEnd;
pIter = taosHashIterate(dbInfo->vgHash, pIter);
vgInfo = NULL;
++i;
}
_return:
if (dbCache) {
ctgRUnlockVgInfo(dbCache);
ctgReleaseDBCache(pCtg, dbCache);
} else if (dbInfo) {
taosHashCleanup(dbInfo->vgHash);
taosMemoryFreeClear(dbInfo);
}
CTG_API_LEAVE(code);
}
int32_t catalogUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo* dbInfo) {
CTG_API_ENTER();
......
source/libs/catalog/test/catalogTests.cpp
浏览文件 @ 05d04dcd
......@@ -778,7 +778,7 @@ void *ctgTestGetDbVgroupThread(void *param) {
int32_t n = 0;
while (!ctgTestStop) {
code = catalogGetDBVgInfo(pCtg, mockPointer, ctgTestDbname, &vgList);
code = catalogGetDBVgList(pCtg, mockPointer, ctgTestDbname, &vgList);
if (code) {
assert(0);
}
......@@ -2063,7 +2063,7 @@ TEST(dbVgroup, getSetDbVgroupCase) {
strcpy(n.dbname, "db1");
strcpy(n.tname, ctgTestTablename);
code = catalogGetDBVgInfo(pCtg, mockPointer, ctgTestDbname, &vgList);
code = catalogGetDBVgList(pCtg, mockPointer, ctgTestDbname, &vgList);
ASSERT_EQ(code, 0);
ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), ctgTestVgNum);
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 05d04dcd
......@@ -677,22 +677,20 @@ typedef struct SFillOperatorInfo {
uint64_t curGroupId; // current handled group id
SExprInfo* pExprInfo;
int32_t numOfExpr;
SExprInfo* pNotFillExprInfo;
int32_t numOfNotFillExpr;
SExprSupp noFillExprSupp;
} SFillOperatorInfo;
typedef struct SGroupbyOperatorInfo {
SOptrBasicInfo binfo;
SAggSupporter aggSup;
SArray* pGroupCols; // group by columns, SArray<SColumn>
SArray* pGroupColVals; // current group column values, SArray<SGroupKeys>
SNode* pCondition;
bool isInit; // denote if current val is initialized or not
char* keyBuf; // group by keys for hash
int32_t groupKeyLen; // total group by column width
SGroupResInfo groupResInfo;
SExprSupp scalarSup;
SArray* pGroupCols; // group by columns, SArray<SColumn>
SArray* pGroupColVals; // current group column values, SArray<SGroupKeys>
SNode* pCondition;
bool isInit; // denote if current val is initialized or not
char* keyBuf; // group by keys for hash
int32_t groupKeyLen; // total group by column width
SGroupResInfo groupResInfo;
SExprSupp scalarSup;
} SGroupbyOperatorInfo;
typedef struct SDataGroupInfo {
......@@ -1016,8 +1014,7 @@ SOperatorInfo* createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFi
int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
int32_t numOfOutput, SArray* pPseudoList);
void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order,
int32_t scanFlag, bool createDummyCol);
void setInputDataBlock(SExprSupp* pExprSupp, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol);
bool isTaskKilled(SExecTaskInfo* pTaskInfo);
int32_t checkForQueryBuf(size_t numOfTables);
......
source/libs/executor/src/cachescanoperator.c
浏览文件 @ 05d04dcd
......@@ -58,7 +58,7 @@ SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SRe
// partition by tbname
if (taosArrayGetSize(pTableList->pGroupList) == taosArrayGetSize(pTableList->pTableList)) {
pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_ALL | CACHESCAN_RETRIEVE_LAST_ROW;
pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_ALL|(pScanNode->ignoreNull? CACHESCAN_RETRIEVE_LAST:CACHESCAN_RETRIEVE_LAST_ROW);
code = tsdbCacherowsReaderOpen(pInfo->readHandle.vnode, pInfo->retrieveType, pTableList->pTableList,
taosArrayGetSize(pInfo->pColMatchInfo), &pInfo->pLastrowReader);
if (code != TSDB_CODE_SUCCESS) {
......@@ -67,8 +67,8 @@ SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SRe
pInfo->pBufferredRes = createOneDataBlock(pInfo->pRes, false);
blockDataEnsureCapacity(pInfo->pBufferredRes, pOperator->resultInfo.capacity);
} else { // by tags
pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_SINGLE | CACHESCAN_RETRIEVE_LAST_ROW;
} else { // by tags
pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_SINGLE|(pScanNode->ignoreNull? CACHESCAN_RETRIEVE_LAST:CACHESCAN_RETRIEVE_LAST_ROW);
}
if (pScanNode->scan.pScanPseudoCols != NULL) {
......@@ -198,16 +198,20 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
if (pInfo->pRes->info.rows > 0) {
if (pInfo->pseudoExprSup.numOfExprs > 0) {
SExprSupp* pSup = &pInfo->pseudoExprSup;
pInfo->pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, 0);
STableKeyInfo* pKeyInfo = taosArrayGet(pGroupTableList, 0);
pInfo->pRes->info.groupId = pKeyInfo->groupId;
code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes,
GET_TASKID(pTaskInfo));
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code;
return NULL;
if (taosArrayGetSize(pInfo->pUidList) > 0) {
ASSERT((pInfo->retrieveType & CACHESCAN_RETRIEVE_LAST_ROW) == CACHESCAN_RETRIEVE_LAST_ROW);
pInfo->pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, 0);
code = addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes,
GET_TASKID(pTaskInfo));
if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code;
return NULL;
}
}
}
......
source/libs/executor/src/executorimpl.c
浏览文件 @ 05d04dcd
......@@ -403,25 +403,24 @@ void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfo
}
}
static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order,
int32_t scanFlag, bool createDummyCol);
static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
bool createDummyCol);
static void doSetInputDataBlockInfo(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock,
int32_t order) {
for (int32_t i = 0; i < pOperator->exprSupp.numOfExprs; ++i) {
static void doSetInputDataBlockInfo(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order) {
SqlFunctionCtx* pCtx = pExprSup->pCtx;
for (int32_t i = 0; i < pExprSup->numOfExprs; ++i) {
pCtx[i].order = order;
pCtx[i].input.numOfRows = pBlock->info.rows;
setBlockSMAInfo(&pCtx[i], &pOperator->exprSupp.pExprInfo[i], pBlock);
setBlockSMAInfo(&pCtx[i], &pExprSup->pExprInfo[i], pBlock);
pCtx[i].pSrcBlock = pBlock;
}
}
void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order,
int32_t scanFlag, bool createDummyCol) {
void setInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol) {
if (pBlock->pBlockAgg != NULL) {
doSetInputDataBlockInfo(pOperator, pCtx, pBlock, order);
doSetInputDataBlockInfo(pExprSup, pBlock, order);
} else {
doSetInputDataBlock(pOperator, pCtx, pBlock, order, scanFlag, createDummyCol);
doSetInputDataBlock(pExprSup, pBlock, order, scanFlag, createDummyCol);
}
}
......@@ -468,11 +467,12 @@ static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunc
return TSDB_CODE_SUCCESS;
}
static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order,
int32_t scanFlag, bool createDummyCol) {
static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
bool createDummyCol) {
int32_t code = TSDB_CODE_SUCCESS;
SqlFunctionCtx* pCtx = pExprSup->pCtx;
for (int32_t i = 0; i < pOperator->exprSupp.numOfExprs; ++i) {
for (int32_t i = 0; i < pExprSup->numOfExprs; ++i) {
pCtx[i].order = order;
pCtx[i].input.numOfRows = pBlock->info.rows;
......@@ -483,7 +483,7 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt
pInput->uid = pBlock->info.uid;
pInput->colDataAggIsSet = false;
SExprInfo* pOneExpr = &pOperator->exprSupp.pExprInfo[i];
SExprInfo* pOneExpr = &pExprSup->pExprInfo[i];
for (int32_t j = 0; j < pOneExpr->base.numOfParams; ++j) {
SFunctParam* pFuncParam = &pOneExpr->base.pParam[j];
if (pFuncParam->type == FUNC_PARAM_TYPE_COLUMN) {
......@@ -2435,7 +2435,7 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
// the pDataBlock are always the same one, no need to call this again
setExecutionContext(pOperator, pOperator->exprSupp.numOfExprs, pBlock->info.groupId);
setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, true);
setInputDataBlock(pSup, pBlock, order, scanFlag, true);
code = doAggregateImpl(pOperator, pSup->pCtx);
if (code != 0) {
T_LONG_JMP(pTaskInfo->env, code);
......@@ -2734,28 +2734,16 @@ static void doHandleRemainBlockFromNewGroup(SOperatorInfo* pOperator, SFillOpera
static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t order, int32_t scanFlag) {
SFillOperatorInfo* pInfo = pOperator->info;
SExprSupp* pSup = &pOperator->exprSupp;
SSDataBlock* pResBlock = pInfo->pFinalRes;
setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, false);
setInputDataBlock(pSup, pBlock, order, scanFlag, false);
projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs, NULL);
pInfo->pRes->info.groupId = pBlock->info.groupId;
SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pInfo->primaryTsCol);
SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, pInfo->primarySrcSlotId);
colDataAssign(pDst, pSrc, pInfo->pRes->info.rows, &pResBlock->info);
for (int32_t i = 0; i < pInfo->numOfNotFillExpr; ++i) {
SFillColInfo* pCol = &pInfo->pFillInfo->pFillCol[i + pInfo->numOfExpr];
ASSERT(pCol->notFillCol);
SExprInfo* pExpr = pCol->pExpr;
int32_t srcSlotId = pExpr->base.pParam[0].pCol->slotId;
int32_t dstSlotId = pExpr->base.resSchema.slotId;
// reset the row value before applying the no-fill functions to the input data block, which is "pBlock" in this case.
pInfo->pRes->info.rows = 0;
SExprSupp* pNoFillSupp = &pInfo->noFillExprSupp;
setInputDataBlock(pNoFillSupp, pBlock, order, scanFlag, false);
SColumnInfoData* pDst1 = taosArrayGet(pInfo->pRes->pDataBlock, dstSlotId);
SColumnInfoData* pSrc1 = taosArrayGet(pBlock->pDataBlock, srcSlotId);
colDataAssign(pDst1, pSrc1, pInfo->pRes->info.rows, &pResBlock->info);
}
projectApplyFunctions(pNoFillSupp->pExprInfo, pInfo->pRes, pBlock, pNoFillSupp->pCtx, pNoFillSupp->numOfExprs, NULL);
pInfo->pRes->info.groupId = pBlock->info.groupId;
}
static SSDataBlock* doFillImpl(SOperatorInfo* pOperator) {
......@@ -3143,10 +3131,7 @@ void destroyFillOperatorInfo(void* param) {
pInfo->pRes = blockDataDestroy(pInfo->pRes);
pInfo->pFinalRes = blockDataDestroy(pInfo->pFinalRes);
if (pInfo->pNotFillExprInfo != NULL) {
destroyExprInfo(pInfo->pNotFillExprInfo, pInfo->numOfNotFillExpr);
taosMemoryFree(pInfo->pNotFillExprInfo);
}
cleanupExprSupp(&pInfo->noFillExprSupp);
taosMemoryFreeClear(pInfo->p);
taosArrayDestroy(pInfo->pColMatchColInfo);
......@@ -3211,11 +3196,12 @@ static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t
}
static bool isWstartColumnExist(SFillOperatorInfo* pInfo) {
if (pInfo->numOfNotFillExpr == 0) {
if (pInfo->noFillExprSupp.numOfExprs == 0) {
return false;
}
for (int32_t i = 0; i < pInfo->numOfNotFillExpr; ++i) {
SExprInfo* exprInfo = pInfo->pNotFillExprInfo + i;
for (int32_t i = 0; i < pInfo->noFillExprSupp.numOfExprs; ++i) {
SExprInfo* exprInfo = pInfo->noFillExprSupp.pExprInfo + i;
if (exprInfo->pExpr->nodeType == QUERY_NODE_COLUMN && exprInfo->base.numOfParams == 1 &&
exprInfo->base.pParam[0].pCol->colType == COLUMN_TYPE_WINDOW_START) {
return true;
......@@ -3224,25 +3210,24 @@ static bool isWstartColumnExist(SFillOperatorInfo* pInfo) {
return false;
}
static int32_t createWStartTsAsNotFillExpr(SFillOperatorInfo* pInfo, SFillPhysiNode* pPhyFillNode) {
static int32_t createPrimaryTsExprIfNeeded(SFillOperatorInfo* pInfo, SFillPhysiNode* pPhyFillNode, SExprSupp* pExprSupp,
const char* idStr) {
bool wstartExist = isWstartColumnExist(pInfo);
if (wstartExist == false) {
if (pPhyFillNode->pWStartTs->type != QUERY_NODE_TARGET) {
qError("pWStartTs of fill physical node is not a target node");
qError("pWStartTs of fill physical node is not a target node, %s", idStr);
return TSDB_CODE_QRY_SYS_ERROR;
}
SExprInfo* notFillExprs =
taosMemoryRealloc(pInfo->pNotFillExprInfo, (pInfo->numOfNotFillExpr + 1) * sizeof(SExprInfo));
if (notFillExprs == NULL) {
SExprInfo* pExpr = taosMemoryRealloc(pExprSupp->pExprInfo, (pExprSupp->numOfExprs + 1) * sizeof(SExprInfo));
if (pExpr == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
createExprFromTargetNode(notFillExprs + pInfo->numOfNotFillExpr, (STargetNode*)pPhyFillNode->pWStartTs);
++pInfo->numOfNotFillExpr;
pInfo->pNotFillExprInfo = notFillExprs;
return TSDB_CODE_SUCCESS;
createExprFromTargetNode(&pExpr[pExprSupp->numOfExprs], (STargetNode*)pPhyFillNode->pWStartTs);
pExprSupp->numOfExprs += 1;
pExprSupp->pExprInfo = pExpr;
}
return TSDB_CODE_SUCCESS;
......@@ -3260,8 +3245,14 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pFillExprs, NULL, &pInfo->numOfExpr);
pOperator->exprSupp.pExprInfo = pExprInfo;
pInfo->pNotFillExprInfo = createExprInfo(pPhyFillNode->pNotFillExprs, NULL, &pInfo->numOfNotFillExpr);
int32_t code = createWStartTsAsNotFillExpr(pInfo, pPhyFillNode);
SExprSupp* pNoFillSupp = &pInfo->noFillExprSupp;
pNoFillSupp->pExprInfo = createExprInfo(pPhyFillNode->pNotFillExprs, NULL, &pNoFillSupp->numOfExprs);
int32_t code = createPrimaryTsExprIfNeeded(pInfo, pPhyFillNode, pNoFillSupp, pTaskInfo->id.str);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
code = initExprSupp(pNoFillSupp, pNoFillSupp->pExprInfo, pNoFillSupp->numOfExprs);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
......@@ -3290,7 +3281,7 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
pInfo->pColMatchColInfo = extractColMatchInfo(pPhyFillNode->pFillExprs, pPhyFillNode->node.pOutputDataBlockDesc,
&numOfOutputCols, COL_MATCH_FROM_SLOT_ID);
code = initFillInfo(pInfo, pExprInfo, pInfo->numOfExpr, pInfo->pNotFillExprInfo, pInfo->numOfNotFillExpr,
code = initFillInfo(pInfo, pExprInfo, pInfo->numOfExpr, pNoFillSupp->pExprInfo, pNoFillSupp->numOfExprs,
(SNodeListNode*)pPhyFillNode->pValues, pPhyFillNode->timeRange, pResultInfo->capacity,
pTaskInfo->id.str, pInterval, type, order);
if (code != TSDB_CODE_SUCCESS) {
......
source/libs/executor/src/groupoperator.c
浏览文件 @ 05d04dcd
......@@ -360,7 +360,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pOperator->exprSupp.pCtx, pBlock, order, scanFlag, true);
setInputDataBlock(&pOperator->exprSupp, pBlock, order, scanFlag, true);
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
if (pInfo->scalarSup.pExprInfo != NULL) {
......
source/libs/executor/src/projectoperator.c
浏览文件 @ 05d04dcd
......@@ -285,7 +285,7 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
T_LONG_JMP(pTaskInfo->env, code);
}
setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, false);
setInputDataBlock(pSup, pBlock, order, scanFlag, false);
blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
code = projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs,
......@@ -446,7 +446,7 @@ static void doHandleDataBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, SOp
}
}
setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, scanFlag, false);
setInputDataBlock(pSup, pBlock, order, scanFlag, false);
blockDataEnsureCapacity(pInfo->pRes, pInfo->pRes->info.rows + pBlock->info.rows);
code = projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs,
......
source/libs/executor/src/tfill.c
浏览文件 @ 05d04dcd
......@@ -272,7 +272,17 @@ static void copyCurrentRowIntoBuf(SFillInfo* pFillInfo, int32_t rowIndex, SArray
char* p = colDataGetData(pSrcCol, rowIndex);
saveColData(pRow, i, p, isNull);
} else if (type == QUERY_NODE_OPERATOR) {
SColumnInfoData* pSrcCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, i);
int32_t srcSlotId = GET_DEST_SLOT_ID(&pFillInfo->pFillCol[i]);
SColumnInfoData* pSrcCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, srcSlotId);
bool isNull = colDataIsNull_s(pSrcCol, rowIndex);
char* p = colDataGetData(pSrcCol, rowIndex);
saveColData(pRow, i, p, isNull);
} else if (type == QUERY_NODE_FUNCTION) {
int32_t srcSlotId = GET_DEST_SLOT_ID(&pFillInfo->pFillCol[i]);
SColumnInfoData* pSrcCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, srcSlotId);
bool isNull = colDataIsNull_s(pSrcCol, rowIndex);
char* p = colDataGetData(pSrcCol, rowIndex);
......@@ -621,8 +631,8 @@ int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, SSDataBlock* p, int32_t ca
int64_t getFillInfoStart(struct SFillInfo* pFillInfo) { return pFillInfo->start; }
SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprInfo* pNotFillExpr,
int32_t numOfNotFillExpr, const struct SNodeListNode* pValNode) {
SFillColInfo* pFillCol = taosMemoryCalloc(numOfFillExpr + numOfNotFillExpr, sizeof(SFillColInfo));
int32_t numOfNoFillExpr, const struct SNodeListNode* pValNode) {
SFillColInfo* pFillCol = taosMemoryCalloc(numOfFillExpr + numOfNoFillExpr, sizeof(SFillColInfo));
if (pFillCol == NULL) {
return NULL;
}
......@@ -643,7 +653,7 @@ SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprIn
}
}
for (int32_t i = 0; i < numOfNotFillExpr; ++i) {
for (int32_t i = 0; i < numOfNoFillExpr; ++i) {
SExprInfo* pExprInfo = &pNotFillExpr[i];
pFillCol[i + numOfFillExpr].pExpr = pExprInfo;
pFillCol[i + numOfFillExpr].notFillCol = true;
......@@ -1403,7 +1413,7 @@ static void doApplyStreamScalarCalculation(SOperatorInfo* pOperator, SSDataBlock
blockDataCleanup(pDstBlock);
blockDataEnsureCapacity(pDstBlock, pSrcBlock->info.rows);
setInputDataBlock(pOperator, pSup->pCtx, pSrcBlock, TSDB_ORDER_ASC, MAIN_SCAN, false);
setInputDataBlock(pSup, pSrcBlock, TSDB_ORDER_ASC, MAIN_SCAN, false);
projectApplyFunctions(pSup->pExprInfo, pDstBlock, pSrcBlock, pSup->pCtx, pSup->numOfExprs, NULL);
pDstBlock->info.groupId = pSrcBlock->info.groupId;
......@@ -1553,8 +1563,8 @@ static SStreamFillSupporter* initStreamFillSup(SStreamFillPhysiNode* pPhyFillNod
}
pFillSup->numOfFillCols = numOfFillCols;
int32_t numOfNotFillCols = 0;
SExprInfo* pNotFillExprInfo = createExprInfo(pPhyFillNode->pNotFillExprs, NULL, &numOfNotFillCols);
pFillSup->pAllColInfo = createFillColInfo(pFillExprInfo, pFillSup->numOfFillCols, pNotFillExprInfo, numOfNotFillCols,
SExprInfo* noFillExprInfo = createExprInfo(pPhyFillNode->pNotFillExprs, NULL, &numOfNotFillCols);
pFillSup->pAllColInfo = createFillColInfo(pFillExprInfo, pFillSup->numOfFillCols, noFillExprInfo, numOfNotFillCols,
(const SNodeListNode*)(pPhyFillNode->pValues));
pFillSup->type = convertFillType(pPhyFillNode->mode);
pFillSup->numOfAllCols = pFillSup->numOfFillCols + numOfNotFillCols;
......
source/libs/executor/src/timewindowoperator.c
浏览文件 @ 05d04dcd
......@@ -1079,7 +1079,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->inputOrder, scanFlag, true);
setInputDataBlock(pSup, pBlock, pInfo->inputOrder, scanFlag, true);
blockDataUpdateTsWindow(pBlock, pInfo->primaryTsIndex);
hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag);
......@@ -1209,7 +1209,7 @@ static int32_t openStateWindowAggOptr(SOperatorInfo* pOperator) {
break;
}
setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true);
setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);
// there is an scalar expression that needs to be calculated right before apply the group aggregation.
......@@ -1944,7 +1944,7 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true);
setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);
doSessionWindowAggImpl(pOperator, pInfo, pBlock);
......@@ -2295,7 +2295,7 @@ static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true);
setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSliceInfo->tsCol.slotId);
for (int32_t i = 0; i < pBlock->info.rows; ++i) {
......@@ -3268,7 +3268,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
SExprSupp* pExprSup = &pInfo->scalarSupp;
projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
}
setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
doStreamIntervalAggImpl(pOperator, pBlock, pBlock->info.groupId, pUpdatedMap);
if (IS_FINAL_OP(pInfo)) {
int32_t chIndex = getChildIndex(pBlock);
......@@ -3286,7 +3286,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
}
SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
SStreamIntervalOperatorInfo* pChInfo = pChildOp->info;
setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
setInputDataBlock(&pChildOp->exprSupp, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
doStreamIntervalAggImpl(pChildOp, pBlock, pBlock->info.groupId, NULL);
}
maxTs = TMAX(maxTs, pBlock->info.window.ekey);
......@@ -4120,7 +4120,8 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
SExprSupp* pExprSup = &pInfo->scalarSupp;
projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
}
setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
doStreamSessionAggImpl(pOperator, pBlock, pStUpdated, pInfo->pStDeleted, IS_FINAL_OP(pInfo));
if (IS_FINAL_OP(pInfo)) {
int32_t chIndex = getChildIndex(pBlock);
......@@ -4135,7 +4136,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
taosArrayPush(pInfo->pChildren, &pChildOp);
}
SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
setInputDataBlock(&pChildOp->exprSupp, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
doStreamSessionAggImpl(pChildOp, pBlock, NULL, NULL, true);
}
maxTs = TMAX(maxTs, pBlock->info.window.ekey);
......@@ -4334,7 +4335,7 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
doStreamSessionAggImpl(pOperator, pBlock, pStUpdated, NULL, false);
maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
}
......@@ -4647,7 +4648,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
}
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
doStreamStateAggImpl(pOperator, pBlock, pSeUpdated, pInfo->pSeDeleted);
maxTs = TMAX(maxTs, pBlock->info.window.ekey);
}
......@@ -4917,7 +4918,7 @@ static void doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
}
getTableScanInfo(pOperator, &pIaInfo->inputOrder, &scanFlag);
setInputDataBlock(pOperator, pSup->pCtx, pBlock, pIaInfo->inputOrder, scanFlag, true);
setInputDataBlock(pSup, pBlock, pIaInfo->inputOrder, scanFlag, true);
doMergeAlignedIntervalAggImpl(pOperator, &pIaInfo->binfo.resultRowInfo, pBlock, pRes);
doFilter(pMiaInfo->pCondition, pRes, NULL, NULL);
......@@ -5246,7 +5247,7 @@ static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) {
}
getTableScanInfo(pOperator, &iaInfo->inputOrder, &scanFlag);
setInputDataBlock(pOperator, pExpSupp->pCtx, pBlock, iaInfo->inputOrder, scanFlag, true);
setInputDataBlock(pExpSupp, pBlock, iaInfo->inputOrder, scanFlag, true);
doMergeIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes);
if (pRes->info.rows >= pOperator->resultInfo.threshold) {
......@@ -5425,7 +5426,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
// The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the
// caller. Note that all the time window are not close till now.
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
if (pInfo->invertible) {
setInverFunction(pSup->pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.type);
}
......@@ -5475,9 +5476,11 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
}
SStreamIntervalPhysiNode* pIntervalPhyNode = (SStreamIntervalPhysiNode*)pPhyNode;
int32_t code = TSDB_CODE_SUCCESS;
int32_t numOfCols = 0;
SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols);
ASSERT(numOfCols > 0);
SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
SInterval interval = {
.interval = pIntervalPhyNode->interval,
......@@ -5487,6 +5490,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
.offset = pIntervalPhyNode->offset,
.precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision,
};
STimeWindowAggSupp twAggSupp = {
.waterMark = pIntervalPhyNode->window.watermark,
.calTrigger = pIntervalPhyNode->window.triggerType,
......@@ -5494,6 +5498,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
.minTs = INT64_MAX,
.deleteMark = INT64_MAX,
};
ASSERT(twAggSupp.calTrigger != STREAM_TRIGGER_MAX_DELAY);
pOperator->pTaskInfo = pTaskInfo;
pInfo->interval = interval;
......@@ -5501,16 +5506,25 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->ignoreExpiredData = pIntervalPhyNode->window.igExpired;
pInfo->isFinal = false;
pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
initResultSizeInfo(&pOperator->resultInfo, 4096);
SExprSupp* pSup = &pOperator->exprSupp;
initBasicInfo(&pInfo->binfo, pResBlock);
initStreamFunciton(pSup->pCtx, pSup->numOfExprs);
initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
initResultSizeInfo(&pOperator->resultInfo, 4096);
if (pIntervalPhyNode->window.pExprs != NULL) {
int32_t numOfScalar = 0;
SExprInfo* pScalarExprInfo = createExprInfo(pIntervalPhyNode->window.pExprs, NULL, &numOfScalar);
code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
}
size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
int32_t code = initAggInfo(pSup, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
code = initAggInfo(pSup, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
......
source/libs/function/src/builtins.c
浏览文件 @ 05d04dcd
......@@ -2405,6 +2405,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.processFunc = cachedLastRowFunction,
.finalizeFunc = firstLastFinalize
},
{
.name = "_cache_last",
.type = FUNCTION_TYPE_CACHE_LAST,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
.translateFunc = translateFirstLast,
.getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup,
.processFunc = lastFunctionMerge,
.finalizeFunc = firstLastFinalize
},
{
.name = "_last_row_partial",
.type = FUNCTION_TYPE_LAST_PARTIAL,
......
source/libs/function/src/builtinsimpl.c
浏览文件 @ 05d04dcd
......@@ -57,16 +57,6 @@ typedef struct SAvgRes {
int16_t type; // store the original input type, used in merge function
} SAvgRes;
typedef struct STuplePos {
union {
struct {
int32_t pageId;
int32_t offset;
};
STupleKey streamTupleKey;
};
} STuplePos;
typedef struct SMinmaxResInfo {
bool assign; // assign the first value or not
int64_t v;
......@@ -93,17 +83,6 @@ typedef struct STopBotRes {
STopBotResItem* pItems;
} STopBotRes;
typedef struct SFirstLastRes {
bool hasResult;
// used for last_row function only, isNullRes in SResultRowEntry can not be passed to downstream.So,
// this attribute is required
bool isNull;
int32_t bytes;
int64_t ts;
STuplePos pos;
char buf[];
} SFirstLastRes;
typedef struct SStddevRes {
double result;
int64_t count;
......
source/libs/function/src/functionMgt.c
浏览文件 @ 05d04dcd
......@@ -16,6 +16,7 @@
#include "functionMgt.h"
#include "builtins.h"
#include "builtinsimpl.h"
#include "functionMgtInt.h"
#include "taos.h"
#include "taoserror.h"
......@@ -314,6 +315,11 @@ bool fmIsSameInOutType(int32_t funcId) {
return res;
}
void getLastCacheDataType(SDataType* pType) {
pType->bytes = getFirstLastInfoSize(pType->bytes) + VARSTR_HEADER_SIZE;
pType->type = TSDB_DATA_TYPE_BINARY;
}
static int32_t getFuncInfo(SFunctionNode* pFunc) {
char msg[128] = {0};
return fmGetFuncInfo(pFunc, msg, sizeof(msg));
......
source/libs/function/src/tudf.c
浏览文件 @ 05d04dcd
......@@ -339,10 +339,11 @@ typedef struct SUdfcProxy {
uv_mutex_t udfStubsMutex;
SArray *udfStubs; // SUdfcFuncStub
uv_mutex_t udfcUvMutex;
int8_t initialized;
} SUdfcProxy;
SUdfcProxy gUdfdProxy = {0};
SUdfcProxy gUdfcProxy = {0};
typedef struct SUdfcUvSession {
SUdfcProxy *udfc;
......@@ -896,23 +897,23 @@ int compareUdfcFuncSub(const void *elem1, const void *elem2) {
int32_t acquireUdfFuncHandle(char *udfName, UdfcFuncHandle *pHandle) {
int32_t code = 0;
uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
uv_mutex_lock(&gUdfcProxy.udfStubsMutex);
SUdfcFuncStub key = {0};
strncpy(key.udfName, udfName, TSDB_FUNC_NAME_LEN);
int32_t stubIndex = taosArraySearchIdx(gUdfdProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ);
int32_t stubIndex = taosArraySearchIdx(gUdfcProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ);
if (stubIndex != -1) {
SUdfcFuncStub *foundStub = taosArrayGet(gUdfdProxy.udfStubs, stubIndex);
SUdfcFuncStub *foundStub = taosArrayGet(gUdfcProxy.udfStubs, stubIndex);
UdfcFuncHandle handle = foundStub->handle;
if (handle != NULL && ((SUdfcUvSession *)handle)->udfUvPipe != NULL) {
*pHandle = foundStub->handle;
++foundStub->refCount;
foundStub->lastRefTime = taosGetTimestampUs();
uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
return 0;
} else {
fnInfo("invalid handle for %s, refCount: %d, last ref time: %" PRId64 ". remove it from cache", udfName,
foundStub->refCount, foundStub->lastRefTime);
taosArrayRemove(gUdfdProxy.udfStubs, stubIndex);
taosArrayRemove(gUdfcProxy.udfStubs, stubIndex);
}
}
*pHandle = NULL;
......@@ -923,46 +924,46 @@ int32_t acquireUdfFuncHandle(char *udfName, UdfcFuncHandle *pHandle) {
stub.handle = *pHandle;
++stub.refCount;
stub.lastRefTime = taosGetTimestampUs();
taosArrayPush(gUdfdProxy.udfStubs, &stub);
taosArraySort(gUdfdProxy.udfStubs, compareUdfcFuncSub);
taosArrayPush(gUdfcProxy.udfStubs, &stub);
taosArraySort(gUdfcProxy.udfStubs, compareUdfcFuncSub);
} else {
*pHandle = NULL;
}
uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
return code;
}
void releaseUdfFuncHandle(char *udfName) {
uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
uv_mutex_lock(&gUdfcProxy.udfStubsMutex);
SUdfcFuncStub key = {0};
strncpy(key.udfName, udfName, TSDB_FUNC_NAME_LEN);
SUdfcFuncStub *foundStub = taosArraySearch(gUdfdProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ);
SUdfcFuncStub *foundStub = taosArraySearch(gUdfcProxy.udfStubs, &key, compareUdfcFuncSub, TD_EQ);
if (!foundStub) {
uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
return;
}
if (foundStub->refCount > 0) {
--foundStub->refCount;
}
uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
}
int32_t cleanUpUdfs() {
int8_t initialized = atomic_load_8(&gUdfdProxy.initialized);
int8_t initialized = atomic_load_8(&gUdfcProxy.initialized);
if (!initialized) {
return TSDB_CODE_SUCCESS;
}
uv_mutex_lock(&gUdfdProxy.udfStubsMutex);
if (gUdfdProxy.udfStubs == NULL || taosArrayGetSize(gUdfdProxy.udfStubs) == 0) {
uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
uv_mutex_lock(&gUdfcProxy.udfStubsMutex);
if (gUdfcProxy.udfStubs == NULL || taosArrayGetSize(gUdfcProxy.udfStubs) == 0) {
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
return TSDB_CODE_SUCCESS;
}
SArray *udfStubs = taosArrayInit(16, sizeof(SUdfcFuncStub));
int32_t i = 0;
while (i < taosArrayGetSize(gUdfdProxy.udfStubs)) {
SUdfcFuncStub *stub = taosArrayGet(gUdfdProxy.udfStubs, i);
while (i < taosArrayGetSize(gUdfcProxy.udfStubs)) {
SUdfcFuncStub *stub = taosArrayGet(gUdfcProxy.udfStubs, i);
if (stub->refCount == 0) {
fnInfo("tear down udf. udf name: %s, handle: %p, ref count: %d", stub->udfName, stub->handle, stub->refCount);
doTeardownUdf(stub->handle);
......@@ -979,9 +980,9 @@ int32_t cleanUpUdfs() {
}
++i;
}
taosArrayDestroy(gUdfdProxy.udfStubs);
gUdfdProxy.udfStubs = udfStubs;
uv_mutex_unlock(&gUdfdProxy.udfStubsMutex);
taosArrayDestroy(gUdfcProxy.udfStubs);
gUdfcProxy.udfStubs = udfStubs;
uv_mutex_unlock(&gUdfcProxy.udfStubsMutex);
return 0;
}
......@@ -1157,9 +1158,11 @@ void onUdfcPipeClose(uv_handle_t *handle) {
QUEUE_REMOVE(&task->procTaskQueue);
uv_sem_post(&task->taskSem);
}
uv_mutex_lock(&gUdfcProxy.udfcUvMutex);
if (conn->session != NULL) {
conn->session->udfUvPipe = NULL;
}
uv_mutex_unlock(&gUdfcProxy.udfcUvMutex);
taosMemoryFree(conn->readBuf.buf);
taosMemoryFree(conn);
taosMemoryFree((uv_pipe_t *)handle);
......@@ -1553,11 +1556,11 @@ void constructUdfService(void *argsThread) {
}
int32_t udfcOpen() {
int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 0, 1);
int8_t old = atomic_val_compare_exchange_8(&gUdfcProxy.initialized, 0, 1);
if (old == 1) {
return 0;
}
SUdfcProxy *proxy = &gUdfdProxy;
SUdfcProxy *proxy = &gUdfcProxy;
getUdfdPipeName(proxy->udfdPipeName, sizeof(proxy->udfdPipeName));
proxy->udfcState = UDFC_STATE_STARTNG;
uv_barrier_init(&proxy->initBarrier, 2);
......@@ -1567,16 +1570,17 @@ int32_t udfcOpen() {
uv_barrier_wait(&proxy->initBarrier);
uv_mutex_init(&proxy->udfStubsMutex);
proxy->udfStubs = taosArrayInit(8, sizeof(SUdfcFuncStub));
uv_mutex_init(&proxy->udfcUvMutex);
fnInfo("udfc initialized") return 0;
}
int32_t udfcClose() {
int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 1, 0);
int8_t old = atomic_val_compare_exchange_8(&gUdfcProxy.initialized, 1, 0);
if (old == 0) {
return 0;
}
SUdfcProxy *udfc = &gUdfdProxy;
SUdfcProxy *udfc = &gUdfcProxy;
udfc->udfcState = UDFC_STATE_STOPPING;
uv_async_send(&udfc->loopStopAsync);
uv_thread_join(&udfc->loopThread);
......@@ -1584,6 +1588,7 @@ int32_t udfcClose() {
uv_barrier_destroy(&udfc->initBarrier);
taosArrayDestroy(udfc->udfStubs);
uv_mutex_destroy(&udfc->udfStubsMutex);
uv_mutex_destroy(&udfc->udfcUvMutex);
udfc->udfcState = UDFC_STATE_INITAL;
fnInfo("udfc is cleaned up");
return 0;
......@@ -1611,13 +1616,13 @@ int32_t udfcRunUdfUvTask(SClientUdfTask *task, int8_t uvTaskType) {
}
int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
if (gUdfdProxy.udfcState != UDFC_STATE_READY) {
if (gUdfcProxy.udfcState != UDFC_STATE_READY) {
return TSDB_CODE_UDF_INVALID_STATE;
}
SClientUdfTask *task = taosMemoryCalloc(1, sizeof(SClientUdfTask));
task->errCode = 0;
task->session = taosMemoryCalloc(1, sizeof(SUdfcUvSession));
task->session->udfc = &gUdfdProxy;
task->session->udfc = &gUdfcProxy;
task->type = UDF_TASK_SETUP;
SUdfSetupRequest *req = &task->_setup.req;
......@@ -1625,7 +1630,7 @@ int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
int32_t errCode = udfcRunUdfUvTask(task, UV_TASK_CONNECT);
if (errCode != 0) {
fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfdProxy)->udfdPipeName);
fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfcProxy)->udfdPipeName);
taosMemoryFree(task->session);
taosMemoryFree(task);
return TSDB_CODE_UDF_PIPE_CONNECT_ERR;
......@@ -1799,10 +1804,12 @@ int32_t doTeardownUdf(UdfcFuncHandle handle) {
fnInfo("tear down udf. udf name: %s, udf func handle: %p", session->udfName, handle);
// TODO: synchronization refactor between libuv event loop and request thread
uv_mutex_lock(&gUdfcProxy.udfcUvMutex);
if (session->udfUvPipe != NULL && session->udfUvPipe->data != NULL) {
SClientUvConn *conn = session->udfUvPipe->data;
conn->session = NULL;
}
uv_mutex_unlock(&gUdfcProxy.udfcUvMutex);
taosMemoryFree(session);
taosMemoryFree(task);
......
source/libs/nodes/src/nodesMsgFuncs.c
浏览文件 @ 05d04dcd
......@@ -1969,7 +1969,7 @@ static int32_t msgToPhysiScanNode(STlvDecoder* pDecoder, void* pObj) {
return code;
}
enum { PHY_LAST_ROW_SCAN_CODE_SCAN = 1, PHY_LAST_ROW_SCAN_CODE_GROUP_TAGS, PHY_LAST_ROW_SCAN_CODE_GROUP_SORT };
enum { PHY_LAST_ROW_SCAN_CODE_SCAN = 1, PHY_LAST_ROW_SCAN_CODE_GROUP_TAGS, PHY_LAST_ROW_SCAN_CODE_GROUP_SORT, PHY_LAST_ROW_SCAN_CODE_IGNULL };
static int32_t physiLastRowScanNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
const SLastRowScanPhysiNode* pNode = (const SLastRowScanPhysiNode*)pObj;
......@@ -1981,6 +1981,9 @@ static int32_t physiLastRowScanNodeToMsg(const void* pObj, STlvEncoder* pEncoder
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, PHY_LAST_ROW_SCAN_CODE_GROUP_SORT, pNode->groupSort);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, PHY_LAST_ROW_SCAN_CODE_IGNULL, pNode->ignoreNull);
}
return code;
}
......@@ -2001,6 +2004,9 @@ static int32_t msgToPhysiLastRowScanNode(STlvDecoder* pDecoder, void* pObj) {
case PHY_LAST_ROW_SCAN_CODE_GROUP_SORT:
code = tlvDecodeBool(pTlv, &pNode->groupSort);
break;
case PHY_LAST_ROW_SCAN_CODE_IGNULL:
code = tlvDecodeBool(pTlv, &pNode->ignoreNull);
break;
default:
break;
}
......
source/libs/parser/src/parTranslater.c
浏览文件 @ 05d04dcd
......@@ -417,11 +417,11 @@ static int32_t getDBVgInfoImpl(STranslateContext* pCxt, const SName* pName, SArr
.requestId = pParCxt->requestId,
.requestObjRefId = pParCxt->requestRid,
.mgmtEps = pParCxt->mgmtEpSet};
code = catalogGetDBVgInfo(pParCxt->pCatalog, &conn, fullDbName, pVgInfo);
code = catalogGetDBVgList(pParCxt->pCatalog, &conn, fullDbName, pVgInfo);
}
}
if (TSDB_CODE_SUCCESS != code) {
parserError("0x%" PRIx64 " catalogGetDBVgInfo error, code:%s, dbFName:%s", pCxt->pParseCxt->requestId,
parserError("0x%" PRIx64 " catalogGetDBVgList error, code:%s, dbFName:%s", pCxt->pParseCxt->requestId,
tstrerror(code), fullDbName);
}
return code;
......
source/libs/parser/test/mockCatalog.cpp
浏览文件 @ 05d04dcd
......@@ -243,8 +243,8 @@ int32_t __catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* ve
return 0;
}
int32_t __catalogGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, SArray** pVgList) {
return g_mockCatalogService->catalogGetDBVgInfo(dbFName, pVgList);
int32_t __catalogGetDBVgList(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, SArray** pVgList) {
return g_mockCatalogService->catalogGetDBVgList(dbFName, pVgList);
}
int32_t __catalogGetDBCfg(SCatalog* pCtg, SRequestConnInfo* pConn, const char* dbFName, SDbCfgInfo* pDbCfg) {
......@@ -293,7 +293,7 @@ void initMetaDataEnv() {
stub.set(catalogGetTableHashVgroup, __catalogGetTableHashVgroup);
stub.set(catalogGetTableDistVgInfo, __catalogGetTableDistVgInfo);
stub.set(catalogGetDBVgVersion, __catalogGetDBVgVersion);
stub.set(catalogGetDBVgInfo, __catalogGetDBVgInfo);
stub.set(catalogGetDBVgList, __catalogGetDBVgList);
stub.set(catalogGetDBCfg, __catalogGetDBCfg);
stub.set(catalogChkAuth, __catalogChkAuth);
stub.set(catalogGetUdfInfo, __catalogGetUdfInfo);
......
source/libs/parser/test/mockCatalogService.cpp
浏览文件 @ 05d04dcd
......@@ -132,7 +132,7 @@ class MockCatalogServiceImpl {
return copyTableVgroup(db, tNameGetTableName(pTableName), vgList);
}
int32_t catalogGetDBVgInfo(const char* pDbFName, SArray** pVgList) const {
int32_t catalogGetDBVgList(const char* pDbFName, SArray** pVgList) const {
std::string dbFName(pDbFName);
DbMetaCache::const_iterator it = meta_.find(dbFName.substr(std::string(pDbFName).find_last_of('.') + 1));
if (meta_.end() == it) {
......@@ -663,8 +663,8 @@ int32_t MockCatalogService::catalogGetTableDistVgInfo(const SName* pTableName, S
return impl_->catalogGetTableDistVgInfo(pTableName, pVgList);
}
int32_t MockCatalogService::catalogGetDBVgInfo(const char* pDbFName, SArray** pVgList) const {
return impl_->catalogGetDBVgInfo(pDbFName, pVgList);
int32_t MockCatalogService::catalogGetDBVgList(const char* pDbFName, SArray** pVgList) const {
return impl_->catalogGetDBVgList(pDbFName, pVgList);
}
int32_t MockCatalogService::catalogGetDBCfg(const char* pDbFName, SDbCfgInfo* pDbCfg) const {
......
source/libs/parser/test/mockCatalogService.h
浏览文件 @ 05d04dcd
......@@ -70,7 +70,7 @@ class MockCatalogService {
int32_t catalogGetTableMeta(const SName* pTableName, STableMeta** pTableMeta) const;
int32_t catalogGetTableHashVgroup(const SName* pTableName, SVgroupInfo* vgInfo) const;
int32_t catalogGetTableDistVgInfo(const SName* pTableName, SArray** pVgList) const;
int32_t catalogGetDBVgInfo(const char* pDbFName, SArray** pVgList) const;
int32_t catalogGetDBVgList(const char* pDbFName, SArray** pVgList) const;
int32_t catalogGetDBCfg(const char* pDbFName, SDbCfgInfo* pDbCfg) const;
int32_t catalogGetUdfInfo(const std::string& funcName, SFuncInfo* pInfo) const;
int32_t catalogGetTableIndex(const SName* pTableName, SArray** pIndexes) const;
......
source/libs/planner/src/planOptimizer.c
浏览文件 @ 05d04dcd
......@@ -2139,6 +2139,46 @@ static int32_t rewriteUniqueOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLog
return rewriteUniqueOptimizeImpl(pCxt, pLogicSubplan, pIndef);
}
typedef struct SLastRowScanOptLastParaCkCxt {
bool hasTag;
bool hasCol;
} SLastRowScanOptLastParaCkCxt;
static EDealRes lastRowScanOptLastParaCheckImpl(SNode* pNode, void* pContext) {
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SLastRowScanOptLastParaCkCxt* pCxt = pContext;
if (COLUMN_TYPE_TAG == ((SColumnNode*)pNode)->colType || COLUMN_TYPE_TBNAME == ((SColumnNode*)pNode)->colType) {
pCxt->hasTag = true;
} else {
pCxt->hasCol = true;
}
return DEAL_RES_END;
}
return DEAL_RES_CONTINUE;
}
static bool lastRowScanOptLastParaCheck(SNode* pExpr) {
SLastRowScanOptLastParaCkCxt cxt = {.hasTag = false, .hasCol = false};
nodesWalkExpr(pExpr, lastRowScanOptLastParaCheckImpl, &cxt);
return !cxt.hasTag && cxt.hasCol;
}
static bool hasSuitableCache(int8_t cacheLastMode, bool hasLastRow, bool hasLast) {
switch (cacheLastMode) {
case TSDB_CACHE_MODEL_NONE:
return false;
case TSDB_CACHE_MODEL_LAST_ROW:
return hasLastRow;
case TSDB_CACHE_MODEL_LAST_VALUE:
return hasLast;
case TSDB_CACHE_MODEL_BOTH:
return true;
default:
break;
}
return false;
}
static bool lastRowScanOptMayBeOptimized(SLogicNode* pNode) {
if (QUERY_NODE_LOGIC_PLAN_AGG != nodeType(pNode) || 1 != LIST_LENGTH(pNode->pChildren) ||
QUERY_NODE_LOGIC_PLAN_SCAN != nodeType(nodesListGetNode(pNode->pChildren, 0))) {
......@@ -2149,16 +2189,27 @@ static bool lastRowScanOptMayBeOptimized(SLogicNode* pNode) {
SScanLogicNode* pScan = (SScanLogicNode*)nodesListGetNode(pNode->pChildren, 0);
// Only one of LAST and LASTROW can appear
if (pAgg->hasLastRow == pAgg->hasLast || NULL != pAgg->pGroupKeys || NULL != pScan->node.pConditions ||
0 == pScan->cacheLastMode || IS_TSWINDOW_SPECIFIED(pScan->scanRange)) {
!hasSuitableCache(pScan->cacheLastMode, pAgg->hasLastRow, pAgg->hasLast) ||
IS_TSWINDOW_SPECIFIED(pScan->scanRange)) {
return false;
}
bool hasLastFunc = false;
bool hasSelectFunc = false;
SNode* pFunc = NULL;
FOREACH(pFunc, ((SAggLogicNode*)pNode)->pAggFuncs) {
if (FUNCTION_TYPE_LAST_ROW != ((SFunctionNode*)pFunc)->funcType &&
// FUNCTION_TYPE_LAST != ((SFunctionNode*)pFunc)->funcType &&
FUNCTION_TYPE_SELECT_VALUE != ((SFunctionNode*)pFunc)->funcType &&
FUNCTION_TYPE_GROUP_KEY != ((SFunctionNode*)pFunc)->funcType) {
SFunctionNode* pAggFunc = (SFunctionNode*)pFunc;
if (FUNCTION_TYPE_LAST == pAggFunc->funcType) {
if (hasSelectFunc || !lastRowScanOptLastParaCheck(nodesListGetNode(pAggFunc->pParameterList, 0))) {
return false;
}
hasLastFunc = true;
} else if (FUNCTION_TYPE_SELECT_VALUE == pAggFunc->funcType || FUNCTION_TYPE_GROUP_KEY == pAggFunc->funcType) {
if (hasLastFunc) {
return false;
}
hasSelectFunc = true;
} else if (FUNCTION_TYPE_LAST_ROW != pAggFunc->funcType) {
return false;
}
}
......@@ -2166,6 +2217,31 @@ static bool lastRowScanOptMayBeOptimized(SLogicNode* pNode) {
return true;
}
typedef struct SLastRowScanOptSetColDataTypeCxt {
bool doAgg;
SNodeList* pLastCols;
} SLastRowScanOptSetColDataTypeCxt;
static EDealRes lastRowScanOptSetColDataType(SNode* pNode, void* pContext) {
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SLastRowScanOptSetColDataTypeCxt* pCxt = pContext;
if (pCxt->doAgg) {
nodesListMakeAppend(&pCxt->pLastCols, pNode);
getLastCacheDataType(&(((SColumnNode*)pNode)->node.resType));
} else {
SNode* pCol = NULL;
FOREACH(pCol, pCxt->pLastCols) {
if (nodesEqualNode(pCol, pNode)) {
getLastCacheDataType(&(((SColumnNode*)pNode)->node.resType));
break;
}
}
}
return DEAL_RES_IGNORE_CHILD;
}
return DEAL_RES_CONTINUE;
}
static int32_t lastRowScanOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan) {
SAggLogicNode* pAgg = (SAggLogicNode*)optFindPossibleNode(pLogicSubplan->pNode, lastRowScanOptMayBeOptimized);
......@@ -2173,22 +2249,36 @@ static int32_t lastRowScanOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogic
return TSDB_CODE_SUCCESS;
}
SNode* pNode = NULL;
SLastRowScanOptSetColDataTypeCxt cxt = {.doAgg = true, .pLastCols = NULL};
SNode* pNode = NULL;
FOREACH(pNode, pAgg->pAggFuncs) {
SFunctionNode* pFunc = (SFunctionNode*)pNode;
if (FUNCTION_TYPE_LAST_ROW == pFunc->funcType || FUNCTION_TYPE_LAST == pFunc->funcType) {
int32_t len = snprintf(pFunc->functionName, sizeof(pFunc->functionName), "_cache_last_row");
int32_t funcType = pFunc->funcType;
if (FUNCTION_TYPE_LAST_ROW == funcType || FUNCTION_TYPE_LAST == funcType) {
int32_t len = snprintf(pFunc->functionName, sizeof(pFunc->functionName),
FUNCTION_TYPE_LAST_ROW == funcType ? "_cache_last_row" : "_cache_last");
pFunc->functionName[len] = '\0';
int32_t code = fmGetFuncInfo(pFunc, NULL, 0);
if (TSDB_CODE_SUCCESS != code) {
nodesClearList(cxt.pLastCols);
return code;
}
if (FUNCTION_TYPE_LAST == funcType) {
nodesWalkExpr(nodesListGetNode(pFunc->pParameterList, 0), lastRowScanOptSetColDataType, &cxt);
}
}
}
SScanLogicNode* pScan = (SScanLogicNode*)nodesListGetNode(pAgg->node.pChildren, 0);
pScan->scanType = SCAN_TYPE_LAST_ROW;
pScan->igLastNull = pAgg->hasLast ? true : false;
if (NULL != cxt.pLastCols) {
cxt.doAgg = false;
nodesWalkExprs(pScan->pScanCols, lastRowScanOptSetColDataType, &cxt);
nodesWalkExprs(pScan->pScanPseudoCols, lastRowScanOptSetColDataType, &cxt);
nodesWalkExprs(pScan->node.pTargets, lastRowScanOptSetColDataType, &cxt);
nodesClearList(cxt.pLastCols);
}
pAgg->hasLastRow = false;
pAgg->hasLast = false;
......
source/libs/planner/test/planBasicTest.cpp
浏览文件 @ 05d04dcd
......@@ -105,24 +105,6 @@ TEST_F(PlanBasicTest, interpFunc) {
run("SELECT _IROWTS, INTERP(c1) FROM t1 RANGE('2017-7-14 18:00:00', '2017-7-14 19:00:00') EVERY(5s) FILL(LINEAR)");
}
TEST_F(PlanBasicTest, lastRowFunc) {
useDb("root", "cache_db");
run("SELECT LAST_ROW(c1) FROM t1");
run("SELECT LAST_ROW(*) FROM t1");
run("SELECT LAST_ROW(c1, c2) FROM t1");
run("SELECT LAST_ROW(c1), c2 FROM t1");
run("SELECT LAST_ROW(c1) FROM st1");
run("SELECT LAST_ROW(c1) FROM st1 PARTITION BY TBNAME");
run("SELECT LAST_ROW(c1), SUM(c3) FROM t1");
}
TEST_F(PlanBasicTest, lastRowFuncWithoutCache) {
useDb("root", "test");
......
source/libs/planner/test/planOptimizeTest.cpp
浏览文件 @ 05d04dcd
......@@ -109,9 +109,28 @@ TEST_F(PlanOptimizeTest, mergeProjects) {
TEST_F(PlanOptimizeTest, pushDownProjectCond) {
useDb("root", "test");
run("select 1-abs(c1) from (select unique(c1) c1 from st1s3) where 1-c1>5 order by 1 nulls first");
}
TEST_F(PlanOptimizeTest, LastRowScan) {
useDb("root", "cache_db");
run("SELECT LAST_ROW(c1), c2 FROM t1");
run("SELECT LAST_ROW(c1), c2, tag1, tbname FROM st1");
run("SELECT LAST_ROW(c1) FROM st1 PARTITION BY TBNAME");
run("SELECT LAST_ROW(c1), SUM(c3) FROM t1");
run("SELECT LAST_ROW(tag1) FROM st1");
run("SELECT LAST(c1) FROM st1");
run("SELECT LAST(c1), c2 FROM st1");
}
TEST_F(PlanOptimizeTest, tagScan) {
useDb("root", "test");
run("select tag1 from st1 group by tag1");
......
source/libs/tdb/src/db/tdbBtree.c
浏览文件 @ 05d04dcd
......@@ -371,7 +371,7 @@ static int tdbDefaultKeyCmprFn(const void *pKey1, int keyLen1, const void *pKey2
}
return cret;
}
/*
static int tdbBtreeOpenImpl(SBTree *pBt) {
// Try to get the root page of the an existing btree
SPgno pgno;
......@@ -400,7 +400,7 @@ static int tdbBtreeOpenImpl(SBTree *pBt) {
pBt->root = pgno;
return 0;
}
*/
int tdbBtreeInitPage(SPage *pPage, void *arg, int init) {
SBTree *pBt;
u8 flags;
......
source/libs/tdb/src/db/tdbPage.c
浏览文件 @ 05d04dcd
......@@ -586,7 +586,7 @@ static inline void setLPageNFree(SPage *pPage, int nFree) {
// cell offset
static inline int getLPageCellOffset(SPage *pPage, int idx) {
ASSERT(idx >= 0 && idx < getPageCellNum(pPage));
ASSERT(idx >= 0 && idx < getLPageCellNum(pPage));
return TDB_GET_U24(pPage->pCellIdx + 3 * idx);
}
......
source/libs/tdb/src/db/tdbPager.c
浏览文件 @ 05d04dcd
......@@ -116,7 +116,7 @@ int tdbPagerClose(SPager *pPager) {
}
return 0;
}
/*
int tdbPagerOpenDB(SPager *pPager, SPgno *ppgno, bool toCreate, SBTree *pBt) {
SPgno pgno;
SPage *pPage;
......@@ -167,7 +167,7 @@ int tdbPagerOpenDB(SPager *pPager, SPgno *ppgno, bool toCreate, SBTree *pBt) {
*ppgno = pgno;
return 0;
}
*/
int tdbPagerWrite(SPager *pPager, SPage *pPage) {
int ret;
SPage **ppPage;
......
source/libs/wal/src/walRead.c
浏览文件 @ 05d04dcd
......@@ -278,12 +278,12 @@ static int32_t walFetchBodyNew(SWalReader *pRead) {
wDebug("vgId:%d, wal starts to fetch body, index:%" PRId64, pRead->pWal->cfg.vgId, ver);
if (pRead->capacity < pReadHead->bodyLen) {
void *ptr = taosMemoryRealloc(pRead->pHead, sizeof(SWalCkHead) + pReadHead->bodyLen);
SWalCkHead *ptr = (SWalCkHead *)taosMemoryRealloc(pRead->pHead, sizeof(SWalCkHead) + pReadHead->bodyLen);
if (ptr == NULL) {
terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
return -1;
}
pRead->pHead = (SWalCkHead *)ptr;
pRead->pHead = ptr;
pReadHead = &pRead->pHead->head;
pRead->capacity = pReadHead->bodyLen;
}
......@@ -398,12 +398,12 @@ int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead) {
int64_t ver = pReadHead->version;
if (pRead->capacity < pReadHead->bodyLen) {
void *ptr = taosMemoryRealloc(*ppHead, sizeof(SWalCkHead) + pReadHead->bodyLen);
SWalCkHead *ptr = (SWalCkHead *)taosMemoryRealloc(*ppHead, sizeof(SWalCkHead) + pReadHead->bodyLen);
if (ptr == NULL) {
terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
return -1;
}
*ppHead = (SWalCkHead *)ptr;
*ppHead = ptr;
pReadHead = &((*ppHead)->head);
pRead->capacity = pReadHead->bodyLen;
}
......@@ -493,13 +493,14 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
}
if (pReader->capacity < pReader->pHead->head.bodyLen) {
void *ptr = taosMemoryRealloc(pReader->pHead, sizeof(SWalCkHead) + pReader->pHead->head.bodyLen);
SWalCkHead *ptr =
(SWalCkHead *)taosMemoryRealloc(pReader->pHead, sizeof(SWalCkHead) + pReader->pHead->head.bodyLen);
if (ptr == NULL) {
terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
taosThreadMutexUnlock(&pReader->mutex);
return -1;
}
pReader->pHead = (SWalCkHead *)ptr;
pReader->pHead = ptr;
pReader->capacity = pReader->pHead->head.bodyLen;
}
......
source/libs/wal/src/walWrite.c
浏览文件 @ 05d04dcd
......@@ -424,7 +424,12 @@ static int32_t walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) {
return -1;
}
ASSERT(taosLSeekFile(pWal->pIdxFile, 0, SEEK_END) == idxOffset + sizeof(SWalIdxEntry) && "Offset of idx entries misaligned");
// check alignment of idx entries
int64_t endOffset = taosLSeekFile(pWal->pIdxFile, 0, SEEK_END);
if (endOffset < 0) {
wFatal("vgId:%d, failed to seek end of idxfile due to %s. ver:%" PRId64 "", pWal->cfg.vgId, strerror(errno), ver);
}
ASSERT(endOffset == idxOffset + sizeof(SWalIdxEntry) && "Offset of idx entries misaligned");
return 0;
}
......
source/util/test/hashTest.cpp
浏览文件 @ 05d04dcd
......@@ -216,7 +216,7 @@ void perfTest() {
char* name = (char*)taosMemoryCalloc(50000000, 9);
for (int64_t i = 0; i < 50000000; ++i) {
sprintf(name + i * 9, "t%08d", i);
sprintf(name + i * 9, "t%08" PRId64, i);
}
for (int64_t i = 0; i < 50; ++i) {
......
tests/script/api/dbTableRoute.c 0 → 100644
浏览文件 @ 05d04dcd
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
// TAOS asynchronous API example
// this example opens multiple tables, insert/retrieve multiple tables
// it is used by TAOS internally for one performance testing
// to compiple: gcc -o asyncdemo asyncdemo.c -ltaos
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include "taos.h"
int rtTables = 20;
char hostName[128];
static void rtExecSQL(TAOS *taos, char *command) {
int i;
int32_t code = -1;
TAOS_RES *pSql = taos_query(taos, command);
code = taos_errno(pSql);
if (code != 0) {
fprintf(stderr, "Failed to run %s, reason: %s\n", command, taos_errstr(pSql));
taos_free_result(pSql);
taos_close(taos);
taos_cleanup();
exit(EXIT_FAILURE);
}
taos_free_result(pSql);
}
static void rtFetchVgId(TAOS *taos, char *sql, int *vgId) {
int i;
int32_t code = -1;
TAOS_RES *pSql = taos_query(taos, sql);
code = taos_errno(pSql);
if (code != 0) {
fprintf(stderr, "Failed to run %s, reason: %s\n", sql, taos_errstr(pSql));
taos_free_result(pSql);
taos_close(taos);
taos_cleanup();
exit(EXIT_FAILURE);
}
TAOS_ROW row = taos_fetch_row(pSql);
*vgId = *(int*)row[0];
taos_free_result(pSql);
}
void rtError(char* prefix, const char* errMsg) {
fprintf(stderr, "%s error: %s\n", prefix, errMsg);
}
void rtExit(char* prefix, const char* errMsg) {
rtError(prefix, errMsg);
exit(1);
}
int rtPrepare(TAOS ** p, int prefix, int suffix) {
char sql[1024] = {0};
int32_t code = 0;
TAOS *taos = taos_connect(hostName, "root", "taosdata", NULL, 0);
if (taos == NULL) rtExit("taos_connect", taos_errstr(NULL));
strcpy(sql, "drop database if exists db1");
rtExecSQL(taos, sql);
sprintf(sql, "create database db1 vgroups 10 table_prefix %d table_suffix %d", prefix, suffix);
rtExecSQL(taos, sql);
strcpy(sql, "use db1");
rtExecSQL(taos, sql);
for (int32_t i = 0; i < rtTables; ++i) {
sprintf(sql, "create table tb%d (ts timestamp, f1 int)", i);
rtExecSQL(taos, sql);
}
*p = taos;
return 0;
}
int rtGetDbRouteInfo(TAOS * taos) {
TAOS_DB_ROUTE_INFO dbInfo;
int code = taos_get_db_route_info(taos, "db1", &dbInfo);
if (code) {
rtExit("taos_get_db_route_info", taos_errstr(NULL));
}
printf("db db1 routeVersion:%d hashPrefix:%d hashSuffix:%d hashMethod:%d vgNum %d\n",
dbInfo.routeVersion, dbInfo.hashPrefix, dbInfo.hashSuffix, dbInfo.hashMethod, dbInfo.vgNum);
for (int32_t i = 0; i < dbInfo.vgNum; ++i) {
printf("%dth vg, id:%d hashBegin:%u hashEnd:%u\n",
i, dbInfo.vgHash[i].vgId, dbInfo.vgHash[i].hashBegin, dbInfo.vgHash[i].hashEnd);
}
return 0;
}
int rtGetTableRouteInfo(TAOS * taos) {
char table[64] = {0};
int vgId1 = 0;
int vgId2 = 0;
char sql[1024] = {0};
for (int32_t i = 0; i < rtTables; ++i) {
sprintf(table, "tb%d", i);
int code = taos_get_table_vgId(taos, "db1", table, &vgId1);
if (code) {
rtExit("taos_get_table_vgId", taos_errstr(NULL));
}
sprintf(sql, "select vgroup_id from information_schema.ins_tables where table_name=\"tb%d\"", i);
rtFetchVgId(taos, sql, &vgId2);
if (vgId1 != vgId2) {
fprintf(stderr, "!!!! table tb%d vgId mis-match, vgId(api):%d, vgId(sys):%d\n", i, vgId1, vgId2);
exit(1);
} else {
printf("table tb%d vgId %d\n", i, vgId1);
}
}
return 0;
}
void rtClose(TAOS * taos) {
taos_close(taos);
}
int rtRunCase1(void) {
TAOS *taos = NULL;
rtPrepare(&taos, 0, 0);
rtGetDbRouteInfo(taos);
rtGetTableRouteInfo(taos);
rtClose(taos);
return 0;
}
int rtRunCase2(void) {
TAOS *taos = NULL;
rtPrepare(&taos, 2, 0);
rtGetTableRouteInfo(taos);
rtGetDbRouteInfo(taos);
rtClose(taos);
return 0;
}
int main(int argc, char *argv[]) {
if (argc != 2) {
printf("usage: %s server-ip\n", argv[0]);
exit(0);
}
srand((unsigned int)time(NULL));
strcpy(hostName, argv[1]);
rtRunCase1();
rtRunCase2();
int32_t l = 5;
while (l) {
printf("%d\n", l--);
sleep(1);
}
return 0;
}
tests/script/api/makefile
浏览文件 @ 05d04dcd
......@@ -13,6 +13,7 @@ all: $(TARGET)
exe:
gcc $(CFLAGS) ./batchprepare.c -o $(ROOT)batchprepare $(LFLAGS)
gcc $(CFLAGS) ./stopquery.c -o $(ROOT)stopquery $(LFLAGS)
gcc $(CFLAGS) ./dbTableRoute.c -o $(ROOT)dbTableRoute $(LFLAGS)
clean:
rm $(ROOT)batchprepare
......
tests/script/tsim/parser/last_cache_query.sim
浏览文件 @ 05d04dcd
......@@ -39,6 +39,7 @@ if $data02 != 5.000000000 then
return -1
endi
if $data03 != 3 then
print expect 3, actual: $data03
return -1
endi
if $data04 != @70-01-01 07:59:57.000@ then
......@@ -210,7 +211,7 @@ if $data01 != 6 then
return -1
endi
if $data02 != 37.000000000 then
print $data02
print expect 37.000000000 actual: $data02
return -1
endi
if $data03 != 27 then
......@@ -233,7 +234,7 @@ if $data01 != 6 then
return -1
endi
if $data02 != 37.000000000 then
print $data02
print expect 37.000000000, acutal: $data02
return -1
endi
if $data03 != 27 then
......
tests/system-test/2-query/last_row.py
浏览文件 @ 05d04dcd
......@@ -13,7 +13,7 @@ class TDTestCase:
def init(self, conn, logSql):
tdLog.debug(f"start to excute {__file__}")
tdSql.init(conn.cursor(), False)
tdSql.init(conn.cursor(), True)
self.tb_nums = 10
self.row_nums = 20
self.ts = 1434938400000
......
tests/system-test/2-query/stablity.py 0 → 100755
浏览文件 @ 05d04dcd
因为 它太大了无法显示 source diff 。你可以改为 查看blob。
tests/system-test/fulltest.sh
浏览文件 @ 05d04dcd
......@@ -127,7 +127,7 @@ python3 ./test.py -f 2-query/leastsquares.py -R
python3 ./test.py -f 2-query/length.py
python3 ./test.py -f 2-query/length.py -R
python3 ./test.py -f 2-query/log.py
# python3 ./test.py -f 2-query/log.py -R
python3 ./test.py -f 2-query/log.py -R
python3 ./test.py -f 2-query/lower.py
python3 ./test.py -f 2-query/lower.py -R
python3 ./test.py -f 2-query/ltrim.py
......@@ -175,7 +175,7 @@ python3 ./test.py -f 2-query/sum.py -R
python3 ./test.py -f 2-query/tail.py
python3 ./test.py -f 2-query/tail.py -R
python3 ./test.py -f 2-query/tan.py
# python3 ./test.py -f 2-query/tan.py -R
python3 ./test.py -f 2-query/tan.py -R
python3 ./test.py -f 2-query/Timediff.py
python3 ./test.py -f 2-query/Timediff.py -R
python3 ./test.py -f 2-query/timetruncate.py
......@@ -221,6 +221,7 @@ python3 ./test.py -f 2-query/json_tag.py
# TD-15983 subquery output duplicate name column.
# Please Xiangyang Guo modify the following script
# python3 ./test.py -f 2-query/nestedQuery_str.py
python3 ./test.py -f 2-query/stablity.py
python3 ./test.py -f 2-query/elapsed.py
python3 ./test.py -f 2-query/csum.py
......
taosws-rs @ 1bdfca39
Subproject commit 1bdfca396cd6730cdc334e06fc7b2156dd1239a0
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或