Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 364132e4 编写于 作者: sangshuduo's avatar sangshuduo
浏览文件
操作

Merge branch '3.0' into feat/sangshuduo/TD-14141-update-taostools-for3.0

上级 52bff98f f80b0669
显示空白变更内容
内联 并排
Showing with 154 addition and 122 deletion
cmake/taosws_CMakeLists.txt.in
浏览文件 @ 364132e4
......@@ -2,7 +2,7 @@
# taosws-rs
ExternalProject_Add(taosws-rs
GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git
GIT_TAG b91b39c
GIT_TAG 6fc47d7
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE
......
docs/en/12-taos-sql/06-select.md
浏览文件 @ 364132e4
......@@ -52,11 +52,6 @@ window_clause: {
| STATE_WINDOW(col)
| INTERVAL(interval_val [, interval_offset]) [SLIDING (sliding_val)] [WATERMARK(watermark_val)] [FILL(fill_mod_and_val)]
changes_option: {
DURATION duration_val
| ROWS rows_val
}
group_by_clause:
GROUP BY expr [, expr] ... HAVING condition
......@@ -126,7 +121,6 @@ SELECT DISTINCT col_name [, col_name ...] FROM tb_name;
1. Configuration parameter `maxNumOfDistinctRes` in `taos.cfg` is used to control the number of rows to output. The minimum configurable value is 100,000, the maximum configurable value is 100,000,000, the default value is 1,000,000. If the actual number of rows exceeds the value of this parameter, only the number of rows specified by this parameter will be output.
2. It can't be guaranteed that the results selected by using `DISTINCT` on columns of `FLOAT` or `DOUBLE` are exactly unique because of the precision errors in floating point numbers.
3. `DISTINCT` can't be used in the sub-query of a nested query statement, and can't be used together with aggregate functions, `GROUP BY` or `JOIN` in the same SQL statement.
:::
......
docs/en/12-taos-sql/10-function.md
浏览文件 @ 364132e4
......@@ -917,7 +917,7 @@ SELECT MAX(field_name) FROM { tb_name | stb_name } [WHERE clause];
**Return value type**:Same as the data type of the column being operated upon
**Applicable data types**: Numeric
**Applicable data types**: Numeric, Timestamp
**Applicable table types**: standard tables and supertables
......@@ -932,7 +932,7 @@ SELECT MIN(field_name) FROM {tb_name | stb_name} [WHERE clause];
**Return value type**:Same as the data type of the column being operated upon
**Applicable data types**: Numeric
**Applicable data types**: Numeric, Timestamp
**Applicable table types**: standard tables and supertables
......
docs/zh/07-develop/_sub_java.mdx
浏览文件 @ 364132e4
```java
{{#include docs/examples/java/src/main/java/com/taos/example/SubscribeDemo.java}}
{{#include docs/examples/java/src/main/java/com/taos/example/MetersDeserializer.java}}
{{#include docs/examples/java/src/main/java/com/taos/example/Meters.java}}
```
```java
{{#include docs/examples/java/src/main/java/com/taos/example/MetersDeserializer.java}}
......
docs/zh/12-taos-sql/06-select.md
浏览文件 @ 364132e4
......@@ -53,11 +53,6 @@ window_clause: {
| STATE_WINDOW(col)
| INTERVAL(interval_val [, interval_offset]) [SLIDING (sliding_val)] [WATERMARK(watermark_val)] [FILL(fill_mod_and_val)]
changes_option: {
DURATION duration_val
| ROWS rows_val
}
group_by_clause:
GROUP BY expr [, expr] ... HAVING condition
......@@ -127,7 +122,6 @@ SELECT DISTINCT col_name [, col_name ...] FROM tb_name;
1. cfg 文件中的配置参数 maxNumOfDistinctRes 将对 DISTINCT 能够输出的数据行数进行限制。其最小值是 100000，最大值是 100000000，默认值是 10000000。如果实际计算结果超出了这个限制，那么会仅输出这个数量范围内的部分。
2. 由于浮点数天然的精度机制原因，在特定情况下，对 FLOAT 和 DOUBLE 列使用 DISTINCT 并不能保证输出值的完全唯一性。
3. 在当前版本下，DISTINCT 不能在嵌套查询的子查询中使用，也不能与聚合函数、GROUP BY、或 JOIN 在同一条语句中混用。
:::
......
include/common/tmsg.h
浏览文件 @ 364132e4
......@@ -1192,6 +1192,7 @@ typedef struct {
int64_t walRetentionSize;
int32_t walRollPeriod;
int64_t walSegmentSize;
int16_t sstTrigger;
} SCreateVnodeReq;
int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
......
source/common/src/systable.c
浏览文件 @ 364132e4
......@@ -102,6 +102,7 @@ static const SSysDbTableSchema userDBSchema[] = {
{.name = "wal_retention_size", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = true},
{.name = "wal_roll_period", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
{.name = "wal_segment_size", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = true},
{.name = "sst_trigger", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT, .sysInfo = true},
};
static const SSysDbTableSchema userFuncSchema[] = {
......@@ -257,10 +258,12 @@ static const SSysDbTableSchema subscriptionSchema[] = {
};
static const SSysDbTableSchema vnodesSchema[] = {
{.name = "dnode_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
{.name = "dnode_endpoint", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
{.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
{.name = "status", .bytes = 20 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = true},
{.name = "replica", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT, .sysInfo = true},
{.name = "status", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
{.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
{.name = "dnode_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
{.name = "dnode_ep", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
};
static const SSysTableMeta infosMeta[] = {
......
source/common/src/tmsg.c
浏览文件 @ 364132e4
......@@ -3766,6 +3766,7 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
if (tEncodeI64(&encoder, pReq->walRetentionSize) < 0) return -1;
if (tEncodeI32(&encoder, pReq->walRollPeriod) < 0) return -1;
if (tEncodeI64(&encoder, pReq->walSegmentSize) < 0) return -1;
if (tEncodeI16(&encoder, pReq->sstTrigger) < 0) return -1;
tEndEncode(&encoder);
......@@ -3838,6 +3839,7 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
if (tDecodeI64(&decoder, &pReq->walRetentionSize) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->walRollPeriod) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->walSegmentSize) < 0) return -1;
if (tDecodeI16(&decoder, &pReq->sstTrigger) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
......
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @ 364132e4
......@@ -167,6 +167,7 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
pCfg->walCfg.segSize = pCreate->walSegmentSize;
pCfg->walCfg.level = pCreate->walLevel;
pCfg->sstTrigger = pCreate->sstTrigger;
pCfg->hashBegin = pCreate->hashBegin;
pCfg->hashEnd = pCreate->hashEnd;
pCfg->hashMethod = pCreate->hashMethod;
......@@ -219,8 +220,9 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
return -1;
}
dDebug("vgId:%d, start to create vnode, tsma:%d standby:%d cacheLast:%d cacheLastSize:%d", createReq.vgId,
createReq.isTsma, createReq.standby, createReq.cacheLast, createReq.cacheLastSize);
dDebug("vgId:%d, start to create vnode, tsma:%d standby:%d cacheLast:%d cacheLastSize:%d sstTrigger:%d",
createReq.vgId, createReq.isTsma, createReq.standby, createReq.cacheLast, createReq.cacheLastSize,
createReq.sstTrigger);
vmGenerateVnodeCfg(&createReq, &vnodeCfg);
if (vmTsmaAdjustDays(&vnodeCfg, &createReq) < 0) {
......
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ 364132e4
......@@ -311,6 +311,7 @@ typedef struct {
int64_t walRetentionSize;
int32_t walRollPeriod;
int64_t walSegmentSize;
int16_t sstTrigger;
} SDbCfg;
typedef struct {
......
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ 364132e4
......@@ -30,7 +30,7 @@
#include "systable.h"
#define DB_VER_NUMBER 1
#define DB_RESERVE_SIZE 64
#define DB_RESERVE_SIZE 62
static SSdbRaw *mndDbActionEncode(SDbObj *pDb);
static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw);
......@@ -124,6 +124,7 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
SDB_SET_INT64(pRaw, dataPos, pDb->cfg.walRetentionSize, _OVER)
SDB_SET_INT32(pRaw, dataPos, pDb->cfg.walRollPeriod, _OVER)
SDB_SET_INT64(pRaw, dataPos, pDb->cfg.walSegmentSize, _OVER)
SDB_SET_INT16(pRaw, dataPos, pDb->cfg.sstTrigger, _OVER)
SDB_SET_RESERVE(pRaw, dataPos, DB_RESERVE_SIZE, _OVER)
SDB_SET_DATALEN(pRaw, dataPos, _OVER)
......@@ -207,6 +208,7 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT64(pRaw, dataPos, &pDb->cfg.walRetentionSize, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pDb->cfg.walRollPeriod, _OVER)
SDB_GET_INT64(pRaw, dataPos, &pDb->cfg.walSegmentSize, _OVER)
SDB_GET_INT16(pRaw, dataPos, &pDb->cfg.sstTrigger, _OVER)
SDB_GET_RESERVE(pRaw, dataPos, DB_RESERVE_SIZE, _OVER)
taosInitRWLatch(&pDb->lock);
......@@ -254,6 +256,7 @@ static int32_t mndDbActionUpdate(SSdb *pSdb, SDbObj *pOld, SDbObj *pNew) {
pOld->cfg.strict = pNew->cfg.strict;
pOld->cfg.cacheLast = pNew->cfg.cacheLast;
pOld->cfg.replications = pNew->cfg.replications;
pOld->cfg.sstTrigger = pNew->cfg.sstTrigger;
taosWUnLockLatch(&pOld->lock);
return 0;
}
......@@ -330,6 +333,7 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
if (pCfg->walRetentionSize < TSDB_DB_MIN_WAL_RETENTION_SIZE) return -1;
if (pCfg->walRollPeriod < TSDB_DB_MIN_WAL_ROLL_PERIOD) return -1;
if (pCfg->walSegmentSize < TSDB_DB_MIN_WAL_SEGMENT_SIZE) return -1;
if (pCfg->sstTrigger < TSDB_MIN_SST_TRIGGER || pCfg->sstTrigger > TSDB_MAX_SST_TRIGGER) return -1;
terrno = 0;
return terrno;
......@@ -363,6 +367,7 @@ static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
pCfg->walRetentionSize = TSDB_REPS_DEF_DB_WAL_RET_SIZE;
if (pCfg->walRollPeriod < 0) pCfg->walRollPeriod = TSDB_REPS_DEF_DB_WAL_ROLL_PERIOD;
if (pCfg->walSegmentSize < 0) pCfg->walSegmentSize = TSDB_DEFAULT_DB_WAL_SEGMENT_SIZE;
if (pCfg->sstTrigger <= 0) pCfg->sstTrigger = TSDB_DEFAULT_SST_TRIGGER;
}
static int32_t mndSetCreateDbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
......@@ -479,6 +484,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
.walRetentionSize = pCreate->walRetentionSize,
.walRollPeriod = pCreate->walRollPeriod,
.walSegmentSize = pCreate->walSegmentSize,
.sstTrigger = pCreate->sstTrigger,
};
dbObj.cfg.numOfRetensions = pCreate->numOfRetensions;
......@@ -1682,6 +1688,9 @@ static void mndDumpDbInfoData(SMnode *pMnode, SSDataBlock *pBlock, SDbObj *pDb,
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.walSegmentSize, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.sstTrigger, false);
}
taosMemoryFree(buf);
......
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @ 364132e4
......@@ -234,6 +234,7 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
createReq.walRetentionSize = pDb->cfg.walRetentionSize;
createReq.walRollPeriod = pDb->cfg.walRollPeriod;
createReq.walSegmentSize = pDb->cfg.walSegmentSize;
createReq.sstTrigger = pDb->cfg.sstTrigger;
for (int32_t v = 0; v < pVgroup->replica; ++v) {
SReplica *pReplica = &createReq.replicas[v];
......@@ -792,31 +793,42 @@ static int32_t mndRetrieveVnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
for (int32_t i = 0; i < pVgroup->replica && numOfRows < rows; ++i) {
SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
SColumnInfoData *pColInfo = NULL;
cols = 0;
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vgId, false);
SName name = {0};
char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&name, pVgroup->dbName, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&name, varDataVal(db));
varDataSetLen(db, strlen(varDataVal(db)));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)db, false);
colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vgId, false);
uint32_t val = 0;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&val, false);
colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->replica, false);
char buf[20] = {0};
STR_TO_VARSTR(buf, syncStr(pVgid->role));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)buf, false);
const char *dbname = mndGetDbStr(pVgroup->dbName);
char b1[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
if (dbname != NULL) {
STR_WITH_MAXSIZE_TO_VARSTR(b1, dbname, TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE);
} else {
STR_WITH_MAXSIZE_TO_VARSTR(b1, "NULL", TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE);
}
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)b1, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pVgid->dnodeId, false);
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
char b2[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0};
if (pDnode != NULL) {
STR_WITH_MAXSIZE_TO_VARSTR(b2, pDnode->ep, TSDB_EP_LEN + VARSTR_HEADER_SIZE);
} else {
STR_WITH_MAXSIZE_TO_VARSTR(b2, "NULL", TSDB_EP_LEN + VARSTR_HEADER_SIZE);
}
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->replica, false); // onlines
colDataAppend(pColInfo, numOfRows, (const char *)b2, false);
numOfRows++;
}
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 364132e4
......@@ -287,6 +287,7 @@ struct SVnodeCfg {
SVnodeStats vndStats;
uint32_t hashBegin;
uint32_t hashEnd;
int16_t sstTrigger;
};
typedef struct {
......
source/dnode/vnode/src/vnd/vnodeCfg.c
浏览文件 @ 364132e4
......@@ -106,6 +106,7 @@ int vnodeEncodeConfig(const void *pObj, SJson *pJson) {
if (tjsonAddIntegerToObject(pJson, "wal.retentionSize", pCfg->walCfg.retentionSize) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "wal.segSize", pCfg->walCfg.segSize) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "wal.level", pCfg->walCfg.level) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "sstTrigger", pCfg->sstTrigger) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "hashBegin", pCfg->hashBegin) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "hashEnd", pCfg->hashEnd) < 0) return -1;
if (tjsonAddIntegerToObject(pJson, "hashMethod", pCfg->hashMethod) < 0) return -1;
......@@ -205,6 +206,8 @@ int vnodeDecodeConfig(const SJson *pJson, void *pObj) {
if (code < 0) return -1;
tjsonGetNumberValue(pJson, "wal.level", pCfg->walCfg.level, code);
if (code < 0) return -1;
tjsonGetNumberValue(pJson, "sstTrigger", pCfg->sstTrigger, code);
if (code < 0) return -1;
tjsonGetNumberValue(pJson, "hashBegin", pCfg->hashBegin, code);
if (code < 0) return -1;
tjsonGetNumberValue(pJson, "hashEnd", pCfg->hashEnd, code);
......
source/libs/index/inc/indexInt.h
浏览文件 @ 364132e4
......@@ -40,26 +40,31 @@ extern "C" {
#define indexTrace(...) do { if (idxDebugFlag & DEBUG_TRACE) { taosPrintLog("IDX", DEBUG_TRACE, idxDebugFlag, __VA_ARGS__);} } while (0)
// clang-format on
extern void* indexQhandle;
typedef enum { LT, LE, GT, GE, CONTAINS, EQ } RangeType;
typedef enum { kTypeValue, kTypeDeletion } STermValueType;
typedef enum { kRebuild, kFinished } SIdxStatus;
typedef struct SIndexStat {
int32_t totalAdded; //
int32_t totalDeled; //
int32_t totalUpdated; //
int32_t totalTerms; //
int32_t distinctCol; // distinct column
int32_t total;
int32_t add; //
int32_t del; //
int32_t update; //
int32_t terms; //
int32_t distCol; // distinct column
} SIndexStat;
struct SIndex {
SIndexOpts opts;
int64_t refId;
void* cache;
void* tindex;
SHashObj* colObj; // < field name, field id>
int64_t suid; // current super table id, -1 is normal table
int32_t cVersion; // current version allocated to cache
int32_t version; // current version allocated to cache
SLRUCache* lru;
char* path;
......@@ -68,7 +73,6 @@ struct SIndex {
TdThreadMutex mtx;
tsem_t sem;
bool quit;
SIndexOpts opts;
};
struct SIndexMultiTermQuery {
......@@ -111,14 +115,15 @@ typedef struct Iterate {
void iterateValueDestroy(IterateValue* iv, bool destroy);
extern void* indexQhandle;
typedef struct TFileCacheKey {
uint64_t suid;
uint8_t colType;
char* colName;
int32_t nColName;
} ICacheKey;
int32_t idxSerialCacheKey(ICacheKey* key, char* buf);
int idxFlushCacheToTFile(SIndex* sIdx, void*, bool quit);
int64_t idxAddRef(void* p);
......@@ -126,10 +131,6 @@ int32_t idxRemoveRef(int64_t ref);
void idxAcquireRef(int64_t ref);
void idxReleaseRef(int64_t ref);
int32_t idxSerialCacheKey(ICacheKey* key, char* buf);
// int32_t indexSerialKey(ICacheKey* key, char* buf);
// int32_t indexSerialTermKey(SIndexTerm* itm, char* buf);
#define IDX_TYPE_CONTAIN_EXTERN_TYPE(ty, exTy) (((ty >> 4) & (exTy)) != 0)
#define IDX_TYPE_GET_TYPE(ty) (ty & 0x0F)
......
source/libs/index/src/index.c
浏览文件 @ 364132e4
......@@ -25,10 +25,6 @@
#include "tref.h"
#include "tsched.h"
#ifdef USE_LUCENE
#include "lucene++/Lucene_c.h"
#endif
#define INDEX_NUM_OF_THREADS 5
#define INDEX_QUEUE_SIZE 200
......@@ -74,7 +70,7 @@ void indexCleanup() {
typedef struct SIdxColInfo {
int colId; // generated by index internal
int cVersion;
int version;
} SIdxColInfo;
static TdThreadOnce isInit = PTHREAD_ONCE_INIT;
......@@ -123,7 +119,7 @@ int indexOpen(SIndexOpts* opts, const char* path, SIndex** index) {
}
idx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
idx->cVersion = 1;
idx->version = 1;
idx->path = tstrdup(path);
taosThreadMutexInit(&idx->mtx, NULL);
tsem_init(&idx->sem, 0, 0);
......
source/libs/index/src/indexCache.c
浏览文件 @ 364132e4
......@@ -566,7 +566,6 @@ int idxCachePut(void* cache, SIndexTerm* term, uint64_t uid) {
taosThreadMutexUnlock(&pCache->mtx);
idxCacheUnRef(pCache);
return 0;
// encode end
}
void idxCacheForceToMerge(void* cache) {
IndexCache* pCache = cache;
......@@ -602,10 +601,10 @@ static int32_t idxQueryMem(MemTable* mem, SIndexTermQuery* query, SIdxTRslt* tr,
}
}
int idxCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* result, STermValueType* s) {
int64_t st = taosGetTimestampUs();
if (cache == NULL) {
return 0;
}
IndexCache* pCache = cache;
MemTable *mem = NULL, *imm = NULL;
......@@ -616,6 +615,8 @@ int idxCacheSearch(void* cache, SIndexTermQuery* query, SIdxTRslt* result, STerm
idxMemRef(imm);
taosThreadMutexUnlock(&pCache->mtx);
int64_t st = taosGetTimestampUs();
int ret = (mem && mem->mem) ? idxQueryMem(mem, query, result, s) : 0;
if (ret == 0 && *s != kTypeDeletion) {
// continue search in imm
......
source/libs/index/src/indexComm.c
浏览文件 @ 364132e4
......@@ -178,9 +178,9 @@ TExeCond tDoCompare(__compar_fn_t func, int8_t comparType, void* a, void* b) {
// optime later
int32_t ret = func(a, b);
switch (comparType) {
case QUERY_LESS_THAN: {
case QUERY_LESS_THAN:
if (ret < 0) return MATCH;
} break;
break;
case QUERY_LESS_EQUAL: {
if (ret <= 0) return MATCH;
break;
......
source/libs/index/src/indexFilter.c
浏览文件 @ 364132e4
......@@ -27,6 +27,44 @@
#define SIF_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define SIF_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
// clang-format on
typedef union {
uint8_t u8;
uint16_t u16;
uint32_t u32;
uint64_t u64;
int8_t i8;
int16_t i16;
int32_t i32;
int64_t i64;
double d;
float f;
} SDataTypeBuf;
#define SIF_DATA_CONVERT(type, val, dst) \
do { \
if (type == TSDB_DATA_TYPE_DOUBLE) \
dst = GET_DOUBLE_VAL(val); \
else if (type == TSDB_DATA_TYPE_BIGINT) \
dst = *(int64_t *)val; \
else if (type == TSDB_DATA_TYPE_INT) \
dst = *(int32_t *)val; \
else if (type == TSDB_DATA_TYPE_SMALLINT) \
dst = *(int16_t *)val; \
else if (type == TSDB_DATA_TYPE_TINYINT) \
dst = *(int8_t *)val; \
else if (type == TSDB_DATA_TYPE_UTINYINT) \
dst = *(uint8_t *)val; \
else if (type == TSDB_DATA_TYPE_USMALLINT) \
dst = *(uint16_t *)val; \
else if (type == TSDB_DATA_TYPE_UINT) \
dst = *(uint32_t *)val; \
else if (type == TSDB_DATA_TYPE_UBIGINT) \
dst = *(uint64_t *)val; \
} while (0);
typedef struct SIFParam {
SHashObj *pFilter;
......@@ -48,7 +86,6 @@ typedef struct SIFCtx {
SHashObj *pRes; /* element is SIFParam */
bool noExec; // true: just iterate condition tree, and add hint to executor plan
SIndexMetaArg arg;
// SIdxFltStatus st;
} SIFCtx;
static int32_t sifGetFuncFromSql(EOperatorType src, EIndexQueryType *dst) {
......@@ -75,11 +112,6 @@ static int32_t sifGetFuncFromSql(EOperatorType src, EIndexQueryType *dst) {
typedef int32_t (*sif_func_t)(SIFParam *left, SIFParam *rigth, SIFParam *output);
static sif_func_t sifNullFunc = NULL;
// typedef struct SIFWalkParm
// construct tag filter operator later
// static void destroyTagFilterOperatorInfo(void *param) {
// STagFilterOperatorInfo *pInfo = (STagFilterOperatorInfo *)param;
//}
static void sifFreeParam(SIFParam *param) {
if (param == NULL) return;
......@@ -365,42 +397,6 @@ static Filter sifGetFilterFunc(EIndexQueryType type, bool *reverse) {
}
return NULL;
}
typedef union {
uint8_t u8;
uint16_t u16;
uint32_t u32;
uint64_t u64;
int8_t i8;
int16_t i16;
int32_t i32;
int64_t i64;
double d;
float f;
} SDataTypeBuf;
#define SIF_DATA_CONVERT(type, val, dst) \
do { \
if (type == TSDB_DATA_TYPE_DOUBLE) \
dst = GET_DOUBLE_VAL(val); \
else if (type == TSDB_DATA_TYPE_BIGINT) \
dst = *(int64_t *)val; \
else if (type == TSDB_DATA_TYPE_INT) \
dst = *(int32_t *)val; \
else if (type == TSDB_DATA_TYPE_SMALLINT) \
dst = *(int16_t *)val; \
else if (type == TSDB_DATA_TYPE_TINYINT) \
dst = *(int8_t *)val; \
else if (type == TSDB_DATA_TYPE_UTINYINT) \
dst = *(uint8_t *)val; \
else if (type == TSDB_DATA_TYPE_USMALLINT) \
dst = *(uint16_t *)val; \
else if (type == TSDB_DATA_TYPE_UINT) \
dst = *(uint32_t *)val; \
else if (type == TSDB_DATA_TYPE_UBIGINT) \
dst = *(uint64_t *)val; \
} while (0);
static void sifSetFltParam(SIFParam *left, SIFParam *right, SDataTypeBuf *typedata, SMetaFltParam *param) {
int8_t ltype = left->colValType, rtype = right->colValType;
......@@ -693,11 +689,8 @@ static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *ou
for (int32_t m = 0; m < node->pParameterList->length; m++) {
if (node->condType == LOGIC_COND_TYPE_AND) {
taosArrayAddAll(output->result, params[m].result);
// taosArrayDestroy(params[m].result);
// params[m].result = NULL;
} else if (node->condType == LOGIC_COND_TYPE_OR) {
taosArrayAddAll(output->result, params[m].result);
// params[m].result = NULL;
} else if (node->condType == LOGIC_COND_TYPE_NOT) {
// taosArrayAddAll(output->result, params[m].result);
}
......
source/libs/index/src/indexFstFile.c
浏览文件 @ 364132e4
......@@ -211,9 +211,7 @@ IdxFstFile* idxFileCreate(void* wrt) {
return cw;
}
void idxFileDestroy(IdxFstFile* cw) {
// free wrt object: close fd or free mem
idxFileFlush(cw);
// idxFileCtxDestroy((IFileCtx *)(cw->wrt));
taosMemoryFree(cw);
}
......@@ -222,9 +220,7 @@ int idxFileWrite(IdxFstFile* write, uint8_t* buf, uint32_t len) {
return 0;
}
// update checksum
// write data to file/socket or mem
IFileCtx* ctx = write->wrt;
int nWrite = ctx->write(ctx, buf, len);
assert(nWrite == len);
write->count += len;
......
source/libs/index/src/indexTfile.c
浏览文件 @ 364132e4
......@@ -183,13 +183,14 @@ TFileReader* tfileReaderCreate(IFileCtx* ctx) {
return NULL;
}
reader->ctx = ctx;
reader->remove = false;
if (0 != tfileReaderVerify(reader)) {
indexError("invalid tfile, suid:%" PRIu64 ", colName:%s", reader->header.suid, reader->header.colName);
tfileReaderDestroy(reader);
return NULL;
}
// T_REF_INC(reader);
if (0 != tfileReaderLoadHeader(reader)) {
indexError("failed to load index header, suid:%" PRIu64 ", colName:%s", reader->header.suid,
reader->header.colName);
......@@ -203,7 +204,6 @@ TFileReader* tfileReaderCreate(IFileCtx* ctx) {
tfileReaderDestroy(reader);
return NULL;
}
reader->remove = false;
return reader;
}
......@@ -211,7 +211,6 @@ void tfileReaderDestroy(TFileReader* reader) {
if (reader == NULL) {
return;
}
// T_REF_INC(reader);
fstDestroy(reader->fst);
if (reader->remove) {
indexInfo("%s is removed", reader->ctx->file.buf);
......@@ -222,6 +221,7 @@ void tfileReaderDestroy(TFileReader* reader) {
taosMemoryFree(reader);
}
static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTRslt* tr) {
int ret = 0;
char* p = tem->colVal;
......@@ -494,7 +494,6 @@ int tfileReaderSearch(TFileReader* reader, SIndexTermQuery* query, SIdxTRslt* tr
TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int64_t version, const char* colName, uint8_t colType) {
char fullname[256] = {0};
tfileGenFileFullName(fullname, path, suid, colName, version);
// indexInfo("open write file name %s", fullname);
IFileCtx* wcx = idxFileCtxCreate(TFILE, fullname, false, 1024 * 1024 * 64);
if (wcx == NULL) {
return NULL;
......@@ -503,8 +502,8 @@ TFileWriter* tfileWriterOpen(char* path, uint64_t suid, int64_t version, const c
TFileHeader tfh = {0};
tfh.suid = suid;
tfh.version = version;
memcpy(tfh.colName, colName, strlen(colName));
tfh.colType = colType;
memcpy(tfh.colName, colName, strlen(colName));
return tfileWriterCreate(wcx, &tfh);
}
......@@ -706,7 +705,6 @@ static bool tfileIteratorNext(Iterate* iiter) {
iv->type = ADD_VALUE; // value in tfile always ADD_VALUE
iv->colVal = colVal;
return true;
// std::string key(ch, sz);
}
static IterateValue* tifileIterateGetValue(Iterate* iter) { return &iter->val; }
......
source/libs/transport/src/transCli.c
浏览文件 @ 364132e4
......@@ -1304,11 +1304,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
pTransInst->cfp(pTransInst->parent, pResp, NULL);
return 0;
}
/*
* no retry
* 1. query conn
* 2. rpc thread already receive quit msg
*/
STransConnCtx* pCtx = pMsg->ctx;
int32_t code = pResp->code;
......
tests/script/tsim/db/basic1.sim
浏览文件 @ 364132e4
......@@ -107,6 +107,37 @@ if $data30 != 12 then
return -1
endi
print =============== show vnodes
sql show vnodes 1
if $rows != 9 then
return -1
endi
if $data(4)[1] != 1 then
return -1
endi
if $data(4)[2] != leader then
return -1
endi
if $data(4)[3] != d2 then
return -1
endi
if $data(4)[4] != 1 then
return -1
endi
if $data(4)[5] != localhost:7100 then
return -1
endi
#sql show vnodes 'localhost:7100'
#if $rows != 9 then
# return -1
#endi
print =============== drop database
sql drop database d2
sql drop database d3
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或