diff --git a/src/client/src/tscSub.c b/src/client/src/tscSub.c index 526f38047557af01f4a04f8cf6f3af4657f2a675..8e554d62ffa1da2cdcefb9eaabf8ea400aef283f 100644 --- a/src/client/src/tscSub.c +++ b/src/client/src/tscSub.c @@ -256,11 +256,12 @@ static int tscUpdateSubscription(STscObj* pObj, SSub* pSub) { } size_t numOfTables = taosArrayGetSize(tables); + SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); SArray* progress = taosArrayInit(numOfTables, sizeof(SSubscriptionProgress)); for( size_t i = 0; i < numOfTables; i++ ) { STidTags* tt = taosArrayGet( tables, i ); SSubscriptionProgress p = { .uid = tt->uid }; - p.key = tscGetSubscriptionProgress(pSub, tt->uid, INT64_MIN); + p.key = tscGetSubscriptionProgress(pSub, tt->uid, pQueryInfo->window.skey); taosArrayPush(progress, &p); } taosArraySort(progress, tscCompareSubscriptionProgress); diff --git a/tests/pytest/subscribe/__init__.py b/tests/pytest/subscribe/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/tests/pytest/subscribe/singlemeter.py b/tests/pytest/subscribe/singlemeter.py new file mode 100644 index 0000000000000000000000000000000000000000..879e0a75ebdf29022990b5e2e250370620c74636 --- /dev/null +++ b/tests/pytest/subscribe/singlemeter.py @@ -0,0 +1,79 @@ +################################################################### + # Copyright (c) 2020 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 taos +import time +from util.log import * +from util.cases import * +from util.sql import * +from util.sub import * + +class TDTestCase: + def init(self, conn, logSql): + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor(), logSql) + self.conn = conn + + def run(self): + sqlstr = "select * from t0" + topic = "test" + now = int(time.time() * 1000) + tdSql.prepare() + + tdLog.info("create a table and insert 10 rows.") + tdSql.execute("create table t0(ts timestamp, a int, b int);") + for i in range(0, 10): + tdSql.execute("insert into t0 values (%d, %d, %d);" % (now + i, i, i)) + + tdLog.info("consumption 01.") + tdSub.init(self.conn.subscribe(True, topic, sqlstr, 0)) + tdSub.consume() + tdSub.checkRows(10) + + tdLog.info("consumption 02: no new rows inserted") + tdSub.consume() + tdSub.checkRows(0) + + tdLog.info("consumption 03: after one new rows inserted") + tdSql.execute("insert into t0 values (%d, 10, 10);" % (now + 10)) + tdSub.consume() + tdSub.checkRows(1) + + tdLog.info("consumption 04: keep progress and continue previous subscription") + tdSub.close(True) + tdSub.init(self.conn.subscribe(False, topic, sqlstr, 0)) + tdSub.consume() + tdSub.checkRows(0) + + tdLog.info("consumption 05: remove progress and continue previous subscription") + tdSub.close(False) + tdSub.init(self.conn.subscribe(False, topic, sqlstr, 0)) + tdSub.consume() + tdSub.checkRows(11) + + tdLog.info("consumption 06: keep progress and restart the subscription") + tdSub.close(True) + tdSub.init(self.conn.subscribe(True, topic, sqlstr, 0)) + tdSub.consume() + tdSub.checkRows(11) + + tdSub.close(True) + + def stop(self): + tdSub.close(False) + tdSql.close() + tdLog.success("%s successfully executed" % __file__) + +tdCases.addWindows(__file__, TDTestCase()) +tdCases.addLinux(__file__, TDTestCase()) diff --git a/tests/pytest/subscribe/supertable.py b/tests/pytest/subscribe/supertable.py new file mode 100644 index 0000000000000000000000000000000000000000..c6cc2969aa57848bf2c43163a26608117e82ad90 --- /dev/null +++ b/tests/pytest/subscribe/supertable.py @@ -0,0 +1,114 @@ +################################################################### + # Copyright (c) 2020 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 taos +import time +from util.log import * +from util.cases import * +from util.sql import * +from util.sub import * + +class TDTestCase: + def init(self, conn, logSql): + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor(), logSql) + self.conn = conn + + def run(self): + sqlstr = "select * from meters" + topic = "test" + now = int(time.time() * 1000) + tdSql.prepare() + + tdLog.info("create a super table and 10 sub-tables, then insert 5 rows into each sub-table.") + tdSql.execute("create table meters(ts timestamp, a int, b int) tags(area int, loc binary(20));") + for i in range(0, 10): + for j in range(0, 5): + tdSql.execute("insert into t%d using meters tags(%d, 'area%d') values (%d, %d, %d);" % (i, i, i, now + j, j, j)) + + tdLog.info("consumption 01.") + tdSub.init(self.conn.subscribe(True, topic, sqlstr, 0)) + tdSub.consume() + tdSub.checkRows(50) + + tdLog.info("consumption 02: no new rows inserted") + tdSub.consume() + tdSub.checkRows(0) + + tdLog.info("consumption 03: after one new rows inserted") + tdSql.execute("insert into t0 values (%d, 10, 10);" % (now + 10)) + tdSub.consume() + tdSub.checkRows(1) + + tdLog.info("consumption 04: keep progress and continue previous subscription") + tdSub.close(True) + tdSub.init(self.conn.subscribe(False, topic, sqlstr, 0)) + tdSub.consume() + tdSub.checkRows(0) + + tdLog.info("consumption 05: remove progress and continue previous subscription") + tdSub.close(False) + tdSub.init(self.conn.subscribe(False, topic, sqlstr, 0)) + tdSub.consume() + tdSub.checkRows(51) + + tdLog.info("consumption 06: keep progress and restart the subscription") + tdSub.close(True) + tdSub.init(self.conn.subscribe(True, topic, sqlstr, 0)) + tdSub.consume() + tdSub.checkRows(51) + + tdLog.info("consumption 07: insert one row to two table then remove one table") + tdSql.execute("insert into t0 values (%d, 11, 11);" % (now + 11)) + tdSql.execute("insert into t1 values (%d, 11, 11);" % (now + 11)) + tdSql.execute("drop table t0") + tdSub.consume() + tdSub.checkRows(1) + + tdLog.info("consumption 08: check timestamp criteria") + tdSub.close(False) + tdSub.init(self.conn.subscribe(True, topic, sqlstr + " where ts > %d" % now, 0)) + tdSub.consume() + tdSub.checkRows(37) + + tdLog.info("consumption 09: insert large timestamp to t2 then insert smaller timestamp to t1") + tdSql.execute("insert into t2 values (%d, 100, 100);" % (now + 100)) + tdSub.consume() + tdSub.checkRows(1) + tdSql.execute("insert into t1 values (%d, 12, 12);" % (now + 12)) + tdSub.consume() + tdSub.checkRows(1) + + tdLog.info("consumption 10: field criteria") + tdSub.close(True) + tdSub.init(self.conn.subscribe(False, topic, sqlstr + " where a > 100", 0)) + tdSql.execute("insert into t2 values (%d, 101, 100);" % (now + 101)) + tdSql.execute("insert into t2 values (%d, 100, 100);" % (now + 102)) + tdSql.execute("insert into t2 values (%d, 102, 100);" % (now + 103)) + tdSub.consume() + tdSub.checkRows(2) + + tdLog.info("consumption 11: two vnodes") + tdSql.execute("insert into t2 values (%d, 102, 100);" % (now + 104)) + tdSql.execute("insert into t9 values (%d, 102, 100);" % (now + 104)) + tdSub.consume() + tdSub.checkRows(2) + + def stop(self): + tdSub.close(False) + tdSql.close() + tdLog.success("%s successfully executed" % __file__) + +tdCases.addWindows(__file__, TDTestCase()) +tdCases.addLinux(__file__, TDTestCase()) diff --git a/tests/pytest/util/sub.py b/tests/pytest/util/sub.py new file mode 100644 index 0000000000000000000000000000000000000000..2e3c2a96b7312c176c25bd35e109e146e5c4593f --- /dev/null +++ b/tests/pytest/util/sub.py @@ -0,0 +1,43 @@ +################################################################### + # Copyright (c) 2020 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 os +import time +import datetime +from util.log import * + +class TDSub: + def __init__(self): + self.consumedRows = 0 + self.consumedCols = 0 + + def init(self, sub): + self.sub = sub + + def close(self, keepProgress): + self.sub.close(keepProgress) + + def consume(self): + self.data = self.sub.consume() + self.consumedRows = len(self.data) + self.consumedCols = len(self.sub.fields) + return self.consumedRows + + def checkRows(self, expectRows): + if self.consumedRows != expectRows: + tdLog.exit("consumed rows:%d != expect:%d" % (self.consumedRows, expectRows)) + tdLog.info("consumed rows:%d == expect:%d" % (self.consumedRows, expectRows)) + + +tdSub = TDSub()