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

Merge branch '3.0' of https://github.com/taosdata/TDengine into enh/TS-5445-3.0

This commit is contained in:
Hongze Cheng 2025-02-24 09:49:31 +08:00
parent 93e17112e1 17b433a2ef
commit f8a0dc8fe0
65 changed files with 1636 additions and 532 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
docs/zh/14-reference/01-components/01-taosd.md
View File

@ -1441,7 +1441,7 @@ charset 的有效值是 UTF-8。
- 取值范围：float/double/none
- 默认值：none，表示关闭无损压缩
- 动态修改：不支持
- 支持版本：从 v3.3.0.0 前支持
- 支持版本：从v3.1.0.0 版本引入，v3.3.0.0 以后废弃
#### ifAdtFse
- 说明：在启用 TSZ 有损压缩时，使用 FSE 算法替换 HUFFMAN 算法，FSE 算法压缩速度更快，但解压稍慢，追求压缩速度可选用此算法
@ -1450,22 +1450,22 @@ charset 的有效值是 UTF-8。
- 最小值：0
- 最大值：1
- 动态修改：支持通过 SQL 修改，重启生效
- 支持版本：从 v3.1.0.0 版本开始引入
- 支持版本：从v3.1.0.0 版本引入，v3.3.0.0 以后废弃
#### maxRange
- 说明：用于有损压缩设置 `内部参数`
- 动态修改：支持通过 SQL 修改，重启生效
- 支持版本：从 v3.1.0.0 版本开始引入
- 支持版本：从v3.1.0.0 版本引入，v3.3.0.0 以后废弃
#### curRange
- 说明：用于有损压缩设置 `内部参数`
- 动态修改：支持通过 SQL 修改，重启生效
- 支持版本：从 v3.1.0.0 版本开始引入
- 支持版本：从v3.1.0.0 版本引入，v3.3.0.0 以后废弃
#### compressor
- 说明：用于有损压缩设置 `内部参数`
- 动态修改：支持通过 SQL 修改，重启生效
- 支持版本：从 v3.1.0.0 版本开始引入
- 支持版本：从v3.1.0.0 版本引入，v3.3.0.0 以后废弃
**补充说明**
1. 在 3.3.5.0 之后，所有配置参数都将被持久化到本地存储，重启数据库服务后，将默认使用持久化的配置参数列表；如果您希望继续使用 config 文件中配置的参数，需设置 forceReadConfig 为 1。

1
include/common/tmsgcb.h
View File

@ -38,6 +38,7 @@ typedef enum {
STREAM_QUEUE,
ARB_QUEUE,
STREAM_CTRL_QUEUE,
STREAM_LONG_EXEC_QUEUE,
QUEUE_MAX,
} EQueueType;

2
include/libs/executor/executor.h
View File

@ -183,7 +183,7 @@ void qCleanExecTaskBlockBuf(qTaskInfo_t tinfo);
*/
int32_t qAsyncKillTask(qTaskInfo_t tinfo, int32_t rspCode);
int32_t qKillTask(qTaskInfo_t tinfo, int32_t rspCode);
int32_t qKillTask(qTaskInfo_t tinfo, int32_t rspCode, int64_t waitDuration);
bool qTaskIsExecuting(qTaskInfo_t qinfo);

1
include/libs/executor/storageapi.h
View File

@ -361,6 +361,7 @@ typedef struct SStateStore {
bool (*streamStateCheck)(SStreamState* pState, const SWinKey* key);
int32_t (*streamStateGetByPos)(SStreamState* pState, void* pos, void** pVal);
void (*streamStateDel)(SStreamState* pState, const SWinKey* key);
void (*streamStateDelByGroupId)(SStreamState* pState, uint64_t groupId);
void (*streamStateClear)(SStreamState* pState);
void (*streamStateSetNumber)(SStreamState* pState, int32_t number, int32_t tsIdex);
void (*streamStateSaveInfo)(SStreamState* pState, void* pKey, int32_t keyLen, void* pVal, int32_t vLen);

1
include/libs/stream/streamState.h
View File

@ -44,6 +44,7 @@ int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, in
bool streamStateCheck(SStreamState* pState, const SWinKey* key);
int32_t streamStateGetByPos(SStreamState* pState, void* pos, void** pVal);
void streamStateDel(SStreamState* pState, const SWinKey* key);
void streamStateDelByGroupId(SStreamState* pState, uint64_t groupId);
void streamStateClear(SStreamState* pState);
void streamStateSetNumber(SStreamState* pState, int32_t number, int32_t tsIdex);
void streamStateSaveInfo(SStreamState* pState, void* pKey, int32_t keyLen, void* pVal, int32_t vLen);

10
include/libs/stream/tstream.h
View File

@ -58,6 +58,7 @@ extern "C" {
#define STREAM_EXEC_T_STOP_ALL_TASKS (-5)
#define STREAM_EXEC_T_RESUME_TASK (-6)
#define STREAM_EXEC_T_ADD_FAILED_TASK (-7)
#define STREAM_EXEC_T_STOP_ONE_TASK (-8)
typedef struct SStreamTask SStreamTask;
typedef struct SStreamQueue SStreamQueue;
@ -768,15 +769,19 @@ void streamMetaCleanup();
int32_t streamMetaOpen(const char* path, void* ahandle, FTaskBuild expandFunc, FTaskExpand expandTaskFn, int32_t vgId,
int64_t stage, startComplete_fn_t fn, SStreamMeta** pMeta);
void streamMetaClose(SStreamMeta* streamMeta);
int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask); // save to stream meta store
int32_t streamMetaRemoveTask(SStreamMeta* pMeta, STaskId* pKey);
int32_t streamMetaSaveTaskInMeta(SStreamMeta* pMeta, SStreamTask* pTask); // save to stream meta store
int32_t streamMetaRemoveTaskInMeta(SStreamMeta* pMeta, STaskId* pKey);
int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTask, bool* pAdded);
int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId);
int32_t streamMetaGetNumOfTasks(SStreamMeta* pMeta);
int32_t streamMetaAcquireTaskNoLock(SStreamMeta* pMeta, int64_t streamId, int32_t taskId, SStreamTask** pTask);
int32_t streamMetaAcquireTaskUnsafe(SStreamMeta* pMeta, STaskId* pId, SStreamTask** pTask);
int32_t streamMetaAcquireTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId, SStreamTask** pTask);
void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask);
void streamMetaClear(SStreamMeta* pMeta);
void streamMetaInitBackend(SStreamMeta* pMeta);
int32_t streamMetaCommit(SStreamMeta* pMeta);
@ -810,6 +815,7 @@ void streamMetaLoadAllTasks(SStreamMeta* pMeta);
int32_t streamMetaStartAllTasks(SStreamMeta* pMeta);
int32_t streamMetaStopAllTasks(SStreamMeta* pMeta);
int32_t streamMetaStartOneTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId);
int32_t streamMetaStopOneTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId);
bool streamMetaAllTasksReady(const SStreamMeta* pMeta);
int32_t streamTaskSendNegotiateChkptIdMsg(SStreamTask* pTask);
int32_t streamTaskCheckIfReqConsenChkptId(SStreamTask* pTask, int64_t ts);

1
include/libs/stream/tstreamFileState.h
View File

@ -59,6 +59,7 @@ int32_t addRowBuffIfNotExist(SStreamFileState* pFileState, void* pKey, int32_t k
int32_t getRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen,
int32_t* pWinCode);
void deleteRowBuff(SStreamFileState* pFileState, const void* pKey, int32_t keyLen);
void deleteRowBuffByGroupId(SStreamFileState* pFileState, uint64_t groupId);
int32_t getRowBuffByPos(SStreamFileState* pFileState, SRowBuffPos* pPos, void** pVal);
bool hasRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen);
int32_t putFreeBuff(SStreamFileState* pFileState, SRowBuffPos* pPos);

3
include/os/osFile.h
View File

@ -129,7 +129,8 @@ size_t taosWriteToCFile(const void *ptr, size_t size, size_t nitems, FILE *strea
int taosCloseCFile(FILE *);
int taosSetAutoDelFile(char *path);
bool lastErrorIsFileNotExist();
FILE *taosOpenFileForStream(const char *path, int32_t tdFileOptions);
bool lastErrorIsFileNotExist();
#ifdef BUILD_WITH_RAND_ERR
#define STUB_RAND_NETWORK_ERR(ret) \

2
include/util/tworker.h
View File

@ -76,7 +76,7 @@ void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue);
int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool);
void tAutoQWorkerCleanup(SAutoQWorkerPool *pool);
STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem fp);
STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem fp, int32_t minNum);
void tAutoQWorkerFreeQueue(SAutoQWorkerPool *pool, STaosQueue *queue);
int32_t tWWorkerInit(SWWorkerPool *pool);

83
source/common/test/CMakeLists.txt
View File

@ -3,37 +3,28 @@ MESSAGE(STATUS "build parser unit test")
# GoogleTest requires at least C++11
SET(CMAKE_CXX_STANDARD 11)
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
ADD_EXECUTABLE(commonTest "")
TARGET_SOURCES(
commonTest
PRIVATE
"commonTests.cpp"
)
TARGET_LINK_LIBRARIES(
commonTest
PUBLIC os util common gtest
)
TARGET_INCLUDE_DIRECTORIES(
if(TD_LINUX)
add_executable(commonTest "commonTests.cpp")
target_link_libraries(commonTest os util common gtest_main)
target_include_directories(
commonTest
PUBLIC "${TD_SOURCE_DIR}/include/libs/common/"
PRIVATE "${TD_SOURCE_DIR}/source/libs/common/inc"
)
)
add_test(
NAME commonTest
COMMAND commonTest
)
endif()
# dataformatTest.cpp
add_executable(dataformatTest "")
target_sources(
dataformatTest
PRIVATE
"dataformatTest.cpp"
)
# dataformatTest.cpp
add_executable(dataformatTest "dataformatTest.cpp")
target_link_libraries(dataformatTest gtest gtest_main util common)
target_include_directories(
dataformatTest
PUBLIC "${TD_SOURCE_DIR}/include/common"
PUBLIC "${TD_SOURCE_DIR}/include/util"
dataformatTest
PUBLIC "${TD_SOURCE_DIR}/include/common"
PUBLIC "${TD_SOURCE_DIR}/include/util"
)
add_test(
NAME dataformatTest
@ -41,17 +32,12 @@ add_test(
)
# cosCpTest.cpp
add_executable(cosCpTest "")
target_sources(
cosCpTest
PRIVATE
"cosCpTest.cpp"
)
add_executable(cosCpTest "cosCpTest.cpp")
target_link_libraries(cosCpTest gtest gtest_main util common)
target_include_directories(
cosCpTest
PUBLIC "${TD_SOURCE_DIR}/include/common"
PUBLIC "${TD_SOURCE_DIR}/include/util"
cosCpTest
PUBLIC "${TD_SOURCE_DIR}/include/common"
PUBLIC "${TD_SOURCE_DIR}/include/util"
)
add_test(
NAME cosCpTest
@ -59,31 +45,24 @@ add_test(
)
if(TD_LINUX)
# cosTest.cpp
add_executable(cosTest "")
target_sources(
cosTest
PRIVATE
"cosTest.cpp"
)
target_link_libraries(cosTest gtest gtest_main util common)
target_include_directories(
# cosTest.cpp
add_executable(cosTest "cosTest.cpp")
target_link_libraries(cosTest gtest gtest_main util common)
target_include_directories(
cosTest
PUBLIC "${TD_SOURCE_DIR}/include/common"
PUBLIC "${TD_SOURCE_DIR}/include/util"
)
add_test(
NAME cosTest
COMMAND cosTest
)
)
add_test(
NAME cosTest
COMMAND cosTest
)
endif()
if (${TD_LINUX})
if(${TD_LINUX})
# tmsg test
add_executable(tmsgTest "")
target_sources(tmsgTest
target_sources(tmsgTest
PRIVATE
"tmsgTest.cpp"
"../src/msg/tmsg.c"
@ -100,4 +79,4 @@ if (${TD_LINUX})
add_custom_command(TARGET tmsgTest POST_BUILD
COMMAND ${CMAKE_COMMAND} -E copy_if_different ${MSG_TBL_FILE} $<TARGET_FILE_DIR:tmsgTest>
)
endif ()
endif()

91
source/common/test/commonTests.cpp
View File

@ -235,9 +235,60 @@ TEST(testCase, toInteger_test) {
ASSERT_EQ(ret, -1);
}
TEST(testCase, Datablock_test_inc) {
{
SColumnInfoData cinfo = {0};
uint32_t row = 0;
bool ret = colDataIsNull_s(&cinfo, row);
EXPECT_EQ(ret, false);
cinfo.hasNull = 1;
cinfo.info.type = TSDB_DATA_TYPE_INT;
ret = colDataIsNull_s(&cinfo, row);
EXPECT_EQ(ret, false);
}
{
SColumnInfoData cinfo = {0};
uint32_t row = 0;
bool isVarType = false;
bool ret = colDataIsNull_t(&cinfo, row, isVarType);
EXPECT_EQ(ret, false);
cinfo.hasNull = 1;
ret = colDataIsNull_t(&cinfo, row, isVarType);
EXPECT_EQ(ret, false);
}
{
SColumnInfoData cinfo = {0};
uint32_t totalRows = 0;
uint32_t row = 0;
SColumnDataAgg colAgg = {0};
bool ret = colDataIsNull(&cinfo, totalRows, row, &colAgg);
EXPECT_EQ(ret, false);
cinfo.hasNull = 1;
ret = colDataIsNull(&cinfo, totalRows, row, &colAgg);
EXPECT_EQ(ret, true);
totalRows = 1;
ret = colDataIsNull(&cinfo, totalRows, row, &colAgg);
EXPECT_EQ(ret, false);
colAgg.colId = -1;
cinfo.info.type = TSDB_DATA_TYPE_INT;
ret = colDataIsNull(&cinfo, totalRows, row, &colAgg);
EXPECT_EQ(ret, false);
}
}
TEST(testCase, Datablock_test) {
SSDataBlock* b = NULL;
int32_t code = createDataBlock(&b);
int32_t code = createDataBlock(&b);
ASSERT(code == 0);
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 1);
@ -288,7 +339,7 @@ TEST(testCase, Datablock_test) {
printf("binary column length:%d\n", *(int32_t*)p1->pData);
ASSERT_EQ(blockDataGetNumOfCols(b), 2);
ASSERT_EQ(blockDataGetNumOfCols(b), 3);
ASSERT_EQ(blockDataGetNumOfRows(b), 40);
char* pData = colDataGetData(p1, 3);
@ -364,7 +415,7 @@ TEST(testCase, var_dataBlock_split_test) {
int32_t numOfRows = 1000000;
SSDataBlock* b = NULL;
int32_t code = createDataBlock(&b);
int32_t code = createDataBlock(&b);
ASSERT(code == 0);
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 1);
@ -468,7 +519,7 @@ TEST(timeTest, timestamp2tm) {
}
void test_ts2char(int64_t ts, const char* format, int32_t precison, const char* expected) {
char buf[256] = {0};
char buf[256] = {0};
int32_t code = TEST_ts2char(format, ts, precison, buf, 256);
ASSERT_EQ(code, 0);
printf("ts: %ld format: %s res: [%s], expected: [%s]\n", ts, format, buf, expected);
@ -639,12 +690,14 @@ TEST(timeTest, char2ts) {
ASSERT_EQ(-2, TEST_char2ts("yyyyMM/dd ", &ts, TSDB_TIME_PRECISION_MICRO, "210011/32"));
ASSERT_EQ(-1, TEST_char2ts("HH12:MI:SS", &ts, TSDB_TIME_PRECISION_MICRO, "21:12:12"));
ASSERT_EQ(-1, TEST_char2ts("yyyy/MM1/dd ", &ts, TSDB_TIME_PRECISION_MICRO, "2100111111111/11/2"));
ASSERT_EQ(-2, TEST_char2ts("yyyy/MM1/ddTZH", &ts, TSDB_TIME_PRECISION_MICRO, "23/11/2-13"));
TEST_char2ts("yyyy/MM1/ddTZH", &ts, TSDB_TIME_PRECISION_MICRO, "23/11/2-13");
// ASSERT_EQ(-2, TEST_char2ts("yyyy/MM1/ddTZH", &ts, TSDB_TIME_PRECISION_MICRO, "23/11/2-13"));
ASSERT_EQ(0, TEST_char2ts("yyyy年 MM/ddTZH", &ts, TSDB_TIME_PRECISION_MICRO, "1970年1/1+0"));
ASSERT_EQ(ts, 0);
// ASSERT_EQ(ts, 0);
ASSERT_EQ(-1, TEST_char2ts("yyyy年a MM/dd", &ts, TSDB_TIME_PRECISION_MICRO, "2023年1/2"));
ASSERT_EQ(0, TEST_char2ts("yyyy年 MM/ddTZH", &ts, TSDB_TIME_PRECISION_MICRO, "1970年 1/1+0"));
ASSERT_EQ(ts, 0);
// ASSERT_EQ(ts, 0);
ASSERT_EQ(0, TEST_char2ts("yyyy年 a a a MM/ddTZH", &ts, TSDB_TIME_PRECISION_MICRO, "1970年 a a a 1/1+0"));
ASSERT_EQ(0, TEST_char2ts("yyyy年 a a a a a a a a a a a a a a a MM/ddTZH", &ts, TSDB_TIME_PRECISION_MICRO,
"1970年 a "));
@ -701,33 +754,33 @@ TEST(timeTest, epSet) {
// Define test cases
TEST(AlreadyAddGroupIdTest, GroupIdAdded) {
// Test case 1: Group ID has been added
char ctbName[64] = "abc123";
char ctbName[64] = "abc123";
int64_t groupId = 123;
bool result = alreadyAddGroupId(ctbName, groupId);
bool result = alreadyAddGroupId(ctbName, groupId);
EXPECT_TRUE(result);
}
TEST(AlreadyAddGroupIdTest, GroupIdNotAdded) {
// Test case 2: Group ID has not been added
char ctbName[64] = "abc456";
char ctbName[64] = "abc456";
int64_t groupId = 123;
bool result = alreadyAddGroupId(ctbName, groupId);
bool result = alreadyAddGroupId(ctbName, groupId);
EXPECT_FALSE(result);
}
TEST(AlreadyAddGroupIdTest, GroupIdAddedAtTheEnd) {
// Test case 3: Group ID has been added at the end
char ctbName[64] = "xyz1";
char ctbName[64] = "xyz1";
int64_t groupId = 1;
bool result = alreadyAddGroupId(ctbName, groupId);
bool result = alreadyAddGroupId(ctbName, groupId);
EXPECT_TRUE(result);
}
TEST(AlreadyAddGroupIdTest, GroupIdAddedWithDifferentLength) {
// Test case 4: Group ID has been added with different length
char ctbName[64] = "def";
char ctbName[64] = "def";
int64_t groupId = 123456;
bool result = alreadyAddGroupId(ctbName, groupId);
bool result = alreadyAddGroupId(ctbName, groupId);
EXPECT_FALSE(result);
}
@ -746,8 +799,8 @@ static int32_t taosSetSlowLogScope(char* pScopeStr, int32_t* pScope) {
int32_t slowScope = 0;
char* scope = NULL;
char *tmp = NULL;
while((scope = strsep(&pScopeStr, "|")) != NULL){
char* tmp = NULL;
while ((scope = strsep(&pScopeStr, "|")) != NULL) {
taosMemoryFreeClear(tmp);
tmp = taosStrdup(scope);
strtrim(tmp);
@ -847,8 +900,8 @@ TEST(TaosSetSlowLogScopeTest, InvalidScopeInput) {
char pScopeStr[] = "invalid";
int32_t scope = 0;
int32_t result = taosSetSlowLogScope(pScopeStr, &scope);
EXPECT_EQ(result, TSDB_CODE_SUCCESS);
EXPECT_EQ(scope, -1);
// EXPECT_EQ(result, TSDB_CODE_SUCCESS);
// EXPECT_EQ(scope, -1);
}
TEST(TaosSetSlowLogScopeTest, MixedScopesInput) {

4
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
View File

@ -32,6 +32,7 @@ typedef struct SVnodeMgmt {
const char *name;
SQueryAutoQWorkerPool queryPool;
SAutoQWorkerPool streamPool;
SAutoQWorkerPool streamLongExecPool;
SWWorkerPool streamCtrlPool;
SWWorkerPool fetchPool;
SSingleWorker mgmtWorker;
@ -75,6 +76,7 @@ typedef struct {
STaosQueue *pQueryQ;
STaosQueue *pStreamQ;
STaosQueue *pStreamCtrlQ;
STaosQueue *pStreamLongExecQ;
STaosQueue *pFetchQ;
STaosQueue *pMultiMgmQ;
} SVnodeObj;
@ -137,6 +139,8 @@ int32_t vmPutMsgToQueryQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToFetchQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToStreamQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToStreamCtrlQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToStreamLongExecQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToMergeQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
int32_t vmPutMsgToMultiMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);

18
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
View File

@ -1008,27 +1008,29 @@ SArray *vmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_CHECK, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_CHECK_RSP, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_PAUSE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RESUME, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_STOP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_CHECK_POINT_SOURCE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECKPOINT_READY, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECKPOINT_READY_RSP, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_TRIGGER, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_TRIGGER_RSP, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_UPDATE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_RESET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_STREAM_HEARTBEAT_RSP, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_STREAM_REQ_CHKPT_RSP, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_STREAM_CHKPT_REPORT_RSP, vmPutMsgToStreamCtrlQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_SCAN_HISTORY, vmPutMsgToStreamLongExecQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_GET_STREAM_PROGRESS, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_UPDATE_CHKPT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_CONSEN_CHKPT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_PAUSE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RESUME, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_STOP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_CHECK_POINT_SOURCE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_UPDATE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TASK_RESET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

8
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
View File

@ -398,10 +398,14 @@ void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode, bool commitAndRemoveWal,
dInfo("vgId:%d, wait for vnode stream queue:%p is empty, %d remains", pVnode->vgId,
pVnode->pStreamQ, taosQueueItemSize(pVnode->pStreamQ));
while (!taosQueueEmpty(pVnode->pStreamQ)) taosMsleep(10);
while (!taosQueueEmpty(pVnode->pStreamQ)) taosMsleep(50);
dInfo("vgId:%d, wait for vnode stream ctrl queue:%p is empty", pVnode->vgId, pVnode->pStreamCtrlQ);
while (!taosQueueEmpty(pVnode->pStreamCtrlQ)) taosMsleep(10);
while (!taosQueueEmpty(pVnode->pStreamCtrlQ)) taosMsleep(50);
dInfo("vgId:%d, wait for vnode stream long-exec queue:%p is empty, %d remains", pVnode->vgId,
pVnode->pStreamLongExecQ, taosQueueItemSize(pVnode->pStreamLongExecQ));
while (!taosQueueEmpty(pVnode->pStreamLongExecQ)) taosMsleep(50);
dInfo("vgId:%d, all vnode queues is empty", pVnode->vgId);

59
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -150,7 +150,7 @@ static void vmProcessStreamCtrlQueue(SQueueInfo *pInfo, STaosQall* pQall, int32_
SRpcMsg *pMsg = pItem;
const STraceId *trace = &pMsg->info.traceId;
dGTrace("vgId:%d, msg:%p get from vnode-ctrl-stream queue", pVnode->vgId, pMsg);
dGTrace("vgId:%d, msg:%p get from vnode-stream-ctrl queue", pVnode->vgId, pMsg);
code = vnodeProcessStreamCtrlMsg(pVnode->pImpl, pMsg, pInfo);
if (code != 0) {
terrno = code;
@ -165,6 +165,26 @@ static void vmProcessStreamCtrlQueue(SQueueInfo *pInfo, STaosQall* pQall, int32_
}
}
static void vmProcessStreamLongExecQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
SVnodeObj *pVnode = pInfo->ahandle;
const STraceId *trace = &pMsg->info.traceId;
int32_t code = 0;
dGTrace("vgId:%d, msg:%p get from vnode-stream long-exec queue", pVnode->vgId, pMsg);
code = vnodeProcessStreamLongExecMsg(pVnode->pImpl, pMsg, pInfo);
if (code != 0) {
terrno = code;
dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),
tstrerror(code));
vmSendRsp(pMsg, code);
}
dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);
rpcFreeCont(pMsg->pCont);
taosFreeQitem(pMsg);
}
static void vmProcessFetchQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
SVnodeObj *pVnode = pInfo->ahandle;
SRpcMsg *pMsg = NULL;
@ -274,9 +294,13 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
code = taosWriteQitem(pVnode->pStreamQ, pMsg);
break;
case STREAM_CTRL_QUEUE:
dGTrace("vgId:%d, msg:%p put into vnode-ctrl-stream queue", pVnode->vgId, pMsg);
dGTrace("vgId:%d, msg:%p put into vnode-stream-ctrl queue", pVnode->vgId, pMsg);
code = taosWriteQitem(pVnode->pStreamCtrlQ, pMsg);
break;
case STREAM_LONG_EXEC_QUEUE:
dGTrace("vgId:%d, msg:%p put into vnode-stream-long-exec queue", pVnode->vgId, pMsg);
code = taosWriteQitem(pVnode->pStreamLongExecQ, pMsg);
break;
case FETCH_QUEUE:
dGTrace("vgId:%d, msg:%p put into vnode-fetch queue", pVnode->vgId, pMsg);
code = taosWriteQitem(pVnode->pFetchQ, pMsg);
@ -335,6 +359,8 @@ int32_t vmPutMsgToStreamQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMs
int32_t vmPutMsgToStreamCtrlQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_CTRL_QUEUE); }
int32_t vmPutMsgToStreamLongExecQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { return vmPutMsgToQueue(pMgmt, pMsg, STREAM_LONG_EXEC_QUEUE); }
int32_t vmPutMsgToMultiMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
const STraceId *trace = &pMsg->info.traceId;
dGTrace("msg:%p, put into vnode-multi-mgmt queue", pMsg);
@ -409,6 +435,10 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {
break;
case STREAM_CTRL_QUEUE:
size = taosQueueItemSize(pVnode->pStreamCtrlQ);
break;
case STREAM_LONG_EXEC_QUEUE:
size = taosQueueItemSize(pVnode->pStreamLongExecQ);
break;
default:
break;
}
@ -451,13 +481,16 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
}
pVnode->pQueryQ = tQueryAutoQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue);
pVnode->pStreamQ = tAutoQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue);
pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue);
// init stream msg processing queue family
pVnode->pStreamQ = tAutoQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue, 2);
pVnode->pStreamCtrlQ = tWWorkerAllocQueue(&pMgmt->streamCtrlPool, pVnode, (FItems)vmProcessStreamCtrlQueue);
pVnode->pStreamLongExecQ = tAutoQWorkerAllocQueue(&pMgmt->streamLongExecPool, pVnode, (FItem)vmProcessStreamLongExecQueue, 1);
if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncRdW.queue == NULL ||
pVnode->pApplyW.queue == NULL || pVnode->pQueryQ == NULL || pVnode->pStreamQ == NULL || pVnode->pFetchQ == NULL
|| pVnode->pStreamCtrlQ == NULL) {
|| pVnode->pStreamCtrlQ == NULL || pVnode->pStreamLongExecQ == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
@ -473,6 +506,7 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
dInfo("vgId:%d, fetch-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pFetchQ,
taosQueueGetThreadId(pVnode->pFetchQ));
dInfo("vgId:%d, stream-queue:%p is alloced", pVnode->vgId, pVnode->pStreamQ);
dInfo("vgId:%d, stream-long-exec-queue:%p is alloced", pVnode->vgId, pVnode->pStreamLongExecQ);
dInfo("vgId:%d, stream-ctrl-queue:%p is alloced, thread:%08" PRId64, pVnode->vgId, pVnode->pStreamCtrlQ,
taosQueueGetThreadId(pVnode->pStreamCtrlQ));
return 0;
@ -481,17 +515,22 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
tQueryAutoQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ);
tAutoQWorkerFreeQueue(&pMgmt->streamPool, pVnode->pStreamQ);
tAutoQWorkerFreeQueue(&pMgmt->streamLongExecPool, pVnode->pStreamLongExecQ);
tWWorkerFreeQueue(&pMgmt->streamCtrlPool, pVnode->pStreamCtrlQ);
tWWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ);
pVnode->pQueryQ = NULL;
pVnode->pFetchQ = NULL;
pVnode->pStreamQ = NULL;
pVnode->pStreamCtrlQ = NULL;
pVnode->pFetchQ = NULL;
pVnode->pStreamLongExecQ = NULL;
dDebug("vgId:%d, queue is freed", pVnode->vgId);
}
int32_t vmStartWorker(SVnodeMgmt *pMgmt) {
int32_t code = 0;
int32_t code = 0;
SQueryAutoQWorkerPool *pQPool = &pMgmt->queryPool;
pQPool->name = "vnode-query";
pQPool->min = tsNumOfVnodeQueryThreads;
@ -505,8 +544,13 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) {
pStreamPool->ratio = tsRatioOfVnodeStreamThreads;
if ((code = tAutoQWorkerInit(pStreamPool)) != 0) return code;
SAutoQWorkerPool *pLongExecPool = &pMgmt->streamLongExecPool;
pLongExecPool->name = "vnode-stream-long-exec";
pLongExecPool->ratio = tsRatioOfVnodeStreamThreads/3;
if ((code = tAutoQWorkerInit(pLongExecPool)) != 0) return code;
SWWorkerPool *pStreamCtrlPool = &pMgmt->streamCtrlPool;
pStreamCtrlPool->name = "vnode-ctrl-stream";
pStreamCtrlPool->name = "vnode-stream-ctrl";
pStreamCtrlPool->max = 1;
if ((code = tWWorkerInit(pStreamCtrlPool)) != 0) return code;
@ -541,6 +585,7 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) {
void vmStopWorker(SVnodeMgmt *pMgmt) {
tQueryAutoQWorkerCleanup(&pMgmt->queryPool);
tAutoQWorkerCleanup(&pMgmt->streamPool);
tAutoQWorkerCleanup(&pMgmt->streamLongExecPool);
tWWorkerCleanup(&pMgmt->streamCtrlPool);
tWWorkerCleanup(&pMgmt->fetchPool);
dDebug("vnode workers are closed");

1
source/dnode/snode/src/snodeInitApi.c
View File

@ -43,6 +43,7 @@ void initStateStoreAPI(SStateStore* pStore) {
pStore->streamStateCheck = streamStateCheck;
pStore->streamStateGetByPos = streamStateGetByPos;
pStore->streamStateDel = streamStateDel;
pStore->streamStateDelByGroupId = streamStateDelByGroupId;
pStore->streamStateClear = streamStateClear;
pStore->streamStateSaveInfo = streamStateSaveInfo;
pStore->streamStateGetInfo = streamStateGetInfo;

1
source/dnode/vnode/inc/vnode.h
View File

@ -113,6 +113,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
int32_t vnodeProcessStreamMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
int32_t vnodeProcessStreamCtrlMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
int32_t vnodeProcessStreamLongExecMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs);
void vnodeApplyWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs);
void vnodeProposeCommitOnNeed(SVnode *pVnode, bool atExit);

2
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -1302,7 +1302,7 @@ _checkpoint:
}
streamMetaWLock(pMeta);
if ((code = streamMetaSaveTask(pMeta, pTask)) != 0) {
if ((code = streamMetaSaveTaskInMeta(pMeta, pTask)) != 0) {
streamMetaWUnLock(pMeta);
taosHashCancelIterate(pInfoHash, infoHash);
TSDB_CHECK_CODE(code, lino, _exit);

9
source/dnode/vnode/src/tqCommon/tqCommon.c
View File

@ -268,13 +268,13 @@ int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pM
// stream do update the nodeEp info, write it into stream meta.
if (updated) {
tqDebug("s-task:%s vgId:%d save task after update epset, and stop task", idstr, vgId);
code = streamMetaSaveTask(pMeta, pTask);
code = streamMetaSaveTaskInMeta(pMeta, pTask);
if (code) {
tqError("s-task:%s vgId:%d failed to save task, code:%s", idstr, vgId, tstrerror(code));
}
if (pHTask != NULL) {
code = streamMetaSaveTask(pMeta, pHTask);
code = streamMetaSaveTaskInMeta(pMeta, pHTask);
if (code) {
tqError("s-task:%s vgId:%d failed to save related history task, code:%s", idstr, vgId, tstrerror(code));
}
@ -751,6 +751,8 @@ int32_t tqStreamTaskProcessDropReq(SStreamMeta* pMeta, char* msg, int32_t msgLen
}
streamMetaWUnLock(pMeta);
tqDebug("vgId:%d process drop task:0x%x completed", vgId, pReq->taskId);
return 0; // always return success
}
@ -865,6 +867,9 @@ int32_t tqStreamTaskProcessRunReq(SStreamMeta* pMeta, SRpcMsg* pMsg, bool isLead
} else if (type == STREAM_EXEC_T_ADD_FAILED_TASK) {
code = streamMetaAddFailedTask(pMeta, req.streamId, req.taskId);
return code;
} else if (type == STREAM_EXEC_T_STOP_ONE_TASK) {
code = streamMetaStopOneTask(pMeta, req.streamId, req.taskId);
return code;
} else if (type == STREAM_EXEC_T_RESUME_TASK) { // task resume to run after idle for a while
SStreamTask* pTask = NULL;
code = streamMetaAcquireTask(pMeta, req.streamId, req.taskId, &pTask);

1
source/dnode/vnode/src/vnd/vnodeInitApi.c
View File

@ -166,6 +166,7 @@ void initStateStoreAPI(SStateStore* pStore) {
pStore->streamStateCheck = streamStateCheck;
pStore->streamStateGetByPos = streamStateGetByPos;
pStore->streamStateDel = streamStateDel;
pStore->streamStateDelByGroupId = streamStateDelByGroupId;
pStore->streamStateClear = streamStateClear;
pStore->streamStateSaveInfo = streamStateSaveInfo;
pStore->streamStateGetInfo = streamStateGetInfo;

22
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -934,9 +934,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
int32_t vnodeProcessStreamMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
vTrace("vgId:%d, msg:%p in stream queue is processing", pVnode->config.vgId, pMsg);
if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG ||
pMsg->msgType == TDMT_VND_BATCH_META) &&
!syncIsReadyForRead(pVnode->sync)) {
if (!syncIsReadyForRead(pVnode->sync)) {
vnodeRedirectRpcMsg(pVnode, pMsg, terrno);
return 0;
}
@ -948,8 +946,6 @@ int32_t vnodeProcessStreamMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo)
return tqProcessTaskRetrieveReq(pVnode->pTq, pMsg);
case TDMT_STREAM_RETRIEVE_RSP:
return tqProcessTaskRetrieveRsp(pVnode->pTq, pMsg);
case TDMT_VND_STREAM_SCAN_HISTORY:
return tqProcessTaskScanHistory(pVnode->pTq, pMsg);
case TDMT_VND_GET_STREAM_PROGRESS:
return tqStreamProgressRetrieveReq(pVnode->pTq, pMsg);
default:
@ -996,6 +992,22 @@ int32_t vnodeProcessStreamCtrlMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pIn
}
}
int32_t vnodeProcessStreamLongExecMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
vTrace("vgId:%d, msg:%p in stream long exec queue is processing", pVnode->config.vgId, pMsg);
if (!syncIsReadyForRead(pVnode->sync)) {
vnodeRedirectRpcMsg(pVnode, pMsg, terrno);
return 0;
}
switch (pMsg->msgType) {
case TDMT_VND_STREAM_SCAN_HISTORY:
return tqProcessTaskScanHistory(pVnode->pTq, pMsg);
default:
vError("unknown msg type:%d in stream long exec queue", pMsg->msgType);
return TSDB_CODE_APP_ERROR;
}
}
void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
int32_t code = tdProcessTSmaInsert(((SVnode *)pVnode)->pSma, smaId, (const char *)data);
if (code) {

41
source/libs/executor/src/executor.c
View File

@ -631,7 +631,7 @@ void qUpdateOperatorParam(qTaskInfo_t tinfo, void* pParam) {
}
int32_t qExecutorInit(void) {
taosThreadOnce(&initPoolOnce, initRefPool);
(void)taosThreadOnce(&initPoolOnce, initRefPool);
return TSDB_CODE_SUCCESS;
}
@ -995,26 +995,43 @@ int32_t qAsyncKillTask(qTaskInfo_t qinfo, int32_t rspCode) {
return TSDB_CODE_SUCCESS;
}
int32_t qKillTask(qTaskInfo_t tinfo, int32_t rspCode) {
int32_t qKillTask(qTaskInfo_t tinfo, int32_t rspCode, int64_t waitDuration) {
int64_t st = taosGetTimestampMs();
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
if (pTaskInfo == NULL) {
return TSDB_CODE_QRY_INVALID_QHANDLE;
}
qDebug("%s sync killed execTask", GET_TASKID(pTaskInfo));
if (waitDuration > 0) {
qDebug("%s sync killed execTask, and waiting for %.2fs", GET_TASKID(pTaskInfo), waitDuration/1000.0);
} else {
qDebug("%s async killed execTask", GET_TASKID(pTaskInfo));
}
setTaskKilled(pTaskInfo, TSDB_CODE_TSC_QUERY_KILLED);
while (1) {
taosWLockLatch(&pTaskInfo->lock);
if (qTaskIsExecuting(pTaskInfo)) { // let's wait for 100 ms and try again
taosWUnLockLatch(&pTaskInfo->lock);
taosMsleep(100);
} else { // not running now
pTaskInfo->code = rspCode;
taosWUnLockLatch(&pTaskInfo->lock);
return TSDB_CODE_SUCCESS;
if (waitDuration > 0) {
while (1) {
taosWLockLatch(&pTaskInfo->lock);
if (qTaskIsExecuting(pTaskInfo)) { // let's wait for 100 ms and try again
taosWUnLockLatch(&pTaskInfo->lock);
taosMsleep(200);
int64_t d = taosGetTimestampMs() - st;
if (d >= waitDuration && waitDuration >= 0) {
qWarn("%s waiting more than %.2fs, not wait anymore", GET_TASKID(pTaskInfo), waitDuration / 1000.0);
return TSDB_CODE_SUCCESS;
}
} else { // not running now
pTaskInfo->code = rspCode;
taosWUnLockLatch(&pTaskInfo->lock);
return TSDB_CODE_SUCCESS;
}
}
}
return TSDB_CODE_SUCCESS;
}
bool qTaskIsExecuting(qTaskInfo_t qinfo) {

8
source/libs/executor/src/projectoperator.c
View File

@ -42,9 +42,7 @@ typedef struct SIndefOperatorInfo {
} SIndefOperatorInfo;
static int32_t doGenerateSourceData(SOperatorInfo* pOperator);
static SSDataBlock* doProjectOperation1(SOperatorInfo* pOperator);
static int32_t doProjectOperation(SOperatorInfo* pOperator, SSDataBlock** pResBlock);
static SSDataBlock* doApplyIndefinitFunction1(SOperatorInfo* pOperator);
static int32_t doApplyIndefinitFunction(SOperatorInfo* pOperator, SSDataBlock** pResBlock);
static int32_t setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols, SArray** pResList);
static int32_t setFunctionResultOutput(SOperatorInfo* pOperator, SOptrBasicInfo* pInfo, SAggSupporter* pSup,
@ -557,12 +555,6 @@ static void doHandleDataBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, SOp
}
}
SSDataBlock* doApplyIndefinitFunction1(SOperatorInfo* pOperator) {
SSDataBlock* pResBlock = NULL;
pOperator->pTaskInfo->code = doApplyIndefinitFunction(pOperator, &pResBlock);
return pResBlock;
}
int32_t doApplyIndefinitFunction(SOperatorInfo* pOperator, SSDataBlock** pResBlock) {
QRY_PARAM_CHECK(pResBlock);
SIndefOperatorInfo* pIndefInfo = pOperator->info;

6
source/libs/executor/src/scanoperator.c
View File

@ -3818,7 +3818,11 @@ FETCH_NEXT_BLOCK:
int32_t deleteNum = 0;
code = deletePartName(pInfo, pBlock, &deleteNum);
QUERY_CHECK_CODE(code, lino, _end);
if (deleteNum == 0) goto FETCH_NEXT_BLOCK;
if (deleteNum == 0) {
printSpecDataBlock(pBlock, getStreamOpName(pOperator->operatorType), "block recv", GET_TASKID(pTaskInfo));
qDebug("===stream=== ignore block type 18, delete num is 0");
goto FETCH_NEXT_BLOCK;
}
} break;
case STREAM_CHECKPOINT: {
qError("stream check point error. msg type: STREAM_INPUT__DATA_BLOCK");

30
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -232,6 +232,29 @@ static void doDeleteWindow(SOperatorInfo* pOperator, TSKEY ts, uint64_t groupId)
static int32_t getChildIndex(SSDataBlock* pBlock) { return pBlock->info.childId; }
static void doDeleteWindowByGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
SStorageAPI* pAPI = &pOperator->pTaskInfo->storageAPI;
SStreamIntervalOperatorInfo* pInfo = pOperator->info;
SColumnInfoData* pGpIdCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
uint64_t* pGroupIdData = (uint64_t*)pGpIdCol->pData;
for (int32_t i = 0; i < pBlock->info.rows; i++) {
uint64_t groupId = pGroupIdData[i];
void* pIte = NULL;
int32_t iter = 0;
while ((pIte = tSimpleHashIterate(pInfo->aggSup.pResultRowHashTable, pIte, &iter)) != NULL) {
size_t keyLen = 0;
SWinKey* pKey = tSimpleHashGetKey(pIte, &keyLen);
if (pKey->groupId == groupId) {
int32_t tmpRes = tSimpleHashIterateRemove(pInfo->aggSup.pResultRowHashTable, pKey, keyLen, &pIte, &iter);
qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
}
}
pAPI->stateStore.streamStateDelByGroupId(pInfo->pState, groupId);
}
}
static int32_t doDeleteWindows(SOperatorInfo* pOperator, SInterval* pInterval, SSDataBlock* pBlock, SArray* pUpWins,
SSHashObj* pUpdatedMap, SHashObj* pInvalidWins) {
int32_t code = TSDB_CODE_SUCCESS;
@ -5443,7 +5466,12 @@ static int32_t doStreamIntervalAggNext(SOperatorInfo* pOperator, SSDataBlock** p
code = getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pInfo->pUpdatedMap);
QUERY_CHECK_CODE(code, lino, _end);
continue;
} else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE || pBlock->info.type == STREAM_DROP_CHILD_TABLE) {
} else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
printDataBlock(pBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
(*ppRes) = pBlock;
return code;
} else if (pBlock->info.type == STREAM_DROP_CHILD_TABLE) {
doDeleteWindowByGroupId(pOperator, pBlock);
printDataBlock(pBlock, getStreamOpName(pOperator->operatorType), GET_TASKID(pTaskInfo));
(*ppRes) = pBlock;
return code;

26
source/libs/stream/src/streamCheckpoint.c
View File

@ -622,19 +622,19 @@ int32_t streamTaskUpdateTaskCheckpointInfo(SStreamTask* pTask, bool restored, SV
pReq->transId);
streamMutexUnlock(&pTask->lock);
{ // destroy the related fill-history tasks
// drop task should not in the meta-lock, and drop the related fill-history task now
if (pReq->dropRelHTask) {
code = streamMetaUnregisterTask(pMeta, pReq->hStreamId, pReq->hTaskId);
int32_t numOfTasks = streamMetaGetNumOfTasks(pMeta);
stDebug("s-task:%s vgId:%d related fill-history task:0x%x dropped in update checkpointInfo, remain tasks:%d",
id, vgId, pReq->taskId, numOfTasks);
}
{ // destroy the related fill-history tasks
// drop task should not in the meta-lock, and drop the related fill-history task now
if (pReq->dropRelHTask) {
code = streamMetaUnregisterTask(pMeta, pReq->hStreamId, pReq->hTaskId);
int32_t numOfTasks = streamMetaGetNumOfTasks(pMeta);
stDebug("s-task:%s vgId:%d related fill-history task:0x%x dropped in update checkpointInfo, remain tasks:%d",
id, vgId, pReq->taskId, numOfTasks);
}
if (pReq->dropRelHTask) {
code = streamMetaCommit(pMeta);
}
}
if (pReq->dropRelHTask) {
code = streamMetaCommit(pMeta);
}
}
// always return true
return TSDB_CODE_SUCCESS;
@ -697,7 +697,7 @@ int32_t streamTaskUpdateTaskCheckpointInfo(SStreamTask* pTask, bool restored, SV
pTask->status.taskStatus = TASK_STATUS__READY;
code = streamMetaSaveTask(pMeta, pTask);
code = streamMetaSaveTaskInMeta(pMeta, pTask);
streamMutexUnlock(&pTask->lock);
if (code != TSDB_CODE_SUCCESS) {

2
source/libs/stream/src/streamExec.c
View File

@ -875,7 +875,7 @@ static int32_t doStreamExecTask(SStreamTask* pTask) {
}
double el = (taosGetTimestampMs() - st) / 1000.0;
if (el > 5.0) { // elapsed more than 5 sec, not occupy the CPU anymore
if (el > 2.0) { // elapsed more than 5 sec, not occupy the CPU anymore
stDebug("s-task:%s occupy more than 5.0s, release the exec threads and idle for 500ms", id);
streamTaskSetIdleInfo(pTask, 500);
return code;

42
source/libs/stream/src/streamMeta.c
View File

@ -633,7 +633,7 @@ void streamMetaCloseImpl(void* arg) {
}
// todo let's check the status for each task
int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask) {
int32_t streamMetaSaveTaskInMeta(SStreamMeta* pMeta, SStreamTask* pTask) {
int32_t vgId = pTask->pMeta->vgId;
void* buf = NULL;
int32_t len;
@ -683,7 +683,7 @@ int32_t streamMetaSaveTask(SStreamMeta* pMeta, SStreamTask* pTask) {
return code;
}
int32_t streamMetaRemoveTask(SStreamMeta* pMeta, STaskId* pTaskId) {
int32_t streamMetaRemoveTaskInMeta(SStreamMeta* pMeta, STaskId* pTaskId) {
int64_t key[2] = {pTaskId->streamId, pTaskId->taskId};
int32_t code = tdbTbDelete(pMeta->pTaskDb, key, STREAM_TASK_KEY_LEN, pMeta->txn);
if (code != 0) {
@ -706,7 +706,7 @@ int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTa
void* p = taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
if (p != NULL) {
stDebug("s-task:%" PRIx64 " already exist in meta, no need to register", id.taskId);
stDebug("s-task:0x%" PRIx64 " already exist in meta, no need to register", id.taskId);
tFreeStreamTask(pTask);
return code;
}
@ -736,7 +736,7 @@ int32_t streamMetaRegisterTask(SStreamMeta* pMeta, int64_t ver, SStreamTask* pTa
return code;
}
if ((code = streamMetaSaveTask(pMeta, pTask)) != 0) {
if ((code = streamMetaSaveTaskInMeta(pMeta, pTask)) != 0) {
int32_t unused = taosHashRemove(pMeta->pTasksMap, &id, sizeof(id));
void* pUnused = taosArrayPop(pMeta->pTaskList);
@ -886,6 +886,8 @@ static void doRemoveIdFromList(SArray* pTaskList, int32_t num, SStreamTaskId* id
static int32_t streamTaskSendTransSuccessMsg(SStreamTask* pTask, void* param) {
int32_t code = 0;
int32_t waitingDuration = 5000;
if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
code = streamTaskSendCheckpointSourceRsp(pTask);
if (code) {
@ -896,7 +898,7 @@ static int32_t streamTaskSendTransSuccessMsg(SStreamTask* pTask, void* param) {
// let's kill the query procedure within stream, to end it ASAP.
if (pTask->info.taskLevel != TASK_LEVEL__SINK && pTask->exec.pExecutor != NULL) {
code = qKillTask(pTask->exec.pExecutor, TSDB_CODE_SUCCESS);
code = qKillTask(pTask->exec.pExecutor, TSDB_CODE_SUCCESS, -1);
if (code != TSDB_CODE_SUCCESS) {
stError("s-task:%s failed to kill task related query handle, code:%s", pTask->id.idStr, tstrerror(code));
}
@ -933,7 +935,7 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t t
code = taosHashRemove(pMeta->pTasksMap, &id, sizeof(id));
doRemoveIdFromList(pMeta->pTaskList, (int32_t)taosArrayGetSize(pMeta->pTaskList), &pTask->id);
code = streamMetaRemoveTask(pMeta, &id);
code = streamMetaRemoveTaskInMeta(pMeta, &id);
if (code) {
stError("vgId:%d failed to remove task:0x%" PRIx64 ", code:%s", pMeta->vgId, id.taskId, tstrerror(code));
}
@ -964,6 +966,32 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t t
return 0;
}
int32_t streamMetaStopOneTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId) {
SStreamTask* pTask = NULL;
int32_t code = 0;
int32_t vgId = pMeta->vgId;
int32_t numOfTasks = 0;
streamMetaWLock(pMeta);
// code = streamMetaUnregisterTask(pMeta, streamId, taskId);
// numOfTasks = streamMetaGetNumOfTasks(pMeta);
// if (code) {
// stError("vgId:%d failed to drop task:0x%x, code:%s", vgId, taskId, tstrerror(code));
// }
//
// code = streamMetaCommit(pMeta);
// if (code) {
// stError("vgId:%d failed to commit after drop task:0x%x, code:%s", vgId, taskId, tstrerror(code));
// } else {
// stDebug("s-task:0x%"PRIx64"-0x%x vgId:%d dropped, remain tasks:%d", streamId, taskId, pMeta->vgId, numOfTasks);
// }
streamMetaWUnLock(pMeta);
return code;
}
int32_t streamMetaBegin(SStreamMeta* pMeta) {
streamMetaWLock(pMeta);
int32_t code = tdbBegin(pMeta->db, &pMeta->txn, tdbDefaultMalloc, tdbDefaultFree, NULL,
@ -1187,7 +1215,7 @@ void streamMetaLoadAllTasks(SStreamMeta* pMeta) {
if (taosArrayGetSize(pRecycleList) > 0) {
for (int32_t i = 0; i < taosArrayGetSize(pRecycleList); ++i) {
STaskId* pId = taosArrayGet(pRecycleList, i);
code = streamMetaRemoveTask(pMeta, pId);
code = streamMetaRemoveTaskInMeta(pMeta, pId);
if (code) {
stError("s-task:0x%" PRIx64 " failed to remove task, code:%s", pId->taskId, tstrerror(code));
}

2
source/libs/stream/src/streamStartHistory.c
View File

@ -76,7 +76,7 @@ int32_t streamStartScanHistoryAsync(SStreamTask* pTask, int8_t igUntreated) {
memcpy(serializedReq, &req, len);
SRpcMsg rpcMsg = {.contLen = len, .pCont = serializedReq, .msgType = TDMT_VND_STREAM_SCAN_HISTORY};
return tmsgPutToQueue(pTask->pMsgCb, STREAM_QUEUE, &rpcMsg);
return tmsgPutToQueue(pTask->pMsgCb, STREAM_LONG_EXEC_QUEUE, &rpcMsg);
}
void streamExecScanHistoryInFuture(SStreamTask* pTask, int32_t idleDuration) {

1
source/libs/stream/src/streamStartTask.c
View File

@ -447,7 +447,6 @@ int32_t streamMetaStopAllTasks(SStreamMeta* pMeta) {
continue;
}
int64_t refId = pTask->id.refId;
int32_t ret = streamTaskStop(pTask);
if (ret) {
stError("s-task:0x%x failed to stop task, code:%s", pTaskId->taskId, tstrerror(ret));

4
source/libs/stream/src/streamState.c
View File

@ -224,6 +224,10 @@ void streamStateDel(SStreamState* pState, const SWinKey* key) {
deleteRowBuff(pState->pFileState, key, sizeof(SWinKey));
}
void streamStateDelByGroupId(SStreamState* pState, uint64_t groupId) {
deleteRowBuffByGroupId(pState->pFileState, groupId);
}
int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) {
return streamStateFillPut_rocksdb(pState, key, value, vLen);
}

8
source/libs/stream/src/streamTask.c
View File

@ -710,7 +710,7 @@ int32_t streamTaskStop(SStreamTask* pTask) {
}
if (pTask->info.taskLevel != TASK_LEVEL__SINK && pTask->exec.pExecutor != NULL) {
code = qKillTask(pTask->exec.pExecutor, TSDB_CODE_SUCCESS);
code = qKillTask(pTask->exec.pExecutor, TSDB_CODE_SUCCESS, 5000);
if (code != TSDB_CODE_SUCCESS) {
stError("s-task:%s failed to kill task related query handle, code:%s", id, tstrerror(code));
}
@ -869,7 +869,7 @@ int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t resetRelHalt) {
pStreamTask->status.taskStatus = TASK_STATUS__READY;
}
code = streamMetaSaveTask(pMeta, pStreamTask);
code = streamMetaSaveTaskInMeta(pMeta, pStreamTask);
streamMutexUnlock(&(pStreamTask->lock));
streamMetaReleaseTask(pMeta, pStreamTask);
@ -1034,7 +1034,7 @@ static int32_t taskPauseCallback(SStreamTask* pTask, void* param) {
// in case of fill-history task, stop the tsdb file scan operation.
if (pTask->info.fillHistory == 1) {
void* pExecutor = pTask->exec.pExecutor;
code = qKillTask(pExecutor, TSDB_CODE_SUCCESS);
code = qKillTask(pExecutor, TSDB_CODE_SUCCESS, 10000);
}
stDebug("vgId:%d s-task:%s set pause flag and pause task", pMeta->vgId, pTask->id.idStr);
@ -1296,6 +1296,8 @@ const char* streamTaskGetExecType(int32_t type) {
return "resume-task-from-idle";
case STREAM_EXEC_T_ADD_FAILED_TASK:
return "record-start-failed-task";
case STREAM_EXEC_T_STOP_ONE_TASK:
return "stop-one-task";
case 0:
return "exec-all-tasks";
default:

26
source/libs/stream/src/tstreamFileState.c
View File

@ -667,6 +667,32 @@ void deleteRowBuff(SStreamFileState* pFileState, const void* pKey, int32_t keyLe
}
}
void deleteRowBuffByGroupId(SStreamFileState* pFileState, uint64_t groupId) {
SSHashObj* pRowMap = pFileState->rowStateBuff;
void* pIte = NULL;
int32_t iter = 0;
while ((pIte = tSimpleHashIterate(pRowMap, pIte, &iter)) != NULL) {
size_t keyLen = 0;
SWinKey* pKey = tSimpleHashGetKey(pIte, &keyLen);
if (pKey->groupId == groupId) {
int32_t tmpRes = tSimpleHashIterateRemove(pRowMap, pKey, keyLen, &pIte, &iter);
qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
}
}
while (1) {
SWinKey tmp = {.ts = INT64_MIN, .groupId = groupId};
SStreamStateCur* pCur = streamStateSeekKeyNext_rocksdb(pFileState->pFileStore, &tmp);
SWinKey delKey = {.groupId = groupId};
int32_t code = streamStateGetGroupKVByCur_rocksdb(pFileState->pFileStore, pCur, &delKey, NULL, 0);
if (code != TSDB_CODE_SUCCESS) {
break;
}
code = streamStateDel_rocksdb(pFileState->pFileStore, &delKey);
qTrace("%s at line %d res:%d", __func__, __LINE__, code);
}
}
static int32_t recoverSessionRowBuff(SStreamFileState* pFileState, SRowBuffPos* pPos) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;

9
source/os/src/osFile.c
View File

@ -1347,9 +1347,6 @@ int64_t taosGetLineFile(TdFilePtr pFile, char **__restrict ptrBuf) {
int64_t ret = -1;
int32_t code = 0;
#if FILE_WITH_LOCK
(void)taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
if (pFile == NULL || ptrBuf == NULL) {
terrno = TSDB_CODE_INVALID_PARA;
goto END;
@ -1363,6 +1360,10 @@ int64_t taosGetLineFile(TdFilePtr pFile, char **__restrict ptrBuf) {
goto END;
}
#if FILE_WITH_LOCK
(void)taosThreadRwlockRdlock(&(pFile->rwlock));
#endif
#ifdef WINDOWS
size_t bufferSize = 512;
*ptrBuf = taosMemoryMalloc(bufferSize);
@ -1619,10 +1620,12 @@ size_t taosReadFromCFile(void *buffer, size_t size, size_t count, FILE *stream)
return fread(buffer, size, count, stream);
}
#if 0
size_t taosWriteToCFile(const void *ptr, size_t size, size_t nitems, FILE *stream) {
STUB_RAND_IO_ERR(terrno)
return fwrite(ptr, size, nitems, stream);
}
#endif
int taosCloseCFile(FILE *f) { return fclose(f); }

6
source/os/test/CMakeLists.txt
View File

@ -30,6 +30,12 @@ add_test(
NAME osDirTests
COMMAND osDirTests
)
add_executable(osFileTests "osFileTests.cpp")
target_link_libraries(osFileTests os util gtest_main)
add_test(
NAME osFileTests
COMMAND osFileTests
)
endif()
if(TD_LINUX)

217
source/os/test/osFileTests.cpp Normal file
View File

@ -0,0 +1,217 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
#include <iostream>
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wwrite-strings"
#pragma GCC diagnostic ignored "-Wunused-function"
#pragma GCC diagnostic ignored "-Wunused-variable"
#pragma GCC diagnostic ignored "-Wsign-compare"
#pragma GCC diagnostic ignored "-Wsign-compare"
#pragma GCC diagnostic ignored "-Wformat"
#pragma GCC diagnostic ignored "-Wint-to-pointer-cast"
#pragma GCC diagnostic ignored "-Wpointer-arith"
#include "os.h"
#include "tlog.h"
TEST(osFileTests, taosGetTmpfilePath) {
char inputTmpDir[100] = "/tmp";
char fileNamePrefix[100] = "txt";
char dstPath[100] = {0};
taosGetTmpfilePath(NULL, fileNamePrefix, dstPath);
taosGetTmpfilePath(inputTmpDir, NULL, dstPath);
taosGetTmpfilePath(inputTmpDir, fileNamePrefix, dstPath);
int32_t ret = taosRemoveFile(NULL);
EXPECT_NE(ret, 0);
ret = taosCloseFile(NULL);
EXPECT_EQ(ret, 0);
ret = taosRenameFile(NULL, "");
EXPECT_NE(ret, 0);
ret = taosRenameFile("", NULL);
EXPECT_NE(ret, 0);
int64_t stDev = 0;
int64_t stIno = 0;
ret = taosDevInoFile(NULL, &stDev, &stIno);
EXPECT_NE(ret, 0);
}
TEST(osFileTests, taosCopyFile) {
char from[100] = {0};
char to[100] = {0};
int64_t ret = taosCopyFile(from, NULL);
EXPECT_EQ(ret, -1);
ret = taosCopyFile(NULL, to);
EXPECT_EQ(ret, -1);
ret = taosCopyFile(from, to);
EXPECT_EQ(ret, -1);
tstrncpy(from, "/tmp/tdengine-test-file", sizeof(from));
TdFilePtr testFilePtr = taosCreateFile(from, TD_FILE_CREATE);
taosWriteFile(testFilePtr, "abcdefg", 9);
int64_t ret64 = taosReadFile(testFilePtr, NULL, 0);
EXPECT_NE(ret64, 0);
ret64 = taosReadFile(NULL, to, 100);
EXPECT_NE(ret64, 0);
ret64 = taosWriteFile(testFilePtr, NULL, 0);
EXPECT_EQ(ret64, 0);
ret64 = taosWriteFile(NULL, to, 100);
EXPECT_EQ(ret64, 0);
ret64 = taosPWriteFile(testFilePtr, NULL, 0, 0);
EXPECT_EQ(ret64, 0);
ret64 = taosPWriteFile(NULL, to, 100, 0);
EXPECT_EQ(ret64, 0);
ret64 = taosLSeekFile(NULL, 0, 0);
EXPECT_EQ(ret64, -1);
ret64 = taosPReadFile(NULL, NULL, 0, 0);
EXPECT_EQ(ret64, -1);
bool retb = taosValidFile(testFilePtr);
EXPECT_TRUE(retb);
retb = taosValidFile(NULL);
EXPECT_FALSE(retb);
retb = taosCheckAccessFile(NULL, 0);
EXPECT_FALSE(retb);
int32_t ret32 = taosFStatFile(NULL, NULL, NULL);
EXPECT_NE(ret32, 0);
ret32 = taosLockFile(NULL);
EXPECT_NE(ret32, 0);
ret32 = taosUnLockFile(NULL);
EXPECT_NE(ret32, 0);
ret32 = taosFtruncateFile(NULL, 0);
EXPECT_NE(ret32, 0);
ret64 = taosFSendFile(NULL, testFilePtr, NULL, 0);
EXPECT_NE(ret64, 0);
ret64 = taosFSendFile(testFilePtr, NULL, NULL, 0);
EXPECT_NE(ret64, 0);
char buf[100] = {0};
ret64 = taosGetLineFile(NULL, (char**)&buf);
EXPECT_EQ(ret64, -1);
ret64 = taosGetLineFile(testFilePtr, NULL);
EXPECT_EQ(ret64, -1);
ret64 = taosGetsFile(testFilePtr, 0, NULL);
EXPECT_NE(ret64, -1);
ret64 = taosGetsFile(NULL, 0, buf);
EXPECT_NE(ret64, -1);
ret32 = taosEOFFile(NULL);
EXPECT_NE(ret64, -1);
taosCloseFile(&testFilePtr);
ret32 = taosFStatFile(testFilePtr, NULL, NULL);
EXPECT_NE(ret32, 0);
ret32 = taosLockFile(testFilePtr);
EXPECT_NE(ret32, 0);
ret32 = taosUnLockFile(testFilePtr);
EXPECT_NE(ret32, 0);
ret32 = taosFtruncateFile(testFilePtr, 0);
EXPECT_NE(ret32, 0);
ret64 = taosFSendFile(testFilePtr, testFilePtr, NULL, 0);
EXPECT_NE(ret64, 0);
ret64 = taosGetLineFile(testFilePtr, NULL);
EXPECT_EQ(ret64, -1);
ret64 = taosGetsFile(testFilePtr, 0, NULL);
EXPECT_NE(ret64, -1);
ret32 = taosEOFFile(testFilePtr);
EXPECT_NE(ret64, -1);
retb = taosValidFile(testFilePtr);
EXPECT_FALSE(retb);
ret = taosCopyFile(from, to);
EXPECT_EQ(ret, -1);
int64_t size = 0;
int64_t mtime = 0;
int64_t atime = 0;
ret = taosStatFile(NULL, &size, &mtime, &atime);
EXPECT_NE(ret, 0);
ret = taosStatFile(from, &size, &mtime, NULL);
EXPECT_EQ(ret, 0);
int64_t diskid = 0;
ret = taosGetFileDiskID(NULL, &diskid);
EXPECT_NE(ret, 0);
ret = taosGetFileDiskID("", &diskid);
EXPECT_NE(ret, 0);
ret = taosGetFileDiskID(from, NULL);
EXPECT_EQ(ret, 0);
ret32 = taosCompressFile(NULL, "");
EXPECT_NE(ret32, 0);
ret32 = taosCompressFile("", NULL);
EXPECT_NE(ret32, 0);
ret32 = taosCompressFile("", "");
EXPECT_NE(ret32, 0);
ret32 = taosCompressFile("/tmp/tdengine-test-file", "");
EXPECT_NE(ret32, 0);
ret32 = taosLinkFile("", "");
EXPECT_NE(ret32, 0);
char mod[8] = {0};
FILE* retptr = taosOpenCFile(NULL, "");
EXPECT_EQ(retptr, nullptr);
retptr = taosOpenCFile("", NULL);
EXPECT_EQ(retptr, nullptr);
retptr = taosOpenCFile("", mod);
EXPECT_EQ(retptr, nullptr);
ret32 = taosSeekCFile(NULL, 0, 0);
EXPECT_NE(ret32, 0);
size_t retsize = taosReadFromCFile(buf, 0, 0, NULL);
EXPECT_EQ(retsize, 0);
retsize = taosReadFromCFile(NULL, 0, 0, NULL);
EXPECT_EQ(retsize, 0);
taosRemoveFile(from);
}
TEST(osFileTests, taosCreateFile) {
char path[100] = {0};
int32_t tdFileOptions = 0;
TdFilePtr ret = taosCreateFile(NULL, 0);
EXPECT_EQ(ret, nullptr);
ret = taosCreateFile(path, 0);
EXPECT_EQ(ret, nullptr);
FILE* retptr = taosOpenFileForStream(NULL, 0);
EXPECT_EQ(retptr, nullptr);
TdFilePtr retptr2 = taosOpenFile(NULL, 0);
EXPECT_EQ(retptr2, nullptr);
}

7
source/util/src/tworker.c
View File

@ -256,7 +256,7 @@ static void *tAutoQWorkerThreadFp(SQueueWorker *worker) {
return NULL;
}
STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem fp) {
STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem fp, int32_t minNum) {
int32_t code;
STaosQueue *queue;
@ -280,7 +280,10 @@ STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem
int32_t queueNum = taosGetQueueNumber(pool->qset);
int32_t curWorkerNum = taosArrayGetSize(pool->workers);
int32_t dstWorkerNum = ceilf(queueNum * pool->ratio);
if (dstWorkerNum < 2) dstWorkerNum = 2;
if (dstWorkerNum < minNum) {
dstWorkerNum = minNum;
}
// spawn a thread to process queue
while (curWorkerNum < dstWorkerNum) {

181
tests/army/sys/checkErrorCode.py
View File

@ -1,181 +0,0 @@
###################################################################
# 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 random
import taos
import frame
import frame.eos
import frame.etime
import frame.etool
from frame.log import *
from frame.sql import *
from frame.cases import *
from frame.caseBase import *
from frame.srvCtl import *
from frame import *
class TDTestCase(TBase):
# parse line
def parseLine(self, line):
line = line.strip()
PRE_DEFINE = "#define TSDB_CODE_"
n = len(PRE_DEFINE)
if line[:n] != PRE_DEFINE:
return None
# TAOS_DEF_ERROR_CODE(0, 0x000B)
pos = line.find("TAOS_DEF_ERROR_CODE(0, 0x", n)
if pos == -1:
tdLog.info(f"not found \"TAOS_DEF_ERROR_CODE(0, \" line={line}")
return None
code = line[pos:].strip()
pos = code.find(")")
if pos == -1:
tdLog.info(f"not found \")\", line={line}")
return None
code = code[:pos]
if len(code) != 4:
tdLog.info(f"code is len not 4 len:{len(code)} subcode={code}\")\", line={line}")
return None
# return
return "0x8000" + code
# ignore error
def ignoreCode(self, code):
ignoreCodes = {"0x00008, 0x000009"}
if code in ignoreCodes:
return True
else:
return False
# read valid code
def readHeadCodes(self, hFile):
codes = []
start = False
# read
with open(hFile) as file:
for line in file:
code = self.parseLine(line)
# invalid
if code == None:
continue
# ignore
if self.ignoreCode(code):
tdLog.info(f"ignore error {code}\n")
# valid
if code == 0:
start = True
if start:
codes.append(code)
# return
return codes
# parse doc lines
def parseDocLine(self, line):
line = line.strip()
PRE_DEFINE = "| 0x8000"
n = len(PRE_DEFINE)
if line[:n] != PRE_DEFINE:
return None
line = line[2:]
cols = line.split("|")
# remove blank
cols = [col.strip() for col in cols]
# return
return cols
# read valid code
def readDocCodes(self, docFile):
codes = []
start = False
# read
with open(docFile) as file:
for line in file:
code = self.parseDocLine(line)
# invalid
if code == None:
continue
# valid
if start:
codes.append(code)
# return
return codes
# check
def checkConsistency(self, docCodes, codes):
diff = False
# len
docLen = len(docCodes)
len = len(codes)
tdLog.info("head file codes = {len} doc file codes={docLen} \n")
if docLen > len:
maxLen = docLen
else:
maxLen = len
for i in range(maxLen):
if i < len and i < docLen:
if codes[i] == docCodes[i][0]:
tdLog.info(f" i={i} same head code: {codes[i]} doc code:{docCodes[i][0]}\n")
else:
tdLog.info(f" i={i} diff head code: {codes[i]} doc code:{docCodes[i][0]}\n")
diff = True
elif i < len:
tdLog.info(f" i={i} diff head code: {codes[i]} doc code: None\n")
diff = True
elif i < docLen:
tdLog.info(f" i={i} diff head code: None doc code: {docCodes[i][0]}\n")
diff = True
# result
if diff:
tdLog.exit("check error code consistency failed.\n")
# run
def run(self):
tdLog.debug(f"start to excute {__file__}")
# read head error code
hFile = "../../include/util/taoserror.h"
codes = self.readHeadCodes(hFile)
# read zh codes
zhDoc = "../../docs/zh/14-reference/09-error-code.md"
zhCodes = self.readDocCodes(zhDoc, codes)
# read en codes
enDoc = "../../docs/en/14-reference/09-error-code.md"
enCodes = self.readDocCodes(enDoc, codes)
# check zh
tdLog.info(f"check zh docs ...\n")
self.checkConsistency(zhCodes, codes)
# check en
tdLog.info(f"check en docs ...\n")
self.checkConsistency(enCodes, codes)
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

11
tests/army/tools/benchmark/basic/bugs.py
View File

@ -21,15 +21,6 @@ from frame.caseBase import *
from frame import *
# reomve single and double quotation
def removeQuotation(origin):
value = ""
for c in origin:
if c != '\'' and c != '"':
value += c
return value
class TDTestCase(TBase):
def caseDescription(self):
"""
@ -111,7 +102,7 @@ class TDTestCase(TBase):
tdSql.query(sql)
if cachemode != None:
value = removeQuotation(cachemode)
value = frame.eutil.removeQuota(cachemode)
tdLog.info(f" deal both origin={cachemode} after={value}")
tdSql.checkData(0, 1, value)

9
tests/army/tools/benchmark/basic/exportCsv.py
View File

@ -23,15 +23,6 @@ from frame.caseBase import *
from frame import *
# reomve single and double quotation
def removeQuotation(origin):
value = ""
for c in origin:
if c != '\'' and c != '"':
value += c
return value
class TDTestCase(TBase):
def caseDescription(self):
"""

133
tests/army/tools/benchmark/basic/insertBasic.py Normal file
View File

@ -0,0 +1,133 @@
###################################################################
# 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 os
import json
import frame
import frame.etool
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.caseBase import *
from frame import *
class TDTestCase(TBase):
def caseDescription(self):
"""
taosBenchmark Insert->Basic test cases
"""
def benchmarkQuery(self, benchmark, jsonFile, keys, options=""):
# exe insert
result = "query.log"
os.system(f"rm -f {result}")
cmd = f"{benchmark} {options} -f {jsonFile} >> {result}"
os.system(cmd)
tdLog.info(cmd)
with open(result) as file:
content = file.read()
for key in keys:
if content.find(key) == -1:
tdLog.exit(f"not found key: {key} in content={content}")
else:
tdLog.info(f"found key:{key} successful.")
def testBenchmarkJson(self, benchmark, jsonFile, options = "", checkTimeStep = False):
# exe insert
cmd = f"{benchmark} {options} -f {jsonFile}"
os.system(cmd)
#
# check insert result
#
with open(jsonFile, "r") as file:
data = json.load(file)
db = data["databases"][0]["dbinfo"]["name"]
stb = data["databases"][0]["super_tables"][0]["name"]
child_count = data["databases"][0]["super_tables"][0]["childtable_count"]
insert_rows = data["databases"][0]["super_tables"][0]["insert_rows"]
timestamp_step = data["databases"][0]["super_tables"][0]["timestamp_step"]
# drop
try:
drop = data["databases"][0]["dbinfo"]["drop"]
except:
drop = "yes"
# command is first
if options.find("-Q") != -1:
drop = "no"
# only support
cmdVG = None
pos = options.find("=")
if pos != -1:
arr = options.split("=")
if arr[0] == "--vgroups":
cmdVG = arr[1]
# vgropus
try:
if cmdVG != None:
# command special vgroups first priority
vgroups = cmdVG
else:
vgroups = data["databases"][0]["dbinfo"]["vgroups"]
except:
vgroups = None
tdLog.info(f"get json info: db={db} stb={stb} child_count={child_count} insert_rows={insert_rows} cmdVG={cmdVG}\n")
# all count insert_rows * child_table_count
sql = f"select * from {db}.{stb}"
tdSql.query(sql)
tdSql.checkRows(child_count * insert_rows)
# timestamp step
if checkTimeStep:
sql = f"select * from (select diff(ts) as dif from {db}.{stb} partition by tbname) where dif != {timestamp_step};"
tdSql.query(sql)
tdSql.checkRows(0)
if drop.lower() == "yes":
# check database optins
sql = f"select `vgroups` from information_schema.ins_databases where name='{db}';"
tdSql.query(sql)
if vgroups != None:
tdLog.info(f" vgroups real={tdSql.getData(0,0)} expect={vgroups}")
tdSql.checkData(0, 0, vgroups, True)
# bugs ts
def checkVGroups(self, benchmark):
# vgroups with command line set
self.testBenchmarkJson(benchmark, "./tools/benchmark/basic/json/insertBasic.json", "--vgroups=3", True)
# vgroups with json file
self.testBenchmarkJson(benchmark, "./tools/benchmark/basic/json/insertBasic.json", "", True)
def run(self):
benchmark = etool.benchMarkFile()
# vgroups
self.checkVGroups(benchmark)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

48
tests/army/tools/benchmark/basic/insertBindVGroup.py Normal file
View File

@ -0,0 +1,48 @@
###################################################################
# 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 os
import json
import frame
import frame.etool
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.caseBase import *
from frame import *
class TDTestCase(TBase):
def caseDescription(self):
"""
taosBenchmark insert->BindVGroup test cases
"""
# bugs ts
def checkBasic(self):
# thread equal vgroups
self.insertBenchJson("./tools/benchmark/basic/json/insertBindVGroup.json", "", True)
# thread is limited
self.insertBenchJson("./tools/benchmark/basic/json/insertBindVGroup.json", "-T 2", True)
def run(self):
# basic
self.checkBasic()
# advance
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

89
tests/army/tools/benchmark/basic/insertPrecision.py Normal file
View File

@ -0,0 +1,89 @@
###################################################################
# 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 os
import json
import frame
import frame.etool
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.caseBase import *
from frame import *
class TDTestCase(TBase):
def caseDescription(self):
"""
taosBenchmark insert->Precision test cases
"""
def testBenchmarkJson(self, benchmark, jsonFile, options = ""):
# exe insert
cmd = f"{benchmark} {options} -f {jsonFile}"
os.system(cmd)
#
# check insert result
#
with open(jsonFile, "r") as file:
data = json.load(file)
db = data["databases"][0]["dbinfo"]["name"]
precison = data["databases"][0]["dbinfo"]["precision"]
stb = data["databases"][0]["super_tables"][0]["name"]
child_count = data["databases"][0]["super_tables"][0]["childtable_count"]
insert_rows = data["databases"][0]["super_tables"][0]["insert_rows"]
timestamp_step = data["databases"][0]["super_tables"][0]["timestamp_step"]
start_timestamp = data["databases"][0]["super_tables"][0]["start_timestamp"]
tdLog.info(f"get json info: db={db} precision={precison} stb={stb} child_count={child_count} insert_rows={insert_rows} "
f"start_timestamp={start_timestamp} timestamp_step={timestamp_step} \n")
# all count insert_rows * child_table_count
sql = f"select * from {db}.{stb}"
tdSql.query(sql)
tdSql.checkRows(child_count * insert_rows)
# timestamp step
sql = f"select * from (select diff(ts) as dif from {db}.{stb} partition by tbname) where dif != {timestamp_step};"
tdSql.query(sql)
tdSql.checkRows(0)
# check last ts
lastTime = start_timestamp + timestamp_step * (insert_rows - 1)
sql = f"select last(ts) from {db}.{stb}"
tdSql.checkAgg(sql, lastTime)
# bugs ts
def checkBasic(self, benchmark):
# MS
self.testBenchmarkJson(benchmark, "./tools/benchmark/basic/json/insertPrecisionMS.json", "")
# US
self.testBenchmarkJson(benchmark, "./tools/benchmark/basic/json/insertPrecisionUS.json", "")
# NS
self.testBenchmarkJson(benchmark, "./tools/benchmark/basic/json/insertPrecisionNS.json", "")
def run(self):
benchmark = etool.benchMarkFile()
# vgroups
self.checkBasic(benchmark)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

64
tests/army/tools/benchmark/basic/json/insertBasic.json Normal file
View File

@ -0,0 +1,64 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos",
"host": "127.0.0.1",
"port": 6030,
"user": "root",
"password": "taosdata",
"num_of_records_per_req": 3000,
"thread_count": 2,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "test",
"drop": "yes",
"precision": "ns",
"vgroups": 2
},
"super_tables": [
{
"name": "meters",
"child_table_exists": "no",
"childtable_count": 2,
"insert_rows": 1000,
"childtable_prefix": "d",
"insert_mode": "taosc",
"insert_interval": 0,
"timestamp_step": 1000,
"start_timestamp":1700000000000000000,
"columns": [
{ "type": "bool", "name": "bc"},
{ "type": "float", "name": "fc", "max": 1, "min": 0 },
{ "type": "double", "name": "dc", "max": 10, "min": 0 },
{ "type": "tinyint", "name": "ti", "max": 100, "min": -100 },
{ "type": "smallint", "name": "si", "max": 100, "min": -50 },
{ "type": "int", "name": "ic", "max": 1000, "min": -1000 },
{ "type": "bigint", "name": "bi", "max": 100, "min": -1000 },
{ "type": "utinyint", "name": "uti", "max": 100, "min": 0 },
{ "type": "usmallint", "name": "usi", "max": 100, "min": 0 },
{ "type": "uint", "name": "ui", "max": 1000, "min": 0 },
{ "type": "ubigint", "name": "ubi", "max": 10000, "min": 0 },
{ "type": "binary", "name": "bin", "len": 4},
{ "type": "nchar", "name": "nch", "len": 8}
],
"tags": [
{ "type": "bool", "name": "tbc"},
{ "type": "float", "name": "tfc", "max": 1, "min": 0 },
{ "type": "double", "name": "tdc", "max": 10, "min": 0 },
{ "type": "tinyint", "name": "tti", "max": 100, "min": -100 },
{ "type": "smallint", "name": "tsi", "max": 100, "min": -50 },
{ "type": "int", "name": "tic", "max": 1000, "min": -1000 },
{ "type": "bigint", "name": "tbi", "max": 100, "min": -1000 },
{ "type": "utinyint", "name": "tuti", "max": 100, "min": 0 },
{ "type": "usmallint", "name": "tusi", "max": 100, "min": 0 },
{ "type": "uint", "name": "tui", "max": 1000, "min": 0 },
{ "type": "ubigint", "name": "tubi", "max": 10000, "min": 0 },
{ "type": "binary", "name": "tbin", "len": 4},
{ "type": "nchar", "name": "tnch", "len": 8}
]
}
]
}
]
}

47
tests/army/tools/benchmark/basic/json/insertBindVGroup.json Normal file
View File

@ -0,0 +1,47 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos",
"host": "127.0.0.1",
"port": 6030,
"user": "root",
"password": "taosdata",
"num_of_records_per_req": 6000,
"thread_count": 10,
"thread_bind_vgroup": "yes",
"create_table_thread_count": 1,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "binddb",
"drop": "yes",
"vgroups": 10
},
"super_tables": [
{
"name": "meters",
"child_table_exists": "no",
"childtable_count": 20,
"insert_rows": 5000,
"childtable_prefix": "d",
"insert_mode": "stmt2",
"timestamp_step": 1000,
"start_timestamp":1600000000000,
"columns": [
{"type": "FLOAT", "name": "current", "count": 1, "max": 12, "min": 8 },
{ "type": "INT", "name": "voltage", "max": 225, "min": 100 },
{ "type": "FLOAT", "name": "phase", "max": 1, "min": 0 }
],
"tags": [
{"type": "tinyint", "name": "groupid","max": 10,"min": 1},
{"type": "binary", "name": "location", "len": 16,
"values": ["San Francisco", "Los Angles", "San Diego",
"San Jose", "Palo Alto", "Campbell", "Mountain View",
"Sunnyvale", "Santa Clara", "Cupertino"]
}
]
}
]
}
]
}

56
tools/taos-tools/case/insertBindVGroup.json → tests/army/tools/benchmark/basic/json/insertPrecisionMS.json
View File

@ -5,16 +5,15 @@
"port": 6030,
"user": "root",
"password": "taosdata",
"num_of_records_per_req": 200,
"thread_count": 20,
"thread_bind_vgroup": "yes",
"create_table_thread_count": 1,
"num_of_records_per_req": 3000,
"thread_count": 2,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "binddb",
"name": "test",
"drop": "yes",
"precision": "ms",
"vgroups": 2
},
"super_tables": [
@ -22,37 +21,44 @@
"name": "meters",
"child_table_exists": "no",
"childtable_count": 4,
"insert_rows": 100,
"interlace_rows": 10,
"insert_rows": 1000,
"childtable_prefix": "d",
"insert_mode": "taosc",
"timestamp_step": 1000,
"start_timestamp":1500000000000,
"interlace_rows": 0,
"timestamp_step": 100,
"start_timestamp":1700000000111,
"columns": [
{ "type": "bool", "name": "bc"},
{ "type": "float", "name": "fc", "max": 1, "min": 0 },
{ "type": "double", "name": "dc", "max": 1, "min": 0 },
{ "type": "tinyint", "name": "ti", "max": 100, "min": 0 },
{ "type": "smallint", "name": "si", "max": 100, "min": 0 },
{ "type": "int", "name": "ic", "max": 100, "min": 0 },
{ "type": "bigint", "name": "bi", "max": 100, "min": 0 },
{ "type": "double", "name": "dc", "max": 10, "min": 0 },
{ "type": "tinyint", "name": "ti", "max": 100, "min": -100 },
{ "type": "smallint", "name": "si", "max": 100, "min": -50 },
{ "type": "int", "name": "ic", "max": 1000, "min": -1000 },
{ "type": "bigint", "name": "bi", "max": 100, "min": -1000 },
{ "type": "utinyint", "name": "uti", "max": 100, "min": 0 },
{ "type": "usmallint", "name": "usi", "max": 100, "min": 0 },
{ "type": "uint", "name": "ui", "max": 100, "min": 0 },
{ "type": "ubigint", "name": "ubi", "max": 100, "min": 0 },
{ "type": "binary", "name": "bin", "len": 32},
{ "type": "nchar", "name": "nch", "len": 64}
{ "type": "uint", "name": "ui", "max": 1000, "min": 0 },
{ "type": "ubigint", "name": "ubi", "max": 10000, "min": 0 },
{ "type": "binary", "name": "bin", "len": 4},
{ "type": "nchar", "name": "nch", "len": 8}
],
"tags": [
{"type": "tinyint", "name": "groupid","max": 10,"min": 1},
{"type": "binary", "name": "location", "len": 16,
"values": ["San Francisco", "Los Angles", "San Diego",
"San Jose", "Palo Alto", "Campbell", "Mountain View",
"Sunnyvale", "Santa Clara", "Cupertino"]
}
{ "type": "bool", "name": "tbc"},
{ "type": "float", "name": "tfc", "max": 1, "min": 0 },
{ "type": "double", "name": "tdc", "max": 10, "min": 0 },
{ "type": "tinyint", "name": "tti", "max": 100, "min": -100 },
{ "type": "smallint", "name": "tsi", "max": 100, "min": -50 },
{ "type": "int", "name": "tic", "max": 1000, "min": -1000 },
{ "type": "bigint", "name": "tbi", "max": 100, "min": -1000 },
{ "type": "utinyint", "name": "tuti", "max": 100, "min": 0 },
{ "type": "usmallint", "name": "tusi", "max": 100, "min": 0 },
{ "type": "uint", "name": "tui", "max": 1000, "min": 0 },
{ "type": "ubigint", "name": "tubi", "max": 10000, "min": 0 },
{ "type": "binary", "name": "tbin", "len": 4},
{ "type": "nchar", "name": "tnch", "len": 8}
]
}
]
}
]
}
}

63
tests/army/tools/benchmark/basic/json/insertPrecisionNS.json Normal file
View File

@ -0,0 +1,63 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos",
"host": "127.0.0.1",
"port": 6030,
"user": "root",
"password": "taosdata",
"num_of_records_per_req": 3000,
"thread_count": 2,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "test",
"drop": "yes",
"precision": "ns",
"vgroups": 2
},
"super_tables": [
{
"name": "meters",
"child_table_exists": "no",
"childtable_count": 3,
"insert_rows": 1000,
"childtable_prefix": "d",
"insert_mode": "stmt2",
"timestamp_step": 10,
"start_timestamp":1700000000000000111,
"columns": [
{ "type": "bool", "name": "bc"},
{ "type": "float", "name": "fc", "max": 1, "min": 0 },
{ "type": "double", "name": "dc", "max": 10, "min": 0 },
{ "type": "tinyint", "name": "ti", "max": 100, "min": -100 },
{ "type": "smallint", "name": "si", "max": 100, "min": -50 },
{ "type": "int", "name": "ic", "max": 1000, "min": -1000 },
{ "type": "bigint", "name": "bi", "max": 100, "min": -1000 },
{ "type": "utinyint", "name": "uti", "max": 100, "min": 0 },
{ "type": "usmallint", "name": "usi", "max": 100, "min": 0 },
{ "type": "uint", "name": "ui", "max": 1000, "min": 0 },
{ "type": "ubigint", "name": "ubi", "max": 10000, "min": 0 },
{ "type": "binary", "name": "bin", "len": 4},
{ "type": "nchar", "name": "nch", "len": 8}
],
"tags": [
{ "type": "bool", "name": "tbc"},
{ "type": "float", "name": "tfc", "max": 1, "min": 0 },
{ "type": "double", "name": "tdc", "max": 10, "min": 0 },
{ "type": "tinyint", "name": "tti", "max": 100, "min": -100 },
{ "type": "smallint", "name": "tsi", "max": 100, "min": -50 },
{ "type": "int", "name": "tic", "max": 1000, "min": -1000 },
{ "type": "bigint", "name": "tbi", "max": 100, "min": -1000 },
{ "type": "utinyint", "name": "tuti", "max": 100, "min": 0 },
{ "type": "usmallint", "name": "tusi", "max": 100, "min": 0 },
{ "type": "uint", "name": "tui", "max": 1000, "min": 0 },
{ "type": "ubigint", "name": "tubi", "max": 10000, "min": 0 },
{ "type": "binary", "name": "tbin", "len": 4},
{ "type": "nchar", "name": "tnch", "len": 8}
]
}
]
}
]
}

64
tests/army/tools/benchmark/basic/json/insertPrecisionUS.json Normal file
View File

@ -0,0 +1,64 @@
{
"filetype": "insert",
"cfgdir": "/etc/taos",
"host": "127.0.0.1",
"port": 6030,
"user": "root",
"password": "taosdata",
"num_of_records_per_req": 3000,
"thread_count": 2,
"confirm_parameter_prompt": "no",
"databases": [
{
"dbinfo": {
"name": "test",
"drop": "yes",
"precision": "us",
"vgroups": 2
},
"super_tables": [
{
"name": "meters",
"child_table_exists": "no",
"childtable_count": 3,
"insert_rows": 5000,
"childtable_prefix": "d",
"insert_mode": "stmt",
"interlace_rows": 100,
"timestamp_step": 10,
"start_timestamp":1700000000000321,
"columns": [
{ "type": "bool", "name": "bc"},
{ "type": "float", "name": "fc", "max": 1, "min": 0 },
{ "type": "double", "name": "dc", "max": 10, "min": 0 },
{ "type": "tinyint", "name": "ti", "max": 100, "min": -100 },
{ "type": "smallint", "name": "si", "max": 100, "min": -50 },
{ "type": "int", "name": "ic", "max": 1000, "min": -1000 },
{ "type": "bigint", "name": "bi", "max": 100, "min": -1000 },
{ "type": "utinyint", "name": "uti", "max": 100, "min": 0 },
{ "type": "usmallint", "name": "usi", "max": 100, "min": 0 },
{ "type": "uint", "name": "ui", "max": 1000, "min": 0 },
{ "type": "ubigint", "name": "ubi", "max": 10000, "min": 0 },
{ "type": "binary", "name": "bin", "len": 4},
{ "type": "nchar", "name": "nch", "len": 8}
],
"tags": [
{ "type": "bool", "name": "tbc"},
{ "type": "float", "name": "tfc", "max": 1, "min": 0 },
{ "type": "double", "name": "tdc", "max": 10, "min": 0 },
{ "type": "tinyint", "name": "tti", "max": 100, "min": -100 },
{ "type": "smallint", "name": "tsi", "max": 100, "min": -50 },
{ "type": "int", "name": "tic", "max": 1000, "min": -1000 },
{ "type": "bigint", "name": "tbi", "max": 100, "min": -1000 },
{ "type": "utinyint", "name": "tuti", "max": 100, "min": 0 },
{ "type": "usmallint", "name": "tusi", "max": 100, "min": 0 },
{ "type": "uint", "name": "tui", "max": 1000, "min": 0 },
{ "type": "ubigint", "name": "tubi", "max": 10000, "min": 0 },
{ "type": "binary", "name": "tbin", "len": 4},
{ "type": "nchar", "name": "tnch", "len": 8}
]
}
]
}
]
}

11
tests/army/tools/benchmark/basic/queryMain.py
View File

@ -28,19 +28,10 @@ from frame.caseBase import *
from frame import *
# reomve single and double quotation
def removeQuotation(origin):
value = ""
for c in origin:
if c != '\'' and c != '"':
value += c
return value
class TDTestCase(TBase):
def caseDescription(self):
"""
[TD-11510] taosBenchmark test cases
taosBenchmark query->Basic test cases
"""
def runSeconds(self, command, timeout = 180):

217
tests/army/tools/taosdump/native/taosdumpCommandline.py Normal file
View File

@ -0,0 +1,217 @@
###################################################################
# 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 os
import json
import frame
import frame.etool
from frame.log import *
from frame.cases import *
from frame.sql import *
from frame.caseBase import *
from frame import *
class TDTestCase(TBase):
def caseDescription(self):
"""
test taosdump support commandline arguments
"""
def exec(self, command):
tdLog.info(command)
return os.system(command)
def clearPath(self, path):
os.system("rm -rf %s/*" % path)
def findPrograme(self):
# taosdump
taosdump = etool.taosDumpFile()
if taosdump == "":
tdLog.exit("taosdump not found!")
else:
tdLog.info("taosdump found in %s" % taosdump)
# taosBenchmark
benchmark = etool.benchMarkFile()
if benchmark == "":
tdLog.exit("benchmark not found!")
else:
tdLog.info("benchmark found in %s" % benchmark)
# tmp dir
tmpdir = "./tmp"
if not os.path.exists(tmpdir):
os.makedirs(tmpdir)
else:
print("directory exists")
self.clearPath(tmpdir)
return taosdump, benchmark,tmpdir
def checkCorrectWithJson(self, jsonFile, newdb = None, checkInterval=False):
#
# check insert result
#
with open(jsonFile, "r") as file:
data = json.load(file)
# db come from arguments
if newdb is None:
db = data["databases"][0]["dbinfo"]["name"]
else:
db = newdb
stb = data["databases"][0]["super_tables"][0]["name"]
child_count = data["databases"][0]["super_tables"][0]["childtable_count"]
insert_rows = data["databases"][0]["super_tables"][0]["insert_rows"]
timestamp_step = data["databases"][0]["super_tables"][0]["timestamp_step"]
tdLog.info(f"get json: db={db} stb={stb} child_count={child_count} insert_rows={insert_rows} \n")
# all count insert_rows * child_table_count
sql = f"select * from {db}.{stb}"
tdSql.query(sql)
tdSql.checkRows(child_count * insert_rows)
# timestamp step
if checkInterval:
sql = f"select * from (select diff(ts) as dif from {db}.{stb} partition by tbname) where dif != {timestamp_step};"
tdSql.query(sql)
tdSql.checkRows(0)
def testBenchmarkJson(self, benchmark, jsonFile, options="", checkInterval=False):
# exe insert
cmd = f"{benchmark} {options} -f {jsonFile}"
self.exec(cmd)
self.checkCorrectWithJson(jsonFile)
def insertData(self, benchmark, json, db):
# insert super table
self.testBenchmarkJson(benchmark, json)
# normal table
sqls = [
f"create table {db}.ntb(st timestamp, c1 int, c2 binary(32))",
f"insert into {db}.ntb values(now, 1, 'abc1')",
f"insert into {db}.ntb values(now, 2, 'abc2')",
f"insert into {db}.ntb values(now, 3, 'abc3')",
f"insert into {db}.ntb values(now, 4, 'abc4')",
f"insert into {db}.ntb values(now, 5, 'abc5')",
]
for sql in sqls:
tdSql.execute(sql)
def dumpOut(self, taosdump, db , outdir):
# dump out
self.exec(f"{taosdump} -D {db} -o {outdir}")
def dumpIn(self, taosdump, db, newdb, indir):
# dump in
self.exec(f'{taosdump} -W "{db}={newdb}" -i {indir}')
def checkSame(self, db, newdb, stb, aggfun):
# sum pk db
sql = f"select {aggfun} from {db}.{stb}"
tdSql.query(sql)
sum1 = tdSql.getData(0,0)
# sum pk newdb
sql = f"select {aggfun} from {newdb}.{stb}"
tdSql.query(sql)
sum2 = tdSql.getData(0,0)
if sum1 == sum2:
tdLog.info(f"{aggfun} source db:{sum1} import db:{sum2} both equal.")
else:
tdLog.exit(f"{aggfun} source db:{sum1} import db:{sum2} not equal.")
def verifyResult(self, db, newdb, json):
# compare with insert json
self.checkCorrectWithJson(json, newdb)
# compare sum(pk)
stb = "meters"
self.checkSame(db, newdb, stb, "sum(pk)")
self.checkSame(db, newdb, stb, "sum(usi)")
self.checkSame(db, newdb, stb, "sum(ic)")
# check normal table
self.checkSame(db, newdb, "ntb", "sum(c1)")
# basic commandline
def basicCommandLine(self, taosdump, tmpdir):
# -h -P -u -p -o
self.exec(taosdump + f" -h 127.0.0.1 -P 6030 -uroot -ptaosdata -A -N -o {tmpdir}")
self.clearPath(tmpdir)
# check except
def checkExcept(self, command):
try:
code = self.exec(command)
if code == 0:
tdLog.exit(f"Failed, not report error cmd:{command}")
else:
tdLog.info(f"Passed, report error code={code} is expect, cmd:{command}")
except:
tdLog.info(f"Passed, catch expect report error for command {command}")
# except commandline
def exceptCommandLine(self, taosdump, tmpdir):
# -o
self.checkExcept(taosdump + " -o= ")
self.checkExcept(taosdump + " -o")
self.checkExcept(taosdump + " -A -o=")
self.checkExcept(taosdump + " -A -o ")
self.checkExcept(taosdump + " -A -o ./noexistpath/")
self.checkExcept(taosdump + f" -d invalidAVRO -o {tmpdir}")
# run
def run(self):
# database
db = "pridb"
newdb = "npridb"
# find
taosdump, benchmark, tmpdir = self.findPrograme()
json = "./tools/taosdump/ws/json/primaryKey.json"
# insert data with taosBenchmark
self.insertData(benchmark, json, db)
# basic commandline
self.basicCommandLine(taosdump, tmpdir)
# except commandline
self.exceptCommandLine(taosdump, tmpdir)
# dump out
#self.dumpOut(taosdump, db, tmpdir)
# dump in
#self.dumpIn(taosdump, db, newdb, tmpdir)
# verify db
#self.verifyResult(db, newdb, json)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

2
tests/ci/func.txt
View File

@ -79,7 +79,7 @@
(void)streamMetaAddFailedTask
(void)streamMetaAddTaskLaunchResult
(void)streamMetaCommit
(void)streamMetaRemoveTask
(void)streamMetaRemoveTaskInMeta
(void)streamMetaSendHbHelper
(void)streamMetaStartAllTasks
(void)streamMetaStartOneTask

124
tests/parallel_test/cases.task
View File

@ -77,73 +77,88 @@
# army/tools
#
# benchmark 64 cases
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/rest_insert_alltypes_json.py -R
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosdemoTestQueryWithJson-mixed-query.py -R
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_sample_csv_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosdemoTestInsertWithJsonStmt-otherPara.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_telnet_insert_alltypes-same-min-max.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/default_tmq_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/reuse-exist-stb.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/sml_interlace.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt2_insert.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_offset_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/json_tag.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline-sml-rest.py -R
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_auto_create_table_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/insert-json-csv.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_insert-table-creating-interval.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/insertMix.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosc_insert-mix.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stream_function_test.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/telnet_tcp.py -R
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_sample_csv_json-subtable.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/from-to-continue.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/sml_json_alltypes-interlace.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/commandline-retry.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/tmq_case.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/limit_offset_json.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/commandline-sml.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/sml_insert_alltypes_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_insert_alltypes_json.py
# benchmark 66 cases
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/sml_taosjson_insert_alltypes-same-min-max.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_insert_alltypes-same-min-max.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/bugs.py -B
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_sample_csv_json-subtable.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/query_json-with-error-sqlfile.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosc_insert-retry-json-global.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/from-to.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/exportCsv.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosdemoTestQueryWithJson.py -R
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline-partial-col-numpy.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/query_json-with-sqlfile.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/query_json.py -B
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_json_alltypes.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/invalid_commandline.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_json_insert_alltypes-same-min-max.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/default_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_sample_csv_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_sample_csv_json_doesnt_use_ts.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosadapter_json.py -B
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/demo.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline-sml-rest.py -R
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline-single-table.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline-supplement-insert.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline-vgroups.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/custom_col_tag.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/default_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/default_tmq_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/demo.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/exportCsv.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/from-to.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/from-to-continue.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/json_tag.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/insert-json-csv.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/insertBasic.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/insertBindVGroup.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/insertMix.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/insertPrecision.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/invalid_commandline.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/query_json.py -B
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/query_json-with-error-sqlfile.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/query_json-with-sqlfile.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/rest_insert_alltypes_json.py -R
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/reuse-exist-stb.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_auto_create_table_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_json_alltypes.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_json_insert_alltypes-same-min-max.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_taosjson_alltypes.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_telnet_alltypes.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_telnet_insert_alltypes-same-min-max.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_auto_create_table_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_insert_alltypes_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_insert_alltypes-same-min-max.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline-vgroups.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosc_insert-retry-json-stb.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_auto_create_table_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_telnet_alltypes.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_offset_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_sample_csv_json_doesnt_use_ts.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_sample_csv_json-subtable.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt_sample_csv_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stmt2_insert.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stream-test.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/sml_taosjson_alltypes.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/commandline-single-table.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stream_function_test.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/stt.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_auto_create_table_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_sample_csv_json-subtable.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_insert_alltypes-same-min-max.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_insert_alltypes_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_insert_alltypes_json-partial-col.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_insert-table-creating-interval.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosc_sample_csv_json.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosdemoTestInsertWithJsonStmt-otherPara.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/taosdemoTestQueryWithJson.py -R
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/telnet_tcp.py -R
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/basic/tmq_case.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/cloud/cloud-test.py
,,y,army,./pytest.sh python3 ./test.py -f tools/benchmark/ws/websocket.py -R
,,n,army,python3 ./test.py -f tools/benchmark/basic/bugs.py -B
,,n,army,python3 ./test.py -f tools/benchmark/basic/commandline-retry.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/commandline-sml.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/sml_json_alltypes-interlace.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/sml_insert_alltypes_json.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/sml_interlace.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/sml_taosjson_insert_alltypes-same-min-max.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosadapter_json.py -B
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosc_insert-mix.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosc_insert-retry-json-global.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosc_insert-retry-json-stb.py
,,n,army,python3 ./test.py -f tools/benchmark/basic/taosdemoTestQueryWithJson-mixed-query.py -R
# taosdump 43 cases
,,y,army,./pytest.sh python3 ./test.py -f tools/taosdump/native/taosdumpCompa.py
,,y,army,./pytest.sh python3 ./test.py -f tools/taosdump/native/taosdumpTest.py
@ -191,6 +206,7 @@
,,n,army,python3 ./test.py -f tools/taosdump/ws/taosdumpTestTypeBool.py -B
,,n,army,python3 ./test.py -f tools/taosdump/ws/taosdumpRetry.py -B
,,n,army,python3 ./test.py -f tools/taosdump/ws/taosdumpTestTypeTinyInt.py -B
,,y,army,./pytest.sh python3 ./test.py -f tools/taosdump/native/taosdumpCommandline.py
#
# system test

2
tools/shell/test/shellTest.cpp
View File

@ -18,7 +18,7 @@
#include "shellAuto.h"
TEST(fieldOptionsArea, autoTabTest) {
printf("hellow world SHELL tab test\n");
printf("hello world SHELL tab test\n");
// str false
const char *s0[] = {

8
tools/taos-tools/CMakeLists.txt
View File

@ -118,3 +118,11 @@ ELSEIF (${CMAKE_SYSTEM_NAME} MATCHES "Windows")
SET(MAKE_INSTALL_SH "${PROJECT_SOURCE_DIR}/packaging/tools/make_install.bat")
INSTALL(CODE "execute_process(COMMAND ${MAKE_INSTALL_SH} :needAdmin ${PROJECT_SOURCE_DIR} ${PROJECT_BINARY_DIR} Windows)")
ENDIF ()
# add test
IF(TD_LINUX)
# unit test
IF(${BUILD_TEST})
ADD_SUBDIRECTORY(test)
ENDIF(${BUILD_TEST})
ENDIF()

10
tools/taos-tools/deps/toolscJson/src/toolscJson.c
View File

@ -222,6 +222,7 @@ static tools_cJSON_bool parse_number(tools_cJSON * const item, parse_buffer * co
unsigned char number_c_string[64];
unsigned char decimal_point = get_decimal_point();
size_t i = 0;
int isFloat = 0;
if ((input_buffer == NULL) || (input_buffer->content == NULL))
{
@ -245,14 +246,18 @@ static tools_cJSON_bool parse_number(tools_cJSON * const item, parse_buffer * co
case '7':
case '8':
case '9':
number_c_string[i] = buffer_at_offset(input_buffer)[i];
break;
case '+':
case '-':
case 'e':
case 'E':
isFloat = 1;
number_c_string[i] = buffer_at_offset(input_buffer)[i];
break;
case '.':
isFloat = 1;
number_c_string[i] = decimal_point;
break;
@ -282,7 +287,10 @@ loop_end:
}
else
{
item->valueint = (int64_t)number;
if(isFloat)
item->valueint = (int64_t)number;
else
item->valueint = strtoll((const char*)number_c_string, (char**)&after_end, 10);
}
item->type = tools_cJSON_Number;

2
tools/taos-tools/inc/bench.h
View File

@ -773,9 +773,7 @@ typedef struct SArguments_S {
int rest_server_ver_major;
bool check_sql;
int suit; // see define SUIT_
#ifdef TD_VER_COMPATIBLE_3_0_0_0
int16_t inputted_vgroups;
#endif
enum CONTINUE_IF_FAIL_MODE continueIfFail;
bool mistMode;
bool escape_character;

9
tools/taos-tools/inc/toolsdef.h
View File

@ -16,9 +16,14 @@
#ifndef __TOOLSDEF_H_
#define __TOOLSDEF_H_
#ifdef __cplusplus
extern "C" {
#endif
#include <stdbool.h>
#include <time.h>
#define TINY_BUFF_LEN 8
#define SMALL_BUFF_LEN 20
#define MIDDLE_BUFF_LEN 64
@ -236,4 +241,8 @@ int setConsoleEcho(bool on);
char *toolsFormatTimestamp(char *buf, int64_t val, int32_t precision);
#ifdef __cplusplus
} /* end extern "C" */
#endif
#endif // __TOOLSDEF_H_

59
tools/taos-tools/src/benchInsert.c
View File

@ -607,14 +607,10 @@ int32_t toolsGetDefaultVGroups() {
infoPrint("check local machine CPU: %d Memory:%d MB \n", cores, (int32_t)(MemKB/1024));
if (MemKB <= 2*1024*1024) { // 2G
return 1;
} else if (MemKB <= 4*1024*1024) { // 4G
} else if (MemKB <= 256*1024*1024) { // 256G
return 2;
} else if (MemKB <= 8*1024*1024) { // 8G
return 3;
} else if (MemKB <= 16*1024*1024) { // 16G
} else if (MemKB <= 512*1024*1024) { // 512G
return 4;
} else if (MemKB <= 32*1024*1024) { // 32G
return 5;
} else {
return cores / 2;
}
@ -623,22 +619,13 @@ int32_t toolsGetDefaultVGroups() {
int geneDbCreateCmd(SDataBase *database, char *command, int remainVnodes) {
int dataLen = 0;
int n;
if (-1 != g_arguments->inputted_vgroups) {
n = snprintf(command + dataLen, SHORT_1K_SQL_BUFF_LEN - dataLen,
g_arguments->escape_character
? "CREATE DATABASE IF NOT EXISTS `%s` VGROUPS %d"
: "CREATE DATABASE IF NOT EXISTS %s VGROUPS %d",
database->dbName,
(-1 != g_arguments->inputted_vgroups)?
g_arguments->inputted_vgroups:
min(remainVnodes, toolsGetNumberOfCores()));
} else {
n = snprintf(command + dataLen, SHORT_1K_SQL_BUFF_LEN - dataLen,
g_arguments->escape_character
? "CREATE DATABASE IF NOT EXISTS `%s`"
: "CREATE DATABASE IF NOT EXISTS %s",
database->dbName);
}
// create database
n = snprintf(command + dataLen, SHORT_1K_SQL_BUFF_LEN - dataLen,
g_arguments->escape_character
? "CREATE DATABASE IF NOT EXISTS `%s`"
: "CREATE DATABASE IF NOT EXISTS %s",
database->dbName);
if (n < 0 || n >= SHORT_1K_SQL_BUFF_LEN - dataLen) {
errorPrint("%s() LN%d snprintf overflow\n",
@ -648,9 +635,24 @@ int geneDbCreateCmd(SDataBase *database, char *command, int remainVnodes) {
dataLen += n;
}
int vgroups = g_arguments->inputted_vgroups;
// append config items
if (database->cfgs) {
for (int i = 0; i < database->cfgs->size; i++) {
SDbCfg* cfg = benchArrayGet(database->cfgs, i);
// check vgroups
if (strcasecmp(cfg->name, "vgroups") == 0) {
if (vgroups > 0) {
// inputted vgroups by commandline
infoPrint("ignore config set vgroups %d\n", cfg->valueint);
} else {
vgroups = cfg->valueint;
}
continue;
}
if (cfg->valuestring) {
n = snprintf(command + dataLen,
TSDB_MAX_ALLOWED_SQL_LEN - dataLen,
@ -670,6 +672,17 @@ int geneDbCreateCmd(SDataBase *database, char *command, int remainVnodes) {
}
}
// benchmark default
if (vgroups < 1) {
vgroups = toolsGetDefaultVGroups();
debugPrint("vgroup set with toolsGetDefaultVGroups(). vgroups=%d\n", vgroups);
}
// not found vgroups
if (vgroups > 0) {
dataLen += snprintf(command + dataLen, TSDB_MAX_ALLOWED_SQL_LEN - dataLen, " VGROUPS %d", vgroups);
}
switch (database->precision) {
case TSDB_TIME_PRECISION_MILLI:
snprintf(command + dataLen, TSDB_MAX_ALLOWED_SQL_LEN - dataLen,
@ -4137,6 +4150,8 @@ int32_t runInsertLimitThread(SDataBase* database, SSuperTable* stbInfo, int32_t
}
}
tmfree(slot);
return 0;
}
#endif

30
tools/taos-tools/src/toolsDir.c
View File

@ -93,36 +93,6 @@ typedef struct dirent TdDirEntry;
#endif
int32_t toolsExpandDir(const char *dirname, char *outname, int32_t maxlen) {
wordexp_t full_path;
switch (wordexp(dirname, &full_path, 0)) {
case 0:
break;
case WRDE_NOSPACE:
wordfree(&full_path);
// printf("failed to expand path:%s since Out of memory\n", dirname);
return -1;
case WRDE_BADCHAR:
// printf("failed to expand path:%s since illegal occurrence of newline or one of |, &, ;, <, >, (, ), {, }\n",
// dirname);
return -1;
case WRDE_SYNTAX:
// printf("failed to expand path:%s since Shell syntax error, such as unbalanced parentheses or unmatched
// quotes\n", dirname);
return -1;
default:
// printf("failed to expand path:%s since %s\n", dirname, strerror(errno));
return -1;
}
if (full_path.we_wordv != NULL && full_path.we_wordv[0] != NULL) {
strncpy(outname, full_path.we_wordv[0], maxlen);
}
wordfree(&full_path);
return 0;
}
TdDirPtr toolsOpenDir(const char *dirname) {
if (dirname == NULL) {

44
tools/taos-tools/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,44 @@
MESSAGE(STATUS "build taos-tools unit test")
IF(TD_LINUX)
# GoogleTest requires at least C++11
SET(CMAKE_CXX_STANDARD 11)
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
# benchmark
ADD_EXECUTABLE(benchmarkTest benchmarkTest.cpp)
TARGET_LINK_LIBRARIES(
benchmarkTest
PRIVATE gtest
)
target_include_directories(
benchmarkTest
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
add_test(
NAME benchmarkTest
COMMAND benchmarkTest
)
# taosdump
ADD_EXECUTABLE(taosdumpTest taosdumpTest.cpp ../src/toolsSys.c)
TARGET_LINK_LIBRARIES(
taosdumpTest
PRIVATE gtest
)
target_include_directories(
taosdumpTest
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
)
add_test(
NAME taosdumpTest
COMMAND taosdumpTest
)
ENDIF()

26
tools/taos-tools/test/benchmarkTest.cpp Normal file
View File

@ -0,0 +1,26 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
#include <iostream>
TEST(jsonTest, taosBenchmarkTest) {
printf("hello world taosBenchmark unit test for C \n");
}
int main(int argc, char **argv) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

36
tools/taos-tools/test/taosdumpTest.cpp Normal file
View File

@ -0,0 +1,36 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h>
#include <iostream>
#include "toolsdef.h"
TEST(taosdump, toolsSys) {
// errorPrintReqArg3
errorPrintReqArg3((char *)"taosdump", (char *)"test parameters");
printf("ut function errorPrintReqArg3 .................... [Passed]\n");
// setConsoleEcho
setConsoleEcho(true);
setConsoleEcho(false);
printf("ut function setConsoleEcho ....................... [Passed]\n");
}
int main(int argc, char **argv) {
printf("hello world taosdump unit test for C\n");
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

4
utils/tsim/src/simParse.c
View File

@ -182,7 +182,9 @@ SScript *simParseScript(char *fileName) {
SCommand *pCmd = NULL;
SScript *script = NULL;
if ((fileName[0] == '.') || (fileName[0] == '/')) {
if (fileName[0] == 0) {
return NULL;
} else if ((fileName[0] == '.') || (fileName[0] == '/')) {
tstrncpy(name, fileName, PATH_MAX);
} else {
snprintf(name, PATH_MAX, "%s" TD_DIRSEP "%s", simScriptDir, fileName);

14
utils/tsim/test/CMakeLists.txt
View File

@ -14,10 +14,12 @@ ENDIF()
INCLUDE_DIRECTORIES(${TD_SOURCE_DIR}/src/util/inc)
ADD_EXECUTABLE(simTests "simTests.cpp")
TARGET_LINK_LIBRARIES(simTests os util tsim_static gtest_main)
IF(TD_LINUX)
ADD_EXECUTABLE(simTests "simTests.cpp")
TARGET_LINK_LIBRARIES(simTests os util tsim_static gtest_main)
ADD_TEST(
NAME simTests
COMMAND simTests
)
ADD_TEST(
NAME simTests
COMMAND simTests
)
ENDIF()

40
utils/tsim/test/simTests.cpp
View File

@ -32,30 +32,30 @@ void simHandleSignal(int32_t signo, void *sigInfo, void *context);
TEST(simTests, parameters) {
int32_t ret = 0;
int32_t argc = 2;
int32_t argc = 3;
char *argv[4] = {0};
simSystemCleanUp();
// argv[1] = "-c";
// ret = simEntry(argc, argv);
// EXPECT_EQ(ret, 0);
argc = 3;
argv[1] = "-f";
argv[2] = "";
ret = simEntry(argc, argv);
EXPECT_EQ(ret, -1);
// argv[1] = "-f";
// ret = simEntry(argc, argv);
// EXPECT_EQ(ret, 0);
argc = 4;
argv[3] = "-v";
ret = simEntry(argc, argv);
EXPECT_EQ(ret, -1);
// argv[1] = "-v";
// ret = simEntry(argc, argv);
// EXPECT_EQ(ret, 0);
argc = 5;
argv[3] = "-c";
argv[4] = "/etc/taos";
ret = simEntry(argc, argv);
EXPECT_EQ(ret, -1);
// argv[1] = "-h";
// ret = simEntry(argc, argv);
// EXPECT_EQ(ret, 0);
argc = 4;
argv[3] = "-h";
ret = simEntry(argc, argv);
EXPECT_EQ(ret, 0);
// simHandleSignal(0, NULL, NULL);
// simDebugFlag = 0;
// argc = 1;
// ret = simEntry(argc, argv);
// EXPECT_EQ(ret, -1);
simHandleSignal(0, NULL, NULL);
}