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

Merge branch '3.0' into fix/TD-20317

This commit is contained in:
haoranchen 2022-11-11 12:49:25 +08:00 committed by GitHub
parent 6a2aa3ebb5 2983f9ba48
commit 798e46cfd9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
54 changed files with 2994 additions and 2623 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
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.

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;

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

@ -76,12 +76,14 @@ typedef void (*RpcDfp)(void *ahandle);
typedef struct SRpcInit {
char localFqdn[TSDB_FQDN_LEN];
uint16_t localPort; // local port
char *label; // for debug purpose
int32_t numOfThreads; // number of threads to handle connections
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
uint16_t localPort; // local port
char *label; // for debug purpose
int32_t numOfThreads; // number of threads to handle connections
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

6
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] "
@ -136,4 +136,4 @@ if [ "$cloudBuild" != "y" ] && [ ${dockerLatest} == 'y' ] ;then
docker push tdengine/tdengine-${dockername}:latest
fi
rm -f ${pkgFile}
rm -f ${pkgFile}

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) {

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

@ -163,10 +163,12 @@ int32_t tsMqRebalanceInterval = 2;
int32_t tsTtlUnit = 86400;
int32_t tsTtlPushInterval = 86400;
int32_t tsGrantHBInterval = 60;
int32_t tsUptimeInterval = 300; // seconds
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/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)

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

4
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++;
tbDataMovePosTo(pTbData, pos, &key, SL_MOVE_FROM_POS);
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:

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

@ -6288,21 +6288,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;
SSelectStmt* pSelect = NULL;
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);
}

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

@ -94,11 +94,11 @@ 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_CONN_TIMEOUT 3 // connect timeout (s)
#define TRANS_READ_TIMEOUT 3000 // read timeout (ms)
#define TRANS_PACKET_LIMIT 1024 * 1024 * 512
//#define TRANS_RETRY_COUNT_LIMIT 100 // retry count limit
//#define TRANS_RETRY_INTERVAL 15 // retry interval (ms)
#define TRANS_CONN_TIMEOUT 3 // connect timeout (s)
#define TRANS_READ_TIMEOUT 3000 // read timeout (ms)
#define TRANS_PACKET_LIMIT 1024 * 1024 * 512
#define TRANS_MAGIC_NUM 0x5f375a86

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

@ -47,8 +47,10 @@ typedef struct {
char label[TSDB_LABEL_LEN];
char user[TSDB_UNI_LEN]; // meter ID
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 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

@ -1288,6 +1288,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;
@ -1299,7 +1300,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) {
@ -1351,7 +1352,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);
@ -1360,7 +1361,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);

6
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
@ -989,4 +993,4 @@ bool taosCheckCurrentInDll() {
#else
return false;
#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;

2
tests/parallel_test/container_build.sh Executable file → Normal file
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"

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)

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;
}