diff --git a/source/dnode/vnode/src/tq/tqScan.c b/source/dnode/vnode/src/tq/tqScan.c index ead040b7a5..d43afc9ebd 100644 --- a/source/dnode/vnode/src/tq/tqScan.c +++ b/source/dnode/vnode/src/tq/tqScan.c @@ -400,29 +400,31 @@ static void preProcessSubmitMsg(STqHandle* pHandle, const SMqPollReq* pRequest, return; } + if (pSubmitTbData->pCreateTbReq == NULL){ + continue; + } + int64_t uid = pSubmitTbData->uid; if (taosHashGet(pRequest->uidHash, &uid, LONG_BYTES) != NULL) { tqDebug("poll rawdata split,uid:%" PRId64 " is already exists", uid); terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT; return; } else { - if (taosHashPut(pRequest->uidHash, &uid, LONG_BYTES, &uid, LONG_BYTES) != 0){ - tqError("failed to add table create time to hash, uid:%"PRId64, uid); + int32_t code = taosHashPut(pRequest->uidHash, &uid, LONG_BYTES, &uid, LONG_BYTES); + if (code != 0){ + tqError("failed to add table uid to hash, code:%d, uid:%"PRId64, code, uid); } } - if (pSubmitTbData->pCreateTbReq == NULL){ - continue; - } - int64_t createTime = 0; int64_t *cTime = (int64_t*)taosHashGet(pHandle->tableCreateTimeHash, &uid, LONG_BYTES); if (cTime != NULL){ createTime = *cTime; } else{ createTime = metaGetTableCreateTime(pReader->pVnodeMeta, uid, 1); - if (taosHashPut(pHandle->tableCreateTimeHash, &uid, LONG_BYTES, &createTime, LONG_BYTES) != 0){ - tqError("failed to add table create time to hash, uid:%"PRId64, uid); + int32_t code = taosHashPut(pHandle->tableCreateTimeHash, &uid, LONG_BYTES, &createTime, LONG_BYTES); + if (code != 0){ + tqError("failed to add table create time to hash,code:%d, uid:%"PRId64, code, uid); } } if (pHandle->fetchMeta == WITH_DATA || pSubmitTbData->ctimeMs > createTime){ @@ -447,8 +449,9 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqData } code = tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver, rawList); TSDB_CHECK_CODE(code, lino, END); - preProcessSubmitMsg(pHandle, pRequest, &rawList); - + if (pRequest->rawData) { + preProcessSubmitMsg(pHandle, pRequest, &rawList); + } // data could not contains same uid data in rawdata mode if (pRequest->rawData != 0 && terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT){ goto END; diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 94bb254302..22ea60cd3d 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -449,6 +449,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq_ts5466.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq_td33504.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq_ts-5473.py +,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq_ts-5776.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq_ts5906.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/td-32187.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/td-33225.py diff --git a/tests/system-test/7-tmq/tmq_ts-5776.py b/tests/system-test/7-tmq/tmq_ts-5776.py new file mode 100644 index 0000000000..738d69701f --- /dev/null +++ b/tests/system-test/7-tmq/tmq_ts-5776.py @@ -0,0 +1,39 @@ + +import taos +import sys +import time +import socket +import os +import threading + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * +from util.common import * +from taos.tmq import * +sys.path.append("./7-tmq") +from tmqCommon import * + +class TDTestCase: + updatecfgDict = {'debugFlag': 135, 'asynclog': 0} + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor()) + #tdSql.init(conn.cursor(), logSql) # output sql.txt file + + def run(self): + buildPath = tdCom.getBuildPath() + cmdStr = '%s/build/bin/tmq_ts5776'%(buildPath) + tdLog.info(cmdStr) + os.system(cmdStr) + + return + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) diff --git a/utils/test/c/CMakeLists.txt b/utils/test/c/CMakeLists.txt index d1c049ef1e..d73f91aad3 100644 --- a/utils/test/c/CMakeLists.txt +++ b/utils/test/c/CMakeLists.txt @@ -6,6 +6,7 @@ add_executable(tmq_taosx_ci tmq_taosx_ci.c) add_executable(tmq_ts5466 tmq_ts5466.c) add_executable(tmq_td32526 tmq_td32526.c) add_executable(tmq_td32187 tmq_td32187.c) +add_executable(tmq_ts5776 tmq_ts5776.c) add_executable(tmq_td32471 tmq_td32471.c) add_executable(tmq_write_raw_test tmq_write_raw_test.c) add_executable(write_raw_block_test write_raw_block_test.c) @@ -87,6 +88,13 @@ target_link_libraries( PUBLIC common PUBLIC os ) +target_link_libraries( + tmq_ts5776 + PUBLIC ${TAOS_LIB} + PUBLIC util + PUBLIC common + PUBLIC os +) target_link_libraries( tmq_taosx_ci PUBLIC ${TAOS_LIB} diff --git a/utils/test/c/tmq_ts5776.c b/utils/test/c/tmq_ts5776.c new file mode 100644 index 0000000000..c50809bccd --- /dev/null +++ b/utils/test/c/tmq_ts5776.c @@ -0,0 +1,255 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#include +#include +#include +#include +#include +#include "cJSON.h" +#include "taos.h" +#include "tmsg.h" +#include "types.h" + +static TAOS* use_db() { + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + if (pConn == NULL) { + return NULL; + } + + TAOS_RES* pRes = taos_query(pConn, "use db_dst"); + if (taos_errno(pRes) != 0) { + printf("error in use db_taosx, reason:%s\n", taos_errstr(pRes)); + return NULL; + } + taos_free_result(pRes); + return pConn; +} + +static void msg_process(TAOS_RES* msg) { + printf("-----------topic-------------: %s\n", tmq_get_topic_name(msg)); + printf("db: %s\n", tmq_get_db_name(msg)); + printf("vg: %d\n", tmq_get_vgroup_id(msg)); + TAOS* pConn = use_db(); + if (tmq_get_res_type(msg) == TMQ_RES_TABLE_META || tmq_get_res_type(msg) == TMQ_RES_METADATA) { + char* result = tmq_get_json_meta(msg); + printf("meta result: %s\n", result); + tmq_free_json_meta(result); + } + + tmq_raw_data raw = {0}; + tmq_get_raw(msg, &raw); + printf("write raw data type: %d\n", raw.raw_type); + int32_t ret = tmq_write_raw(pConn, raw); + printf("write raw data: %s\n", tmq_err2str(ret)); + ASSERT(ret == 0); + + tmq_free_raw(raw); + taos_close(pConn); +} + +int buildDatabase(TAOS* pConn, TAOS_RES* pRes) { + pRes = taos_query(pConn, + "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 " + "nchar(8), t4 bool)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query( + pConn, + "insert into ct3 using st1(t1) tags(3000) values(1626006833600, 5, 6, 'c') ct1 using st1(t1) tags(2000) values(1626006833601, 2, 3, 'sds') (1626006833602, 4, 5, " + "'ddd') ct0 using st1 tags(1000, \"ttt\", true) values(1626006833603, 4, 3, 'hwj') ct1 using st1(t1) tags(2000) values(now+5s, 23, 32, 's21ds')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query( + pConn, + "insert into ct3 values(1626006839602, 5, 6, 'c')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query( + pConn, + "insert into ct3 using st1(t1) tags(3000) values(1626006833602, 5, 6, 'c') ct1 values(1626006833611, 2, 3, 'sds') (1626006833612, 4, 5, " + "'ddd') ct10 using st1 tags(1000, \"ttt\", true) values(1626006833603, 4, 3, 'hwj')"); + if (taos_errno(pRes) != 0) { + printf("failed to insert into ct3, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + return 0; +} + +int32_t init_env() { + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + if (pConn == NULL) { + return -1; + } + + TAOS_RES* pRes = taos_query(pConn, "drop database if exists db_dst"); + if (taos_errno(pRes) != 0) { + printf("error in drop db_taosx, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create database if not exists db_dst vgroups 1 wal_retention_period 3600"); + if (taos_errno(pRes) != 0) { + printf("error in create db_taosx, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "drop topic if exists topic_db"); + if (taos_errno(pRes) != 0) { + printf("error in drop topic, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "drop database if exists db_src"); + if (taos_errno(pRes) != 0) { + printf("error in drop db, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create database if not exists db_src vgroups 1 wal_retention_period 3600"); + if (taos_errno(pRes) != 0) { + printf("error in create db, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "use db_src"); + if (taos_errno(pRes) != 0) { + printf("error in use db, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + buildDatabase(pConn, pRes); + + taos_close(pConn); + return 0; +} + +int32_t create_topic() { + printf("create topic\n"); + TAOS_RES* pRes; + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + if (pConn == NULL) { + return -1; + } + + pRes = taos_query(pConn, "create topic topic_db with meta as database db_src"); + if (taos_errno(pRes) != 0) { + printf("failed to create topic topic_db, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + taos_close(pConn); + return 0; +} + +void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { + printf("commit %d tmq %p param %p\n", code, tmq, param); +} + +tmq_t* build_consumer() { + tmq_conf_t* conf = tmq_conf_new(); + tmq_conf_set(conf, "group.id", "tg2"); + tmq_conf_set(conf, "client.id", "my app 1"); + tmq_conf_set(conf, "td.connect.user", "root"); + tmq_conf_set(conf, "td.connect.pass", "taosdata"); + tmq_conf_set(conf, "msg.with.table.name", "true"); + tmq_conf_set(conf, "enable.auto.commit", "true"); + tmq_conf_set(conf, "auto.offset.reset", "earliest"); + tmq_conf_set(conf, "msg.consume.rawdata", "1"); + + tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL); + tmq_t* tmq = tmq_consumer_new(conf, NULL, 0); + assert(tmq); + tmq_conf_destroy(conf); + return tmq; +} + +tmq_list_t* build_topic_list() { + tmq_list_t* topic_list = tmq_list_new(); + tmq_list_append(topic_list, "topic_db"); + return topic_list; +} + +void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) { + int32_t code; + + if ((code = tmq_subscribe(tmq, topics))) { + fprintf(stderr, "%% Failed to start consuming topics: %s\n", tmq_err2str(code)); + printf("subscribe err\n"); + return; + } + int32_t cnt = 0; + while (1) { + TAOS_RES* tmqmessage = tmq_consumer_poll(tmq, 5000); + if (tmqmessage) { + cnt++; + msg_process(tmqmessage); + taos_free_result(tmqmessage); + } else { + break; + } + } + + code = tmq_consumer_close(tmq); + if (code) + fprintf(stderr, "%% Failed to close consumer: %s\n", tmq_err2str(code)); + else + fprintf(stderr, "%% Consumer closed\n"); +} + +void check_result(){ + TAOS* taos = use_db(); + TAOS_RES *res = taos_query(taos, "select * from st1"); + int code = taos_errno(res); + assert(code == 0); + int affectedRows = taos_affected_rows(res); + printf("affected rows %d\n", affectedRows); + assert(affectedRows == 10); + taos_free_result(res); + taos_close(taos); + +} +int main(int argc, char* argv[]) { + if (init_env() < 0) { + return -1; + } + create_topic(); + + tmq_t* tmq = build_consumer(); + tmq_list_t* topic_list = build_topic_list(); + basic_consume_loop(tmq, topic_list); + tmq_list_destroy(topic_list); +}