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

first test case passed

This commit is contained in:
Bomin Zhang 2020-06-03 16:16:41 +08:00
parent 40c1d665b5
commit 81ea4a8a86
5 changed files with 166 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -404,6 +404,7 @@ TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port,
void *param, void **taos);
void waitForQueryRsp(void *param, TAOS_RES *tres, int code) ;
int doAsyncParseSql(SSqlObj* pSql, const char* sqlstr, size_t sqlLen);
void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, void (*fp)(), void *param, const char *sqlstr, size_t sqlLen);
void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql);

37
src/client/src/tscAsync.c
View File

@ -40,30 +40,23 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
static void tscAsyncFetchRowsProxy(void *param, TAOS_RES *tres, int numOfRows);
static void tscAsyncFetchSingleRowProxy(void *param, TAOS_RES *tres, int numOfRows);
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen) {
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
pSql->signature = pSql;
pSql->param = param;
pSql->pTscObj = pObj;
pSql->maxRetry = TSDB_MAX_REPLICA_NUM;
pSql->fp = fp;
sem_init(&pSql->rspSem, 0, 0);
if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) {
int doAsyncParseSql(SSqlObj* pSql, const char* sqlstr, size_t sqlLen) {
SSqlCmd* pCmd = &pSql->cmd;
SSqlRes* pRes = &pSql->res;
int32_t code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
if (code != TSDB_CODE_SUCCESS) {
tscError("failed to malloc payload");
tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
return;
tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_CLI_OUT_OF_MEMORY);
return code;
}
// todo check for OOM problem
pSql->sqlstr = calloc(1, sqlLen + 1);
if (pSql->sqlstr == NULL) {
tscError("%p failed to malloc sql string buffer", pSql);
tscQueueAsyncError(fp, param, TSDB_CODE_CLI_OUT_OF_MEMORY);
tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_CLI_OUT_OF_MEMORY);
free(pCmd->payload);
return;
return TSDB_CODE_CLI_OUT_OF_MEMORY;
}
pRes->qhandle = 0;
@ -72,7 +65,17 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const
strtolower(pSql->sqlstr, sqlstr);
tscDump("%p SQL: %s", pSql, pSql->sqlstr);
int32_t code = tsParseSql(pSql, true);
return tsParseSql(pSql, true);
}
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen) {
pSql->signature = pSql;
pSql->param = param;
pSql->pTscObj = pObj;
pSql->maxRetry = TSDB_MAX_REPLICA_NUM;
pSql->fp = fp;
int32_t code = doAsyncParseSql(pSql, sqlstr, sqlLen);
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
if (code != TSDB_CODE_SUCCESS) {

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

@ -497,46 +497,18 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
}
pSql->signature = pSql;
pSql->param = pSql;
pSql->pTscObj = pObj;
pSql->fp = asyncCallback;
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
int ret = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
if (TSDB_CODE_SUCCESS != ret) {
setErrorInfo(pSql, ret, NULL);
free(pSql);
return NULL;
}
pSql->sqlstr = strdup(sqlstr);
if (pSql->sqlstr == NULL) {
setErrorInfo(pSql, TSDB_CODE_CLI_OUT_OF_MEMORY, NULL);
tfree(pSql);
return NULL;
}
tsem_init(&pSql->rspSem, 0, 0);
SSqlInfo SQLInfo = {0};
tSQLParse(&SQLInfo, pSql->sqlstr);
tscResetSqlCmdObj(&pSql->cmd);
ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
if (TSDB_CODE_SUCCESS != ret) {
setErrorInfo(pSql, ret, NULL);
tscError("%p open stream failed, sql:%s, code:%d", pSql, sqlstr, TSDB_CODE_CLI_OUT_OF_MEMORY);
tscFreeSqlObj(pSql);
return NULL;
}
pSql->param = pSql;
pSql->fp = asyncCallback;
pRes->code = tscToSQLCmd(pSql, &SQLInfo);
if (pRes->code == TSDB_CODE_ACTION_IN_PROGRESS) {
int32_t code = doAsyncParseSql(pSql, sqlstr, strlen(sqlstr));
if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
sem_wait(&pSql->rspSem);
}
SQLInfoDestroy(&SQLInfo);
if (pRes->code != TSDB_CODE_SUCCESS) {
setErrorInfo(pSql, pRes->code, pCmd->payload);
@ -575,6 +547,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
pStream->stime = tscGetStreamStartTimestamp(pSql, pStream, stime);
int64_t starttime = tscGetLaunchTimestamp(pStream);
pCmd->command = TSDB_SQL_SELECT;
taosTmrReset(tscProcessStreamTimer, starttime, pStream, tscTmr, &pStream->pTimer);
tscTrace("%p stream:%p is opened, query on:%s, interval:%" PRId64 ", sliding:%" PRId64 ", first launched in:%" PRId64 ", sql:%s", pSql,

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

@ -75,7 +75,14 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
strcpy(pContext->user, pCfg->user);
strcpy(pContext->pass, pCfg->pass);
strcpy(pContext->db, pCfg->db);
const char* db = pCfg->db;
for (const char* p = db; *p != 0; p++) {
if (*p == '.') {
db = p + 1;
break;
}
}
strcpy(pContext->db, db);
pContext->vgId = pCfg->vgId;
pContext->cqWrite = pCfg->cqWrite;
pContext->ahandle = ahandle;

131
tests/pytest/stream/stream1.py Normal file
View File

@ -0,0 +1,131 @@
###################################################################
# 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):
tbNum = 10
rowNum = 20
tdSql.prepare()
tdLog.info("===== step1 =====")
tdSql.execute("create table stb0(ts timestamp, col1 int, col2 float) tags(tgcol int)")
for i in range(tbNum):
tdSql.execute("create table tb%d using stb0 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)
tdLog.info("===== step2 =====")
tdSql.query("select count(*), count(col1), count(col2) from tb0 interval(1d)")
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
tdSql.checkData(0, 3, rowNum)
tdSql.query("show tables")
tdSql.checkRows(tbNum)
tdSql.execute("create table s0 as select count(*), count(col1), count(col2) from tb0 interval(1d)")
tdSql.query("show tables")
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step3 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
tdSql.checkData(0, 3, rowNum)
tdLog.info("===== step4 =====")
tdSql.execute("drop table s0")
tdSql.query("show tables")
tdSql.checkRows(tbNum)
tdLog.info("===== step5 =====")
tdSql.error("select * from s0")
tdLog.info("===== step6 =====")
time.sleep(0.1)
tdSql.execute("create table s0 as select count(*), count(col1), count(col2) from tb0 interval(1d)")
tdSql.query("show tables")
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step7 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s0")
tdSql.checkData(0, 1, rowNum)
tdSql.checkData(0, 2, rowNum)
tdSql.checkData(0, 3, rowNum)
tdLog.info("===== step8 =====")
tdSql.query("select count(*), count(col1), count(col2) from stb0 interval(1d)")
tdSql.checkData(0, 1, rowNum * tbNum)
tdSql.checkData(0, 2, rowNum * tbNum)
tdSql.checkData(0, 3, rowNum * tbNum)
tdSql.query("show tables")
tdSql.checkRows(tbNum + 1)
tdSql.execute("create table s1 as select count(*), count(col1), count(col2) from stb0 interval(1d)")
tdSql.query("show tables")
tdSql.checkRows(tbNum + 2)
tdLog.info("===== step9 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.checkData(0, 1, rowNum * tbNum)
tdSql.checkData(0, 2, rowNum * tbNum)
tdSql.checkData(0, 3, rowNum * tbNum)
tdLog.info("===== step10 =====")
tdSql.execute("drop table s1")
tdSql.query("show tables")
tdSql.checkRows(tbNum + 1)
tdLog.info("===== step11 =====")
tdSql.error("select * from s1")
tdLog.info("===== step12 =====")
tdSql.execute("create table s1 as select count(*), count(col1), count(col2) from stb0 interval(1d)")
tdSql.query("show tables")
tdSql.checkRows(tbNum + 2)
tdLog.info("===== step13 =====")
tdLog.info("sleeping 120 seconds")
time.sleep(120)
tdSql.query("select * from s1")
tdSql.checkData(0, 1, rowNum * tbNum)
tdSql.checkData(0, 2, rowNum * tbNum)
tdSql.checkData(0, 3, rowNum * tbNum)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())