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

Merge pull request #2605 from taosdata/bugfix/stream 

Bugfix/stream
This commit is contained in:
Shengliang Guan 2020-07-09 11:18:57 +08:00 committed by GitHub
parent c98767fdf0 27925040f3
commit 1afcd83065
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 519 additions and 1211 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/client/inc/tsclient.h
View File

@ -320,6 +320,8 @@ typedef struct SSqlStream {
SSqlObj *pSql;
uint32_t streamId;
char listed;
bool isProject;
int16_t precision;
int64_t num; // number of computing count
/*
@ -334,7 +336,6 @@ typedef struct SSqlStream {
int64_t etime; // stream end query time, when time is larger then etime, the stream will be closed
int64_t interval;
int64_t slidingTime;
int16_t precision;
void * pTimer;
void (*fp)();

147
src/client/src/tscStream.c
View File

@ -71,6 +71,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
pSql->fp = tscProcessStreamQueryCallback;
pSql->param = pStream;
pSql->res.completed = false;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
@ -86,7 +87,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
// failed to get meter/metric meta, retry in 10sec.
if (code != TSDB_CODE_SUCCESS) {
int64_t retryDelayTime = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
tscError("%p stream:%p,get metermeta failed, retry in %" PRId64 "ms", pStream->pSql, pStream, retryDelayTime);
tscDebug("%p stream:%p,get metermeta failed, retry in %" PRId64 "ms", pStream->pSql, pStream, retryDelayTime);
tscSetRetryTimer(pStream, pSql, retryDelayTime);
} else {
@ -108,7 +109,7 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
tscDebug("%p add into timer", pSql);
if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
/*
* pQueryInfo->window.ekey, which is the start time, does not change in case of
* repeat first execution, once the first execution failed.
@ -121,7 +122,19 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
}
} else {
pQueryInfo->window.skey = pStream->stime - pStream->interval;
pQueryInfo->window.ekey = pStream->stime - 1;
int64_t etime = taosGetTimestamp(pStream->precision);
// delay to wait all data in last time window
if (pStream->precision == TSDB_TIME_PRECISION_MICRO) {
etime -= tsMaxStreamComputDelay * 1000l;
} else {
etime -= tsMaxStreamComputDelay;
}
if (etime > pStream->etime) {
etime = pStream->etime;
} else {
etime = pStream->stime + (etime - pStream->stime) / pStream->interval * pStream->interval;
}
pQueryInfo->window.ekey = etime;
}
// launch stream computing in a new thread
@ -137,7 +150,7 @@ static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOf
SSqlStream *pStream = (SSqlStream *)param;
if (tres == NULL || numOfRows < 0) {
int64_t retryDelay = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
tscError("%p stream:%p, query data failed, code:%d, retry in %" PRId64 "ms", pStream->pSql, pStream, numOfRows,
tscError("%p stream:%p, query data failed, code:0x%08x, retry in %" PRId64 "ms", pStream->pSql, pStream, numOfRows,
retryDelay);
STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pStream->pSql->cmd, 0, 0);
@ -151,17 +164,45 @@ static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOf
taos_fetch_rows_a(tres, tscProcessStreamRetrieveResult, param);
}
static void tscSetTimestampForRes(SSqlStream *pStream, SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res;
int64_t timestamp = *(int64_t *)pRes->data;
int64_t actualTimestamp = pStream->stime - pStream->interval;
if (timestamp != actualTimestamp) {
// reset the timestamp of each agg point by using start time of each interval
*((int64_t *)pRes->data) = actualTimestamp;
tscWarn("%p stream:%p, timestamp of points is:%" PRId64 ", reset to %" PRId64, pSql, pStream, timestamp, actualTimestamp);
// no need to be called as this is alreay done in the query
static void tscStreamFillTimeGap(SSqlStream* pStream, TSKEY ts) {
#if 0
SSqlObj * pSql = pStream->pSql;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (pQueryInfo->fillType != TSDB_FILL_SET_VALUE && pQueryInfo->fillType != TSDB_FILL_NULL) {
return;
}
SSqlRes *pRes = &pSql->res;
/* failed to retrieve any result in this retrieve */
pSql->res.numOfRows = 1;
void *row[TSDB_MAX_COLUMNS] = {0};
char tmpRes[TSDB_MAX_BYTES_PER_ROW] = {0};
void *oldPtr = pSql->res.data;
pSql->res.data = tmpRes;
int32_t rowNum = 0;
while (pStream->stime + pStream->slidingTime < ts) {
pStream->stime += pStream->slidingTime;
*(TSKEY*)row[0] = pStream->stime;
for (int32_t i = 1; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
int16_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->fillVal[i]), pField->bytes, pField->type);
row[i] = pSql->res.data + offset;
}
(*pStream->fp)(pStream->param, pSql, row);
++rowNum;
}
if (rowNum > 0) {
tscDebug("%p stream:%p %d rows padded", pSql, pStream, rowNum);
}
pRes->numOfRows = 0;
pRes->data = oldPtr;
#endif
}
static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOfRows) {
@ -170,7 +211,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
if (pSql == NULL || numOfRows < 0) {
int64_t retryDelayTime = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision);
tscError("%p stream:%p, retrieve data failed, code:%d, retry in %" PRId64 "ms", pSql, pStream, numOfRows, retryDelayTime);
tscError("%p stream:%p, retrieve data failed, code:0x%08x, retry in %" PRId64 "ms", pSql, pStream, numOfRows, retryDelayTime);
tscSetRetryTimer(pStream, pStream->pSql, retryDelayTime);
return;
@ -180,16 +221,11 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
if (numOfRows > 0) { // when reaching here the first execution of stream computing is successful.
pStream->numOfRes += numOfRows;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
for(int32_t i = 0; i < numOfRows; ++i) {
TAOS_ROW row = taos_fetch_row(res);
tscDebug("%p stream:%p fetch result", pSql, pStream);
if (isProjectStream(pQueryInfo)) {
pStream->stime = *(TSKEY *)row[0];
} else {
tscSetTimestampForRes(pStream, pSql);
}
tscStreamFillTimeGap(pStream, *(TSKEY*)row[0]);
pStream->stime = *(TSKEY *)row[0];
// user callback function
(*pStream->fp)(pStream->param, res, row);
@ -199,55 +235,18 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
taos_fetch_rows_a(res, tscProcessStreamRetrieveResult, pStream);
} else { // numOfRows == 0, all data has been retrieved
pStream->useconds += pSql->res.useconds;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (pStream->numOfRes == 0) {
if (pQueryInfo->fillType == TSDB_FILL_SET_VALUE || pQueryInfo->fillType == TSDB_FILL_NULL) {
SSqlRes *pRes = &pSql->res;
/* failed to retrieve any result in this retrieve */
pSql->res.numOfRows = 1;
void *row[TSDB_MAX_COLUMNS] = {0};
char tmpRes[TSDB_MAX_BYTES_PER_ROW] = {0};
void *oldPtr = pSql->res.data;
pSql->res.data = tmpRes;
for (int32_t i = 1; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
int16_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->fillVal[i]), pField->bytes, pField->type);
row[i] = pSql->res.data + offset;
}
tscSetTimestampForRes(pStream, pSql);
row[0] = pRes->data;
// char result[512] = {0};
// taos_print_row(result, row, pQueryInfo->fieldsInfo.pFields, pQueryInfo->fieldsInfo.numOfOutput);
// tscInfo("%p stream:%p query result: %s", pSql, pStream, result);
tscDebug("%p stream:%p fetch result", pSql, pStream);
// user callback function
(*pStream->fp)(pStream->param, res, row);
pRes->numOfRows = 0;
pRes->data = oldPtr;
} else if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
/* no resuls in the query range, retry */
// todo set retry dynamic time
int32_t retry = tsProjectExecInterval;
tscError("%p stream:%p, retrieve no data, code:%d, retry in %" PRId32 "ms", pSql, pStream, numOfRows, retry);
tscError("%p stream:%p, retrieve no data, code:0x%08x, retry in %" PRId32 "ms", pSql, pStream, numOfRows, retry);
tscSetRetryTimer(pStream, pStream->pSql, retry);
return;
}
} else {
if (isProjectStream(pQueryInfo)) {
pStream->stime += 1;
}
} else if (pStream->isProject) {
pStream->stime += 1;
}
tscDebug("%p stream:%p, query on:%s, fetch result completed, fetched rows:%" PRId64, pSql, pStream, pTableMetaInfo->name,
@ -262,10 +261,9 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
}
static void tscSetRetryTimer(SSqlStream *pStream, SSqlObj *pSql, int64_t timer) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
int64_t delay = getDelayValueAfterTimewindowClosed(pStream, timer);
if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
int64_t now = taosGetTimestamp(pStream->precision);
int64_t etime = now > pStream->etime ? pStream->etime : now;
@ -323,8 +321,7 @@ static int64_t getLaunchTimeDelay(const SSqlStream* pStream) {
static void tscSetNextLaunchTimer(SSqlStream *pStream, SSqlObj *pSql) {
int64_t timer = 0;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
/*
* for project query, no mater fetch data successfully or not, next launch will issue
* more than the sliding time window
@ -342,7 +339,6 @@ static void tscSetNextLaunchTimer(SSqlStream *pStream, SSqlObj *pSql) {
return;
}
} else {
pStream->stime += pStream->slidingTime;
if ((pStream->stime - pStream->interval) >= pStream->etime) {
tscDebug("%p stream:%p, stime:%" PRId64 " is larger than end time: %" PRId64 ", stop the stream", pStream->pSql, pStream,
pStream->stime, pStream->etime);
@ -409,14 +405,16 @@ static void tscSetSlidingWindowInfo(SSqlObj *pSql, SSqlStream *pStream) {
pStream->slidingTime = pQueryInfo->slidingTime;
pQueryInfo->intervalTime = 0; // clear the interval value to avoid the force time window split by query processor
pQueryInfo->slidingTime = 0;
if (pStream->isProject) {
pQueryInfo->intervalTime = 0; // clear the interval value to avoid the force time window split by query processor
pQueryInfo->slidingTime = 0;
}
}
static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, int64_t stime) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (isProjectStream(pQueryInfo)) {
if (pStream->isProject) {
// no data in table, flush all data till now to destination meter, 10sec delay
pStream->interval = tsProjectExecInterval;
pStream->slidingTime = tsProjectExecInterval;
@ -489,7 +487,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
SSqlStream *pStream = (SSqlStream *)calloc(1, sizeof(SSqlStream));
if (pStream == NULL) {
tscError("%p open stream failed, sql:%s, reason:%s, code:%d", pSql, sqlstr, pCmd->payload, pRes->code);
tscError("%p open stream failed, sql:%s, reason:%s, code:0x%08x", pSql, sqlstr, pCmd->payload, pRes->code);
tscFreeSqlObj(pSql);
return NULL;
}
@ -514,7 +512,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
if (pRes->code != TSDB_CODE_SUCCESS) {
setErrorInfo(pSql, pRes->code, pCmd->payload);
tscError("%p open stream failed, sql:%s, reason:%s, code:%d", pSql, sqlstr, pCmd->payload, pRes->code);
tscError("%p open stream failed, sql:%s, reason:%s, code:0x%08x", pSql, sqlstr, pCmd->payload, pRes->code);
tscFreeSqlObj(pSql);
return NULL;
}
@ -523,6 +521,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
pStream->isProject = isProjectStream(pQueryInfo);
pStream->fp = fp;
pStream->callback = callback;
pStream->param = param;
@ -565,6 +564,8 @@ void taos_close_stream(TAOS_STREAM *handle) {
taosTmrStopA(&(pStream->pTimer));
tscDebug("%p stream:%p is closed", pSql, pStream);
// notify CQ to release the pStream object
pStream->fp(pStream->param, NULL, NULL);
tscFreeSqlObj(pSql);
pStream->pSql = NULL;

6
src/cq/src/cqMain.c
View File

@ -109,6 +109,8 @@ void cqClose(void *handle) {
while (pObj) {
SCqObj *pTemp = pObj;
pObj = pObj->next;
tdFreeSchema(pTemp->pSchema);
tfree(pTemp->sqlStr);
free(pTemp);
}
@ -242,6 +244,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
SCqObj *pObj = (SCqObj *)param;
if (tres == NULL && row == NULL) {
pObj->pStream = NULL;
return;
}
SCqContext *pContext = pObj->pContext;
STSchema *pSchema = pObj->pSchema;
if (pObj->pStream == NULL) return;

2
src/query/src/qExecutor.c
View File

@ -5822,7 +5822,7 @@ static int32_t initQInfo(SQueryTableMsg *pQueryMsg, void *tsdb, int32_t vgId, SQ
qDebug("QInfo:%p no result in time range %" PRId64 "-%" PRId64 ", order %d", pQInfo, pQuery->window.skey,
pQuery->window.ekey, pQuery->order.order);
setQueryStatus(pQuery, QUERY_COMPLETED);
pQInfo->tableqinfoGroupInfo.numOfTables = 0;
sem_post(&pQInfo->dataReady);
return TSDB_CODE_SUCCESS;
}

3
tests/pytest/fulltest.sh
View File

@ -145,8 +145,11 @@ python3 ./test.py -f query/queryJoin.py
python3 ./test.py -f query/select_last_crash.py
#stream
python3 ./test.py -f stream/metric_1.py
python3 ./test.py -f stream/new.py
python3 ./test.py -f stream/stream1.py
python3 ./test.py -f stream/stream2.py
python3 ./test.py -f stream/parser.py
#alter table
python3 ./test.py -f alter/alter_table_crash.py

104
tests/pytest/stream/metric_1.py Normal file
View File

@ -0,0 +1,104 @@
###################################################################
# 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 sys
import time
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def createFuncStream(self, expr, suffix, value):
tbname = "strm_" + suffix
tdLog.info("create stream table %s" % tbname)
tdSql.query("select %s from stb interval(1d)" % expr)
tdSql.checkData(0, 1, value)
tdSql.execute("create table %s as select %s from stb interval(1d)" % (tbname, expr))
def checkStreamData(self, suffix, value):
sql = "select * from strm_" + suffix
tdSql.waitedQuery(sql, 1, 120)
tdSql.checkData(0, 1, value)
def run(self):
tbNum = 10
rowNum = 20
tdSql.prepare()
tdLog.info("===== preparing data =====")
tdSql.execute(
"create table stb(ts timestamp, tbcol int, tbcol2 float) tags(tgcol int)")
for i in range(tbNum):
tdSql.execute("create table tb%d using stb tags(%d)" % (i, i))
for j in range(rowNum):
tdSql.execute(
"insert into tb%d values (now - %dm, %d, %d)" %
(i, 1440 - j, j, j))
time.sleep(0.1)
self.createFuncStream("count(*)", "c1", 200)
self.createFuncStream("count(tbcol)", "c2", 200)
self.createFuncStream("count(tbcol2)", "c3", 200)
self.createFuncStream("avg(tbcol)", "av", 9.5)
self.createFuncStream("sum(tbcol)", "su", 1900)
self.createFuncStream("min(tbcol)", "mi", 0)
self.createFuncStream("max(tbcol)", "ma", 19)
self.createFuncStream("first(tbcol)", "fi", 0)
self.createFuncStream("last(tbcol)", "la", 19)
#tdSql.query("select stddev(tbcol) from stb interval(1d)")
#tdSql.query("select leastsquares(tbcol, 1, 1) from stb interval(1d)")
tdSql.query("select top(tbcol, 1) from stb interval(1d)")
tdSql.query("select bottom(tbcol, 1) from stb interval(1d)")
#tdSql.query("select percentile(tbcol, 1) from stb interval(1d)")
#tdSql.query("select diff(tbcol) from stb interval(1d)")
tdSql.query("select count(tbcol) from stb where ts < now + 4m interval(1d)")
tdSql.checkData(0, 1, 200)
#tdSql.execute("create table strm_wh as select count(tbcol) from stb where ts < now + 4m interval(1d)")
self.createFuncStream("count(tbcol)", "as", 200)
tdSql.query("select count(tbcol) from stb interval(1d) group by tgcol")
tdSql.checkData(0, 1, 20)
tdSql.query("select count(tbcol) from stb where ts < now + 4m interval(1d) group by tgcol")
tdSql.checkData(0, 1, 20)
self.checkStreamData("c1", 200)
self.checkStreamData("c2", 200)
self.checkStreamData("c3", 200)
self.checkStreamData("av", 9.5)
self.checkStreamData("su", 1900)
self.checkStreamData("mi", 0)
self.checkStreamData("ma", 19)
self.checkStreamData("fi", 0)
self.checkStreamData("la", 19)
#self.checkStreamData("wh", 200)
self.checkStreamData("as", 200)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

71
tests/pytest/stream/new.py Normal file
View File

@ -0,0 +1,71 @@
###################################################################
# 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 sys
import time
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
rowNum = 200
totalNum = 200
tdSql.prepare()
tdLog.info("=============== step1")
tdSql.execute("create table mt(ts timestamp, tbcol int, tbcol2 float) TAGS(tgcol int)")
for i in range(5):
tdSql.execute("create table tb%d using mt tags(%d)" % (i, i))
for j in range(rowNum):
tdSql.execute("insert into tb%d values(now + %ds, %d, %d)" % (i, j, j, j))
time.sleep(0.1)
tdLog.info("=============== step2")
tdSql.query("select count(*), count(tbcol), count(tbcol2) from mt interval(10s)")
tdSql.execute("create table st as select count(*), count(tbcol), count(tbcol2) from mt interval(10s)")
tdLog.info("=============== step3")
tdSql.waitedQuery("select * from st", 1, 120)
v = tdSql.getData(0, 3)
if v >= 51:
tdLog.exit("value is %d, which is larger than 51" % v)
tdLog.info("=============== step4")
for i in range(5, 10):
tdSql.execute("create table tb%d using mt tags(%d)" % (i, i))
for j in range(rowNum):
tdSql.execute("insert into tb%d values(now + %ds, %d, %d)" % (i, j, j, j))
tdLog.info("=============== step5")
tdLog.sleep(30)
tdSql.waitedQuery("select * from st order by ts desc", 1, 120)
v = tdSql.getData(0, 3)
if v <= 51:
tdLog.exit("value is %d, which is smaller than 51" % v)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

182
tests/pytest/stream/parser.py Normal file
View File

@ -0,0 +1,182 @@
###################################################################
# 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 sys
import time
import taos
from util.log import tdLog
from util.cases import tdCases
from util.sql import tdSql
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
'''
def bug2222(self):
tdSql.prepare()
tdSql.execute("create table superreal(ts timestamp, addr binary(5), val float) tags (deviceNo binary(20))")
tdSql.execute("create table real_001 using superreal tags('001')")
tdSql.execute("create table tj_001 as select sum(val) from real_001 interval(1m)")
t = datetime.datetime.now()
for i in range(60):
ts = t.strftime("%Y-%m-%d %H:%M")
t += datetime.timedelta(minutes=1)
sql = "insert into real_001 values('%s:0%d', '1', %d)" % (ts, 0, i)
for j in range(4):
sql += ",('%s:0%d', '%d', %d)" % (ts, j + 1, j + 1, i)
tdSql.execute(sql)
time.sleep(60 + random.random() * 60 - 30)
'''
def tbase300(self):
tdLog.debug("begin tbase300")
tdSql.prepare()
tdSql.execute("create table mt(ts timestamp, c1 int, c2 int) tags(t1 int)")
tdSql.execute("create table tb1 using mt tags(1)");
tdSql.execute("create table tb2 using mt tags(2)");
tdSql.execute("create table strm as select count(*), avg(c1), sum(c2), max(c1), min(c2),first(c1), last(c2) from mt interval(4s) sliding(2s)")
#tdSql.execute("create table strm as select count(*), avg(c1), sum(c2), max(c1), min(c2), first(c1) from mt interval(4s) sliding(2s)")
tdLog.sleep(10)
tdSql.execute("insert into tb2 values(now, 1, 1)");
tdSql.execute("insert into tb1 values(now, 1, 1)");
tdLog.sleep(4)
tdSql.query("select * from mt")
tdSql.query("select * from strm")
tdSql.execute("drop table tb1")
tdSql.waitedQuery("select * from strm", 1, 100)
if tdSql.queryRows < 1 or tdSql.queryRows > 2:
tdLog.exit("rows should be 1 or 2")
tdSql.execute("drop table tb2")
tdSql.execute("drop table mt")
tdSql.execute("drop table strm")
def tbase304(self):
tdLog.debug("begin tbase304")
# we cannot reset query cache in server side, as a workaround,
# set super table name to mt304, need to change back to mt later
tdSql.execute("create table mt304 (ts timestamp, c1 int) tags(t1 int, t2 int)")
tdSql.execute("create table tb1 using mt304 tags(1, 1)")
tdSql.execute("create table tb2 using mt304 tags(1, -1)")
time.sleep(0.1)
tdSql.execute("create table strm as select count(*), avg(c1) from mt304 where t2 >= 0 interval(4s) sliding(2s)")
tdSql.execute("insert into tb1 values (now,1)")
tdSql.execute("insert into tb2 values (now,2)")
tdSql.waitedQuery("select * from strm", 1, 100)
if tdSql.queryRows < 1 or tdSql.queryRows > 2:
tdLog.exit("rows should be 1 or 2")
tdSql.checkData(0, 1, 1)
tdSql.checkData(0, 2, 1.000000000)
tdSql.execute("alter table mt304 drop tag t2")
tdSql.execute("insert into tb2 values (now,2)")
tdSql.execute("insert into tb1 values (now,1)")
tdSql.query("select * from strm")
tdSql.execute("alter table mt304 add tag t2 int")
tdLog.sleep(1)
tdSql.query("select * from strm")
def wildcardFilterOnTags(self):
tdLog.debug("begin wildcardFilterOnTag")
tdSql.prepare()
tdSql.execute("create table stb (ts timestamp, c1 int, c2 binary(10)) tags(t1 binary(10))")
tdSql.execute("create table tb1 using stb tags('a1')")
tdSql.execute("create table tb2 using stb tags('b2')")
tdSql.execute("create table tb3 using stb tags('a3')")
tdSql.execute("create table strm as select count(*), avg(c1), first(c2) from stb where t1 like 'a%' interval(4s) sliding(2s)")
tdSql.query("describe strm")
tdSql.checkRows(4)
tdLog.sleep(1)
tdSql.execute("insert into tb1 values (now, 0, 'tb1')")
tdLog.sleep(4)
tdSql.execute("insert into tb2 values (now, 2, 'tb2')")
tdLog.sleep(4)
tdSql.execute("insert into tb3 values (now, 0, 'tb3')")
tdSql.waitedQuery("select * from strm", 4, 60)
tdSql.checkRows(4)
tdSql.checkData(0, 2, 0.000000000)
if tdSql.getData(0, 3) == 'tb2':
tdLog.exit("unexpected value of data03")
if tdSql.getData(1, 3) == 'tb2':
tdLog.exit("unexpected value of data13")
if tdSql.getData(2, 3) == 'tb2':
tdLog.exit("unexpected value of data23")
if tdSql.getData(3, 3) == 'tb2':
tdLog.exit("unexpected value of data33")
tdLog.info("add table tb4 to see if stream still works correctly")
# The vnode client needs to refresh metadata cache to allow strm calculate tb4's data.
# But the current refreshing frequency is every 10 min
# commented out the case below to save running time
tdSql.execute("create table tb4 using stb tags('a4')")
tdSql.execute("insert into tb4 values(now, 4, 'tb4')")
tdSql.waitedQuery("select * from strm order by ts desc", 6, 60)
tdSql.checkRows(6)
tdSql.checkData(0, 2, 4)
tdSql.checkData(0, 3, "tb4")
tdLog.info("change tag values to see if stream still works correctly")
tdSql.execute("alter table tb4 set tag t1='b4'")
tdLog.sleep(3)
tdSql.execute("insert into tb1 values (now, 1, 'tb1_a1')")
tdLog.sleep(4)
tdSql.execute("insert into tb4 values (now, -4, 'tb4_b4')")
tdSql.waitedQuery("select * from strm order by ts desc", 8, 100)
tdSql.checkRows(8)
tdSql.checkData(0, 2, 1)
tdSql.checkData(0, 3, "tb1_a1")
def datatypes(self):
tdLog.debug("begin data types")
tdSql.prepare()
tdSql.execute("create table stb3 (ts timestamp, c1 int, c2 bigint, c3 float, c4 double, c5 binary(15), c6 nchar(15), c7 bool) tags(t1 int, t2 binary(15))")
tdSql.execute("create table tb0 using stb3 tags(0, 'tb0')")
tdSql.execute("create table tb1 using stb3 tags(1, 'tb1')")
tdSql.execute("create table tb2 using stb3 tags(2, 'tb2')")
tdSql.execute("create table tb3 using stb3 tags(3, 'tb3')")
tdSql.execute("create table tb4 using stb3 tags(4, 'tb4')")
tdSql.execute("create table strm0 as select count(ts), count(c1), max(c2), min(c4), first(c5), last(c6) from stb3 where ts < now + 30s interval(4s) sliding(2s)")
#tdSql.execute("create table strm0 as select count(ts), count(c1), max(c2), min(c4), first(c5) from stb where ts < now + 30s interval(4s) sliding(2s)")
tdLog.sleep(1)
tdSql.execute("insert into tb0 values (now, 0, 0, 0, 0, 'binary0', '涛思0', true) tb1 values (now, 1, 1, 1, 1, 'binary1', '涛思1', false) tb2 values (now, 2, 2, 2, 2, 'binary2', '涛思2', true) tb3 values (now, 3, 3, 3, 3, 'binary3', '涛思3', false) tb4 values (now, 4, 4, 4, 4, 'binary4', '涛思4', true) ")
tdSql.waitedQuery("select * from strm0 order by ts desc", 2, 120)
tdSql.checkRows(2)
tdSql.execute("insert into tb0 values (now, 10, 10, 10, 10, 'binary0', '涛思0', true) tb1 values (now, 11, 11, 11, 11, 'binary1', '涛思1', false) tb2 values (now, 12, 12, 12, 12, 'binary2', '涛思2', true) tb3 values (now, 13, 13, 13, 13, 'binary3', '涛思3', false) tb4 values (now, 14, 14, 14, 14, 'binary4', '涛思4', true) ")
tdSql.waitedQuery("select * from strm0 order by ts desc", 4, 120)
tdSql.checkRows(4)
def run(self):
self.tbase300()
self.tbase304()
self.wildcardFilterOnTags()
self.datatypes()
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

19
tests/pytest/stream/stream1.py
View File

@ -55,10 +55,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step3 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.waitedQuery("select * from s0", 1, 120)
try:
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
@ -82,10 +79,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step7 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.waitedQuery("select * from s0", 1, 120)
try:
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
@ -108,10 +102,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 2)
tdLog.info("===== step9 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.waitedQuery("select * from s1", 1, 120)
try:
tdSql.checkData(0, 1, rowNum * tbNum)
tdSql.checkData(0, 2, rowNum * tbNum)
@ -134,9 +125,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 2)
tdLog.info("===== step13 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.waitedQuery("select * from s1", 1, 120)
try:
tdSql.checkData(0, 1, rowNum * tbNum)
tdSql.checkData(0, 2, rowNum * tbNum)

12
tests/pytest/stream/stream2.py
View File

@ -53,8 +53,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step3 =====")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.waitedQuery("select * from s0", 1, 120)
try:
tdSql.checkData(0, 1, rowNum)
except Exception as e:
@ -81,8 +80,7 @@ class TDTestCase:
tdLog.info(repr(e))
tdLog.info("===== step7 =====")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.waitedQuery("select * from s0", 1, 120)
try:
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
@ -107,8 +105,7 @@ class TDTestCase:
tdSql.checkRows(tbNum + 2)
tdLog.info("===== step9 =====")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.waitedQuery("select * from s1", 1, 120)
try:
tdSql.checkData(0, 1, totalNum)
tdSql.checkData(0, 2, totalNum)
@ -137,8 +134,7 @@ class TDTestCase:
tdLog.info(repr(e))
tdLog.info("===== step13 =====")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.waitedQuery("select * from s1", 1, 120)
try:
tdSql.checkData(0, 1, totalNum)
#tdSql.checkData(0, 2, None)

167
tests/pytest/util/sql.py
View File

@ -29,10 +29,8 @@ class TDSql:
self.cursor = cursor
if (log):
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
self.cursor.log(callerFilename + ".sql")
caller = inspect.getframeinfo(inspect.stack()[1][0])
self.cursor.log(caller.filename + ".sql")
def close(self):
self.cursor.close()
@ -55,12 +53,8 @@ class TDSql:
except BaseException:
expectErrNotOccured = False
if expectErrNotOccured:
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
tdLog.exit(
"%s failed: sql:%s, expect error not occured" %
(callerFilename, sql))
caller = inspect.getframeinfo(inspect.stack()[1][0])
tdLog.exit("%s(%d) failed: sql:%s, expect error not occured" % (caller.filename, caller.lineno, sql))
else:
self.queryRows = 0
self.queryCols = 0
@ -69,75 +63,68 @@ class TDSql:
def query(self, sql):
self.sql = sql
self.cursor.execute(sql)
self.queryResult = self.cursor.fetchall()
self.queryRows = len(self.queryResult)
self.queryCols = len(self.cursor.description)
# if self.queryRows == 1 and self.queryCols == 1:
# tdLog.info("sql:%s, rows:%d cols:%d data:%s" % (self.sql, self.queryRows, self.queryCols, self.queryResult[0][0]))
# else:
# tdLog.info("sql:%s, rows:%d cols:%d" % (self.sql, self.queryRows, self.queryCols))
try:
self.cursor.execute(sql)
self.queryResult = self.cursor.fetchall()
self.queryRows = len(self.queryResult)
self.queryCols = len(self.cursor.description)
except Exception as e:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, sql, repr(e))
tdLog.exit("%s(%d) failed: sql:%s, %s" % args)
return self.queryRows
def waitedQuery(self, sql, expectRows, timeout):
tdLog.info("sql: %s, try to retrieve %d rows in %d seconds" % (sql, expectRows, timeout))
self.sql = sql
try:
for i in range(timeout):
self.cursor.execute(sql)
self.queryResult = self.cursor.fetchall()
self.queryRows = len(self.queryResult)
self.queryCols = len(self.cursor.description)
if self.queryRows >= expectRows:
return (self.queryRows, i)
time.sleep(1)
except Exception as e:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, sql, repr(e))
tdLog.exit("%s(%d) failed: sql:%s, %s" % args)
return (self.queryRows, timeout)
def checkRows(self, expectRows):
if self.queryRows != expectRows:
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
tdLog.exit(
"%s failed: sql:%s, queryRows:%d != expect:%d" %
(callerFilename, self.sql, self.queryRows, expectRows))
tdLog.info("sql:%s, queryRows:%d == expect:%d" %
(self.sql, self.queryRows, expectRows))
if self.queryRows == expectRows:
tdLog.info("sql:%s, queryRows:%d == expect:%d" % (self.sql, self.queryRows, expectRows))
else:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, self.sql, self.queryRows, expectRows)
tdLog.exit("%s(%d) failed: sql:%s, queryRows:%d != expect:%d" % args)
def checkRowCol(self, row, col):
caller = inspect.getframeinfo(inspect.stack()[2][0])
if row < 0:
args = (caller.filename, caller.lineno, self.sql, row)
tdLog.exit("%s(%d) failed: sql:%s, row:%d is smaller than zero" % args)
if col < 0:
args = (caller.filename, caller.lineno, self.sql, row)
tdLog.exit("%s(%d) failed: sql:%s, col:%d is smaller than zero" % args)
if row > self.queryRows:
args = (caller.filename, caller.lineno, self.sql, row, self.queryRows)
tdLog.exit("%s(%d) failed: sql:%s, row:%d is larger than queryRows:%d" % args)
if col > self.queryCols:
args = (caller.filename, caller.lineno, self.sql, col, self.queryCols)
tdLog.exit("%s(%d) failed: sql:%s, col:%d is larger than queryCols:%d" % args)
def checkDataType(self, row, col, dataType):
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
if row < 0:
tdLog.exit(
"%s failed: sql:%s, row:%d is smaller than zero" %
(callerFilename, self.sql, row))
if col < 0:
tdLog.exit(
"%s failed: sql:%s, col:%d is smaller than zero" %
(callerFilename, self.sql, col))
if row > self.queryRows:
tdLog.exit(
"%s failed: sql:%s, row:%d is larger than queryRows:%d" %
(callerFilename, self.sql, row, self.queryRows))
if col > self.queryCols:
tdLog.exit(
"%s failed: sql:%s, col:%d is larger than queryCols:%d" %
(callerFilename, self.sql, col, self.queryCols))
self.checkRowCol(row, col)
return self.cursor.istype(col, dataType)
def checkData(self, row, col, data):
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
if row < 0:
tdLog.exit(
"%s failed: sql:%s, row:%d is smaller than zero" %
(callerFilename, self.sql, row))
if col < 0:
tdLog.exit(
"%s failed: sql:%s, col:%d is smaller than zero" %
(callerFilename, self.sql, col))
if row > self.queryRows:
tdLog.exit(
"%s failed: sql:%s, row:%d is larger than queryRows:%d" %
(callerFilename, self.sql, row, self.queryRows))
if col > self.queryCols:
tdLog.exit(
"%s failed: sql:%s, col:%d is larger than queryCols:%d" %
(callerFilename, self.sql, col, self.queryCols))
self.checkRowCol(row, col)
if self.queryResult[row][col] != data:
tdLog.exit("%s failed: sql:%s row:%d col:%d data:%s != expect:%s" % (
callerFilename, self.sql, row, col, self.queryResult[row][col], data))
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, self.sql, row, col, self.queryResult[row][col], data)
tdLog.exit("%s(%d) failed: sql:%s row:%d col:%d data:%s != expect:%s" % args)
if data is None:
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" %
@ -153,26 +140,7 @@ class TDSql:
(self.sql, row, col, self.queryResult[row][col], data))
def getData(self, row, col):
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
if row < 0:
tdLog.exit(
"%s failed: sql:%s, row:%d is smaller than zero" %
(callerFilename, self.sql, row))
if col < 0:
tdLog.exit(
"%s failed: sql:%s, col:%d is smaller than zero" %
(callerFilename, self.sql, col))
if row > self.queryRows:
tdLog.exit(
"%s failed: sql:%s, row:%d is larger than queryRows:%d" %
(callerFilename, self.sql, row, self.queryRows))
if col > self.queryCols:
tdLog.exit(
"%s failed: sql:%s, col:%d is larger than queryCols:%d" %
(callerFilename, self.sql, col, self.queryCols))
self.checkRowCol(row, col)
return self.queryResult[row][col]
def executeTimes(self, sql, times):
@ -185,20 +153,21 @@ class TDSql:
def execute(self, sql):
self.sql = sql
self.affectedRows = self.cursor.execute(sql)
try:
self.affectedRows = self.cursor.execute(sql)
except Exception as e:
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, sql, repr(e))
tdLog.exit("%s(%d) failed: sql:%s, %s" % args)
return self.affectedRows
def checkAffectedRows(self, expectAffectedRows):
if self.affectedRows != expectAffectedRows:
frame = inspect.stack()[1]
callerModule = inspect.getmodule(frame[0])
callerFilename = callerModule.__file__
caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, self.sql, self.affectedRows, expectAffectedRows)
tdLog.exit("%s(%d) failed: sql:%s, affectedRows:%d != expect:%d" % args)
tdLog.exit(
"%s failed: sql:%s, affectedRows:%d != expect:%d" %
(callerFilename, self.sql, self.affectedRows, expectAffectedRows))
tdLog.info("sql:%s, affectedRows:%d == expect:%d" %
(self.sql, self.affectedRows, expectAffectedRows))
tdLog.info("sql:%s, affectedRows:%d == expect:%d" % (self.sql, self.affectedRows, expectAffectedRows))
tdSql = TDSql()

212
tests/script/general/parser/stream.sim
View File

@ -1,212 +0,0 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 0
system sh/cfg.sh -n dnode1 -c tableMetaKeepTimer -v 5
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
print ======================== stream.sim
sleep 2000
$db = strm_db
$tb = tb
$mt = mt
$strm = strm
$tbNum = 10
$rowNum = 20
$totalNum = 200
print =============== set up DB
$i = 0
sql drop database if exists $db
sql create database $db
sql use $db
## [TBASE300]
print ====== TBASE-300
sql create table mt (ts timestamp, c1 int, c2 int) tags(t1 int)
sql create table tb1 using mt tags(1)
sql create table tb2 using mt tags(2)
sql create table strm as select count(*), avg(c1), sum(c2), max(c1), min(c2),first(c1), last(c2) from mt interval(4s) sliding(2s)
sleep 10000
sql insert into tb2 values(now,1,1)
sql insert into tb1 values(now,1,1)
sleep 4000
sql select * from mt
sql select * from strm
sql drop table tb1
sleep 100000
sql select * from strm
if $rows != 2 then
if $rows != 1 then
return -1
endi
endi
sql drop table tb2
sql drop table mt
sql drop table strm
## [TBASE304]
print ====== TBASE-304
sleep 10000
# we cannot reset query cache in server side, as a workaround,
# set super table name to mt304, need to change back to mt later
print create mt304
sql create table mt304 (ts timestamp, c1 int) tags(t1 int, t2 int)
print create tb1
sql create table tb1 using mt304 tags(1, 1)
print create tb2
sql create table tb2 using mt304 tags(1, -1)
print create strm
sql create table strm as select count(*), avg(c1) from mt304 where t2 >= 0 interval(4s) sliding(2s)
sql insert into tb1 values (now,1)
sql insert into tb2 values (now,2)
sleep 100000
sql select * from strm;
if $rows != 2 then
print ==== expect rows = 2, actually returned rows = $rows
return -1
endi
if $data01 != 1 then
return -1
endi
print data02 = $data02
if $data02 != 1.000000000 then
return -1
endi
sql alter table mt304 drop tag t2;
sql insert into tb2 values (now,2);
sql insert into tb1 values (now,1);
sql select * from strm;
sql alter table mt304 add tag t2 int;
sleep 10000
sql select * from strm
print ================= create a stream with a wildcard filter on tags of a STable
sql drop database $db
sql create database $db
sql use $db
sql create table stb (ts timestamp, c1 int, c2 binary(10)) tags(t1 binary(10))
sql create table tb1 using stb tags('a1')
sql create table tb2 using stb tags('b2')
sql create table tb3 using stb tags('a3')
sql create table strm as select count(*), avg(c1), first(c2) from stb where t1 like 'a%' interval(4s) sliding(2s)
sleep 11000
sql insert into tb1 values (now, 0, 'tb1')
sleep 4000
sql insert into tb2 values (now, 2, 'tb2')
sleep 4000
sql insert into tb3 values (now, 0, 'tb3')
sleep 60000
sql describe strm
if $rows == 0 then
return -1
endi
sql select * from strm
sleep 1000
print ======== data0: $data00 $data01 $data02 $data03
print ======== data1: $data10 $data11 $data12 $data13
print ======== data2: $data20 $data21 $data22 $data23
print ======== data3: $data30 $data31 $data32 $data33
if $rows != 4 then
print ==== expect rows = 4, actually returned rows = $rows
return -1
endi
if $data02 != 0.000000000 then
return -1
endi
if $data03 == tb2 then
return -1
endi
if $data13 == tb2 then
return -1
endi
if $data23 == tb2 then
return -1
endi
if $data33 == tb2 then
return -1
endi
## The vnode client needs to refresh metadata cache to allow strm calculate tb4's data. But the current refreshing frequency is every 10 min
## commented out the case below to save running time
sql create table tb4 using stb tags('a4')
sql insert into tb4 values(now, 4, 'tb4')
sleep 60000
sql select * from strm order by ts desc
print ======== data0: $data00 $data01 $data02 $data03
#print ======== data1: $data10 $data11 $data12 $data13
#print ======== data2: $data20 $data21 $data22 $data23
#print ======== data3: $data30 $data31 $data32 $data33
if $rows != 6 then
print ==== expect rows = 6, actually returned rows = $rows
return -1
endi
if $data02 != 4.000000000 then
return -1
endi
if $data03 != tb4 then
return -1
endi
print =============== change tag values to see if stream still works correctly
sql alter table tb4 set tag t1='b4'
sleep 3000 # waiting for new tag valid
sql insert into tb1 values (now, 1, 'tb1_a1')
sleep 4000
sql insert into tb4 values (now, -4, 'tb4_b4')
sleep 100000
sql select * from strm order by ts desc
sleep 1000
print ======== data0: $data00 $data01 $data02 $data03
#print ======== data1: $data10 $data11 $data12 $data13
#print ======== data2: $data20 $data21 $data22 $data23
#print ======== data3: $data30 $data31 $data32 $data33
if $rows != 8 then
print ==== expect rows = 8, actually returned rows = $rows
return -1
endi
if $data02 != 1.000000000 then
return -1
endi
if $data03 != tb1_a1 then
return -1
endi
sql drop database if exists $db
sql drop database if exists strm_db_0
sql create database strm_db_0
sql use strm_db_0
sql create table stb (ts timestamp, c1 int, c2 bigint, c3 float, c4 double, c5 binary(15), c6 nchar(15), c7 bool) tags(t1 int, t2 binary(15))
sql create table tb0 using stb tags(0, 'tb0')
sql create table tb1 using stb tags(1, 'tb1')
sql create table tb2 using stb tags(2, 'tb2')
sql create table tb3 using stb tags(3, 'tb3')
sql create table tb4 using stb tags(4, 'tb4')
sql create table strm0 as select count(ts), count(c1), max(c2), min(c4), first(c5), last(c6) from stb where ts < now + 30s interval(4s) sliding(2s)
sleep 1000
sql insert into tb0 values (now, 0, 0, 0, 0, 'binary0', '涛思0', true) tb1 values (now, 1, 1, 1, 1, 'binary1', '涛思1', false) tb2 values (now, 2, 2, 2, 2, 'binary2', '涛思2', true) tb3 values (now, 3, 3, 3, 3, 'binary3', '涛思3', false) tb4 values (now, 4, 4, 4, 4, 'binary4', '涛思4', true)
sleep 30000
sql select * from strm0 order by ts desc
sleep 1000
if $rows != 2 then
print ==== expect rows = 2, actually returned rows = $rows
return -1
endi
sql insert into tb0 values (now, 10, 10, 10, 10, 'binary0', '涛思0', true) tb1 values (now, 11, 11, 11, 11, 'binary1', '涛思1', false) tb2 values (now, 12, 12, 12, 12, 'binary2', '涛思2', true) tb3 values (now, 13, 13, 13, 13, 'binary3', '涛思3', false) tb4 values (now, 14, 14, 14, 14, 'binary4', '涛思4', true)
sleep 30000
sql select * from strm0 order by ts desc
sleep 10000
if $rows == 4 then
print ==== actually returned rows = $rows, expect always not equal to 4
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT

287
tests/script/general/stream/metrics_1.sim
View File

@ -1,287 +0,0 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 0
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
print ======================== dnode1 start
$dbPrefix = m1_db
$tbPrefix = m1_tb
$mtPrefix = m1_mt
$stPrefix = m1_st
$tbNum = 10
$rowNum = 20
$totalNum = 200
print =============== step1
$i = 0
$db = $dbPrefix . $i
$mt = $mtPrefix . $i
$st = $stPrefix . $i
sql drop databae $db -x step1
step1:
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int, tbcol2 float) TAGS(tgcol int)
$i = 0
while $i < $tbNum
$tb = $tbPrefix . $i
sql create table $tb using $mt tags( $i )
$x = -1440
$y = 0
while $y < $rowNum
$ms = $x . m
sql insert into $tb values (now $ms , $y , $y )
$x = $x + 1
$y = $y + 1
endw
$i = $i + 1
endw
sleep 100
print =============== step2 c1
$i = 0
$mt = $mtPrefix . $i
sql select count(*) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 200 then
return -1
endi
$st = $stPrefix . c1
sql create table $st as select count(*) from $mt interval(1d)
print =============== step3 c2
sql select count(tbcol) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 200 then
return -1
endi
$st = $stPrefix . c2
sql create table $st as select count(tbcol) from $mt interval(1d)
print =============== step4 c3
sql select count(tbcol2) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 200 then
return -1
endi
$st = $stPrefix . c3
sql create table $st as select count(tbcol2) from $mt interval(1d)
print =============== step5 avg
sql select avg(tbcol) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 9.500000000 then
return -1
endi
$st = $stPrefix . av
print create table $st as select avg(tbcol) from $mt interval(1d)
sql create table $st as select avg(tbcol) from $mt interval(1d)
print =============== step6 su
sql select sum(tbcol) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 1900 then
return -1
endi
$st = $stPrefix . su
sql create table $st as select sum(tbcol) from $mt interval(1d)
print =============== step7 mi
sql select min(tbcol) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 0 then
return -1
endi
$st = $stPrefix . mi
sql create table $st as select min(tbcol) from $mt interval(1d)
print =============== step8 ma
sql select max(tbcol) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 19 then
return -1
endi
$st = $stPrefix . ma
sql create table $st as select max(tbcol) from $mt interval(1d)
print =============== step9 fi
sql select first(tbcol) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 0 then
return -1
endi
$st = $stPrefix . fi
sql create table $st as select first(tbcol) from $mt interval(1d)
print =============== step10 la
sql select last(tbcol) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 19 then
return -1
endi
$st = $stPrefix . la
sql create table $st as select last(tbcol) from $mt interval(1d)
print =============== step11 st
sql select stddev(tbcol) from $mt interval(1d) -x step11
return -1
step11:
print =============== step12 le
sql select leastsquares(tbcol, 1, 1) from $mt interval(1d) -x step12
return -1
step12:
print =============== step13
sql select top(tbcol, 1) from $mt interval(1d)
print =============== step14
sql select bottom(tbcol, 1) from $mt interval(1d)
print =============== step15 pe
sql select percentile(tbcol, 1) from $mt interval(1d) -x step15
return -1
step15:
print =============== step16
sql select diff(tbcol) from $mt interval(1d) -x step16
return -1
step16:
print =============== step17 wh
sql select count(tbcol) from $mt where ts < now + 4m interval(1d)
print ===> $data00 $data01
if $data01 != 200 then
return -1
endi
$st = $stPrefix . wh
#sql create table $st as select count(tbcol) from $mt where ts < now + 4m interval(1d)
print =============== step18 as
sql select count(tbcol) from $mt interval(1d)
print ===> $data00 $data01
if $data01 != 200 then
return -1
endi
$st = $stPrefix . as
sql create table $st as select count(tbcol) as c from $mt interval(1d)
print =============== step19 gb
sql select count(tbcol) from $mt interval(1d) group by tgcol
print ===> $data00 $data01
if $data01 != 20 then
return -1
endi
print =============== step20 x
sql select count(tbcol) from $mt where ts < now + 4m interval(1d) group by tgcol
print ===> $data00 $data01
if $data01 != 20 then
return -1
endi
print =============== step21
print sleep 120 seconds
sleep 120000
print =============== step22
$st = $stPrefix . c1
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 200 then
return -1
endi
$st = $stPrefix . c2
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 200 then
return -1
endi
$st = $stPrefix . c3
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 200 then
return -1
endi
$st = $stPrefix . av
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 9.500000000 then
return -1
endi
$st = $stPrefix . su
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 1900 then
return -1
endi
$st = $stPrefix . mi
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 0 then
return -1
endi
$st = $stPrefix . ma
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 19 then
return -1
endi
$st = $stPrefix . fi
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 0 then
return -1
endi
$st = $stPrefix . la
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 19 then
return -1
endi
#$st = $stPrefix . wh
#sql select * from $st
#print ===> select * from $st ===> $data00 $data01
#if $data01 != 200 then
# return -1
#endi
$st = $stPrefix . as
sql select * from $st
print ===> select * from $st ===> $data00 $data01
if $data01 != 200 then
return -1
endi

106
tests/script/general/stream/new_stream.sim
View File

@ -1,106 +0,0 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 0
system sh/cfg.sh -n dnode1 -c tableMetaKeepTimer -v 10
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
print ======================== dnode1 start
$dbPrefix = ns_db
$tbPrefix = ns_tb
$mtPrefix = ns_mt
$stPrefix = ns_st
$tbNum = 5
$rowNum = 200
$totalNum = 200
print =============== step1
$i = 0
$db = $dbPrefix
$mt = $mtPrefix
$st = $stPrefix
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int, tbcol2 float) TAGS(tgcol int)
$i = 0
while $i < $tbNum
$tb = $tbPrefix . $i
sql create table $tb using $mt tags( $i )
$x = 0
$y = 0
while $y < $rowNum
$ms = $x . s
sql insert into $tb values (now + $ms , $y , $y )
$x = $x + 1
$y = $y + 1
endw
$i = $i + 1
endw
sleep 100
print =============== step2
sql select count(*), count(tbcol), count(tbcol2) from $mt interval(10s)
print $data00 $data01 $data02 $data03
sql create table $st as select count(*), count(tbcol), count(tbcol2) from $mt interval(10s)
print =============== step3
print sleep 120 seconds
sleep 120000
print =============== step4
sql select * from $st
print $st ==> $rows1 $data00 $data01 $data02 $data03
if $data03 >= 51 then
return -1
endi
print =============== step5
$tbNum = 10
while $i < $tbNum
$tb = $tbPrefix . $i
sql create table $tb using $mt tags( $i )
if $i == 0 then
sleep 2000
endi
$x = 0
$y = 0
while $y < $rowNum
$ms = $x . s
sql insert into $tb values (now + $ms , $y , $y )
$x = $x + 1
$y = $y + 1
endw
$i = $i + 1
endw
print =============== step6
print sleep 120 seconds
sleep 120000
print =============== step7
sql select * from $st order by ts desc
print $st ==> $rows1 $data00 $data01 $data02 $data03
if $data03 <= 51 then
return -1
endi

206
tests/script/general/stream/stream_1.sim
View File

@ -1,206 +0,0 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 0
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
print ======================== dnode1 start
$dbPrefix = s1_db
$tbPrefix = s1_tb
$mtPrefix = s1_mt
$stPrefix = s1_st
$tbNum = 10
$rowNum = 20
$totalNum = 200
print =============== step1
$i = 0
$db = $dbPrefix . $i
$mt = $mtPrefix . $i
$st = $stPrefix . $i
sql drop databae $db -x step1
step1:
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int, tbcol2 float) TAGS(tgcol int)
$i = 0
while $i < $tbNum
$tb = $tbPrefix . $i
sql create table $tb using $mt tags( $i )
$x = -1440
$y = 0
while $y < $rowNum
$ms = $x . m
sql insert into $tb values (now $ms , $y , $y )
$x = $x + 1
$y = $y + 1
endw
$i = $i + 1
endw
sleep 100
print =============== step2
$i = 0
$tb = $tbPrefix . $i
$st = $stPrefix . $i
sql select count(*), count(tbcol), count(tbcol2) from $tb interval(1d)
print select count(*), count(tbcol), count(tbcol2) from $tb interval(1d) ===> $data00 $data01 $data02, $data03
if $data01 != $rowNum then
return -1
endi
if $data02 != $rowNum then
return -1
endi
if $data03 != $rowNum then
return -1
endi
sql show tables
if $rows != 10 then
return -1
endi
sql create table $st as select count(*), count(tbcol), count(tbcol2) from $tb interval(1d)
sql show tables
if $rows != 11 then
return -1
endi
print =============== step3
print sleep 120 seconds
sleep 120000
sql select * from $st
print select * from $st => $data01
if $data01 != 20 then
return -1
endi
if $data02 != 20 then
return -1
endi
if $data03 != 20 then
return -1
endi
print =============== step4
sql drop table $st
sql show tables
if $rows != 10 then
return -1
endi
print =============== step5
sql select * from $st -x step4
return -1
step4:
print =============== step6
sql create table $st as select count(*), count(tbcol), count(tbcol2) from $tb interval(1d)
sql show tables
if $rows != 11 then
return -1
endi
print =============== step7
print sleep 120 seconds
sleep 120000
sql select * from $st
print select * from $st => $data01
if $data01 != 20 then
return -1
endi
if $data02 != 20 then
return -1
endi
if $data03 != 20 then
return -1
endi
print =============== step8
$i = 1
$st = $stPrefix . $i
sql select count(*), count(tbcol), count(tbcol2) from $mt interval(1d)
print select count(*), count(tbcol), count(tbcol2) from $mt interval(1d) ===> $data00 $data01 $data02, $data03
if $data01 != 200 then
return -1
endi
if $data02 != 200 then
return -1
endi
if $data03 != 200 then
return -1
endi
sql show tables
if $rows != 11 then
return -1
endi
sql create table $st as select count(*), count(tbcol), count(tbcol2) from $mt interval(1d)
sql show tables
if $rows != 12 then
return -1
endi
print =============== step9
print sleep 120 seconds
sleep 120000
sql select * from $st
print select * from $st => $data01
if $data01 != 200 then
return -1
endi
if $data02 != 200 then
return -1
endi
if $data03 != 200 then
return -1
endi
print =============== step10
sql drop table $st
sql show tables
if $rows != 11 then
return -1
endi
print =============== step11
sql select * from $st -x step10
return -1
step10:
print =============== step12
sql create table $st as select count(*), count(tbcol), count(tbcol2) from $mt interval(1d)
sql show tables
if $rows != 12 then
return -1
endi
print =============== step13
print sleep 120 seconds
sleep 120000
sql select * from $st
print select * from $st => $data01
if $data01 != 200 then
return -1
endi
if $data02 != 200 then
return -1
endi
if $data03 != 200 then
return -1
endi

194
tests/script/general/stream/stream_2.sim
View File

@ -1,194 +0,0 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 0
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
print ======================== dnode1 start
$dbPrefix = s2_db
$tbPrefix = s2_tb
$mtPrefix = s2_mt
$stPrefix = s2_st
$tbNum = 10
$rowNum = 20
$totalNum = 200
print =============== step1
$i = 0
$db = $dbPrefix . $i
$mt = $mtPrefix . $i
$st = $stPrefix . $i
sql drop databae $db -x step1
step1:
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int, tbcol2 float) TAGS(tgcol int)
$i = 0
while $i < $tbNum
$tb = $tbPrefix . $i
sql create table $tb using $mt tags( $i )
$x = -1440
$y = 0
while $y < $rowNum
$ms = $x . m
sql insert into $tb values (now $ms , $y , $y )
$x = $x + 1
$y = $y + 1
endw
$i = $i + 1
endw
sleep 100
print =============== step2
$i = 0
$tb = $tbPrefix . $i
$st = $stPrefix . $i
sql select count(tbcol) from $tb interval(1d)
print select count(tbcol) from $tb interval(1d) ===> $data00 $data01 $data02, $data03
if $data01 != $rowNum then
return -1
endi
sql show tables
if $rows != 10 then
return -1
endi
sql create table $st as select count(tbcol) from $tb interval(1d)
sql show tables
if $rows != 11 then
return -1
endi
print =============== step3
print sleep 120 seconds
sleep 120000
sql select * from $st
print select * from $st => $data01
if $data01 != 20 then
return -1
endi
print =============== step4
sql drop table $st
sql show tables
if $rows != 10 then
return -1
endi
print =============== step5
sql select * from $st -x step4
return -1
step4:
print =============== step6
sql create table $st as select count(*), count(tbcol), count(tbcol2) from $tb interval(1d)
sql show tables
if $rows != 11 then
return -1
endi
print =============== step7
print sleep 120 seconds
sleep 120000
sql select * from $st
print select * from $st => $data01
if $data01 != 20 then
return -1
endi
if $data02 != 20 then
return -1
endi
if $data03 != 20 then
return -1
endi
print =============== step8
$i = 1
$st = $stPrefix . $i
sql select count(*), count(tbcol), count(tbcol2) from $mt interval(1d)
print select count(*), count(tbcol), count(tbcol2) from $mt interval(1d) ===> $data00 $data01 $data02, $data03
if $data01 != 200 then
return -1
endi
if $data02 != 200 then
return -1
endi
if $data03 != 200 then
return -1
endi
sql show tables
if $rows != 11 then
return -1
endi
sql create table $st as select count(*), count(tbcol), count(tbcol2) from $mt interval(1d)
sql show tables
if $rows != 12 then
return -1
endi
print =============== step9
print sleep 120 seconds
sleep 120000
sql select * from $st
print select * from $st => $data01 $data02, $data03
if $data01 != 200 then
return -1
endi
if $data02 != 200 then
return -1
endi
if $data03 != 200 then
return -1
endi
print =============== step10
sql drop table $st
sql show tables
if $rows != 11 then
return -1
endi
print =============== step11
sql select * from $st -x step10
return -1
step10:
print =============== step12
sql create table $st as select count(tbcol) from $mt interval(1d)
sql show tables
if $rows != 12 then
return -1
endi
print =============== step13
print sleep 120 seconds
sleep 120000
sql select * from $st
print select * from $st => $data01 $data02, $data03
if $data01 != 200 then
return -1
endi
if $data02 != null then
return -1
endi
if $data03 != null then
return -1
endi

3
tests/script/general/stream/testSuite.sim
View File

@ -1,9 +1,6 @@
run general/stream/stream_1.sim
run general/stream/stream_2.sim
run general/stream/stream_3.sim
run general/stream/stream_restart.sim
run general/stream/table_1.sim
run general/stream/metrics_1.sim
run general/stream/table_n.sim
run general/stream/metrics_n.sim
run general/stream/table_del.sim

5
tests/script/jenkins/basic.txt
View File

@ -313,17 +313,12 @@ cd ../../../debug; make
# stream still has bugs
#./test.sh -f general/parser/stream_on_sys.sim
#./test.sh -f general/parser/stream.sim
#./test.sh -f general/parser/repeatStream.sim
#./test.sh -f general/stream/new_stream.sim
./test.sh -f general/stream/metrics_1.sim
./test.sh -f general/stream/metrics_del.sim
./test.sh -f general/stream/metrics_n.sim
./test.sh -f general/stream/metrics_replica1_vnoden.sim
./test.sh -f general/stream/restart_stream.sim
./test.sh -f general/stream/stream_1.sim
./test.sh -f general/stream/stream_2.sim
./test.sh -f general/stream/stream_3.sim
./test.sh -f general/stream/stream_restart.sim
./test.sh -f general/stream/table_1.sim

1
tests/script/jenkins/simple.txt
View File

@ -148,7 +148,6 @@ cd ../../../debug; make
./test.sh -f general/parser/binary_escapeCharacter.sim
./test.sh -f general/parser/bug.sim
#./test.sh -f general/parser/stream_on_sys.sim
./test.sh -f general/parser/stream.sim
./test.sh -f general/parser/repeatAlter.sim
#./test.sh -f general/parser/repeatStream.sim