Skip to content

T
TDengine

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

通知 1192
Star 22018
Fork 4786
T
TDengine
未验证 提交 757e4c2d 编写于 作者: L Liu Jicong 提交者： GitHub
浏览文件
操作

Merge pull request #13040 from taosdata/feature/stream 

fix(stream): double free
上级 5028c5d7 d3a2e98d
显示空白变更内容
内联 并排
Showing with 139 addition and 278 deletion
include/libs/stream/tstream.h
浏览文件 @ 757e4c2d
...@@ -84,15 +84,16 @@ typedef struct { ...@@ -84,15 +84,16 @@ typedef struct {
} SStreamCheckpoint; } SStreamCheckpoint;
static FORCE_INLINE SStreamDataSubmit* streamDataSubmitNew(SSubmitReq* pReq) { static FORCE_INLINE SStreamDataSubmit* streamDataSubmitNew(SSubmitReq* pReq) {
SStreamDataSubmit* pDataSubmit = (SStreamDataSubmit*)taosMemoryCalloc(1, sizeof(SStreamDataSubmit)); SStreamDataSubmit* pDataSubmit = (SStreamDataSubmit*)taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM);
if (pDataSubmit == NULL) return NULL; if (pDataSubmit == NULL) return NULL;
pDataSubmit->data = pReq;
pDataSubmit->dataRef = (int32_t*)taosMemoryMalloc(sizeof(int32_t)); pDataSubmit->dataRef = (int32_t*)taosMemoryMalloc(sizeof(int32_t));
if (pDataSubmit->data == NULL) goto FAIL; if (pDataSubmit->dataRef == NULL) goto FAIL;
pDataSubmit->data = pReq;
*pDataSubmit->dataRef = 1; *pDataSubmit->dataRef = 1;
pDataSubmit->type = STREAM_INPUT__DATA_SUBMIT;
return pDataSubmit; return pDataSubmit;
FAIL: FAIL:
taosMemoryFree(pDataSubmit); taosFreeQitem(pDataSubmit);
return NULL; return NULL;
} }
...@@ -107,7 +108,6 @@ static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit) ...@@ -107,7 +108,6 @@ static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit)
if (ref == 0) { if (ref == 0) {
taosMemoryFree(pDataSubmit->data); taosMemoryFree(pDataSubmit->data);
taosMemoryFree(pDataSubmit->dataRef); taosMemoryFree(pDataSubmit->dataRef);
taosFreeQitem(pDataSubmit);
} }
} }
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 757e4c2d
...@@ -831,25 +831,13 @@ FAIL: ...@@ -831,25 +831,13 @@ FAIL:
int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) { int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
void* pIter = NULL; void* pIter = NULL;
bool failed = false; bool failed = false;
SStreamDataSubmit* pSubmit = NULL;
SStreamDataSubmit* pSubmit = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM); pSubmit = streamDataSubmitNew(pReq);
if (pSubmit == NULL) { if (pSubmit == NULL) {
failed = true; failed = true;
goto SET_TASK_FAIL;
} }
pSubmit->dataRef = taosMemoryMalloc(sizeof(int32_t));
if (pSubmit->dataRef == NULL) {
failed = true;
goto SET_TASK_FAIL;
}
pSubmit->type = STREAM_INPUT__DATA_SUBMIT;
/*pSubmit->sourceVer = ver;*/
/*pSubmit->sourceVg = pTq->pVnode->config.vgId;*/
pSubmit->data = pReq;
*pSubmit->dataRef = 1;
SET_TASK_FAIL:
while (1) { while (1) {
pIter = taosHashIterate(pTq->pStreamTasks, pIter); pIter = taosHashIterate(pTq->pStreamTasks, pIter);
if (pIter == NULL) break; if (pIter == NULL) break;
...@@ -864,7 +852,9 @@ SET_TASK_FAIL: ...@@ -864,7 +852,9 @@ SET_TASK_FAIL:
} }
streamDataSubmitRefInc(pSubmit); streamDataSubmitRefInc(pSubmit);
taosWriteQitem(pTask->inputQ, pSubmit); SStreamDataSubmit* pSubmitClone = taosAllocateQitem(sizeof(SStreamDataSubmit), DEF_QITEM);
memcpy(pSubmitClone, pSubmit, sizeof(SStreamDataSubmit));
taosWriteQitem(pTask->inputQ, pSubmitClone);
int8_t execStatus = atomic_load_8(&pTask->status); int8_t execStatus = atomic_load_8(&pTask->status);
if (execStatus == TASK_STATUS__IDLE || execStatus == TASK_STATUS__CLOSING) { if (execStatus == TASK_STATUS__IDLE || execStatus == TASK_STATUS__CLOSING) {
...@@ -887,18 +877,12 @@ SET_TASK_FAIL: ...@@ -887,18 +877,12 @@ SET_TASK_FAIL:
} }
} }
if (!failed) {
streamDataSubmitRefDec(pSubmit);
return 0;
} else {
if (pSubmit) { if (pSubmit) {
if (pSubmit->dataRef) { streamDataSubmitRefDec(pSubmit);
taosMemoryFree(pSubmit->dataRef);
}
taosFreeQitem(pSubmit); taosFreeQitem(pSubmit);
} }
return -1;
} return failed ? -1 : 0;
} }
int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) { int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {
......
source/libs/stream/src/tstream.c
浏览文件 @ 757e4c2d
...@@ -166,6 +166,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) ...@@ -166,6 +166,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
// destroy // destroy
if (pTask->inputType == STREAM_INPUT__DATA_SUBMIT) { if (pTask->inputType == STREAM_INPUT__DATA_SUBMIT) {
streamDataSubmitRefDec((SStreamDataSubmit*)data); streamDataSubmitRefDec((SStreamDataSubmit*)data);
taosFreeQitem(data);
} else { } else {
taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)tDeleteSSDataBlock); taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)tDeleteSSDataBlock);
taosFreeQitem(data); taosFreeQitem(data);
...@@ -173,15 +174,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) ...@@ -173,15 +174,7 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
return 0; return 0;
} }
// TODO: handle version static SArray* streamExecForQall(SStreamTask* pTask, SArray* pRes) {
int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) {
SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
if (pRes == NULL) return -1;
while (1) {
int8_t execStatus = atomic_val_compare_exchange_8(&pTask->status, TASK_STATUS__IDLE, TASK_STATUS__EXECUTING);
void* exec = pTask->exec.executor;
if (execStatus == TASK_STATUS__IDLE) {
// first run, from qall, handle failure from last exec
while (1) { while (1) {
void* data = NULL; void* data = NULL;
taosGetQitem(pTask->inputQAll, &data); taosGetQitem(pTask->inputQAll, &data);
...@@ -189,81 +182,41 @@ int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) { ...@@ -189,81 +182,41 @@ int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) {
streamTaskExecImpl(pTask, data, pRes); streamTaskExecImpl(pTask, data, pRes);
/*taosFreeQitem(data);*/
if (taosArrayGetSize(pRes) != 0) { if (taosArrayGetSize(pRes) != 0) {
SStreamDataBlock* resQ = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM); SStreamDataBlock* qRes = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM);
resQ->type = STREAM_INPUT__DATA_BLOCK; qRes->type = STREAM_INPUT__DATA_BLOCK;
resQ->blocks = pRes; qRes->blocks = pRes;
taosWriteQitem(pTask->outputQ, resQ); taosWriteQitem(pTask->outputQ, qRes);
pRes = taosArrayInit(0, sizeof(SSDataBlock)); return taosArrayInit(0, sizeof(SSDataBlock));
if (pRes == NULL) goto FAIL;
} }
} }
// second run, from inputQ return pRes;
taosReadAllQitems(pTask->inputQ, pTask->inputQAll); }
while (1) {
void* data = NULL;
taosGetQitem(pTask->inputQAll, &data);
if (data == NULL) break;
streamTaskExecImpl(pTask, data, pRes);
/*taosFreeQitem(data);*/
if (taosArrayGetSize(pRes) != 0) { // TODO: handle version
SStreamDataBlock* resQ = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM); int32_t streamExec(SStreamTask* pTask, SMsgCb* pMsgCb) {
resQ->type = STREAM_INPUT__DATA_BLOCK; SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
resQ->blocks = pRes; if (pRes == NULL) return -1;
taosWriteQitem(pTask->outputQ, resQ);
pRes = taosArrayInit(0, sizeof(SSDataBlock));
if (pRes == NULL) goto FAIL;
}
}
// set status closing
atomic_store_8(&pTask->status, TASK_STATUS__CLOSING);
// third run, make sure all inputQ is cleared
taosReadAllQitems(pTask->inputQ, pTask->inputQAll);
while (1) { while (1) {
void* data = NULL; int8_t execStatus = atomic_val_compare_exchange_8(&pTask->status, TASK_STATUS__IDLE, TASK_STATUS__EXECUTING);
taosGetQitem(pTask->inputQAll, &data); void* exec = pTask->exec.executor;
if (data == NULL) break; if (execStatus == TASK_STATUS__IDLE) {
// first run, from qall, handle failure from last exec
streamTaskExecImpl(pTask, data, pRes); pRes = streamExecForQall(pTask, pRes);
if (pRes == NULL) goto FAIL;
/*taosFreeQitem(data);*/
if (taosArrayGetSize(pRes) != 0) { // second run, from inputQ
SStreamDataBlock* resQ = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM); taosReadAllQitems(pTask->inputQ, pTask->inputQAll);
resQ->type = STREAM_INPUT__DATA_BLOCK; pRes = streamExecForQall(pTask, pRes);
resQ->blocks = pRes;
taosWriteQitem(pTask->outputQ, resQ);
pRes = taosArrayInit(0, sizeof(SSDataBlock));
if (pRes == NULL) goto FAIL; if (pRes == NULL) goto FAIL;
}
}
// set status closing // set status closing
atomic_store_8(&pTask->status, TASK_STATUS__CLOSING); atomic_store_8(&pTask->status, TASK_STATUS__CLOSING);
// third run, make sure all inputQ is cleared
taosReadAllQitems(pTask->inputQ, pTask->inputQAll);
while (1) {
void* data = NULL;
taosGetQitem(pTask->inputQAll, &data);
if (data == NULL) break;
streamTaskExecImpl(pTask, data, pRes); // third run, make sure inputQ and qall are cleared
taosReadAllQitems(pTask->inputQ, pTask->inputQAll);
taosFreeQitem(data); pRes = streamExecForQall(pTask, pRes);
if (taosArrayGetSize(pRes) != 0) {
SStreamDataBlock* resQ = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM);
resQ->type = STREAM_INPUT__DATA_BLOCK;
resQ->blocks = pRes;
taosWriteQitem(pTask->outputQ, resQ);
pRes = taosArrayInit(0, sizeof(SSDataBlock));
if (pRes == NULL) goto FAIL; if (pRes == NULL) goto FAIL;
}
}
atomic_store_8(&pTask->status, TASK_STATUS__IDLE); atomic_store_8(&pTask->status, TASK_STATUS__IDLE);
break; break;
......
tests/pytest/cq.py 已删除 100644 → 0
浏览文件 @ 5028c5d7
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import threading
import taos
import sys
import json
import time
import random
# query sql
query_sql = [
# first supertable
"select count(*) from test.meters ;",
"select count(*) from test.meters where t3 > 2;",
"select count(*) from test.meters where ts <> '2020-05-13 10:00:00.002';",
"select count(*) from test.meters where t7 like 'taos_1%';",
"select count(*) from test.meters where t7 like '_____2';",
"select count(*) from test.meters where t8 like '%思%';",
"select count(*) from test.meters interval(1n) order by ts desc;",
#"select max(c0) from test.meters group by tbname",
"select first(ts) from test.meters where t5 >5000 and t5<5100;",
"select last(ts) from test.meters where t5 >5000 and t5<5100;",
"select last_row(*) from test.meters;",
"select twa(c1) from test.t1 where ts > 1500000001000 and ts < 1500000101000" ,
"select avg(c1) from test.meters where t5 >5000 and t5<5100;",
"select bottom(c1, 2) from test.t1;",
"select diff(c1) from test.t1;",
"select leastsquares(c1, 1, 1) from test.t1 ;",
"select max(c1) from test.meters where t5 >5000 and t5<5100;",
"select min(c1) from test.meters where t5 >5000 and t5<5100;",
"select c1 + c2 + c1 / c5 + c4 + c2 from test.t1;",
"select percentile(c1, 50) from test.t1;",
"select spread(c1) from test.t1 ;",
"select stddev(c1) from test.t1;",
"select sum(c1) from test.meters where t5 >5000 and t5<5100;",
"select top(c1, 2) from test.meters where t5 >5000 and t5<5100;"
"select twa(c4) from test.t1 where ts > 1500000001000 and ts < 1500000101000" ,
"select avg(c4) from test.meters where t5 >5000 and t5<5100;",
"select bottom(c4, 2) from test.t1 where t5 >5000 and t5<5100;",
"select diff(c4) from test.t1 where t5 >5000 and t5<5100;",
"select leastsquares(c4, 1, 1) from test.t1 ;",
"select max(c4) from test.meters where t5 >5000 and t5<5100;",
"select min(c4) from test.meters where t5 >5000 and t5<5100;",
"select c5 + c2 + c4 / c5 + c4 + c2 from test.t1 ;",
"select percentile(c5, 50) from test.t1;",
"select spread(c5) from test.t1 ;",
"select stddev(c5) from test.t1 where t5 >5000 and t5<5100;",
"select sum(c5) from test.meters where t5 >5000 and t5<5100;",
"select top(c5, 2) from test.meters where t5 >5000 and t5<5100;",
#all vnode
"select count(*) from test.meters where t5 >5000 and t5<5100",
"select max(c0),avg(c1) from test.meters where t5 >5000 and t5<5100",
"select sum(c5),avg(c1) from test.meters where t5 >5000 and t5<5100",
"select max(c0),min(c5) from test.meters where t5 >5000 and t5<5100",
"select min(c0),avg(c5) from test.meters where t5 >5000 and t5<5100",
# second supertable
"select count(*) from test.meters1 where t3 > 2;",
"select count(*) from test.meters1 where ts <> '2020-05-13 10:00:00.002';",
"select count(*) from test.meters where t7 like 'taos_1%';",
"select count(*) from test.meters where t7 like '_____2';",
"select count(*) from test.meters where t8 like '%思%';",
"select count(*) from test.meters1 interval(1n) order by ts desc;",
#"select max(c0) from test.meters1 group by tbname",
"select first(ts) from test.meters1 where t5 >5000 and t5<5100;",
"select last(ts) from test.meters1 where t5 >5000 and t5<5100;",
"select last_row(*) from test.meters1 ;",
"select twa(c1) from test.m1 where ts > 1500000001000 and ts < 1500000101000" ,
"select avg(c1) from test.meters1 where t5 >5000 and t5<5100;",
"select bottom(c1, 2) from test.m1 where t5 >5000 and t5<5100;",
"select diff(c1) from test.m1 ;",
"select leastsquares(c1, 1, 1) from test.m1 ;",
"select max(c1) from test.meters1 where t5 >5000 and t5<5100;",
"select min(c1) from test.meters1 where t5 >5000 and t5<5100;",
"select c1 + c2 + c1 / c0 + c2 from test.m1 ;",
"select percentile(c1, 50) from test.m1;",
"select spread(c1) from test.m1 ;",
"select stddev(c1) from test.m1;",
"select sum(c1) from test.meters1 where t5 >5000 and t5<5100;",
"select top(c1, 2) from test.meters1 where t5 >5000 and t5<5100;",
"select twa(c5) from test.m1 where ts > 1500000001000 and ts < 1500000101000" ,
"select avg(c5) from test.meters1 where t5 >5000 and t5<5100;",
"select bottom(c5, 2) from test.m1;",
"select diff(c5) from test.m1;",
"select leastsquares(c5, 1, 1) from test.m1 ;",
"select max(c5) from test.meters1 where t5 >5000 and t5<5100;",
"select min(c5) from test.meters1 where t5 >5000 and t5<5100;",
"select c5 + c2 + c4 / c5 + c0 from test.m1;",
"select percentile(c4, 50) from test.m1;",
"select spread(c4) from test.m1 ;",
"select stddev(c4) from test.m1;",
"select sum(c4) from test.meters1 where t5 >5100 and t5<5300;",
"select top(c4, 2) from test.meters1 where t5 >5100 and t5<5300;",
"select count(*) from test.meters1 where t5 >5100 and t5<5300",
#all vnode
"select count(*) from test.meters1 where t5 >5100 and t5<5300",
"select max(c0),avg(c1) from test.meters1 where t5 >5000 and t5<5100",
"select sum(c5),avg(c1) from test.meters1 where t5 >5000 and t5<5100",
"select max(c0),min(c5) from test.meters1 where t5 >5000 and t5<5100",
"select min(c0),avg(c5) from test.meters1 where t5 >5000 and t5<5100",
#join
# "select * from meters,meters1 where meters.ts = meters1.ts and meters.t5 = meters1.t5",
# "select * from meters,meters1 where meters.ts = meters1.ts and meters.t7 = meters1.t7",
# "select * from meters,meters1 where meters.ts = meters1.ts and meters.t8 = meters1.t8",
# "select meters.ts,meters1.c2 from meters,meters1 where meters.ts = meters1.ts and meters.t8 = meters1.t8"
]
class ConcurrentInquiry:
def initConnection(self):
self.numOfTherads = 50
self.ts=1500000001000
def SetThreadsNum(self,num):
self.numOfTherads=num
def query_thread(self,threadID):
host = "10.211.55.14"
user = "root"
password = "taosdata"
conn = taos.connect(
host,
user,
password,
)
cl = conn.cursor()
cl.execute("use test;")
print("Thread %d: starting" % threadID)
while True:
ran_query_sql=query_sql
random.shuffle(ran_query_sql)
for i in ran_query_sql:
print("Thread %d : %s"% (threadID,i))
try:
start = time.time()
cl.execute(i)
cl.fetchall()
end = time.time()
print("time cost :",end-start)
except Exception as e:
print(
"Failure thread%d, sql: %s,exception: %s" %
(threadID, str(i),str(e)))
exit(-1)
print("Thread %d: finishing" % threadID)
def run(self):
threads = []
for i in range(self.numOfTherads):
thread = threading.Thread(target=self.query_thread, args=(i,))
threads.append(thread)
thread.start()
q = ConcurrentInquiry()
q.initConnection()
q.run()
tests/pytest/stream/test2.py 0 → 100644
浏览文件 @ 757e4c2d
# -*- coding: utf-8 -*-
import sys
from util.log import *
from util.cases import *
from util.sql import *
from util.common import tdCom
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
#for i in range(100):
tdSql.prepare()
dbname = tdCom.getLongName(10, "letters")
tdSql.execute('show databases')
tdSql.execute('drop database if exists ttxkbrzmpo')
tdSql.execute('create database if not exists ttxkbrzmpo vgroups 1')
tdSql.execute('use ttxkbrzmpo')
tdSql.execute('create table if not exists downsampling_stb (ts timestamp, c1 int, c2 double, c3 varchar(100), c4 bool) tags (t1 int, t2 double, t3 varchar(100), t4 bool);')
tdSql.execute('create table downsampling_ct1 using downsampling_stb tags(10, 10.1, "Beijing", True);')
tdSql.execute('create table if not exists scalar_stb (ts timestamp, c1 int, c2 double, c3 binary(20), c4 nchar(20), c5 nchar(20)) tags (t1 int);')
tdSql.execute('create table scalar_ct1 using scalar_stb tags(10);')
tdSql.execute('create stream downsampling_stream into output_downsampling_stb as select _wstartts AS start, min(c1), max(c2), sum(c1) from downsampling_stb interval(10m);')
tdSql.execute('insert into downsampling_ct1 values (1653547828591, 100, 100.1, "Beijing", True);')
tdSql.execute('insert into downsampling_ct1 values (1653547828591+1s, -100, -100.1, "Tianjin", False);')
tdSql.execute('insert into downsampling_ct1 values (1653547828591+2s, 50, 50.3, "HeBei", False);')
tdSql.execute('select * from output_downsampling_stb;')
tdSql.execute('select start, `min(c1)`, `max(c2)`, `sum(c1)` from output_downsampling_stb;')
tdSql.execute('select _wstartts AS start, min(c1), max(c2), sum(c1) from downsampling_stb interval(10m);')
tdSql.execute('insert into downsampling_ct1 values (1653547828591+10m, 60, 60.3, "heilongjiang", True);')
tdSql.execute('insert into downsampling_ct1 values (1653547828591+11m, 70, 70.3, "JiLin", True);')
tdSql.execute('select * from output_downsampling_stb;')
tdSql.execute('select start, `min(c1)`, `max(c2)`, `sum(c1)` from output_downsampling_stb;')
tdSql.execute('select _wstartts AS start, min(c1), max(c2), sum(c1) from downsampling_stb interval(10m);')
tdSql.execute('insert into downsampling_ct1 values (1653547828591+21m, 70, 70.3, "JiLin", True);')
tdSql.execute('select * from output_downsampling_stb;')
tdSql.execute('select * from output_downsampling_stb;')
tdSql.execute('select start, `min(c1)`, `max(c2)`, `sum(c1)` from output_downsampling_stb;')
tdSql.execute('select _wstartts AS start, min(c1), max(c2), sum(c1) from downsampling_stb interval(10m);')
tdSql.execute('create stream abs_stream into output_abs_stb as select ts, abs(c1), abs(c2), c3 from scalar_stb;')
tdSql.query('describe output_abs_stb')
tdSql.execute('create stream acos_stream into output_acos_stb as select ts, acos(c1), acos(c2), c3 from scalar_stb;')
tdSql.query('describe output_acos_stb')
tdSql.execute('create stream asin_stream into output_asin_stb as select ts, asin(c1), asin(c2), c3 from scalar_stb;')
tdSql.query('describe output_asin_stb')
tdSql.execute('create stream atan_stream into output_atan_stb as select ts, atan(c1), atan(c2), c3 from scalar_stb;')
tdSql.query('describe output_atan_stb')
tdSql.execute('create stream ceil_stream into output_ceil_stb as select ts, ceil(c1), ceil(c2), c3 from scalar_stb;')
tdSql.query('describe output_ceil_stb')
tdSql.execute('create stream cos_stream into output_cos_stb as select ts, cos(c1), cos(c2), c3 from scalar_stb;')
tdSql.query('describe output_cos_stb')
tdSql.execute('create stream floor_stream into output_floor_stb as select ts, floor(c1), floor(c2), c3 from scalar_stb;')
tdSql.query('describe output_floor_stb')
tdSql.execute('create stream log_stream into output_log_stb as select ts, log(c1, 2), log(c2, 2), c3 from scalar_stb;')
tdSql.query('describe output_log_stb')
tdSql.execute('create stream pow_stream into output_pow_stb as select ts, pow(c1, 2), pow(c2, 2), c3 from scalar_stb;')
tdSql.query('describe output_pow_stb')
tdSql.execute('create stream round_stream into output_round_stb as select ts, round(c1), round(c2), c3 from scalar_stb;')
tdSql.query('describe output_round_stb')
tdSql.execute('create stream sin_stream into output_sin_stb as select ts, sin(c1), sin(c2), c3 from scalar_stb;')
tdSql.query('describe output_sin_stb')
tdSql.execute('create stream sqrt_stream into output_sqrt_stb as select ts, sqrt(c1), sqrt(c2), c3 from scalar_stb;')
tdSql.query('describe output_sqrt_stb')
tdSql.execute('create stream tan_stream into output_tan_stb as select ts, tan(c1), tan(c2), c3 from scalar_stb;')
tdSql.query('describe output_tan_stb')
tdSql.execute('create stream char_length_stream into output_char_length_stb as select ts, char_length(c3), char_length(c4), char_length(c5) from scalar_stb;')
tdSql.query('describe output_char_length_stb')
tdSql.execute('create stream concat_stream into output_concat_stb as select ts, concat(c3, c4), concat(c3, c5), concat(c4, c5), concat(c3, c4, c5) from scalar_stb;')
tdSql.execute('create stream concat_ws_stream into output_concat_ws_stb as select ts, concat_ws("aND", c3, c4), concat_ws("and", c3, c5), concat_ws("And", c4, c5), concat_ws("AND", c3, c4, c5) from scalar_stb;')
tdSql.execute('create stream length_stream into output_length_stb as select ts, length(c3), length(c4), length(c5) from scalar_stb;')
tdSql.query('describe output_length_stb')
tdSql.execute('create stream lower_stream into output_lower_stb as select ts, lower(c3), lower(c4), lower(c5) from scalar_stb;')
tdSql.query('describe output_lower_stb')
tdSql.execute('create stream ltrim_stream into output_ltrim_stb as select ts, ltrim(c3), ltrim(c4), ltrim(c5) from scalar_stb;')
tdSql.query('describe output_ltrim_stb')
tdSql.execute('create stream rtrim_stream into output_rtrim_stb as select ts, rtrim(c3), rtrim(c4), rtrim(c5) from scalar_stb;')
tdSql.query('describe output_rtrim_stb')
tdSql.execute('create stream substr_stream into output_substr_stb as select ts, substr(c3, 2), substr(c3, 2, 2), substr(c4, 5, 1), substr(c5, 3, 4) from scalar_stb;')
tdSql.query('describe output_substr_stb')
tdSql.execute('create stream upper_stream into output_upper_stb as select ts, upper(c3), upper(c4), upper(c5) from scalar_stb;')
tdSql.query('describe output_upper_stb')
tdSql.execute('insert into scalar_ct1 values (1653560440733, 100, 100.1, "beijing", "taos", "Taos");')
tdSql.execute('insert into scalar_ct1 values (1653560440733+1s, -50, -50.1, "tianjin", "taosdata", "Taosdata");')
tdSql.execute('insert into scalar_ct1 values (1653560440733+2s, 0, Null, "hebei", "TDengine", Null);')
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或