提交 c23043c0 编写于 作者: S Shuduo Sang

fix child table exists or auto create logic.

上级 6fe3b972
...@@ -240,13 +240,13 @@ typedef struct SColumn_S { ...@@ -240,13 +240,13 @@ typedef struct SColumn_S {
typedef struct SSuperTable_S { typedef struct SSuperTable_S {
char sTblName[MAX_TB_NAME_SIZE+1]; char sTblName[MAX_TB_NAME_SIZE+1];
char dataSource[MAX_TB_NAME_SIZE+1]; // rand_gen or sample
char childTblPrefix[MAX_TB_NAME_SIZE];
char insertMode[MAX_TB_NAME_SIZE]; // taosc, rest
uint16_t childTblExists;
int64_t childTblCount; int64_t childTblCount;
bool childTblExists; // 0: no, 1: yes
uint64_t batchCreateTableNum; // 0: no batch, > 0: batch table number in one sql uint64_t batchCreateTableNum; // 0: no batch, > 0: batch table number in one sql
uint8_t autoCreateTable; // 0: create sub table, 1: auto create sub table uint8_t autoCreateTable; // 0: create sub table, 1: auto create sub table
char childTblPrefix[MAX_TB_NAME_SIZE];
char dataSource[MAX_TB_NAME_SIZE+1]; // rand_gen or sample
char insertMode[MAX_TB_NAME_SIZE]; // taosc, rest
int64_t childTblLimit; int64_t childTblLimit;
uint64_t childTblOffset; uint64_t childTblOffset;
...@@ -793,7 +793,7 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) { ...@@ -793,7 +793,7 @@ static void parse_args(int argc, char *argv[], SArguments *arguments) {
if ((argc == i+1) if ((argc == i+1)
|| (!isStringNumber(argv[i+1]))) { || (!isStringNumber(argv[i+1]))) {
printHelp(); printHelp();
errorPrint("%s", "\n\t-q need a number following!\nQuery mode -- 0: SYNC, 1: ASYNC. Default is SYNC.\n"); errorPrint("%s", "\n\t-q need a number following!\nQuery mode -- 0: SYNC, not-0: ASYNC. Default is SYNC.\n");
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
arguments->async_mode = atoi(argv[++i]); arguments->async_mode = atoi(argv[++i]);
...@@ -1417,7 +1417,8 @@ static int printfInsertMeta() { ...@@ -1417,7 +1417,8 @@ static int printfInsertMeta() {
if (PRE_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) { if (PRE_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) {
printf(" autoCreateTable: \033[33m%s\033[0m\n", "no"); printf(" autoCreateTable: \033[33m%s\033[0m\n", "no");
} else if (AUTO_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) { } else if (AUTO_CREATE_SUBTBL ==
g_Dbs.db[i].superTbls[j].autoCreateTable) {
printf(" autoCreateTable: \033[33m%s\033[0m\n", "yes"); printf(" autoCreateTable: \033[33m%s\033[0m\n", "yes");
} else { } else {
printf(" autoCreateTable: \033[33m%s\033[0m\n", "error"); printf(" autoCreateTable: \033[33m%s\033[0m\n", "error");
...@@ -3012,10 +3013,10 @@ static void createChildTables() { ...@@ -3012,10 +3013,10 @@ static void createChildTables() {
if (g_Dbs.db[i].superTblCount > 0) { if (g_Dbs.db[i].superTblCount > 0) {
// with super table // with super table
for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) { for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
if (AUTO_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) { if ((AUTO_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable)
|| (TBL_ALREADY_EXISTS == g_Dbs.db[i].superTbls[j].childTblExists)) {
continue; continue;
} }
verbosePrint("%s() LN%d: %s\n", __func__, __LINE__, verbosePrint("%s() LN%d: %s\n", __func__, __LINE__,
g_Dbs.db[i].superTbls[j].colsOfCreateChildTable); g_Dbs.db[i].superTbls[j].colsOfCreateChildTable);
uint64_t startFrom = 0; uint64_t startFrom = 0;
...@@ -3023,6 +3024,7 @@ static void createChildTables() { ...@@ -3023,6 +3024,7 @@ static void createChildTables() {
verbosePrint("%s() LN%d: create %"PRId64" child tables from %"PRIu64"\n", verbosePrint("%s() LN%d: create %"PRId64" child tables from %"PRIu64"\n",
__func__, __LINE__, g_totalChildTables, startFrom); __func__, __LINE__, g_totalChildTables, startFrom);
startMultiThreadCreateChildTable( startMultiThreadCreateChildTable(
g_Dbs.db[i].superTbls[j].colsOfCreateChildTable, g_Dbs.db[i].superTbls[j].colsOfCreateChildTable,
g_Dbs.threadCountByCreateTbl, g_Dbs.threadCountByCreateTbl,
...@@ -4864,14 +4866,13 @@ static void getTableName(char *pTblName, ...@@ -4864,14 +4866,13 @@ static void getTableName(char *pTblName,
threadInfo* pThreadInfo, uint64_t tableSeq) threadInfo* pThreadInfo, uint64_t tableSeq)
{ {
SSuperTable* superTblInfo = pThreadInfo->superTblInfo; SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
if ((superTblInfo) if (superTblInfo) {
&& (AUTO_CREATE_SUBTBL != superTblInfo->autoCreateTable)) { if (AUTO_CREATE_SUBTBL != superTblInfo->autoCreateTable) {
if (superTblInfo->childTblLimit > 0) { if (superTblInfo->childTblLimit > 0) {
snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s", snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s",
superTblInfo->childTblName + superTblInfo->childTblName +
(tableSeq - superTblInfo->childTblOffset) * TSDB_TABLE_NAME_LEN); (tableSeq - superTblInfo->childTblOffset) * TSDB_TABLE_NAME_LEN);
} else { } else {
verbosePrint("[%d] %s() LN%d: from=%"PRIu64" count=%"PRId64" seq=%"PRIu64"\n", verbosePrint("[%d] %s() LN%d: from=%"PRIu64" count=%"PRId64" seq=%"PRIu64"\n",
pThreadInfo->threadID, __func__, __LINE__, pThreadInfo->threadID, __func__, __LINE__,
pThreadInfo->start_table_from, pThreadInfo->start_table_from,
...@@ -4879,6 +4880,10 @@ static void getTableName(char *pTblName, ...@@ -4879,6 +4880,10 @@ static void getTableName(char *pTblName,
snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s", snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s",
superTblInfo->childTblName + tableSeq * TSDB_TABLE_NAME_LEN); superTblInfo->childTblName + tableSeq * TSDB_TABLE_NAME_LEN);
} }
} else {
snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s%"PRIu64"",
superTblInfo->childTblPrefix, tableSeq);
}
} else { } else {
snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s%"PRIu64"", snprintf(pTblName, TSDB_TABLE_NAME_LEN, "%s%"PRIu64"",
g_args.tb_prefix, tableSeq); g_args.tb_prefix, tableSeq);
...@@ -6195,7 +6200,7 @@ static int insertTestProcess() { ...@@ -6195,7 +6200,7 @@ static int insertTestProcess() {
} }
} }
taosMsleep(1000); //taosMsleep(1000);
// create sub threads for inserting data // create sub threads for inserting data
//start = taosGetTimestampMs(); //start = taosGetTimestampMs();
for (int i = 0; i < g_Dbs.dbCount; i++) { for (int i = 0; i < g_Dbs.dbCount; i++) {
...@@ -6839,11 +6844,14 @@ static void *specifiedSubscribe(void *sarg) { ...@@ -6839,11 +6844,14 @@ static void *specifiedSubscribe(void *sarg) {
g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID] = taos_consume( g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID] = taos_consume(
g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID]); g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID]);
if (g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID]) { if (g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID]) {
if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq][0] != 0) { if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq][0]
!= 0) {
sprintf(pThreadInfo->filePath, "%s-%d", sprintf(pThreadInfo->filePath, "%s-%d",
g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq], g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq],
pThreadInfo->threadID); pThreadInfo->threadID);
fetchResult(g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID], pThreadInfo); fetchResult(
g_queryInfo.specifiedQueryInfo.res[pThreadInfo->threadID],
pThreadInfo);
} }
g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] ++; g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] ++;
...@@ -6856,7 +6864,8 @@ static void *specifiedSubscribe(void *sarg) { ...@@ -6856,7 +6864,8 @@ static void *specifiedSubscribe(void *sarg) {
g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] = 0; g_queryInfo.specifiedQueryInfo.consumed[pThreadInfo->threadID] = 0;
taos_unsubscribe(g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID], taos_unsubscribe(g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID],
g_queryInfo.specifiedQueryInfo.subscribeKeepProgress); g_queryInfo.specifiedQueryInfo.subscribeKeepProgress);
g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID] = subscribeImpl( g_queryInfo.specifiedQueryInfo.tsub[pThreadInfo->threadID] =
subscribeImpl(
SPECIFIED_CLASS, SPECIFIED_CLASS,
pThreadInfo, pThreadInfo,
g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq], g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq],
...@@ -7078,7 +7087,7 @@ static void setParaFromArg(){ ...@@ -7078,7 +7087,7 @@ static void setParaFromArg(){
g_Dbs.threadCountByCreateTbl = g_args.num_of_threads; g_Dbs.threadCountByCreateTbl = g_args.num_of_threads;
g_Dbs.dbCount = 1; g_Dbs.dbCount = 1;
g_Dbs.db[0].drop = 1; g_Dbs.db[0].drop = true;
tstrncpy(g_Dbs.db[0].dbName, g_args.database, MAX_DB_NAME_SIZE); tstrncpy(g_Dbs.db[0].dbName, g_args.database, MAX_DB_NAME_SIZE);
g_Dbs.db[0].dbCfg.replica = g_args.replica; g_Dbs.db[0].dbCfg.replica = g_args.replica;
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册