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

conn timeout refactor

This commit is contained in:
yihaoDeng 2022-11-11 16:11:37 +08:00
parent 50f148abac 672384eca1
commit 8fa39accdf
84 changed files with 3266 additions and 2810 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
cmake/cmake.platform
View File

@ -102,6 +102,12 @@ IF ("${CPUTYPE}" STREQUAL "")
SET(TD_ARM_64 TRUE)
ADD_DEFINITIONS("-D_TD_ARM_")
ADD_DEFINITIONS("-D_TD_ARM_64")
ELSEIF (CMAKE_SYSTEM_PROCESSOR MATCHES "loongarch64")
MESSAGE(STATUS "The current platform is loongarch64")
SET(PLATFORM_ARCH_STR "loongarch64")
SET(TD_LOONGARCH_64 TRUE)
ADD_DEFINITIONS("-D_TD_LOONGARCH_")
ADD_DEFINITIONS("-D_TD_LOONGARCH_64")
ENDIF ()
ELSE ()
# if generate ARM version:
@ -118,6 +124,12 @@ ELSE ()
ADD_DEFINITIONS("-D_TD_ARM_")
ADD_DEFINITIONS("-D_TD_ARM_64")
SET(TD_ARM_64 TRUE)
ELSEIF (${CPUTYPE} MATCHES "loongarch64")
SET(PLATFORM_ARCH_STR "loongarch64")
MESSAGE(STATUS "input cpuType: loongarch64")
SET(TD_LOONGARCH_64 TRUE)
ADD_DEFINITIONS("-D_TD_LOONGARCH_")
ADD_DEFINITIONS("-D_TD_LOONGARCH_64")
ELSEIF (${CPUTYPE} MATCHES "mips64")
SET(PLATFORM_ARCH_STR "mips")
MESSAGE(STATUS "input cpuType: mips64")

2
cmake/taostools_CMakeLists.txt.in
View File

@ -2,7 +2,7 @@
# taos-tools
ExternalProject_Add(taos-tools
GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
GIT_TAG a921bd4
GIT_TAG 23e2b73
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
BINARY_DIR ""
#BUILD_IN_SOURCE TRUE

2
docs/en/14-reference/09-support-platform/index.md
View File

@ -14,7 +14,7 @@ Note: ● means officially tested and verified, ○ means unofficially tested an
## List of supported platforms for TDengine clients and connectors
TDengine's connector can support a wide range of platforms, including X64/X86/ARM64/ARM32/MIPS/Alpha hardware platforms and Linux/Win64/Win32/macOS development environments.
TDengine's connector can support a wide range of platforms, including X64/X86/ARM64/ARM32/MIPS/Alpha/LoongArch64 hardware platforms and Linux/Win64/Win32/macOS development environments.
The comparison matrix is as follows.

6
docs/zh/05-get-started/index.md
View File

@ -13,3 +13,9 @@ import {useCurrentSidebarCategory} from '@docusaurus/theme-common';
<DocCardList items={useCurrentSidebarCategory().items}/>
```
### 开发者技术交流群
微信扫描下面二维码，加“小 T”为好友，即可加入“物联网大数据技术前沿群”，与大家共同交流物联网大数据技术应用、TDengine 使用问题和技巧等话题。
<img src="./tdengine.webp" width="200" />

BIN
docs/zh/05-get-started/tdengine.webp Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 24 KiB

2
docs/zh/14-reference/09-support-platform/index.md
View File

@ -16,7 +16,7 @@ description: "TDengine 服务端、客户端和连接器支持的平台列表"
## TDengine 客户端和连接器支持的平台列表
目前 TDengine 的连接器可支持的平台广泛，目前包括：X64/X86/ARM64/ARM32/MIPS/Alpha 等硬件平台，以及 Linux/Win64/Win32/macOS 等开发环境。
目前 TDengine 的连接器可支持的平台广泛，目前包括：X64/X86/ARM64/ARM32/MIPS/LoongArch64 等硬件平台，以及 Linux/Win64/Win32/macOS 等开发环境。
对照矩阵如下：

3
include/common/tglobal.h
View File

@ -135,6 +135,9 @@ extern int32_t tsTtlPushInterval;
extern int32_t tsGrantHBInterval;
extern int32_t tsUptimeInterval;
extern int32_t tsRpcRetryLimit;
extern int32_t tsRpcRetryInterval;
//#define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd,

110
include/common/ttokendef.h
View File

@ -174,61 +174,61 @@
#define TK_SUBSCRIPTIONS 156
#define TK_VNODES 157
#define TK_LIKE 158
#define TK_INDEX 159
#define TK_FUNCTION 160
#define TK_INTERVAL 161
#define TK_TOPIC 162
#define TK_AS 163
#define TK_WITH 164
#define TK_META 165
#define TK_CONSUMER 166
#define TK_GROUP 167
#define TK_DESC 168
#define TK_DESCRIBE 169
#define TK_RESET 170
#define TK_QUERY 171
#define TK_CACHE 172
#define TK_EXPLAIN 173
#define TK_ANALYZE 174
#define TK_VERBOSE 175
#define TK_NK_BOOL 176
#define TK_RATIO 177
#define TK_NK_FLOAT 178
#define TK_OUTPUTTYPE 179
#define TK_AGGREGATE 180
#define TK_BUFSIZE 181
#define TK_STREAM 182
#define TK_INTO 183
#define TK_TRIGGER 184
#define TK_AT_ONCE 185
#define TK_WINDOW_CLOSE 186
#define TK_IGNORE 187
#define TK_EXPIRED 188
#define TK_FILL_HISTORY 189
#define TK_SUBTABLE 190
#define TK_KILL 191
#define TK_CONNECTION 192
#define TK_TRANSACTION 193
#define TK_BALANCE 194
#define TK_VGROUP 195
#define TK_MERGE 196
#define TK_REDISTRIBUTE 197
#define TK_SPLIT 198
#define TK_DELETE 199
#define TK_INSERT 200
#define TK_NULL 201
#define TK_NK_QUESTION 202
#define TK_NK_ARROW 203
#define TK_ROWTS 204
#define TK_TBNAME 205
#define TK_QSTART 206
#define TK_QEND 207
#define TK_QDURATION 208
#define TK_WSTART 209
#define TK_WEND 210
#define TK_WDURATION 211
#define TK_IROWTS 212
#define TK_QTAGS 213
#define TK_TBNAME 159
#define TK_QTAGS 160
#define TK_AS 161
#define TK_INDEX 162
#define TK_FUNCTION 163
#define TK_INTERVAL 164
#define TK_TOPIC 165
#define TK_WITH 166
#define TK_META 167
#define TK_CONSUMER 168
#define TK_GROUP 169
#define TK_DESC 170
#define TK_DESCRIBE 171
#define TK_RESET 172
#define TK_QUERY 173
#define TK_CACHE 174
#define TK_EXPLAIN 175
#define TK_ANALYZE 176
#define TK_VERBOSE 177
#define TK_NK_BOOL 178
#define TK_RATIO 179
#define TK_NK_FLOAT 180
#define TK_OUTPUTTYPE 181
#define TK_AGGREGATE 182
#define TK_BUFSIZE 183
#define TK_STREAM 184
#define TK_INTO 185
#define TK_TRIGGER 186
#define TK_AT_ONCE 187
#define TK_WINDOW_CLOSE 188
#define TK_IGNORE 189
#define TK_EXPIRED 190
#define TK_FILL_HISTORY 191
#define TK_SUBTABLE 192
#define TK_KILL 193
#define TK_CONNECTION 194
#define TK_TRANSACTION 195
#define TK_BALANCE 196
#define TK_VGROUP 197
#define TK_MERGE 198
#define TK_REDISTRIBUTE 199
#define TK_SPLIT 200
#define TK_DELETE 201
#define TK_INSERT 202
#define TK_NULL 203
#define TK_NK_QUESTION 204
#define TK_NK_ARROW 205
#define TK_ROWTS 206
#define TK_QSTART 207
#define TK_QEND 208
#define TK_QDURATION 209
#define TK_WSTART 210
#define TK_WEND 211
#define TK_WDURATION 212
#define TK_IROWTS 213
#define TK_CAST 214
#define TK_NOW 215
#define TK_TODAY 216

7
include/libs/nodes/cmdnodes.h
View File

@ -284,6 +284,13 @@ typedef struct SShowVnodesStmt {
SNode* pDnodeEndpoint;
} SShowVnodesStmt;
typedef struct SShowTableTagsStmt {
ENodeType type;
SNode* pDbName; // SValueNode
SNode* pTbName; // SValueNode
SNodeList* pTags;
} SShowTableTagsStmt;
typedef enum EIndexType { INDEX_TYPE_SMA = 1, INDEX_TYPE_FULLTEXT } EIndexType;
typedef struct SIndexOptions {

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

@ -297,6 +297,7 @@ typedef struct SSelectStmt {
bool hasStateKey;
bool onlyHasKeepOrderFunc;
bool groupSort;
bool tagScan;
} SSelectStmt;
typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;

2
include/libs/transport/trpc.h
View File

@ -82,6 +82,8 @@ typedef struct SRpcInit {
int32_t sessions; // number of sessions allowed
int8_t connType; // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
int32_t idleTime; // milliseconds, 0 means idle timer is disabled
int32_t retryLimit; // retry limit
int32_t retryInterval; // retry interval ms
int32_t compressSize; // -1: no compress, 0 : all data compressed, size: compress data if larger than size
int8_t encryption; // encrypt or not

2
include/util/tdef.h
View File

@ -26,7 +26,7 @@ extern "C" {
#define TSKEY int64_t
#define TSKEY_MIN INT64_MIN
#define TSKEY_MAX (INT64_MAX - 1)
#define TSKEY_MAX INT64_MAX
#define TSKEY_INITIAL_VAL TSKEY_MIN
#define TD_VER_MAX UINT64_MAX // TODO: use the real max version from query handle

4
packaging/docker/dockerbuild.sh
View File

@ -5,7 +5,7 @@ set -e
#set -x
# dockerbuild.sh
# -c [aarch32 | aarch64 | amd64 | x86 | mips64 ...]
# -c [aarch32 | aarch64 | amd64 | x86 | mips64 | loongarch64...]
# -n [version number]
# -p [password for docker hub]
# -V [stable | beta]
@ -57,7 +57,7 @@ do
dockerLatest=$(echo $OPTARG)
;;
h)
echo "Usage: `basename $0` -c [aarch32 | aarch64 | amd64 | x86 | mips64 ...] "
echo "Usage: `basename $0` -c [aarch32 | aarch64 | amd64 | x86 | mips64 | loongarch64...] "
echo " -n [version number] "
echo " -p [password for docker hub] "
echo " -V [stable | beta] "

4
packaging/docker/dockerbuildi.sh
View File

@ -5,7 +5,7 @@ set -e
#set -x
# dockerbuild.sh
# -c [aarch32 | aarch64 | amd64 | x86 | mips64 ...]
# -c [aarch32 | aarch64 | amd64 | x86 | mips64 | loongarch64...]
# -n [version number]
# -p [password for docker hub]
@ -30,7 +30,7 @@ do
passWord=$(echo $OPTARG)
;;
h)
echo "Usage: `basename $0` -c [aarch32 | aarch64 | amd64 | x86 | mips64 ...] "
echo "Usage: `basename $0` -c [aarch32 | aarch64 | amd64 | x86 | mips64 | loongarch64...] "
echo " -n [version number] "
echo " -p [password for docker hub] "
exit 0

8
packaging/release.sh
View File

@ -6,7 +6,7 @@ set -e
#set -x
# release.sh -v [cluster | edge]
# -c [aarch32 | aarch64 | x64 | x86 | mips64 ...]
# -c [aarch32 | aarch64 | x64 | x86 | mips64 | loongarch64...]
# -o [Linux | Kylin | Alpine | Raspberrypi | Darwin | Windows | Ningsi60 | Ningsi80 |...]
# -V [stable | beta]
# -l [full | lite]
@ -19,7 +19,7 @@ set -e
# set parameters by default value
verMode=edge # [cluster, edge, cloud]
verType=stable # [stable, beta]
cpuType=x64 # [aarch32 | aarch64 | x64 | x86 | mips64 ...]
cpuType=x64 # [aarch32 | aarch64 | x64 | x86 | mips64 loongarch64...]
osType=Linux # [Linux | Kylin | Alpine | Raspberrypi | Darwin | Windows | Ningsi60 | Ningsi80 |...]
pagMode=full # [full | lite]
soMode=dynamic # [static | dynamic]
@ -77,7 +77,7 @@ while getopts "hv:V:c:o:l:s:d:a:n:m:H:" arg; do
;;
h)
echo "Usage: $(basename $0) -v [cluster | edge] "
echo " -c [aarch32 | aarch64 | x64 | x86 | mips64 ...] "
echo " -c [aarch32 | aarch64 | x64 | x86 | mips64 | loongarch64 ...] "
echo " -o [Linux | Kylin | Alpine | Raspberrypi | Darwin | Windows | Ningsi60 | Ningsi80 |...] "
echo " -V [stable | beta] "
echo " -l [full | lite] "
@ -216,7 +216,7 @@ else
fi
# check support cpu type
if [[ "$cpuType" == "x64" ]] || [[ "$cpuType" == "aarch64" ]] || [[ "$cpuType" == "aarch32" ]] || [[ "$cpuType" == "arm64" ]] || [[ "$cpuType" == "arm32" ]] || [[ "$cpuType" == "mips64" ]]; then
if [[ "$cpuType" == "x64" ]] || [[ "$cpuType" == "aarch64" ]] || [[ "$cpuType" == "aarch32" ]] || [[ "$cpuType" == "arm64" ]] || [[ "$cpuType" == "arm32" ]] || [[ "$cpuType" == "mips64" ]] || [[ "$cpuType" == "loongarch64" ]] ; then
if [ "$verMode" == "edge" ]; then
# community-version compile
cmake ../ -DCPUTYPE=${cpuType} -DWEBSOCKET=true -DOSTYPE=${osType} -DSOMODE=${soMode} -DDBNAME=${dbName} -DVERTYPE=${verType} -DVERDATE="${build_time}" -DGITINFO=${gitinfo} -DGITINFOI=${gitinfoOfInternal} -DVERNUMBER=${verNumber} -DVERCOMPATIBLE=${verNumberComp} -DPAGMODE=${pagMode} -DBUILD_HTTP=${BUILD_HTTP} -DBUILD_TOOLS=${BUILD_TOOLS} ${allocator_macro}

2
source/client/src/clientEnv.c
View File

@ -146,6 +146,8 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.compressSize = tsCompressMsgSize;
rpcInit.dfp = destroyAhandle;
rpcInit.retryLimit = tsRpcRetryLimit;
rpcInit.retryInterval = tsRpcRetryInterval;
void *pDnodeConn = rpcOpen(&rpcInit);
if (pDnodeConn == NULL) {

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

@ -1971,6 +1971,8 @@ TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* de
rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.compressSize = tsCompressMsgSize;
rpcInit.user = "_dnd";
rpcInit.retryLimit = tsRpcRetryLimit;
rpcInit.retryInterval = tsRpcRetryInterval;
clientRpc = rpcOpen(&rpcInit);
if (clientRpc == NULL) {

2
source/client/src/clientSml.c
View File

@ -2082,7 +2082,7 @@ static int32_t smlParseJSONString(SSmlHandle *info, cJSON *root, SSmlTableInfo *
static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql, const int len) {
SSmlLineInfo elements = {0};
uDebug("SML:0x%" PRIx64 " smlParseInfluxLine sql:%s, hello", info->id, sql);
uDebug("SML:0x%" PRIx64 " smlParseInfluxLine sql", info->id);
int ret = smlParseInfluxString(sql, sql + len, &elements, &info->msgBuf);
if (ret != TSDB_CODE_SUCCESS) {

14
source/common/src/tglobal.c
View File

@ -167,6 +167,8 @@ int32_t tsUptimeInterval = 300; // seconds
char tsUdfdResFuncs[512] = ""; // udfd resident funcs that teardown when udfd exits
char tsUdfdLdLibPath[512] = "";
int32_t tsRpcRetryLimit = 100;
int32_t tsRpcRetryInterval = 15;
#ifndef _STORAGE
int32_t taosSetTfsCfg(SConfig *pCfg) {
SConfigItem *pItem = cfgGetItem(pCfg, "dataDir");
@ -297,6 +299,8 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, 1) != 0) return -1;
if (cfgAddBool(pCfg, "smlDataFormat", tsSmlDataFormat, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "maxMemUsedByInsert", tsMaxMemUsedByInsert, 1, INT32_MAX, true) != 0) return -1;
if (cfgAddInt32(pCfg, "rpcRetryLimit", tsRpcRetryLimit, 1, 100000, 0) != 0) return -1;
if (cfgAddInt32(pCfg, "rpcRetryInterval", tsRpcRetryInterval, 1, 100000, 0) != 0) return -1;
tsNumOfTaskQueueThreads = tsNumOfCores / 2;
tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4);
@ -422,6 +426,10 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddBool(pCfg, "udf", tsStartUdfd, 0) != 0) return -1;
if (cfgAddString(pCfg, "udfdResFuncs", tsUdfdResFuncs, 0) != 0) return -1;
if (cfgAddString(pCfg, "udfdLdLibPath", tsUdfdLdLibPath, 0) != 0) return -1;
if (cfgAddInt32(pCfg, "rpcRetryLimit", tsRpcRetryLimit, 1, 100000, 0) != 0) return -1;
if (cfgAddInt32(pCfg, "rpcRetryInterval", tsRpcRetryInterval, 1, 100000, 0) != 0) return -1;
GRANT_CFG_ADD;
return 0;
}
@ -634,6 +642,9 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
tsQueryNodeChunkSize = cfgGetItem(pCfg, "queryNodeChunkSize")->i32;
tsQueryUseNodeAllocator = cfgGetItem(pCfg, "queryUseNodeAllocator")->bval;
tsKeepColumnName = cfgGetItem(pCfg, "keepColumnName")->bval;
tsRpcRetryLimit = cfgGetItem(pCfg, "rpcRetryLimit")->i32;
tsRpcRetryInterval = cfgGetItem(pCfg, "rpcRetryInterval")->i32;
return 0;
}
@ -708,6 +719,9 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
if (tsQueryBufferSize >= 0) {
tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;
}
tsRpcRetryLimit = cfgGetItem(pCfg, "rpcRetryLimit")->i32;
tsRpcRetryInterval = cfgGetItem(pCfg, "rpcRetryInterval")->i32;
GRANT_CFG_GET;
return 0;
}

2
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -258,6 +258,8 @@ int32_t dmInitClient(SDnode *pDnode) {
rpcInit.parent = pDnode;
rpcInit.rfp = rpcRfp;
rpcInit.compressSize = tsCompressMsgSize;
rpcInit.retryLimit = tsRpcRetryLimit;
rpcInit.retryInterval = tsRpcRetryInterval;
pTrans->clientRpc = rpcOpen(&rpcInit);
if (pTrans->clientRpc == NULL) {

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

@ -532,6 +532,7 @@ static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCrea
streamObj.sql = strdup(pCreate->sql);
streamObj.smaId = smaObj.uid;
streamObj.watermark = pCreate->watermark;
streamObj.fillHistory = STREAM_FILL_HISTORY_ON;
streamObj.trigger = STREAM_TRIGGER_WINDOW_CLOSE;
streamObj.triggerParam = pCreate->maxDelay;
streamObj.ast = strdup(smaObj.ast);

37
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -637,6 +637,7 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
if (pIter == NULL) break;
if (pConsumer->status == MQ_CONSUMER_STATUS__LOST_REBD) continue;
int32_t sz = taosArrayGetSize(pConsumer->assignedTopics);
for (int32_t i = 0; i < sz; i++) {
char *name = taosArrayGetP(pConsumer->assignedTopics, i);
@ -649,6 +650,33 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
return -1;
}
}
sz = taosArrayGetSize(pConsumer->rebNewTopics);
for (int32_t i = 0; i < sz; i++) {
char *name = taosArrayGetP(pConsumer->rebNewTopics, i);
if (strcmp(name, pTopic->name) == 0) {
mndReleaseConsumer(pMnode, pConsumer);
mndReleaseTopic(pMnode, pTopic);
terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb new)",
dropReq.name, pConsumer->consumerId, pConsumer->cgroup);
return -1;
}
}
sz = taosArrayGetSize(pConsumer->rebRemovedTopics);
for (int32_t i = 0; i < sz; i++) {
char *name = taosArrayGetP(pConsumer->rebRemovedTopics, i);
if (strcmp(name, pTopic->name) == 0) {
mndReleaseConsumer(pMnode, pConsumer);
mndReleaseTopic(pMnode, pTopic);
terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb remove)",
dropReq.name, pConsumer->consumerId, pConsumer->cgroup);
return -1;
}
}
sdbRelease(pSdb, pConsumer);
}
@ -675,15 +703,6 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
mInfo("trans:%d, used to drop topic:%s", pTrans->id, pTopic->name);
#if 0
if (mndDropOffsetByTopic(pMnode, pTrans, dropReq.name) < 0) {
ASSERT(0);
mndTransDrop(pTrans);
mndReleaseTopic(pMnode, pTopic);
return -1;
}
#endif
// TODO check if rebalancing
if (mndDropSubByTopic(pMnode, pTrans, dropReq.name) < 0) {
/*ASSERT(0);*/

1
source/dnode/vnode/src/inc/meta.h
View File

@ -70,6 +70,7 @@ int32_t metaCacheDrop(SMeta* pMeta, int64_t uid);
int32_t metaStatsCacheUpsert(SMeta* pMeta, SMetaStbStats* pInfo);
int32_t metaStatsCacheDrop(SMeta* pMeta, int64_t uid);
int32_t metaStatsCacheGet(SMeta* pMeta, int64_t uid, SMetaStbStats* pInfo);
void metaUpdateStbStats(SMeta *pMeta, int64_t uid, int64_t delta);
struct SMeta {
TdThreadRwlock lock;

10
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -1445,3 +1445,13 @@ int32_t metaGetStbStats(SMeta *pMeta, int64_t uid, SMetaStbStats *pInfo) {
_exit:
return code;
}
void metaUpdateStbStats(SMeta *pMeta, int64_t uid, int64_t delta) {
SMetaStbStats stats = {0};
if (metaStatsCacheGet(pMeta, uid, &stats) == TSDB_CODE_SUCCESS) {
stats.ctbNum += delta;
metaStatsCacheUpsert(pMeta, &stats);
}
}

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

@ -371,7 +371,7 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
// update uid index
metaUpdateUidIdx(pMeta, &nStbEntry);
metaStatsCacheDrop(pMeta, nStbEntry.uid);
// metaStatsCacheDrop(pMeta, nStbEntry.uid);
metaULock(pMeta);
@ -450,6 +450,10 @@ int metaCreateTable(SMeta *pMeta, int64_t version, SVCreateTbReq *pReq, STableMe
#endif
++pMeta->pVnode->config.vndStats.numOfCTables;
metaWLock(pMeta);
metaUpdateStbStats(pMeta, me.ctbEntry.suid, 1);
metaULock(pMeta);
} else {
me.ntbEntry.ctime = pReq->ctime;
me.ntbEntry.ttlDays = pReq->ttl;
@ -670,6 +674,8 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
tdbTbDelete(pMeta->pCtbIdx, &(SCtbIdxKey){.suid = e.ctbEntry.suid, .uid = uid}, sizeof(SCtbIdxKey), &pMeta->txn);
--pMeta->pVnode->config.vndStats.numOfCTables;
metaUpdateStbStats(pMeta, e.ctbEntry.suid, -1);
} else if (e.type == TSDB_NORMAL_TABLE) {
// drop schema.db (todo)

6
source/dnode/vnode/src/tq/tq.c
View File

@ -582,10 +582,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
code = -1;
}
tqDebug("tmq poll: consumer %" PRId64 ", subkey %s, vg %d, send data blockNum:%d, offset type:%d, uid:%" PRId64
", version:%" PRId64 "",
tqDebug("tmq poll: consumer %" PRId64
", subkey %s, vg %d, send data blockNum:%d, offset type:%d, uid/version:%" PRId64 ", ts:%" PRId64 "",
consumerId, pHandle->subKey, TD_VID(pTq->pVnode), dataRsp.blockNum, dataRsp.rspOffset.type,
dataRsp.rspOffset.uid, dataRsp.rspOffset.version);
dataRsp.rspOffset.uid, dataRsp.rspOffset.ts);
tDeleteSMqDataRsp(&dataRsp);
return code;

7
source/dnode/vnode/src/tq/tqSink.c
View File

@ -244,7 +244,7 @@ SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchem
int32_t rows = pDataBlock->info.rows;
tqDebug("tq sink, convert block %d, rows: %d", i, rows);
tqDebug("tq sink, convert block1 %d, rows: %d", i, rows);
int32_t dataLen = 0;
int32_t schemaLen = 0;
@ -486,7 +486,7 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
blkHead->uid = 0;
blkHead->schemaLen = 0;
tqDebug("tq sink, convert block %d, rows: %d", i, rows);
tqDebug("tq sink, convert block2 %d, rows: %d", i, rows);
int32_t dataLen = 0;
void* blkSchema = POINTER_SHIFT(blkHead, sizeof(SSubmitBlk));
@ -514,6 +514,9 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, pColumn->offset, k);
} else {
void* colData = colDataGetData(pColData, j);
if (k == 0) {
tqDebug("tq sink, row %d ts %" PRId64, j, *(int64_t*)colData);
}
tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, colData, true, pColumn->offset, k);
}
}

4
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -495,6 +495,10 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
pCommitter->commitFid = tsdbKeyFid(pCommitter->nextKey, pCommitter->minutes, pCommitter->precision);
tsdbFidKeyRange(pCommitter->commitFid, pCommitter->minutes, pCommitter->precision, &pCommitter->minKey,
&pCommitter->maxKey);
#if 0
ASSERT(pCommitter->minKey <= pCommitter->nextKey && pCommitter->maxKey >= pCommitter->nextKey);
#endif
pCommitter->nextKey = TSKEY_MAX;
// Reader

2
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -568,7 +568,9 @@ static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, i
do {
key.ts = row.pTSRow->ts;
nRow++;
if (SL_NODE_FORWARD(pos[0], 0) != pTbData->sl.pTail) {
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS);
}
code = tbDataDoPut(pMemTable, pTbData, pos, &row, 1);
if (code) {
goto _err;

4
source/dnode/vnode/src/tsdb/tsdbUtil.c
View File

@ -505,8 +505,8 @@ int32_t tsdbKeyFid(TSKEY key, int32_t minutes, int8_t precision) {
}
void tsdbFidKeyRange(int32_t fid, int32_t minutes, int8_t precision, TSKEY *minKey, TSKEY *maxKey) {
*minKey = fid * minutes * tsTickPerMin[precision];
*maxKey = *minKey + minutes * tsTickPerMin[precision] - 1;
*minKey = tsTickPerMin[precision] * fid * minutes;
*maxKey = *minKey + tsTickPerMin[precision] * minutes - 1;
}
int32_t tsdbFidLevel(int32_t fid, STsdbKeepCfg *pKeepCfg, int64_t now) {

11
source/dnode/vnode/src/tsdb/tsdbWrite.c
View File

@ -15,6 +15,15 @@
#include "tsdb.h"
/**
* @brief max key by precision
* approximately calculation:
* ms: 3600*1000*8765*1000 // 1970 + 1000 years
* us: 3600*1000000*8765*1000 // 1970 + 1000 years
* ns: 3600*1000000000*8765*292 // 1970 + 292 years
*/
static int64_t tsMaxKeyByPrecision[] = {31556995200000L, 31556995200000000L, 9214646400000000000L};
// static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg);
int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
@ -97,7 +106,7 @@ int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
STsdbKeepCfg *pCfg = &pTsdb->keepCfg;
TSKEY now = taosGetTimestamp(pCfg->precision);
TSKEY minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep2;
TSKEY maxKey = now + tsTickPerMin[pCfg->precision] * pCfg->days;
TSKEY maxKey = tsMaxKeyByPrecision[pCfg->precision];
terrno = TSDB_CODE_SUCCESS;
// pMsg->length = htonl(pMsg->length);

1
source/libs/executor/inc/executorimpl.h
View File

@ -981,6 +981,7 @@ void setTaskKilled(SExecTaskInfo* pTaskInfo);
void queryCostStatis(SExecTaskInfo* pTaskInfo);
void doDestroyTask(SExecTaskInfo* pTaskInfo);
void destroyOperatorInfo(SOperatorInfo* pOperator);
int32_t getMaximumIdleDurationSec();
/*

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

@ -91,9 +91,6 @@ static void destroyAggOperatorInfo(void* param);
static void destroyIntervalOperatorInfo(void* param);
static void destroyOperatorInfo(SOperatorInfo* pOperator);
void setOperatorCompleted(SOperatorInfo* pOperator) {
pOperator->status = OP_EXEC_DONE;
ASSERT(pOperator->pTaskInfo != NULL);
@ -2172,7 +2169,7 @@ void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs) {
}
}
static void destroyOperatorInfo(SOperatorInfo* pOperator) {
void destroyOperatorInfo(SOperatorInfo* pOperator) {
if (pOperator == NULL) {
return;
}

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

@ -816,12 +816,12 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
goto _error;
}
setOperatorInfo(pOperator, "PartitionOperator", QUERY_NODE_PHYSICAL_PLAN_PARTITION, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "PartitionOperator", QUERY_NODE_PHYSICAL_PLAN_PARTITION, false, OP_NOT_OPENED, pInfo,
pTaskInfo);
pOperator->exprSupp.numOfExprs = numOfCols;
pOperator->exprSupp.pExprInfo = pExprInfo;
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, hashPartition, NULL, destroyPartitionOperatorInfo, NULL);
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, hashPartition, NULL, destroyPartitionOperatorInfo, NULL);
code = appendDownstream(pOperator, &downstream, 1);
return pOperator;
@ -900,7 +900,7 @@ static SSDataBlock* buildStreamPartitionResult(SOperatorInfo* pOperator) {
// TODO check tbname validity
if (pData != (void*)-1) {
memset(pDest->info.parTbName, 0, TSDB_TABLE_NAME_LEN);
int32_t len = TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN);
int32_t len = TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN - 1);
memcpy(pDest->info.parTbName, varDataVal(pData), len);
/*pDest->info.parTbName[len + 1] = 0;*/
} else {
@ -1099,11 +1099,12 @@ SOperatorInfo* createStreamPartitionOperatorInfo(SOperatorInfo* downstream, SStr
int32_t numOfCols = 0;
SExprInfo* pExprInfo = createExprInfo(pPartNode->part.pTargets, NULL, &numOfCols);
setOperatorInfo(pOperator, "StreamPartitionOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "StreamPartitionOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION, false, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->exprSupp.numOfExprs = numOfCols;
pOperator->exprSupp.pExprInfo = pExprInfo;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamHashPartition, NULL,
destroyStreamPartitionOperatorInfo, NULL);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doStreamHashPartition, NULL, destroyStreamPartitionOperatorInfo, NULL);
initParDownStream(downstream, &pInfo->partitionSup, &pInfo->scalarSup);
code = appendDownstream(pOperator, &downstream, 1);

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

@ -946,7 +946,8 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
pInfo->currentGroupId = -1;
pInfo->assignBlockUid = pTableScanNode->assignBlockUid;
setOperatorInfo(pOperator, "TableScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "TableScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo,
pTaskInfo);
pOperator->exprSupp.numOfExprs = numOfCols;
pInfo->metaCache.pTableMetaEntryCache = taosLRUCacheInit(1024 * 128, -1, .5);
@ -980,7 +981,8 @@ SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo*
pInfo->dataReader = pReadHandle;
// pInfo->prevGroupId = -1;
setOperatorInfo(pOperator, "TableSeqScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "TableSeqScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN, false, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScanImpl, NULL, NULL, NULL);
return pOperator;
}
@ -1136,8 +1138,10 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
goto _error;
}
setOperatorInfo(pOperator, "DataBlockDistScanOperator", QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, NULL);
setOperatorInfo(pOperator, "DataBlockDistScanOperator", QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN, false,
OP_NOT_OPENED, pInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, NULL);
return pOperator;
_error:
@ -1581,7 +1585,7 @@ static void calBlockTbName(SExprSupp* pTbNameCalSup, SSDataBlock* pBlock) {
// TODO check tbname validation
if (pData != (void*)-1 && pData != NULL) {
memset(pBlock->info.parTbName, 0, TSDB_TABLE_NAME_LEN);
int32_t len = TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN);
int32_t len = TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN - 1);
memcpy(pBlock->info.parTbName, varDataVal(pData), len);
/*pBlock->info.parTbName[len + 1] = 0;*/
} else {
@ -2351,7 +2355,8 @@ SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pT
pInfo->vnode = pHandle->vnode;
pInfo->sContext = pHandle->sContext;
setOperatorInfo(pOperator, "RawScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "RawScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo,
pTaskInfo);
pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, destroyRawScanOperatorInfo, NULL);
return pOperator;
@ -2366,9 +2371,7 @@ _end:
static void destroyStreamScanOperatorInfo(void* param) {
SStreamScanInfo* pStreamScan = (SStreamScanInfo*)param;
if (pStreamScan->pTableScanOp && pStreamScan->pTableScanOp->info) {
STableScanInfo* pTableScanInfo = pStreamScan->pTableScanOp->info;
destroyTableScanOperatorInfo(pTableScanInfo);
taosMemoryFreeClear(pStreamScan->pTableScanOp);
destroyOperatorInfo(pStreamScan->pTableScanOp);
}
if (pStreamScan->tqReader) {
tqCloseReader(pStreamScan->tqReader);
@ -2537,7 +2540,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
pInfo->assignBlockUid = pTableScanNode->assignBlockUid;
pInfo->partitionSup.needCalc = false;
setOperatorInfo(pOperator, "StreamScanOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "StreamScanOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, false, OP_NOT_OPENED, pInfo,
pTaskInfo);
pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
__optr_fn_t nextFn = pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM ? doStreamScan : doQueueScan;
@ -4175,7 +4179,8 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan
pInfo->readHandle = *(SReadHandle*)readHandle;
}
setOperatorInfo(pOperator, "SysTableScanOperator", QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "SysTableScanOperator", QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN, false, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, destroySysScanOperator, NULL);
return pOperator;
@ -4305,7 +4310,8 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
pInfo->readHandle = *pReadHandle;
pInfo->curPos = 0;
setOperatorInfo(pOperator, "TagScanOperator", QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "TagScanOperator", QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN, false, OP_NOT_OPENED, pInfo,
pTaskInfo);
initResultSizeInfo(&pOperator->resultInfo, 4096);
blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
@ -4815,11 +4821,12 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
int32_t rowSize = pInfo->pResBlock->info.rowSize;
pInfo->bufPageSize = getProperSortPageSize(rowSize);
setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "TableMergeScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN, false, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->exprSupp.numOfExprs = numOfCols;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableMergeScan, NULL,
destroyTableMergeScanOperatorInfo, getTableMergeScanExplainExecInfo);
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableMergeScan, NULL, destroyTableMergeScanOperatorInfo,
getTableMergeScanExplainExecInfo);
pOperator->cost.openCost = 0;
return pOperator;

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

@ -680,9 +680,9 @@ SResultCellData* getResultCell(SResultRowData* pRaw, int32_t index) {
void* destroyFillColumnInfo(SFillColInfo* pFillCol, int32_t start, int32_t end) {
for (int32_t i = start; i < end; i++) {
destroyExprInfo(pFillCol[i].pExpr, 1);
taosMemoryFreeClear(pFillCol[i].pExpr);
taosVariantDestroy(&pFillCol[i].fillVal);
}
taosMemoryFreeClear(pFillCol[start].pExpr);
taosMemoryFree(pFillCol);
return NULL;
}

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

@ -1776,7 +1776,8 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPh
}
initResultRowInfo(&pInfo->binfo.resultRowInfo);
setOperatorInfo(pOperator, "TimeIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL, true, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "TimeIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, NULL, destroyIntervalOperatorInfo, NULL);
@ -2551,9 +2552,9 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode
pScanInfo->cond.twindows = pInfo->win;
pScanInfo->cond.type = TIMEWINDOW_RANGE_EXTERNAL;
setOperatorInfo(pOperator, "TimeSliceOperator", QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC, false, OP_NOT_OPENED, pInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doTimeslice, NULL, destroyTimeSliceOperatorInfo, NULL);
setOperatorInfo(pOperator, "TimeSliceOperator", QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC, false, OP_NOT_OPENED, pInfo,
pTaskInfo);
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTimeslice, NULL, destroyTimeSliceOperatorInfo, NULL);
blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
@ -2622,7 +2623,8 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWi
pInfo->tsSlotId = tsSlotId;
setOperatorInfo(pOperator, "StateWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE, true, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "StateWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE, true, OP_NOT_OPENED, pInfo,
pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(openStateWindowAggOptr, doStateWindowAgg, NULL, destroyStateWindowOperatorInfo, NULL);
@ -2694,7 +2696,8 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionW
goto _error;
}
setOperatorInfo(pOperator, "SessionWindowAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION, true, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "SessionWindowAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doSessionWindowAgg, NULL, destroySWindowOperatorInfo, NULL);
pOperator->pTaskInfo = pTaskInfo;
@ -3822,7 +3825,7 @@ static int32_t copyUpdateResult(SSHashObj* pStUpdated, SArray* pUpdated) {
return TSDB_CODE_SUCCESS;
}
void doBuildDeleteDataBlock(SSHashObj* pStDeleted, SSDataBlock* pBlock, void** Ite) {
void doBuildDeleteDataBlock(SOperatorInfo* pOp, SSHashObj* pStDeleted, SSDataBlock* pBlock, void** Ite) {
blockDataCleanup(pBlock);
int32_t size = tSimpleHashGetSize(pStDeleted);
if (size == 0) {
@ -3848,6 +3851,26 @@ void doBuildDeleteDataBlock(SSHashObj* pStDeleted, SSDataBlock* pBlock, void** I
colDataAppendNULL(pCalStCol, pBlock->info.rows);
SColumnInfoData* pCalEdCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
colDataAppendNULL(pCalEdCol, pBlock->info.rows);
SHashObj* pGroupIdTbNameMap = NULL;
if (pOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
pOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
SStreamSessionAggOperatorInfo* pInfo = pOp->info;
pGroupIdTbNameMap = pInfo->pGroupIdTbNameMap;
} else if (pOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
SStreamStateAggOperatorInfo* pInfo = pOp->info;
pGroupIdTbNameMap = pInfo->pGroupIdTbNameMap;
}
char* tbname = taosHashGet(pGroupIdTbNameMap, &res->groupId, sizeof(int64_t));
SColumnInfoData* pTableCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
if (tbname == NULL) {
colDataAppendNULL(pTableCol, pBlock->info.rows);
} else {
char parTbName[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN];
STR_WITH_MAXSIZE_TO_VARSTR(parTbName, tbname, sizeof(parTbName));
colDataAppend(pTableCol, pBlock->info.rows, (const char*)parTbName, false);
}
pBlock->info.rows += 1;
}
if ((*Ite) == NULL) {
@ -3994,7 +4017,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
} else if (pOperator->status == OP_RES_TO_RETURN) {
doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
return pInfo->pDelRes;
@ -4099,7 +4122,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
taosMemoryFree(pBuf);
#endif
doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
return pInfo->pDelRes;
@ -4223,7 +4246,7 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
return pBInfo->pRes;
}
doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "semi session delete");
return pInfo->pDelRes;
@ -4303,7 +4326,7 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
return pBInfo->pRes;
}
doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "semi session delete");
return pInfo->pDelRes;
@ -4327,7 +4350,7 @@ SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream
SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
pInfo->isFinal = (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION);
char* name = (pInfo->isFinal)? "StreamSessionFinalAggOperator":"StreamSessionSemiAggOperator";
char* name = (pInfo->isFinal) ? "StreamSessionFinalAggOperator" : "StreamSessionSemiAggOperator";
if (pPhyNode->type != QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR);
@ -4336,7 +4359,7 @@ SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream
destroyStreamSessionAggOperatorInfo, NULL);
}
setOperatorInfo(pOperator, name, pPhyNode->type , false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, name, pPhyNode->type, false, OP_NOT_OPENED, pInfo, pTaskInfo);
pInfo->pGroupIdTbNameMap =
taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
@ -4555,7 +4578,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
SOptrBasicInfo* pBInfo = &pInfo->binfo;
int64_t maxTs = INT64_MIN;
if (pOperator->status == OP_RES_TO_RETURN) {
doBuildDeleteDataBlock(pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
doBuildDeleteDataBlock(pOperator, pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "single state delete");
return pInfo->pDelRes;
@ -4622,7 +4645,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
initGroupResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
doBuildDeleteDataBlock(pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
doBuildDeleteDataBlock(pOperator, pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "single state delete");
return pInfo->pDelRes;
@ -4698,7 +4721,8 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->pGroupIdTbNameMap =
taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
setOperatorInfo(pOperator, "StreamStateAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE, true, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "StreamStateAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo, NULL);
initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType,
@ -4973,7 +4997,8 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream,
initResultRowInfo(&iaInfo->binfo.resultRowInfo);
blockDataEnsureCapacity(iaInfo->binfo.pRes, pOperator->resultInfo.capacity);
setOperatorInfo(pOperator, "TimeMergeAlignedIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL, false, OP_NOT_OPENED, miaInfo, pTaskInfo);
setOperatorInfo(pOperator, "TimeMergeAlignedIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL,
false, OP_NOT_OPENED, miaInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, mergeAlignedIntervalAgg, NULL, destroyMAIOperatorInfo, NULL);
@ -5279,7 +5304,8 @@ SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMerge
}
initResultRowInfo(&pIntervalInfo->binfo.resultRowInfo);
setOperatorInfo(pOperator, "TimeMergeIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL, false, OP_NOT_OPENED, pMergeIntervalInfo, pTaskInfo);
setOperatorInfo(pOperator, "TimeMergeIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL, false,
OP_NOT_OPENED, pMergeIntervalInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doMergeIntervalAgg, NULL, destroyMergeIntervalOperatorInfo, NULL);
@ -5509,9 +5535,10 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->pGroupIdTbNameMap =
taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
setOperatorInfo(pOperator, "StreamIntervalOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, true, OP_NOT_OPENED, pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamIntervalAgg, NULL,
destroyStreamFinalIntervalOperatorInfo, NULL);
setOperatorInfo(pOperator, "StreamIntervalOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doStreamIntervalAgg, NULL, destroyStreamFinalIntervalOperatorInfo, NULL);
initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup);
code = appendDownstream(pOperator, &downstream, 1);

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

@ -424,8 +424,9 @@ SNode* nodesMakeNode(ENodeType type) {
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
case QUERY_NODE_SHOW_TAGS_STMT:
case QUERY_NODE_SHOW_TABLE_TAGS_STMT:
return makeNode(type, sizeof(SShowStmt));
case QUERY_NODE_SHOW_TABLE_TAGS_STMT:
return makeNode(type, sizeof(SShowTableTagsStmt));
case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
return makeNode(type, sizeof(SShowDnodeVariablesStmt));
case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
@ -942,13 +943,19 @@ void nodesDestroyNode(SNode* pNode) {
case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
case QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT:
case QUERY_NODE_SHOW_TAGS_STMT:
case QUERY_NODE_SHOW_TABLE_TAGS_STMT: {
case QUERY_NODE_SHOW_TAGS_STMT: {
SShowStmt* pStmt = (SShowStmt*)pNode;
nodesDestroyNode(pStmt->pDbName);
nodesDestroyNode(pStmt->pTbName);
break;
}
case QUERY_NODE_SHOW_TABLE_TAGS_STMT: {
SShowTableTagsStmt* pStmt = (SShowTableTagsStmt*)pNode;
nodesDestroyNode(pStmt->pDbName);
nodesDestroyNode(pStmt->pTbName);
nodesDestroyList(pStmt->pTags);
break;
}
case QUERY_NODE_SHOW_DNODE_VARIABLES_STMT:
nodesDestroyNode(((SShowDnodeVariablesStmt*)pNode)->pDnodeId);
nodesDestroyNode(((SShowDnodeVariablesStmt*)pNode)->pLikePattern);

1
source/libs/parser/inc/parAst.h
View File

@ -178,6 +178,7 @@ SNode* createShowCreateTableStmt(SAstCreateContext* pCxt, ENodeType type, SNode*
SNode* createShowTableDistributedStmt(SAstCreateContext* pCxt, SNode* pRealTable);
SNode* createShowDnodeVariablesStmt(SAstCreateContext* pCxt, SNode* pDnodeId, SNode* pLikePattern);
SNode* createShowVnodesStmt(SAstCreateContext* pCxt, SNode* pDnodeId, SNode* pDnodeEndpoint);
SNode* createShowTableTagsStmt(SAstCreateContext* pCxt, SNode* pTbName, SNode* pDbName, SNodeList* pTags);
SNode* createCreateUserStmt(SAstCreateContext* pCxt, SToken* pUserName, const SToken* pPassword, int8_t sysinfo);
SNode* createAlterUserStmt(SAstCreateContext* pCxt, SToken* pUserName, int8_t alterType, const SToken* pVal);
SNode* createDropUserStmt(SAstCreateContext* pCxt, SToken* pUserName);

14
source/libs/parser/inc/sql.y
View File

@ -426,7 +426,7 @@ cmd ::= SHOW TABLE DISTRIBUTED full_table_name(A).
cmd ::= SHOW CONSUMERS. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_CONSUMERS_STMT); }
cmd ::= SHOW SUBSCRIPTIONS. { pCxt->pRootNode = createShowStmt(pCxt, QUERY_NODE_SHOW_SUBSCRIPTIONS_STMT); }
cmd ::= SHOW TAGS FROM table_name_cond(A) from_db_opt(B). { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TAGS_STMT, B, A, OP_TYPE_EQUAL); }
cmd ::= SHOW TABLE TAGS FROM table_name_cond(A) from_db_opt(B). { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TABLE_TAGS_STMT, B, A, OP_TYPE_EQUAL); }
cmd ::= SHOW TABLE TAGS tag_list_opt(C) FROM table_name_cond(A) from_db_opt(B). { pCxt->pRootNode = createShowTableTagsStmt(pCxt, A, B, C); }
cmd ::= SHOW VNODES NK_INTEGER(A). { pCxt->pRootNode = createShowVnodesStmt(pCxt, createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &A), NULL); }
cmd ::= SHOW VNODES NK_STRING(A). { pCxt->pRootNode = createShowVnodesStmt(pCxt, NULL, createValueNode(pCxt, TSDB_DATA_TYPE_VARCHAR, &A)); }
@ -441,6 +441,18 @@ table_name_cond(A) ::= table_name(B).
from_db_opt(A) ::= . { A = createDefaultDatabaseCondValue(pCxt); }
from_db_opt(A) ::= FROM db_name(B). { A = createIdentifierValueNode(pCxt, &B); }
%type tag_list_opt { SNodeList* }
%destructor tag_list_opt { nodesDestroyList($$); }
tag_list_opt(A) ::= . { A = NULL; }
tag_list_opt(A) ::= tag_item(B). { A = createNodeList(pCxt, B); }
tag_list_opt(A) ::= tag_list_opt(B) NK_COMMA tag_item(C). { A = addNodeToList(pCxt, B, C); }
tag_item(A) ::= TBNAME(B). { A = setProjectionAlias(pCxt, createFunctionNode(pCxt, &B, NULL), &B); }
tag_item(A) ::= QTAGS(B). { A = createFunctionNode(pCxt, &B, NULL); }
tag_item(A) ::= column_name(B). { A = createColumnNode(pCxt, NULL, &B); }
tag_item(A) ::= column_name(B) column_alias(C). { A = setProjectionAlias(pCxt, createColumnNode(pCxt, NULL, &B), &C); }
tag_item(A) ::= column_name(B) AS column_alias(C). { A = setProjectionAlias(pCxt, createColumnNode(pCxt, NULL, &B), &C); }
/************************************************ create index ********************************************************/
cmd ::= CREATE SMA INDEX not_exists_opt(D)
full_table_name(A) ON full_table_name(B) index_options(C). { pCxt->pRootNode = createCreateIndexStmt(pCxt, INDEX_TYPE_SMA, D, A, B, NULL, C); }

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

@ -1390,6 +1390,21 @@ SNode* createShowVnodesStmt(SAstCreateContext* pCxt, SNode* pDnodeId, SNode* pDn
return (SNode*)pStmt;
}
SNode* createShowTableTagsStmt(SAstCreateContext* pCxt, SNode* pTbName, SNode* pDbName, SNodeList* pTags) {
CHECK_PARSER_STATUS(pCxt);
if (NULL == pDbName) {
snprintf(pCxt->pQueryCxt->pMsg, pCxt->pQueryCxt->msgLen, "database not specified");
pCxt->errCode = TSDB_CODE_PAR_SYNTAX_ERROR;
return NULL;
}
SShowTableTagsStmt* pStmt = (SShowTableTagsStmt*)nodesMakeNode(QUERY_NODE_SHOW_TABLE_TAGS_STMT);
CHECK_OUT_OF_MEM(pStmt);
pStmt->pDbName = pDbName;
pStmt->pTbName = pTbName;
pStmt->pTags = pTags;
return (SNode*)pStmt;
}
SNode* createCreateUserStmt(SAstCreateContext* pCxt, SToken* pUserName, const SToken* pPassword, int8_t sysinfo) {
CHECK_PARSER_STATUS(pCxt);
char password[TSDB_USET_PASSWORD_LEN] = {0};

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

@ -440,7 +440,7 @@ static int32_t collectMetaKeyFromShowTags(SCollectMetaKeyCxt* pCxt, SShowStmt* p
return code;
}
static int32_t collectMetaKeyFromShowStableTags(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
static int32_t collectMetaKeyFromShowStableTags(SCollectMetaKeyCxt* pCxt, SShowTableTagsStmt* pStmt) {
return collectMetaKeyFromRealTableImpl(pCxt, ((SValueNode*)pStmt->pDbName)->literal,
((SValueNode*)pStmt->pTbName)->literal, AUTH_TYPE_READ);
}
@ -627,7 +627,7 @@ static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
case QUERY_NODE_SHOW_TAGS_STMT:
return collectMetaKeyFromShowTags(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_TABLE_TAGS_STMT:
return collectMetaKeyFromShowStableTags(pCxt, (SShowStmt*)pStmt);
return collectMetaKeyFromShowStableTags(pCxt, (SShowTableTagsStmt*)pStmt);
case QUERY_NODE_SHOW_USERS_STMT:
return collectMetaKeyFromShowUsers(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_LICENCES_STMT:

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

@ -1878,6 +1878,7 @@ static int32_t rewriteIsTrue(SNode* pSrc, SNode** pIsTrue) {
static EDealRes translateCaseWhen(STranslateContext* pCxt, SCaseWhenNode* pCaseWhen) {
bool first = true;
bool allNullThen = true;
SNode* pNode = NULL;
FOREACH(pNode, pCaseWhen->pWhenThenList) {
SWhenThenNode* pWhenThen = (SWhenThenNode*)pNode;
@ -1889,12 +1890,28 @@ static EDealRes translateCaseWhen(STranslateContext* pCxt, SCaseWhenNode* pCaseW
}
pWhenThen->pWhen = pIsTrue;
}
if (first || dataTypeComp(&pCaseWhen->node.resType, &((SExprNode*)pNode)->resType) < 0) {
pCaseWhen->node.resType = ((SExprNode*)pNode)->resType;
SExprNode* pThenExpr = (SExprNode*)pNode;
if (TSDB_DATA_TYPE_NULL == pThenExpr->resType.type) {
continue;
}
allNullThen = false;
if (first || dataTypeComp(&pCaseWhen->node.resType, &pThenExpr->resType) < 0) {
pCaseWhen->node.resType = pThenExpr->resType;
}
first = false;
}
if (allNullThen) {
if (NULL != pCaseWhen->pElse) {
pCaseWhen->node.resType = ((SExprNode*)pCaseWhen->pElse)->resType;
} else {
pCaseWhen->node.resType.type = TSDB_DATA_TYPE_NULL;
pCaseWhen->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_NULL].bytes;
return DEAL_RES_CONTINUE;
}
}
FOREACH(pNode, pCaseWhen->pWhenThenList) {
SWhenThenNode* pWhenThen = (SWhenThenNode*)pNode;
if (!dataTypeEqual(&pCaseWhen->node.resType, &((SExprNode*)pNode)->resType)) {
@ -6288,21 +6305,29 @@ static SNode* createTagsFunction() {
return (SNode*)pFunc;
}
static int32_t createShowTableTagsProjections(SNodeList** pProjections, SNodeList** pTags) {
if (NULL != *pTags) {
TSWAP(*pProjections, *pTags);
return TSDB_CODE_SUCCESS;
}
int32_t code = nodesListMakeStrictAppend(pProjections, createTbnameFunction());
if (TSDB_CODE_SUCCESS == code) {
code = nodesListStrictAppend(*pProjections, createTagsFunction());
}
return code;
}
static int32_t rewriteShowStableTags(STranslateContext* pCxt, SQuery* pQuery) {
const char* cols[] = {"tbname", "_tags"};
SShowStmt* pShow = (SShowStmt*)pQuery->pRoot;
SShowTableTagsStmt* pShow = (SShowTableTagsStmt*)pQuery->pRoot;
SSelectStmt* pSelect = NULL;
int32_t code = createSimpleSelectStmt(((SValueNode*)pShow->pDbName)->literal, ((SValueNode*)pShow->pTbName)->literal,
-1, NULL, &pSelect);
if (TSDB_CODE_SUCCESS == code) {
code = nodesListMakeStrictAppend(&pSelect->pProjectionList, createTbnameFunction());
code = createShowTableTagsProjections(&pSelect->pProjectionList, &pShow->pTags);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodesListMakeStrictAppend(&pSelect->pProjectionList, createTagsFunction());
}
if (TSDB_CODE_SUCCESS == code) {
pSelect->isDistinct = true;
pQuery->showRewrite = true;
pSelect->tagScan = true;
nodesDestroyNode(pQuery->pRoot);
pQuery->pRoot = (SNode*)pSelect;
} else {

4851
source/libs/parser/src/sql.c
View File

File diff suppressed because it is too large Load Diff

10
source/libs/parser/test/parShowToUse.cpp
View File

@ -192,6 +192,16 @@ TEST_F(ParserShowToUseTest, showTableDistributed) {
run("SHOW TABLE DISTRIBUTED st1");
}
TEST_F(ParserShowToUseTest, showTableTags) {
useDb("root", "test");
run("SHOW TABLE TAGS FROM st1");
run("SHOW TABLE TAGS tag1, tag2 FROM st1");
run("SHOW TABLE TAGS TBNAME, _TAGS, tag3 FROM st1");
}
TEST_F(ParserShowToUseTest, showTags) {
useDb("root", "test");

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

@ -189,7 +189,7 @@ static int32_t createSelectRootLogicNode(SLogicPlanContext* pCxt, SSelectStmt* p
}
static EScanType getScanType(SLogicPlanContext* pCxt, SNodeList* pScanPseudoCols, SNodeList* pScanCols,
int8_t tableType) {
int8_t tableType, bool tagScan) {
if (pCxt->pPlanCxt->topicQuery || pCxt->pPlanCxt->streamQuery) {
return SCAN_TYPE_STREAM;
}
@ -198,6 +198,10 @@ static EScanType getScanType(SLogicPlanContext* pCxt, SNodeList* pScanPseudoCols
return SCAN_TYPE_SYSTEM_TABLE;
}
if (tagScan) {
return SCAN_TYPE_TAG;
}
if (NULL == pScanCols) {
return NULL == pScanPseudoCols
? SCAN_TYPE_TABLE
@ -310,7 +314,7 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
code = rewriteExprsForSelect(pScan->pScanPseudoCols, pSelect, SQL_CLAUSE_FROM);
}
pScan->scanType = getScanType(pCxt, pScan->pScanPseudoCols, pScan->pScanCols, pScan->tableType);
pScan->scanType = getScanType(pCxt, pScan->pScanPseudoCols, pScan->pScanCols, pScan->tableType, pSelect->tagScan);
if (NULL != pScan->pScanCols) {
pScan->hasNormalCols = true;

4
source/libs/stream/src/streamMeta.c
View File

@ -187,11 +187,9 @@ int32_t streamMetaRemoveTask(SStreamMeta* pMeta, int32_t taskId) {
while (1) {
int8_t schedStatus =
atomic_val_compare_exchange_8(&pTask->schedStatus, TASK_SCHED_STATUS__INACTIVE, TASK_SCHED_STATUS__DROPPING);
if (schedStatus == TASK_SCHED_STATUS__INACTIVE) {
if (schedStatus != TASK_SCHED_STATUS__ACTIVE) {
tFreeSStreamTask(pTask);
break;
} else if (schedStatus == TASK_SCHED_STATUS__DROPPING) {
break;
}
taosMsleep(10);
}

2
source/libs/sync/inc/syncEnv.h
View File

@ -30,8 +30,6 @@ extern "C" {
#define ELECT_TIMER_MS_RANGE (ELECT_TIMER_MS_MAX - ELECT_TIMER_MS_MIN)
#define HEARTBEAT_TIMER_MS 1000
#define EMPTY_RAFT_ID ((SRaftId){.addr = 0, .vgId = 0})
typedef struct SSyncEnv {
uint8_t isStart;

9
source/libs/sync/inc/syncRaftCfg.h
View File

@ -23,10 +23,8 @@ extern "C" {
#include "syncInt.h"
#define CONFIG_FILE_LEN 2048
#define MAX_CONFIG_INDEX_COUNT 256
// SRaftCfgIndex ------------------------------------------
typedef struct SRaftCfgIndex {
TdFilePtr pFile;
char path[TSDB_FILENAME_LEN * 2];
@ -44,11 +42,8 @@ cJSON *raftCfgIndex2Json(SRaftCfgIndex *pRaftCfgIndex);
char *raftCfgIndex2Str(SRaftCfgIndex *pRaftCfgIndex);
int32_t raftCfgIndexFromJson(const cJSON *pRoot, SRaftCfgIndex *pRaftCfgIndex);
int32_t raftCfgIndexFromStr(const char *s, SRaftCfgIndex *pRaftCfgIndex);
int32_t raftCfgIndexCreateFile(const char *path);
// ---------------------------------------------------------
typedef struct SRaftCfg {
SSyncCfg cfg;
TdFilePtr pFile;
@ -68,11 +63,9 @@ int32_t raftCfgClose(SRaftCfg *pRaftCfg);
int32_t raftCfgPersist(SRaftCfg *pRaftCfg);
int32_t raftCfgAddConfigIndex(SRaftCfg *pRaftCfg, SyncIndex configIndex);
cJSON *syncCfg2Json(SSyncCfg *pSyncCfg);
char *syncCfg2Str(SSyncCfg *pSyncCfg);
void syncCfg2SimpleStr(const SSyncCfg *pCfg, char *str, int32_t bufLen);
cJSON *syncCfg2Json(SSyncCfg *pSyncCfg);
int32_t syncCfgFromJson(const cJSON *pRoot, SSyncCfg *pSyncCfg);
int32_t syncCfgFromStr(const char *s, SSyncCfg *pSyncCfg);
cJSON *raftCfg2Json(SRaftCfg *pRaftCfg);
char *raftCfg2Str(SRaftCfg *pRaftCfg);

1
source/libs/sync/inc/syncRaftLog.h
View File

@ -38,7 +38,6 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode);
void logStoreDestory(SSyncLogStore* pLogStore);
SyncIndex logStoreFirstIndex(SSyncLogStore* pLogStore);
SyncIndex logStoreWalCommitVer(SSyncLogStore* pLogStore);
SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore);

11
source/libs/sync/inc/syncRaftStore.h
View File

@ -25,6 +25,8 @@ extern "C" {
#define RAFT_STORE_BLOCK_SIZE 512
#define RAFT_STORE_PATH_LEN (TSDB_FILENAME_LEN * 2)
#define EMPTY_RAFT_ID ((SRaftId){.addr = 0, .vgId = 0})
typedef struct SRaftStore {
SyncTerm currentTerm;
SRaftId voteFor;
@ -43,15 +45,6 @@ void raftStoreVote(SRaftStore *pRaftStore, SRaftId *pRaftId);
void raftStoreClearVote(SRaftStore *pRaftStore);
void raftStoreNextTerm(SRaftStore *pRaftStore);
void raftStoreSetTerm(SRaftStore *pRaftStore, SyncTerm term);
int32_t raftStoreFromJson(SRaftStore *pRaftStore, cJSON *pJson);
cJSON *raftStore2Json(SRaftStore *pRaftStore);
char *raftStore2Str(SRaftStore *pRaftStore);
// for debug -------------------
void raftStorePrint(SRaftStore *pObj);
void raftStorePrint2(char *s, SRaftStore *pObj);
void raftStoreLog(SRaftStore *pObj);
void raftStoreLog2(char *s, SRaftStore *pObj);
#ifdef __cplusplus
}

1
source/libs/sync/inc/syncRequestVote.h
View File

@ -21,7 +21,6 @@ extern "C" {
#endif
#include "syncInt.h"
#include "syncMessage.h"
// TLA+ Spec
// HandleRequestVoteRequest(i, j, m) ==

1
source/libs/sync/inc/syncRequestVoteReply.h
View File

@ -21,7 +21,6 @@ extern "C" {
#endif
#include "syncInt.h"
#include "syncMessage.h"
// TLA+ Spec
// HandleRequestVoteResponse(i, j, m) ==

22
source/libs/sync/src/syncRaftCfg.c
View File

@ -13,10 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "syncRaftCfg.h"
#include "cJSON.h"
#include "syncEnv.h"
#include "syncUtil.h"
// file must already exist!
SRaftCfgIndex *raftCfgIndexOpen(const char *path) {
@ -242,13 +240,6 @@ cJSON *syncCfg2Json(SSyncCfg *pSyncCfg) {
return pRoot;
}
char *syncCfg2Str(SSyncCfg *pSyncCfg) {
cJSON *pJson = syncCfg2Json(pSyncCfg);
char *serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
int32_t syncCfgFromJson(const cJSON *pRoot, SSyncCfg *pSyncCfg) {
memset(pSyncCfg, 0, sizeof(SSyncCfg));
// cJSON *pJson = cJSON_GetObjectItem(pRoot, "SSyncCfg");
@ -283,17 +274,6 @@ int32_t syncCfgFromJson(const cJSON *pRoot, SSyncCfg *pSyncCfg) {
return 0;
}
int32_t syncCfgFromStr(const char *s, SSyncCfg *pSyncCfg) {
cJSON *pRoot = cJSON_Parse(s);
ASSERT(pRoot != NULL);
int32_t ret = syncCfgFromJson(pRoot, pSyncCfg);
ASSERT(ret == 0);
cJSON_Delete(pRoot);
return 0;
}
cJSON *raftCfg2Json(SRaftCfg *pRaftCfg) {
cJSON *pRoot = cJSON_CreateObject();
cJSON_AddItemToObject(pRoot, "SSyncCfg", syncCfg2Json(&(pRaftCfg->cfg)));

3
source/libs/sync/src/syncRaftEntry.c
View File

@ -102,7 +102,6 @@ void syncEntry2OriginalRpc(const SSyncRaftEntry* pEntry, SRpcMsg* pRpcMsg) {
memcpy(pRpcMsg->pCont, pEntry->data, pRpcMsg->contLen);
}
//-----------------------------------
SRaftEntryHashCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) {
SRaftEntryHashCache* pCache = taosMemoryMalloc(sizeof(SRaftEntryHashCache));
if (pCache == NULL) {
@ -256,8 +255,6 @@ int32_t raftCacheClear(struct SRaftEntryHashCache* pCache) {
return 0;
}
//-----------------------------------
static char* keyFn(const void* pData) {
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)pData;
return (char*)(&(pEntry->index));

15
source/libs/sync/src/syncRaftLog.c
View File

@ -13,33 +13,33 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "syncRaftLog.h"
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
//-------------------------------
// log[m .. n]
// public function
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex);
static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);
static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index);
static int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index);
static SyncIndex raftlogCommitIndex(SSyncLogStore* pLogStore);
static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry);
//-------------------------------
SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
SSyncLogStore* pLogStore = taosMemoryMalloc(sizeof(SSyncLogStore));
ASSERT(pLogStore != NULL);
SSyncLogStore* pLogStore = taosMemoryCalloc(1, sizeof(SSyncLogStore));
if (pLogStore == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pLogStore->pCache = taosLRUCacheInit(10 * 1024 * 1024, 1, .5);
if (pLogStore->pCache == NULL) {
terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
taosMemoryFree(pLogStore);
terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
return NULL;
}
@ -96,7 +96,6 @@ void logStoreDestory(SSyncLogStore* pLogStore) {
}
}
//-------------------------------
// log[m .. n]
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) {
ASSERT(snapshotIndex >= 0);

82
source/libs/sync/src/syncRaftStore.c
View File

@ -13,9 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "syncRaftStore.h"
#include "cJSON.h"
#include "syncEnv.h"
#include "syncUtil.h"
// private function
@ -26,22 +25,19 @@ static bool raftStoreFileExist(char *path);
SRaftStore *raftStoreOpen(const char *path) {
int32_t ret;
SRaftStore *pRaftStore = taosMemoryMalloc(sizeof(SRaftStore));
SRaftStore *pRaftStore = taosMemoryCalloc(1, sizeof(SRaftStore));
if (pRaftStore == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
memset(pRaftStore, 0, sizeof(*pRaftStore));
snprintf(pRaftStore->path, sizeof(pRaftStore->path), "%s", path);
char storeBuf[RAFT_STORE_BLOCK_SIZE] = {0};
memset(storeBuf, 0, sizeof(storeBuf));
if (!raftStoreFileExist(pRaftStore->path)) {
ret = raftStoreInit(pRaftStore);
ASSERT(ret == 0);
}
char storeBuf[RAFT_STORE_BLOCK_SIZE] = {0};
pRaftStore->pFile = taosOpenFile(path, TD_FILE_READ | TD_FILE_WRITE);
ASSERT(pRaftStore->pFile != NULL);
@ -72,9 +68,7 @@ static int32_t raftStoreInit(SRaftStore *pRaftStore) {
}
int32_t raftStoreClose(SRaftStore *pRaftStore) {
if (pRaftStore == NULL) {
return 0;
}
if (pRaftStore == NULL) return 0;
taosCloseFile(&pRaftStore->pFile);
taosMemoryFree(pRaftStore);
@ -182,69 +176,3 @@ void raftStoreSetTerm(SRaftStore *pRaftStore, SyncTerm term) {
pRaftStore->currentTerm = term;
raftStorePersist(pRaftStore);
}
int32_t raftStoreFromJson(SRaftStore *pRaftStore, cJSON *pJson) { return 0; }
cJSON *raftStore2Json(SRaftStore *pRaftStore) {
char u64buf[128] = {0};
cJSON *pRoot = cJSON_CreateObject();
if (pRaftStore != NULL) {
snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pRaftStore->currentTerm);
cJSON_AddStringToObject(pRoot, "currentTerm", u64buf);
cJSON *pVoteFor = cJSON_CreateObject();
snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pRaftStore->voteFor.addr);
cJSON_AddStringToObject(pVoteFor, "addr", u64buf);
{
uint64_t u64 = pRaftStore->voteFor.addr;
char host[128] = {0};
uint16_t port;
syncUtilU642Addr(u64, host, sizeof(host), &port);
cJSON_AddStringToObject(pVoteFor, "addr_host", host);
cJSON_AddNumberToObject(pVoteFor, "addr_port", port);
}
cJSON_AddNumberToObject(pVoteFor, "vgId", pRaftStore->voteFor.vgId);
cJSON_AddItemToObject(pRoot, "voteFor", pVoteFor);
int hasVoted = raftStoreHasVoted(pRaftStore);
cJSON_AddNumberToObject(pRoot, "hasVoted", hasVoted);
}
cJSON *pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SRaftStore", pRoot);
return pJson;
}
char *raftStore2Str(SRaftStore *pRaftStore) {
cJSON *pJson = raftStore2Json(pRaftStore);
char *serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
// for debug -------------------
void raftStorePrint(SRaftStore *pObj) {
char *serialized = raftStore2Str(pObj);
printf("raftStorePrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftStorePrint2(char *s, SRaftStore *pObj) {
char *serialized = raftStore2Str(pObj);
printf("raftStorePrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftStoreLog(SRaftStore *pObj) {
char *serialized = raftStore2Str(pObj);
sTrace("raftStoreLog | len:%d | %s", (int32_t)strlen(serialized), serialized);
taosMemoryFree(serialized);
}
void raftStoreLog2(char *s, SRaftStore *pObj) {
char *serialized = raftStore2Str(pObj);
sTrace("raftStoreLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}

5
source/libs/sync/src/syncReplication.c
View File

@ -13,14 +13,11 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "syncReplication.h"
#include "syncIndexMgr.h"
#include "syncMessage.h"
#include "syncRaftCfg.h"
#include "syncRaftEntry.h"
#include "syncRaftLog.h"
#include "syncRaftStore.h"
#include "syncSnapshot.h"
#include "syncUtil.h"
// TLA+ Spec

2
source/libs/sync/src/syncRequestVote.c
View File

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "syncRequestVote.h"
#include "syncInt.h"
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
#include "syncUtil.h"

4
source/libs/sync/src/syncRequestVoteReply.c
View File

@ -13,11 +13,9 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "syncRequestVoteReply.h"
#include "syncInt.h"
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
#include "syncUtil.h"
#include "syncVoteMgr.h"
// TLA+ Spec

3
source/libs/sync/test/syncConfigChangeSnapshotTest.cpp
View File

@ -143,7 +143,8 @@ int32_t SnapshotDoWrite(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_
void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta* cbMeta) {
sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%" PRId64 ", code:%d, currentTerm:%" PRIu64 ", term:%" PRIu64,
sTrace("==callback== ==ReConfigCb== flag:0x%" PRIx64 ", index:%" PRId64 ", code:%d, currentTerm:%" PRIu64
", term:%" PRIu64,
cbMeta->flag, cbMeta->index, cbMeta->code, cbMeta->currentTerm, cbMeta->term);
}

3
source/libs/sync/test/syncConfigChangeTest.cpp
View File

@ -73,7 +73,8 @@ int32_t GetSnapshotCb(const struct SSyncFSM* pFsm, SSnapshot* pSnapshot) {
void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFinishCb=="); }
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta* cbMeta) {
sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%" PRId64 ", code:%d, currentTerm:%" PRIu64 ", term:%" PRIu64,
sTrace("==callback== ==ReConfigCb== flag:%" PRIx64 ", index:%" PRId64 ", code:%d, currentTerm:%" PRIu64
", term:%" PRIu64,
cbMeta->flag, cbMeta->index, cbMeta->code, cbMeta->currentTerm, cbMeta->term);
}

4
source/libs/sync/test/syncTestTool.cpp
View File

@ -148,8 +148,8 @@ void RestoreFinishCb(struct SSyncFSM* pFsm) { sTrace("==callback== ==RestoreFini
void ReConfigCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta* cbMeta) {
char* s = syncCfg2Str(&(cbMeta->newCfg));
sTrace("==callback== ==ReConfigCb== flag:0x%lX, index:%" PRId64 ", code:%d, currentTerm:%" PRIu64 ", term:%" PRIu64
", newCfg:%s",
sTrace("==callback== ==ReConfigCb== flag:%" PRIx64 ", index:%" PRId64 ", code:%d, currentTerm:%" PRIu64
", term:%" PRIu64 ", newCfg:%s",
cbMeta->flag, cbMeta->index, cbMeta->code, cbMeta->currentTerm, cbMeta->term, s);
taosMemoryFree(s);
}

13
source/libs/sync/test/sync_test_lib/inc/syncTest.h
View File

@ -49,6 +49,9 @@ void syncEntryPrint2(char* s, const SSyncRaftEntry* pObj);
void syncEntryLog(const SSyncRaftEntry* pObj);
void syncEntryLog2(char* s, const SSyncRaftEntry* pObj);
char* syncCfg2Str(SSyncCfg* pSyncCfg);
int32_t syncCfgFromStr(const char* s, SSyncCfg* pSyncCfg);
cJSON* raftCache2Json(SRaftEntryHashCache* pObj);
char* raftCache2Str(SRaftEntryHashCache* pObj);
void raftCachePrint(SRaftEntryHashCache* pObj);
@ -63,6 +66,14 @@ void raftEntryCachePrint2(char* s, SRaftEntryCache* pObj);
void raftEntryCacheLog(SRaftEntryCache* pObj);
void raftEntryCacheLog2(char* s, SRaftEntryCache* pObj);
int32_t raftStoreFromJson(SRaftStore* pRaftStore, cJSON* pJson);
cJSON* raftStore2Json(SRaftStore* pRaftStore);
char* raftStore2Str(SRaftStore* pRaftStore);
void raftStorePrint(SRaftStore* pObj);
void raftStorePrint2(char* s, SRaftStore* pObj);
void raftStoreLog(SRaftStore* pObj);
void raftStoreLog2(char* s, SRaftStore* pObj);
cJSON* syncAppendEntriesBatch2Json(const SyncAppendEntriesBatch* pMsg);
char* syncAppendEntriesBatch2Str(const SyncAppendEntriesBatch* pMsg);
void syncAppendEntriesBatchPrint(const SyncAppendEntriesBatch* pMsg);
@ -99,6 +110,8 @@ char* snapshotSender2Str(SSyncSnapshotSender* pSender);
cJSON* snapshotReceiver2Json(SSyncSnapshotReceiver* pReceiver);
char* snapshotReceiver2Str(SSyncSnapshotReceiver* pReceiver);
#ifdef __cplusplus
}
#endif

16
source/libs/sync/test/sync_test_lib/src/syncRaftCfgDebug.c
View File

@ -16,4 +16,20 @@
#define _DEFAULT_SOURCE
#include "syncTest.h"
char *syncCfg2Str(SSyncCfg *pSyncCfg) {
cJSON *pJson = syncCfg2Json(pSyncCfg);
char *serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
int32_t syncCfgFromStr(const char *s, SSyncCfg *pSyncCfg) {
cJSON *pRoot = cJSON_Parse(s);
ASSERT(pRoot != NULL);
int32_t ret = syncCfgFromJson(pRoot, pSyncCfg);
ASSERT(ret == 0);
cJSON_Delete(pRoot);
return 0;
}

84
source/libs/sync/test/sync_test_lib/src/syncRaftStoreDebug.c Normal file
View File

@ -0,0 +1,84 @@
/*
* 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/>.
*/
#define _DEFAULT_SOURCE
#include "syncTest.h"
#include "cJSON.h"
int32_t raftStoreFromJson(SRaftStore *pRaftStore, cJSON *pJson) { return 0; }
cJSON *raftStore2Json(SRaftStore *pRaftStore) {
char u64buf[128] = {0};
cJSON *pRoot = cJSON_CreateObject();
if (pRaftStore != NULL) {
snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pRaftStore->currentTerm);
cJSON_AddStringToObject(pRoot, "currentTerm", u64buf);
cJSON *pVoteFor = cJSON_CreateObject();
snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pRaftStore->voteFor.addr);
cJSON_AddStringToObject(pVoteFor, "addr", u64buf);
{
uint64_t u64 = pRaftStore->voteFor.addr;
char host[128] = {0};
uint16_t port;
syncUtilU642Addr(u64, host, sizeof(host), &port);
cJSON_AddStringToObject(pVoteFor, "addr_host", host);
cJSON_AddNumberToObject(pVoteFor, "addr_port", port);
}
cJSON_AddNumberToObject(pVoteFor, "vgId", pRaftStore->voteFor.vgId);
cJSON_AddItemToObject(pRoot, "voteFor", pVoteFor);
int hasVoted = raftStoreHasVoted(pRaftStore);
cJSON_AddNumberToObject(pRoot, "hasVoted", hasVoted);
}
cJSON *pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SRaftStore", pRoot);
return pJson;
}
char *raftStore2Str(SRaftStore *pRaftStore) {
cJSON *pJson = raftStore2Json(pRaftStore);
char *serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
// for debug -------------------
void raftStorePrint(SRaftStore *pObj) {
char *serialized = raftStore2Str(pObj);
printf("raftStorePrint | len:%d | %s \n", (int32_t)strlen(serialized), serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftStorePrint2(char *s, SRaftStore *pObj) {
char *serialized = raftStore2Str(pObj);
printf("raftStorePrint2 | len:%d | %s | %s \n", (int32_t)strlen(serialized), s, serialized);
fflush(NULL);
taosMemoryFree(serialized);
}
void raftStoreLog(SRaftStore *pObj) {
char *serialized = raftStore2Str(pObj);
sTrace("raftStoreLog | len:%d | %s", (int32_t)strlen(serialized), serialized);
taosMemoryFree(serialized);
}
void raftStoreLog2(char *s, SRaftStore *pObj) {
char *serialized = raftStore2Str(pObj);
sTrace("raftStoreLog2 | len:%d | %s | %s", (int32_t)strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}

4
source/libs/transport/inc/transComm.h
View File

@ -94,8 +94,8 @@ typedef void* queue[2];
/* Return the structure holding the given element. */
#define QUEUE_DATA(e, type, field) ((type*)((void*)((char*)(e)-offsetof(type, field))))
#define TRANS_RETRY_COUNT_LIMIT 100 // retry count limit
#define TRANS_RETRY_INTERVAL 15 // retry interval (ms)
//#define TRANS_RETRY_COUNT_LIMIT 100 // retry count limit
//#define TRANS_RETRY_INTERVAL 15 // retry interval (ms)
#define TRANS_CONN_TIMEOUT 3000 // connect timeout (s)
#define TRANS_READ_TIMEOUT 3000 // read timeout (ms)
#define TRANS_PACKET_LIMIT 1024 * 1024 * 512

2
source/libs/transport/inc/transportInt.h
View File

@ -49,6 +49,8 @@ typedef struct {
int32_t compressSize; // -1: no compress, 0 : all data compressed, size: compress data if larger than size
int8_t encryption; // encrypt or not
int32_t retryLimit; // retry limit
int32_t retryInterval; // retry interval ms
void (*cfp)(void* parent, SRpcMsg*, SEpSet*);
bool (*retry)(int32_t code, tmsg_t msgType);

2
source/libs/transport/src/trans.c
View File

@ -48,6 +48,8 @@ void* rpcOpen(const SRpcInit* pInit) {
pRpc->compressSize = pInit->compressSize;
pRpc->encryption = pInit->encryption;
pRpc->retryLimit = pInit->retryLimit;
pRpc->retryInterval = pInit->retryInterval;
// register callback handle
pRpc->cfp = pInit->cfp;

7
source/libs/transport/src/transCli.c
View File

@ -1336,6 +1336,7 @@ static void doCloseIdleConn(void* param) {
}
static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) {
STrans* pTransInst = pThrd->pTransInst;
STransConnCtx* pCtx = pMsg->ctx;
STraceId* trace = &pMsg->msg.info.traceId;
@ -1347,7 +1348,7 @@ static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) {
STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg));
arg->param1 = pMsg;
arg->param2 = pThrd;
transDQSched(pThrd->delayQueue, doDelayTask, arg, TRANS_RETRY_INTERVAL);
transDQSched(pThrd->delayQueue, doDelayTask, arg, pTransInst->retryInterval);
}
FORCE_INLINE void cliCompareAndSwap(int8_t* val, int8_t exp, int8_t newVal) {
@ -1399,7 +1400,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
pMsg->sent = 0;
pCtx->retryCnt += 1;
if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_BROKEN_LINK) {
cliCompareAndSwap(&pCtx->retryLimit, TRANS_RETRY_COUNT_LIMIT, EPSET_GET_SIZE(&pCtx->epSet) * 3);
cliCompareAndSwap(&pCtx->retryLimit, pTransInst->retryLimit, EPSET_GET_SIZE(&pCtx->epSet) * 3);
if (pCtx->retryCnt < pCtx->retryLimit) {
transUnrefCliHandle(pConn);
EPSET_FORWARD_INUSE(&pCtx->epSet);
@ -1408,7 +1409,7 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
return -1;
}
} else {
cliCompareAndSwap(&pCtx->retryLimit, TRANS_RETRY_COUNT_LIMIT, TRANS_RETRY_COUNT_LIMIT);
cliCompareAndSwap(&pCtx->retryLimit, pTransInst->retryLimit, pTransInst->retryLimit);
if (pCtx->retryCnt < pCtx->retryLimit) {
if (pResp->contLen == 0) {
EPSET_FORWARD_INUSE(&pCtx->epSet);

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

@ -902,9 +902,11 @@ void taosSetCoreDump(bool enable) {
old_len = sizeof(old_usespid);
#ifndef __loongarch64
if (syscall(SYS__sysctl, &args) == -1) {
// printf("_sysctl(kern_core_uses_pid) set fail: %s", strerror(errno));
}
#endif
// printf("The old core_uses_pid[%" PRIu64 "]: %d", old_len, old_usespid);
@ -918,9 +920,11 @@ void taosSetCoreDump(bool enable) {
old_len = sizeof(old_usespid);
#ifndef __loongarch64
if (syscall(SYS__sysctl, &args) == -1) {
// printf("_sysctl(kern_core_uses_pid) get fail: %s", strerror(errno));
}
#endif
// printf("The new core_uses_pid[%" PRIu64 "]: %d", old_len, old_usespid);
#endif

10
source/util/src/tcrc32c.c
View File

@ -19,7 +19,7 @@
*/
#define _DEFAULT_SOURCE
#if !defined(_TD_ARM_) && !defined(_TD_MIPS_)
#if !defined(_TD_ARM_) && !defined(_TD_MIPS_) && !defined(_TD_LOONGARCH_)
#include <nmmintrin.h>
#endif
@ -512,7 +512,7 @@ static uint32_t table[16][256] = {
0x9c221d09, 0x6e2e10f7, 0x7dd67004, 0x8fda7dfa}
};
#if !defined(_TD_ARM_) && !defined(_TD_MIPS_)
#if !defined(_TD_ARM_) && !defined(_TD_MIPS_) && !defined(_TD_LOONGARCH_)
static uint32_t long_shifts[4][256] = {
{0x00000000, 0xe040e0ac, 0xc56db7a9, 0x252d5705, 0x8f3719a3, 0x6f77f90f, 0x4a5aae0a, 0xaa1a4ea6, 0x1b8245b7,
0xfbc2a51b, 0xdeeff21e, 0x3eaf12b2, 0x94b55c14, 0x74f5bcb8, 0x51d8ebbd, 0xb1980b11, 0x37048b6e, 0xd7446bc2,
@ -846,7 +846,7 @@ uint32_t crc32c_sf(uint32_t crci, crc_stream input, size_t length) {
}
return (uint32_t)crc ^ 0xffffffff;
}
#if !defined(_TD_ARM_) && !defined(_TD_MIPS_)
#if !defined(_TD_ARM_) && !defined(_TD_MIPS_) && !defined(_TD_LOONGARCH_)
/* Apply the zeros operator table to crc. */
static uint32_t shift_crc(uint32_t shift_table[][256], uint32_t crc) {
return shift_table[0][crc & 0xff] ^ shift_table[1][(crc >> 8) & 0xff] ^ shift_table[2][(crc >> 16) & 0xff] ^
@ -857,7 +857,7 @@ static uint32_t shift_crc(uint32_t shift_table[][256], uint32_t crc) {
version. Otherwise, use the software version. */
uint32_t (*crc32c)(uint32_t crci, crc_stream bytes, size_t len) = crc32c_sf;
#if !defined(_TD_ARM_) && !defined(_TD_MIPS_)
#if !defined(_TD_ARM_) && !defined(_TD_MIPS_) && !defined(_TD_LOONGARCH_)
/* Compute CRC-32C using the Intel hardware instruction. */
uint32_t crc32c_hw(uint32_t crc, crc_stream buf, size_t len) {
crc_stream next = buf;
@ -1012,7 +1012,7 @@ uint32_t crc32c_hw(uint32_t crc, crc_stream buf, size_t len) {
#endif // #ifndef _TD_ARM_
void taosResolveCRC() {
#if defined _TD_ARM_ || defined _TD_MIPS_ || defined WINDOWS
#if defined _TD_ARM_ || defined _TD_MIPS_ || defined WINDOWS || defined _TD_LOONGARCH_
crc32c = crc32c_sf;
#else
int32_t sse42;

5
tests/parallel_test/cases.task
View File

@ -6,8 +6,8 @@
,,y,unit-test,bash test.sh
#tsim test
,,y,script,./test.sh -f tsim/user/basic.sim
,,y,script,./test.sh -f tsim/user/password.sim
,,,script,./test.sh -f tsim/user/basic.sim
,,,script,./test.sh -f tsim/user/password.sim
,,,script,./test.sh -f tsim/user/privilege_db.sim
,,,script,./test.sh -f tsim/user/privilege_sysinfo.sim
,,,script,./test.sh -f tsim/db/alter_option.sim
@ -714,6 +714,7 @@
,,,system-test,python3 ./test.py -f 7-tmq/tmq_taosx.py
,,,system-test,python3 ./test.py -f 7-tmq/stbTagFilter-multiCtb.py
,,,system-test,python3 ./test.py -f 99-TDcase/TD-19201.py
,,,system-test,python3 ./test.py -f 7-tmq/tmqSubscribeStb-r3.py -N 5
,,,system-test,python3 ./test.py -f 2-query/between.py -Q 2
,,,system-test,python3 ./test.py -f 2-query/distinct.py -Q 2
,,,system-test,python3 ./test.py -f 2-query/varchar.py -Q 2

4
tests/parallel_test/container_build.sh
View File

@ -54,7 +54,7 @@ fi
date
docker run \
-v $REP_MOUNT_PARAM \
--rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true;make -j $THREAD_COUNT"
--rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true;make -j $THREAD_COUNT || exit 1"
if [[ -d ${WORKDIR}/debugNoSan ]] ;then
echo "delete ${WORKDIR}/debugNoSan"
@ -69,7 +69,7 @@ mv ${REP_REAL_PATH}/debug ${WORKDIR}/debugNoSan
date
docker run \
-v $REP_MOUNT_PARAM \
--rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true -DSANITIZER=true;make -j $THREAD_COUNT"
--rm --ulimit core=-1 taos_test:v1.0 sh -c "cd $REP_DIR;rm -rf debug;mkdir -p debug;cd debug;cmake .. -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_TEST=true -DWEBSOCKET=true -DSANITIZER=true;make -j $THREAD_COUNT || exit 1 "
mv ${REP_REAL_PATH}/debug ${WORKDIR}/debugSan

39
tests/script/sh/checkAsan.sh Executable file
View File

@ -0,0 +1,39 @@
#!/bin/bash
set +e
#set -x
SCRIPT_DIR=`dirname $0`
cd $SCRIPT_DIR/../
SCRIPT_DIR=`pwd`
IN_TDINTERNAL="community"
if [[ "$SCRIPT_DIR" == *"$IN_TDINTERNAL"* ]]; then
cd ../../..
else
cd ../../
fi
TAOS_DIR=`pwd`
LOG_DIR=$TAOS_DIR/sim/tsim/asan
error_num=`cat ${LOG_DIR}/tsim.asan | grep "ERROR" | wc -l`
memory_leak=`cat ${LOG_DIR}/tsim.asan | grep "Direct leak" | wc -l`
indirect_leak=`cat ${LOG_DIR}/tsim.asan | grep "Indirect leak" | wc -l`
runtime_error=`cat ${LOG_DIR}/tsim.asan | grep "runtime error" | wc -l`
echo -e "\033[44;32;1m"asan error_num: $error_num"\033[0m"
echo -e "\033[44;32;1m"asan memory_leak: $memory_leak"\033[0m"
echo -e "\033[44;32;1m"asan indirect_leak: $indirect_leak"\033[0m"
echo -e "\033[44;32;1m"asan runtime error: $runtime_error"\033[0m"
let "errors=$error_num+$memory_leak+$indirect_leak+$runtime_error"
if [ $errors -eq 0 ]; then
echo -e "\033[44;32;1m"no asan errors"\033[0m"
exit 0
else
echo -e "\033[44;31;1m"asan total errors: $errors"\033[0m"
cat ${LOG_DIR}/tsim.asan
exit 1
fi

4
tests/script/sh/exec.sh
View File

@ -80,7 +80,7 @@ LOG_DIR=$NODE_DIR/log
DATA_DIR=$NODE_DIR/data
MGMT_DIR=$NODE_DIR/data/mgmt
TSDB_DIR=$NODE_DIR/data/tsdb
ASAN_DIR=$SIM_DIR/tsim/asan
TAOS_CFG=$NODE_DIR/cfg/taos.cfg
echo ------------ $EXEC_OPTON $NODE_NAME
@ -105,7 +105,7 @@ if [ "$EXEC_OPTON" = "start" ]; then
nohup valgrind --log-file=${LOG_DIR}/valgrind-taosd-${NODE_NAME}-${TT}.log --tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all --num-callers=20 -v -v --workaround-gcc296-bugs=yes $EXE_DIR/taosd -c $CFG_DIR > /dev/null 2>&1 &
else
echo "nohup $EXE_DIR/taosd -c $CFG_DIR > /dev/null 2>&1 &"
nohup $EXE_DIR/taosd -c $CFG_DIR > /dev/null 2>&1 &
nohup $EXE_DIR/taosd -c $CFG_DIR > /dev/null 2> $ASAN_DIR/$NODE_NAME.asan &
fi
else

7
tests/script/test.sh
View File

@ -74,6 +74,7 @@ PRG_DIR=$SIM_DIR/tsim
CFG_DIR=$PRG_DIR/cfg
LOG_DIR=$PRG_DIR/log
DATA_DIR=$PRG_DIR/data
ASAN_DIR=$PRG_DIR/asan
chmod -R 777 $PRG_DIR
echo "------------------------------------------------------------------------"
@ -82,14 +83,17 @@ echo "BUILD_DIR: $BUILD_DIR"
echo "SIM_DIR : $SIM_DIR"
echo "CODE_DIR : $CODE_DIR"
echo "CFG_DIR : $CFG_DIR"
echo "ASAN_DIR : $ASAN_DIR"
rm -rf $SIM_DIR/*
rm -rf $LOG_DIR
rm -rf $CFG_DIR
rm -rf $ASAN_DIR
mkdir -p $PRG_DIR
mkdir -p $LOG_DIR
mkdir -p $CFG_DIR
mkdir -p $ASAN_DIR
TAOS_CFG=$PRG_DIR/cfg/taos.cfg
touch -f $TAOS_CFG
@ -133,7 +137,8 @@ if [ -n "$FILE_NAME" ]; then
$PROGRAM -c $CFG_DIR -f $FILE_NAME -v
else
echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f $FILE_NAME
$PROGRAM -c $CFG_DIR -f $FILE_NAME
$PROGRAM -c $CFG_DIR -f $FILE_NAME 2> $ASAN_DIR/tsim.asan
$CODE_DIR/sh/checkAsan.sh
fi
else
echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f basicSuite.sim

4
tests/script/tsim/db/commit.sim
View File

@ -89,8 +89,8 @@ if $data01 != 40 then
endi
print ======== step4 import new data
sql_error import into tb values (now + 30d , 30 )
sql_error import into tb values (now + 31d , 31 )
#sql_error import into tb values (now + 30d , 30 )
#sql_error import into tb values (now + 31d , 31 )
sql select * from tb order by ts desc
print ===> rows $rows

54
tests/script/tsim/stream/drop_stream.sim
View File

@ -216,6 +216,60 @@ sql insert into scalar_tb values (1656668180503+1s, -50, 50.1, "beiJing", "TDeng
print ========== step6 repeat
sql drop database test;
print ========== interval\session\state window
sql CREATE DATABASE test1 BUFFER 96 CACHESIZE 1 CACHEMODEL 'none' COMP 2 DURATION 14400m WAL_FSYNC_PERIOD 3000 MAXROWS 4096 MINROWS 100 KEEP 5256000m,5256000m,5256000m PAGES 256 PAGESIZE 4 PRECISION 'ms' REPLICA 1 STRICT 'off' WAL_LEVEL 1 VGROUPS 2 SINGLE_STABLE 0;
sql use test1;
sql CREATE STABLE st (time TIMESTAMP, ca DOUBLE, cb DOUBLE, cc int) TAGS (ta VARCHAR(10) );
print ========== create table before stream
sql CREATE TABLE t1 using st TAGS ('aaa');
sql CREATE TABLE t2 using st TAGS ('bbb');
sql CREATE TABLE t3 using st TAGS ('ccc');
sql CREATE TABLE t4 using st TAGS ('ddd');
print ========== stable
sql create stream streamd1 into streamt1 as select ca, _wstart,_wend, count(*) as total from st where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca interval(60m) fill(null);
sql create stream streamd2 into streamt2 as select ca, _wstart,_wend, count(*), max(ca), max(cb) from st where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca interval(60m) fill(linear);
sql create stream streamd3 into streamt3 as select ca, _wstart,_wend, count(*) as total from st where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca session(time, 60m);
sql create stream streamd4 into streamt4 as select ta, _wstart,_wend, count(*) as total from st where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ta session(time, 60m);
sql_error create stream streamd5 into streamt5 as select ca, _wstart,_wend, count(*) as total from st where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca state_window(cc);
sql_error create stream streamd6 into streamt6 as select ta, _wstart,_wend, count(*) as total from st where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ta state_window(cc);
print ========== table
sql create stream streamd7 into streamt7 as select ca, _wstart,_wend, count(*) as total from t1 where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca interval(60m) fill(null);
sql create stream streamd8 into streamt8 as select ca, _wstart,_wend, count(*), max(ca), max(cb) from t1 where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca interval(60m) fill(linear);
sql create stream streamd9 into streamt9 as select ca, _wstart,_wend, count(*) as total from t1 where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca session(time, 60m);
sql create stream streamd10 into streamt10 as select ta, _wstart,_wend, count(*) as total from t1 where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ta session(time, 60m);
sql create stream streamd11 into streamt11 as select ca, _wstart,_wend, count(*) as total from t1 where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ca state_window(cc);
sql create stream streamd12 into streamt12 as select ta, _wstart,_wend, count(*) as total from t1 where time > "2022-01-01 00:00:00" and time < "2032-01-01 00:00:00" partition by ta state_window(cc);
print ========== create table after stream
sql CREATE TABLE t5 using st TAGS ('eee');
sql CREATE TABLE t6 using st TAGS ('fff');
sql CREATE TABLE t7 using st TAGS ('ggg');
sql CREATE TABLE t8 using st TAGS ('fff');
sleep 1000
print ========== drop stream
sql drop stream if exists streamd1;
sql drop stream if exists streamd2;
sql drop stream if exists streamd3;
sql drop stream if exists streamd4;
#sql drop stream if exists streamd5;
#sql drop stream if exists streamd6;
sql drop stream if exists streamd7;
sql drop stream if exists streamd8;
sql drop stream if exists streamd9;
sql drop stream if exists streamd10;
sql drop stream if exists streamd11;
sql drop stream if exists streamd12;
print ========== step7
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT
system sh/exec.sh -n dnode3 -s stop -x SIGINT

166
tests/script/tsim/table/date.sim
View File

@ -9,7 +9,7 @@ $tbPrefix = lm_da_tb
$db = $dbPrefix . $i
$tb = $tbPrefix . $i
print =============== step1
print =============== step1 ms db
sql create database $db
sql use $db
@ -23,7 +23,7 @@ if $data00 != @17-01-01 08:00:00.001@ then
return -1
endi
print =============== step2
print =============== step2 ms db
sql_error insert into $tb values ('2017-08-28 00:23:46.429+ 1a', 2)
sql_error insert into $tb values ('2017-08-28 00:23:46cd .429', 2)
sql select ts from $tb
@ -31,7 +31,7 @@ if $rows != 1 then
return -1
endi
print =============== step3
print =============== step3 ms db
sql_error insert into $tb values ('1970-01-01 08:00:00.000', 3)
sql_error insert into $tb values ('1970-01-01 08:00:00.000', 3)
sql select ts from $tb
@ -39,39 +39,48 @@ if $rows != 1 then
return -1
endi
print =============== step4
print =============== step4 ms db
sql insert into $tb values(now, 4);
sql insert into $tb values(now+1a, 5);
sql insert into $tb values(now+1s, 6);
sql insert into $tb values(now+1m, 7);
sql insert into $tb values(now+1h, 8);
sql insert into $tb values(now+1d, 9);
sql_error insert into $tb values(now+3w, 10);
sql_error insert into $tb values(now+1n, 11);
sql_error insert into $tb values(now+1y, 12);
sql insert into $tb values(now+3w, 10);
sql insert into $tb values(31556995200000, 11);
sql insert into $tb values('2970-01-01 00:00:00.000', 12);
print =============== step5
sql_error insert into $tb values ('9999-12-31 213:59:59.999', 13)
sql_error insert into $tb values(now+1n, 20);
sql_error insert into $tb values(now+1y, 21);
sql_error insert into $tb values(31556995200001, 22);
sql_error insert into $tb values('2970-01-02 00:00:00.000', 23);
sql_error insert into $tb values(9223372036854775807, 24);
sql_error insert into $tb values(9223372036854775808, 25);
sql_error insert into $tb values(92233720368547758088, 26);
print =============== step5 ms db
sql_error insert into $tb values ('9999-12-31 213:59:59.999', 27)
sql select ts from $tb
print $rows
if $rows != 7 then
if $rows != 10 then
return -1
endi
print =============== step6
sql_error insert into $tb values ('9999-12-99 23:59:59.999', 13)
print =============== step6 ms db
sql_error insert into $tb values ('9999-12-99 23:59:59.999', 28)
sql select ts from $tb
if $rows != 7 then
if $rows != 10 then
return -1
endi
print =============== step7
print =============== step7 ms db
$i = 1
$tb = $tbPrefix . $i
sql create table $tb (ts timestamp, ts2 timestamp)
print =============== step8
print =============== step8 ms db
sql insert into $tb values (now, now)
sql select * from $tb
if $rows != 1 then
@ -84,4 +93,131 @@ if $rows != 2 then
return -1
endi
print =============== step20 us db
sql create database $db precision 'us' keep 365000d;
sql create table $tb (ts timestamp, speed int)
sql insert into $tb values ('2017-01-01 08:00:00.001', 1)
sql select ts from $tb
if $rows != 1 then
return -1
endi
if $data00 != @17-01-01 08:00:00.001000@ then
print data00 = $data00
return -1
endi
print =============== step21 us db
sql_error insert into $tb values ('2017-08-28 00:23:46.429+ 1a', 2)
sql_error insert into $tb values ('2017-08-28 00:23:46cd .429', 2)
sql select ts from $tb
if $rows != 1 then
return -1
endi
print =============== step22 us db
sql_error insert into $tb values ('970-01-01 08:00:00.000', 3)
sql_error insert into $tb values ('970-01-01 08:00:00.000', 3)
sql select ts from $tb
if $rows != 1 then
return -1
endi
print =============== step23 us db
sql insert into $tb values(now, 4);
sql insert into $tb values(now+1a, 5);
sql insert into $tb values(now+1s, 6);
sql insert into $tb values(now+1m, 7);
sql insert into $tb values(now+1h, 8);
sql insert into $tb values(now+1d, 9);
sql insert into $tb values(now+3w, 10);
sql insert into $tb values(31556995200000000, 11);
sql insert into $tb values('2970-01-01 00:00:00.000000', 12);
sql_error insert into $tb values(now+1n, 20);
sql_error insert into $tb values(now+1y, 21);
sql_error insert into $tb values(31556995200000001, 22);
sql_error insert into $tb values('2970-01-02 00:00:00.000000', 23);
sql_error insert into $tb values(9223372036854775807, 24);
sql_error insert into $tb values(9223372036854775808, 25);
sql_error insert into $tb values(92233720368547758088, 26);
sql_error insert into $tb values ('9999-12-31 213:59:59.999', 27)
print =============== step24 us db
sql select ts from $tb
print $rows
if $rows != 10 then
return -1
endi
sql drop database $db
sql select * from information_schema.ins_databases
if $rows != 2 then
return -1
endi
print =============== step30 ns db
sql create database $db precision 'ns' keep 36500d;
sql create table $tb (ts timestamp, speed int)
sql insert into $tb values ('2017-01-01 08:00:00.001', 1)
sql select ts from $tb
if $rows != 1 then
return -1
endi
if $data00 != @17-01-01 08:00:00.001000000@ then
print data00 = $data00
return -1
endi
print =============== step31 ns db
sql_error insert into $tb values ('2017-08-28 00:23:46.429+ 1a', 2)
sql_error insert into $tb values ('2017-08-28 00:23:46cd .429', 2)
sql select ts from $tb
if $rows != 1 then
return -1
endi
print =============== step32 ns db
#sql_error insert into $tb values ('970-01-01 08:00:00.000000000', 3)
#sql_error insert into $tb values ('970-01-01 08:00:00.000000000', 3)
sql select ts from $tb
if $rows != 1 then
return -1
endi
print =============== step33 ns db
sql insert into $tb values(now, 4);
sql insert into $tb values(now+1a, 5);
sql insert into $tb values(now+1s, 6);
sql insert into $tb values(now+1m, 7);
sql insert into $tb values(now+1h, 8);
sql insert into $tb values(now+1d, 9);
sql insert into $tb values(now+3w, 10);
sql insert into $tb values(9214646400000000000, 11);
sql insert into $tb values('2262-01-01 00:00:00.000000000', 12);
sql_error insert into $tb values(now+1n, 20);
sql_error insert into $tb values(now+1y, 21);
sql_error insert into $tb values(9214646400000000001, 22);
sql_error insert into $tb values('2262-01-02 00:00:00.000000000', 23);
sql_error insert into $tb values(9223372036854775807, 24);
sql_error insert into $tb values(9223372036854775808, 25);
sql_error insert into $tb values(92233720368547758088, 26);
sql_error insert into $tb values ('9999-12-31 213:59:59.999', 27)
print =============== step34 ns db
sql select ts from $tb
print $rows
if $rows != 10 then
return -1
endi
sql drop database $db
sql select * from information_schema.ins_databases
if $rows != 2 then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT

8
tests/system-test/0-others/taosdShell.py
View File

@ -2,6 +2,7 @@
import taos
import sys
import time
from datetime import datetime
import socket
import os
import platform
@ -100,8 +101,11 @@ class TDTestCase:
processName="taosd"
taosdCmd = taosdCmdRun + startAction
tdLog.printNoPrefix("%s"%taosdCmd)
os.system(f"nohup {taosdCmd} & ")
logTime=datetime.now().strftime('%Y%m%d_%H%M%S_%f')
os.system(f"nohup {taosdCmd} > {logTime}.log 2>&1 & ")
self.checkAndstopPro(processName,startAction)
os.system(f"rm -rf {logTime}.log")
def taosdCommandExe(self,startAction,taosdCmdRun):
taosdCmd = taosdCmdRun + startAction
@ -207,7 +211,7 @@ class TDTestCase:
os.system(" mkdir -p taosdCaseTmp ")
os.system("echo \'TAOS_QUERY_POLICY=3\' > taosdCaseTmp/.env ")
self.taosdCommandStop(startAction,taosdCmdRun)
os.system(" rm -rf taosdCaseTmp/.env ")
os.system(" rm -rf taosdCaseTmp ")
startAction = " -V"
tdLog.printNoPrefix("================================ parameter: %s"%startAction)

30
tests/system-test/7-tmq/tmqSubscribeStb-r3.py
View File

@ -1,26 +1,27 @@
from distutils.log import error
import taos
import sys
import time
import socket
import os
import threading
import subprocess
import platform
import socket
import subprocess
import sys
import threading
import time
from distutils.log import error
from util.log import *
from util.sql import *
import taos
from util.cases import *
from util.dnodes import *
from util.dnodes import TDDnodes
from util.dnodes import TDDnode
from util.cluster import *
from util.common import *
from util.dnodes import *
from util.dnodes import TDDnode, TDDnodes
from util.log import *
from util.sql import *
sys.path.append("./6-cluster")
sys.path.append("./7-tmq")
from tmqCommon import *
from clusterCommonCreate import *
from clusterCommonCheck import clusterComCheck
from clusterCommonCreate import *
from tmqCommon import *
class TDTestCase:
def __init__(self):
@ -265,6 +266,7 @@ class TDTestCase:
tdLog.info("start consume processor")
tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot'])
tmqCom.getStartConsumeNotifyFromTmqsim()
tdLog.info("================= restart dnode 2===========================")
cluster.dnodes[1].stoptaosd()
cluster.dnodes[1].starttaosd()

17
tools/shell/src/shellEngine.c
View File

@ -540,11 +540,20 @@ void shellPrintField(const char *val, TAOS_FIELD *field, int32_t width, int32_t
}
}
bool shellIsLimitQuery(const char *sql) {
// todo refactor
// show whole result for this query return true, like limit or describe
bool shellIsShowWhole(const char *sql) {
// limit
if (taosStrCaseStr(sql, " limit ") != NULL) {
return true;
}
// describe
if (taosStrCaseStr(sql, "describe ") != NULL) {
return true;
}
// show
if (taosStrCaseStr(sql, "show ") != NULL) {
return true;
}
return false;
}
@ -578,7 +587,7 @@ int32_t shellVerticalPrintResult(TAOS_RES *tres, const char *sql) {
uint64_t resShowMaxNum = UINT64_MAX;
if (shell.args.commands == NULL && shell.args.file[0] == 0 && !shellIsLimitQuery(sql)) {
if (shell.args.commands == NULL && shell.args.file[0] == 0 && !shellIsShowWhole(sql)) {
resShowMaxNum = SHELL_DEFAULT_RES_SHOW_NUM;
}
@ -723,7 +732,7 @@ int32_t shellHorizontalPrintResult(TAOS_RES *tres, const char *sql) {
uint64_t resShowMaxNum = UINT64_MAX;
if (shell.args.commands == NULL && shell.args.file[0] == 0 && !shellIsLimitQuery(sql)) {
if (shell.args.commands == NULL && shell.args.file[0] == 0 && !shellIsShowWhole(sql)) {
resShowMaxNum = SHELL_DEFAULT_RES_SHOW_NUM;
}

8
utils/test/c/sml_test.c
View File

@ -1097,18 +1097,12 @@ int sml_time_Test() {
pRes = taos_query(taos, "use sml_db");
taos_free_result(pRes);
char* tmp = (char*)taosMemoryCalloc(1024, 1);
memcpy(tmp, sql[0], strlen(sql[0]));
*(char*)(tmp+44) = 0;
int32_t totalRows = 0;
pRes = taos_schemaless_insert_raw(taos, tmp, strlen(sql[0]), &totalRows, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS);
pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS);
ASSERT(totalRows == 3);
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
taos_close(taos);
taosMemoryFree(tmp);
return code;
}