Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
未验证 提交 c1e168e6 编写于 作者: H huili 提交者： GitHub
浏览文件
操作

Merge pull request #4246 from taosdata/hotfix/test 

[NONE]
上级 1af6ea5c 021ea48b
隐藏空白更改
内联 并排
Showing with 102 addition and 66 deletion
src/kit/taosdump/taosdump.c
浏览文件 @ c1e168e6
...@@ -40,19 +40,22 @@ typedef struct { ...@@ -40,19 +40,22 @@ typedef struct {
enum _show_db_index { enum _show_db_index {
TSDB_SHOW_DB_NAME_INDEX, TSDB_SHOW_DB_NAME_INDEX,
TSDB_SHOW_DB_CREATED_TIME_INDEX, TSDB_SHOW_DB_CREATED_TIME_INDEX,
TSDB_SHOW_DB_VGROUPS_INDEX,
TSDB_SHOW_DB_NTABLES_INDEX, TSDB_SHOW_DB_NTABLES_INDEX,
TSDB_SHOW_DB_VGROUPS_INDEX,
TSDB_SHOW_DB_REPLICA_INDEX, TSDB_SHOW_DB_REPLICA_INDEX,
TSDB_SHOW_DB_QUORUM_INDEX,
TSDB_SHOW_DB_DAYS_INDEX, TSDB_SHOW_DB_DAYS_INDEX,
TSDB_SHOW_DB_KEEP_INDEX, TSDB_SHOW_DB_KEEP_INDEX,
TSDB_SHOW_DB_TABLES_INDEX,
TSDB_SHOW_DB_ROWS_INDEX,
TSDB_SHOW_DB_CACHE_INDEX, TSDB_SHOW_DB_CACHE_INDEX,
TSDB_SHOW_DB_ABLOCKS_INDEX, TSDB_SHOW_DB_BLOCKS_INDEX,
TSDB_SHOW_DB_TBLOCKS_INDEX, TSDB_SHOW_DB_MINROWS_INDEX,
TSDB_SHOW_DB_CTIME_INDEX, TSDB_SHOW_DB_MAXROWS_INDEX,
TSDB_SHOW_DB_CLOG_INDEX, TSDB_SHOW_DB_WALLEVEL_INDEX,
TSDB_SHOW_DB_FSYNC_INDEX,
TSDB_SHOW_DB_COMP_INDEX, TSDB_SHOW_DB_COMP_INDEX,
TSDB_SHOW_DB_PRECISION_INDEX,
TSDB_SHOW_DB_UPDATE_INDEX,
TSDB_SHOW_DB_STATUS_INDEX,
TSDB_MAX_SHOW_DB TSDB_MAX_SHOW_DB
}; };
...@@ -90,17 +93,23 @@ extern char version[]; ...@@ -90,17 +93,23 @@ extern char version[];
typedef struct { typedef struct {
char name[TSDB_DB_NAME_LEN + 1]; char name[TSDB_DB_NAME_LEN + 1];
int32_t replica; int32_t tables;
int32_t days; int32_t vgroups;
int32_t keep; int16_t replications;
int32_t tables; int16_t quorum;
int32_t rows; int16_t daysPerFile;
int32_t cache; int16_t daysToKeep;
int32_t ablocks; int16_t daysToKeep1;
int32_t tblocks; int16_t daysToKeep2;
int32_t ctime; int32_t cacheBlockSize; //MB
int32_t clog; int32_t totalBlocks;
int32_t comp; int32_t minRowsPerFileBlock;
int32_t maxRowsPerFileBlock;
int8_t walLevel;
int32_t fsyncPeriod;
int8_t compression;
int8_t precision; // time resolution
int8_t update;
} SDbInfo; } SDbInfo;
typedef struct { typedef struct {
...@@ -173,6 +182,7 @@ static struct argp_option options[] = { ...@@ -173,6 +182,7 @@ static struct argp_option options[] = {
{"start-time", 'S', "START_TIME", 0, "Start time to dump.", 3}, {"start-time", 'S', "START_TIME", 0, "Start time to dump.", 3},
{"end-time", 'E', "END_TIME", 0, "End time to dump.", 3}, {"end-time", 'E', "END_TIME", 0, "End time to dump.", 3},
{"data-batch", 'N', "DATA_BATCH", 0, "Number of data point per insert statement. Default is 1.", 3}, {"data-batch", 'N', "DATA_BATCH", 0, "Number of data point per insert statement. Default is 1.", 3},
{"max-sql-len", 'L', "SQL_LEN", 0, "Max length of one sql. Default is 65480.", 3},
{"table-batch", 't', "TABLE_BATCH", 0, "Number of table dumpout into one output file. Default is 1.", 3}, {"table-batch", 't', "TABLE_BATCH", 0, "Number of table dumpout into one output file. Default is 1.", 3},
{"thread_num", 'T', "THREAD_NUM", 0, "Number of thread for dump in file. Default is 5.", 3}, {"thread_num", 'T', "THREAD_NUM", 0, "Number of thread for dump in file. Default is 5.", 3},
{"allow-sys", 'a', 0, 0, "Allow to dump sys database", 3}, {"allow-sys", 'a', 0, 0, "Allow to dump sys database", 3},
...@@ -200,6 +210,7 @@ struct arguments { ...@@ -200,6 +210,7 @@ struct arguments {
int64_t start_time; int64_t start_time;
int64_t end_time; int64_t end_time;
int32_t data_batch; int32_t data_batch;
int32_t max_sql_len;
int32_t table_batch; // num of table which will be dump into one output file. int32_t table_batch; // num of table which will be dump into one output file.
bool allow_sys; bool allow_sys;
// other options // other options
...@@ -298,6 +309,17 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) { ...@@ -298,6 +309,17 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
case 'N': case 'N':
arguments->data_batch = atoi(arg); arguments->data_batch = atoi(arg);
break; break;
case 'L':
{
int32_t len = atoi(arg);
if (len > TSDB_MAX_ALLOWED_SQL_LEN) {
len = TSDB_MAX_ALLOWED_SQL_LEN;
} else if (len < TSDB_MAX_SQL_LEN) {
len = TSDB_MAX_SQL_LEN;
}
arguments->max_sql_len = len;
break;
}
case 't': case 't':
arguments->table_batch = atoi(arg); arguments->table_batch = atoi(arg);
break; break;
...@@ -360,6 +382,7 @@ struct arguments tsArguments = { ...@@ -360,6 +382,7 @@ struct arguments tsArguments = {
0, 0,
INT64_MAX, INT64_MAX,
1, 1,
TSDB_MAX_SQL_LEN,
1, 1,
false, false,
// other options // other options
...@@ -415,7 +438,9 @@ int main(int argc, char *argv[]) { ...@@ -415,7 +438,9 @@ int main(int argc, char *argv[]) {
printf("start_time: %" PRId64 "\n", tsArguments.start_time); printf("start_time: %" PRId64 "\n", tsArguments.start_time);
printf("end_time: %" PRId64 "\n", tsArguments.end_time); printf("end_time: %" PRId64 "\n", tsArguments.end_time);
printf("data_batch: %d\n", tsArguments.data_batch); printf("data_batch: %d\n", tsArguments.data_batch);
printf("max_sql_len: %d\n", tsArguments.max_sql_len);
printf("table_batch: %d\n", tsArguments.table_batch); printf("table_batch: %d\n", tsArguments.table_batch);
printf("thread_num: %d\n", tsArguments.thread_num);
printf("allow_sys: %d\n", tsArguments.allow_sys); printf("allow_sys: %d\n", tsArguments.allow_sys);
printf("abort: %d\n", tsArguments.abort); printf("abort: %d\n", tsArguments.abort);
printf("isDumpIn: %d\n", tsArguments.isDumpIn); printf("isDumpIn: %d\n", tsArguments.isDumpIn);
...@@ -682,8 +707,8 @@ int taosDumpOut(struct arguments *arguments) { ...@@ -682,8 +707,8 @@ int taosDumpOut(struct arguments *arguments) {
TAOS_FIELD *fields = taos_fetch_fields(result); TAOS_FIELD *fields = taos_fetch_fields(result);
while ((row = taos_fetch_row(result)) != NULL) { while ((row = taos_fetch_row(result)) != NULL) {
// sys database name : 'monitor', but subsequent version changed to 'log' // sys database name : 'log', but subsequent version changed to 'log'
if (strncasecmp(row[TSDB_SHOW_DB_NAME_INDEX], "monitor", fields[TSDB_SHOW_DB_NAME_INDEX].bytes) == 0 && if (strncasecmp(row[TSDB_SHOW_DB_NAME_INDEX], "log", fields[TSDB_SHOW_DB_NAME_INDEX].bytes) == 0 &&
(!arguments->allow_sys)) (!arguments->allow_sys))
continue; continue;
...@@ -711,20 +736,27 @@ int taosDumpOut(struct arguments *arguments) { ...@@ -711,20 +736,27 @@ int taosDumpOut(struct arguments *arguments) {
} }
strncpy(dbInfos[count]->name, (char *)row[TSDB_SHOW_DB_NAME_INDEX], fields[TSDB_SHOW_DB_NAME_INDEX].bytes); strncpy(dbInfos[count]->name, (char *)row[TSDB_SHOW_DB_NAME_INDEX], fields[TSDB_SHOW_DB_NAME_INDEX].bytes);
#if 0 #if 0
dbInfos[count]->replica = (int)(*((int16_t *)row[TSDB_SHOW_DB_REPLICA_INDEX])); if (arguments->with_property) {
dbInfos[count]->days = (int)(*((int16_t *)row[TSDB_SHOW_DB_DAYS_INDEX])); dbInfos[count]->tables = *((int32_t *)row[TSDB_SHOW_DB_NTABLES_INDEX]);
dbInfos[count]->keep = *((int *)row[TSDB_SHOW_DB_KEEP_INDEX]); dbInfos[count]->vgroups = *((int32_t *)row[TSDB_SHOW_DB_VGROUPS_INDEX]);
dbInfos[count]->tables = *((int *)row[TSDB_SHOW_DB_TABLES_INDEX]); dbInfos[count]->replications = *((int16_t *)row[TSDB_SHOW_DB_REPLICA_INDEX]);
dbInfos[count]->rows = *((int *)row[TSDB_SHOW_DB_ROWS_INDEX]); dbInfos[count]->quorum = *((int16_t *)row[TSDB_SHOW_DB_QUORUM_INDEX]);
dbInfos[count]->cache = *((int *)row[TSDB_SHOW_DB_CACHE_INDEX]); dbInfos[count]->daysPerFile = *((int16_t *)row[TSDB_SHOW_DB_DAYS_INDEX]);
dbInfos[count]->ablocks = *((int *)row[TSDB_SHOW_DB_ABLOCKS_INDEX]); dbInfos[count]->daysToKeep = *((int16_t *)row[TSDB_SHOW_DB_KEEP_INDEX]);
dbInfos[count]->tblocks = (int)(*((int16_t *)row[TSDB_SHOW_DB_TBLOCKS_INDEX])); dbInfos[count]->daysToKeep1;
dbInfos[count]->ctime = *((int *)row[TSDB_SHOW_DB_CTIME_INDEX]); dbInfos[count]->daysToKeep2;
dbInfos[count]->clog = (int)(*((int8_t *)row[TSDB_SHOW_DB_CLOG_INDEX])); dbInfos[count]->cacheBlockSize = *((int32_t *)row[TSDB_SHOW_DB_CACHE_INDEX]);
dbInfos[count]->comp = (int)(*((int8_t *)row[TSDB_SHOW_DB_COMP_INDEX])); dbInfos[count]->totalBlocks = *((int32_t *)row[TSDB_SHOW_DB_BLOCKS_INDEX]);
dbInfos[count]->minRowsPerFileBlock = *((int32_t *)row[TSDB_SHOW_DB_MINROWS_INDEX]);
dbInfos[count]->maxRowsPerFileBlock = *((int32_t *)row[TSDB_SHOW_DB_MAXROWS_INDEX]);
dbInfos[count]->walLevel = *((int8_t *)row[TSDB_SHOW_DB_WALLEVEL_INDEX]);
dbInfos[count]->fsyncPeriod = *((int32_t *)row[TSDB_SHOW_DB_FSYNC_INDEX]);
dbInfos[count]->compression = (int8_t)(*((int8_t *)row[TSDB_SHOW_DB_COMP_INDEX]));
dbInfos[count]->precision = *((int8_t *)row[TSDB_SHOW_DB_PRECISION_INDEX]);
dbInfos[count]->update = *((int8_t *)row[TSDB_SHOW_DB_UPDATE_INDEX]);
}
#endif #endif
count++; count++;
if (arguments->databases) { if (arguments->databases) {
...@@ -1037,10 +1069,13 @@ void taosDumpCreateDbClause(SDbInfo *dbInfo, bool isDumpProperty, FILE *fp) { ...@@ -1037,10 +1069,13 @@ void taosDumpCreateDbClause(SDbInfo *dbInfo, bool isDumpProperty, FILE *fp) {
pstr += sprintf(pstr, "CREATE DATABASE IF NOT EXISTS %s", dbInfo->name); pstr += sprintf(pstr, "CREATE DATABASE IF NOT EXISTS %s", dbInfo->name);
if (isDumpProperty) { if (isDumpProperty) {
#if 0
pstr += sprintf(pstr, pstr += sprintf(pstr,
" REPLICA %d DAYS %d KEEP %d TABLES %d ROWS %d CACHE %d ABLOCKS %d TBLOCKS %d CTIME %d CLOG %d COMP %d", "TABLES %d vgroups %d REPLICA %d quorum %d DAYS %d KEEP %d CACHE %d BLOCKS %d MINROWS %d MAXROWS %d WALLEVEL %d FYNC %d COMP %d PRECISION %s UPDATE %d",
dbInfo->replica, dbInfo->days, dbInfo->keep, dbInfo->tables, dbInfo->rows, dbInfo->cache, dbInfo->tables, dbInfo->vgroups, dbInfo->replications, dbInfo->quorum, dbInfo->daysPerFile, dbInfo->daysToKeep, dbInfo->cacheBlockSize,
dbInfo->ablocks, dbInfo->tblocks, dbInfo->ctime, dbInfo->clog, dbInfo->comp); dbInfo->totalBlocks, dbInfo->minRowsPerFileBlock, dbInfo->maxRowsPerFileBlock, dbInfo->walLevel, dbInfo->fsyncPeriod, dbInfo->compression,
dbInfo->precision, dbInfo->update);
#endif
} }
pstr += sprintf(pstr, ";"); pstr += sprintf(pstr, ";");
...@@ -1459,7 +1494,8 @@ int taosDumpTableData(FILE *fp, char *tbname, struct arguments *arguments, TAOS* ...@@ -1459,7 +1494,8 @@ int taosDumpTableData(FILE *fp, char *tbname, struct arguments *arguments, TAOS*
return -1; return -1;
} }
char* tmpBuffer = (char *)calloc(1, COMMAND_SIZE); int32_t sql_buf_len = arguments->max_sql_len;
char* tmpBuffer = (char *)calloc(1, sql_buf_len + 128);
if (tmpBuffer == NULL) { if (tmpBuffer == NULL) {
fprintf(stderr, "failed to allocate memory\n"); fprintf(stderr, "failed to allocate memory\n");
free(tmpCommand); free(tmpCommand);
...@@ -1502,85 +1538,83 @@ int taosDumpTableData(FILE *fp, char *tbname, struct arguments *arguments, TAOS* ...@@ -1502,85 +1538,83 @@ int taosDumpTableData(FILE *fp, char *tbname, struct arguments *arguments, TAOS*
return -1; return -1;
} }
char sqlStr[8] = "\0";
if (arguments->mysqlFlag) {
sprintf(sqlStr, "INSERT");
} else {
sprintf(sqlStr, "IMPORT");
}
int rowFlag = 0; int rowFlag = 0;
int32_t curr_sqlstr_len = 0;
int32_t total_sqlstr_len = 0;
count = 0; count = 0;
while ((row = taos_fetch_row(tmpResult)) != NULL) { while ((row = taos_fetch_row(tmpResult)) != NULL) {
pstr = tmpBuffer; pstr = tmpBuffer;
curr_sqlstr_len = 0;
int32_t* length = taos_fetch_lengths(tmpResult); // act len int32_t* length = taos_fetch_lengths(tmpResult); // act len
if (count == 0) { if (count == 0) {
pstr += sprintf(pstr, "%s INTO %s VALUES (", sqlStr, tbname); total_sqlstr_len = 0;
} else { curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "INSERT INTO %s VALUES (", tbname);
} else {
if (arguments->mysqlFlag) { if (arguments->mysqlFlag) {
if (0 == rowFlag) { if (0 == rowFlag) {
pstr += sprintf(pstr, "("); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "(");
rowFlag++; rowFlag++;
} else { } else {
pstr += sprintf(pstr, ", ("); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, ", (");
} }
} else { } else {
pstr += sprintf(pstr, "("); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "(");
} }
} }
for (int col = 0; col < numFields; col++) { for (int col = 0; col < numFields; col++) {
if (col != 0) pstr += sprintf(pstr, ", "); if (col != 0) curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, ", ");
if (row[col] == NULL) { if (row[col] == NULL) {
pstr += sprintf(pstr, "NULL"); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "NULL");
continue; continue;
} }
switch (fields[col].type) { switch (fields[col].type) {
case TSDB_DATA_TYPE_BOOL: case TSDB_DATA_TYPE_BOOL:
pstr += sprintf(pstr, "%d", ((((int32_t)(*((char *)row[col]))) == 1) ? 1 : 0)); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "%d", ((((int32_t)(*((char *)row[col]))) == 1) ? 1 : 0));
break; break;
case TSDB_DATA_TYPE_TINYINT: case TSDB_DATA_TYPE_TINYINT:
pstr += sprintf(pstr, "%d", *((int8_t *)row[col])); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "%d", *((int8_t *)row[col]));
break; break;
case TSDB_DATA_TYPE_SMALLINT: case TSDB_DATA_TYPE_SMALLINT:
pstr += sprintf(pstr, "%d", *((int16_t *)row[col])); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "%d", *((int16_t *)row[col]));
break; break;
case TSDB_DATA_TYPE_INT: case TSDB_DATA_TYPE_INT:
pstr += sprintf(pstr, "%d", *((int32_t *)row[col])); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "%d", *((int32_t *)row[col]));
break; break;
case TSDB_DATA_TYPE_BIGINT: case TSDB_DATA_TYPE_BIGINT:
pstr += sprintf(pstr, "%" PRId64 "", *((int64_t *)row[col])); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "%" PRId64 "", *((int64_t *)row[col]));
break; break;
case TSDB_DATA_TYPE_FLOAT: case TSDB_DATA_TYPE_FLOAT:
pstr += sprintf(pstr, "%f", GET_FLOAT_VAL(row[col])); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "%f", GET_FLOAT_VAL(row[col]));
break; break;
case TSDB_DATA_TYPE_DOUBLE: case TSDB_DATA_TYPE_DOUBLE:
pstr += sprintf(pstr, "%f", GET_DOUBLE_VAL(row[col])); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "%f", GET_DOUBLE_VAL(row[col]));
break; break;
case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_BINARY:
*(pstr++) = '\''; //*(pstr++) = '\'';
converStringToReadable((char *)row[col], length[col], tbuf, COMMAND_SIZE); converStringToReadable((char *)row[col], length[col], tbuf, COMMAND_SIZE);
pstr = stpcpy(pstr, tbuf); //pstr = stpcpy(pstr, tbuf);
*(pstr++) = '\''; //*(pstr++) = '\'';
pstr += sprintf(pstr + curr_sqlstr_len, "\'%s\'", tbuf);
break; break;
case TSDB_DATA_TYPE_NCHAR: case TSDB_DATA_TYPE_NCHAR:
convertNCharToReadable((char *)row[col], length[col], tbuf, COMMAND_SIZE); convertNCharToReadable((char *)row[col], length[col], tbuf, COMMAND_SIZE);
pstr += sprintf(pstr, "\'%s\'", tbuf); pstr += sprintf(pstr + curr_sqlstr_len, "\'%s\'", tbuf);
break; break;
case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_TIMESTAMP:
if (!arguments->mysqlFlag) { if (!arguments->mysqlFlag) {
pstr += sprintf(pstr, "%" PRId64 "", *(int64_t *)row[col]); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "%" PRId64 "", *(int64_t *)row[col]);
} else { } else {
char buf[64] = "\0"; char buf[64] = "\0";
int64_t ts = *((int64_t *)row[col]); int64_t ts = *((int64_t *)row[col]);
time_t tt = (time_t)(ts / 1000); time_t tt = (time_t)(ts / 1000);
struct tm *ptm = localtime(&tt); struct tm *ptm = localtime(&tt);
strftime(buf, 64, "%y-%m-%d %H:%M:%S", ptm); strftime(buf, 64, "%y-%m-%d %H:%M:%S", ptm);
pstr += sprintf(pstr, "\'%s.%03d\'", buf, (int)(ts % 1000)); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, "\'%s.%03d\'", buf, (int)(ts % 1000));
} }
break; break;
default: default:
...@@ -1588,13 +1622,15 @@ int taosDumpTableData(FILE *fp, char *tbname, struct arguments *arguments, TAOS* ...@@ -1588,13 +1622,15 @@ int taosDumpTableData(FILE *fp, char *tbname, struct arguments *arguments, TAOS*
} }
} }
pstr += sprintf(pstr, ") "); curr_sqlstr_len += sprintf(pstr + curr_sqlstr_len, ") ");
totalRows++; totalRows++;
count++; count++;
fprintf(fp, "%s", tmpBuffer); fprintf(fp, "%s", tmpBuffer);
if (count >= arguments->data_batch) { total_sqlstr_len += curr_sqlstr_len;
if ((count >= arguments->data_batch) || (sql_buf_len - total_sqlstr_len < TSDB_MAX_BYTES_PER_ROW)) {
fprintf(fp, ";\n"); fprintf(fp, ";\n");
count = 0; count = 0;
} //else { } //else {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或