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

[TD-3580] <fix>: taosdump support human readable time format. (#5674) 

Co-authored-by: Shuduo Sang <sdsang@taosdata.com>
This commit is contained in:
Shuduo Sang 2021-04-03 20:12:11 +08:00 committed by GitHub
parent d741ca2d73
commit a8d53d1f90
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 180 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

211
src/kit/taosdump/taosdump.c
View File

@ -39,6 +39,22 @@ typedef struct {
int8_t type; int8_t type;
} SOColInfo; } SOColInfo;
#define debugPrint(fmt, ...) \
do { if (g_args.debug_print || g_args.verbose_print) \
fprintf(stderr, "DEBG: "fmt, __VA_ARGS__); } while(0)
#define verbosePrint(fmt, ...) \
do { if (g_args.verbose_print) \
fprintf(stderr, "VERB: "fmt, __VA_ARGS__); } while(0)
#define performancePrint(fmt, ...) \
do { if (g_args.performance_print) \
fprintf(stderr, "VERB: "fmt, __VA_ARGS__); } while(0)
#define errorPrint(fmt, ...) \
do { fprintf(stderr, "ERROR: "fmt, __VA_ARGS__); } while(0)
// -------------------------- SHOW DATABASE INTERFACE----------------------- // -------------------------- SHOW DATABASE INTERFACE-----------------------
enum _show_db_index { enum _show_db_index {
TSDB_SHOW_DB_NAME_INDEX, TSDB_SHOW_DB_NAME_INDEX,
@ -199,16 +215,18 @@ static struct argp_option options[] = {
{"schemaonly", 's', 0, 0, "Only dump schema.", 3}, {"schemaonly", 's', 0, 0, "Only dump schema.", 3},
{"with-property", 'M', 0, 0, "Dump schema with properties.", 3}, {"with-property", 'M', 0, 0, "Dump schema with properties.", 3},
{"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. Epoch or ISO8601/RFC3339 format is acceptable. For example: 2017-10-01T18:00:00+0800", 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}, {"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},
{"debug", 'g', 0, 0, "Print debug info.", 1},
{"verbose", 'v', 0, 0, "Print verbose debug info.", 1},
{0}}; {0}};
/* Used by main to communicate with parse_opt. */ /* Used by main to communicate with parse_opt. */
struct arguments { typedef struct arguments {
// connection option // connection option
char *host; char *host;
char *user; char *user;
@ -240,7 +258,10 @@ struct arguments {
char **arg_list; char **arg_list;
int arg_list_len; int arg_list_len;
bool isDumpIn; bool isDumpIn;
}; bool debug_print;
bool verbose_print;
bool performance_print;
} SArguments;
/* Parse a single option. */ /* Parse a single option. */
static error_t parse_opt(int key, char *arg, struct argp_state *state) { static error_t parse_opt(int key, char *arg, struct argp_state *state) {
@ -286,6 +307,9 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
tstrncpy(arguments->outpath, full_path.we_wordv[0], TSDB_FILENAME_LEN); tstrncpy(arguments->outpath, full_path.we_wordv[0], TSDB_FILENAME_LEN);
wordfree(&full_path); wordfree(&full_path);
break; break;
case 'g':
arguments->debug_print = true;
break;
case 'i': case 'i':
arguments->isDumpIn = true; arguments->isDumpIn = true;
if (wordexp(arg, &full_path, 0) != 0) { if (wordexp(arg, &full_path, 0) != 0) {
@ -387,7 +411,7 @@ int taosCheckParam(struct arguments *arguments);
void taosFreeDbInfos(); void taosFreeDbInfos();
static void taosStartDumpOutWorkThreads(void* taosCon, struct arguments* args, int32_t numOfThread, char *dbName); static void taosStartDumpOutWorkThreads(void* taosCon, struct arguments* args, int32_t numOfThread, char *dbName);
struct arguments tsArguments = { struct arguments g_args = {
// connection option // connection option
NULL, NULL,
"root", "root",
@ -421,7 +445,10 @@ struct arguments tsArguments = {
0, 0,
NULL, NULL,
0, 0,
false false,
false, // debug_print
false, // verbose_print
false // performance_print
}; };
static int queryDbImpl(TAOS *taos, char *command) { static int queryDbImpl(TAOS *taos, char *command) {
@ -453,13 +480,40 @@ static int queryDbImpl(TAOS *taos, char *command) {
return 0; return 0;
} }
static void parse_args(int argc, char *argv[], SArguments *arguments) {
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "-E") == 0) {
char *tmp = argv[++i];
int64_t tmpEpoch;
if (strchr(tmp, ':') && strchr(tmp, '-')) {
if (TSDB_CODE_SUCCESS != taosParseTime(
tmp, &tmpEpoch, strlen(tmp), TSDB_TIME_PRECISION_MILLI, 0)) {
fprintf(stderr, "Input end time error!\n");
return;
}
} else {
tmpEpoch = atoll(tmp);
}
sprintf(argv[i], "%"PRId64"", tmpEpoch);
debugPrint("%s() LN%d, tmp is: %s, argv[%d]: %s\n",
__func__, __LINE__, tmp, i, argv[i]);
} else if (strcmp(argv[i], "-g") == 0) {
arguments->debug_print = true;
}
}
}
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
/* Parse our arguments; every option seen by parse_opt will be /* Parse our arguments; every option seen by parse_opt will be
reflected in arguments. */ reflected in arguments. */
argp_parse(&argp, argc, argv, 0, 0, &tsArguments); parse_args(argc, argv, &g_args);
if (tsArguments.abort) { argp_parse(&argp, argc, argv, 0, 0, &g_args);
if (g_args.abort) {
#ifndef _ALPINE #ifndef _ALPINE
error(10, 0, "ABORTED"); error(10, 0, "ABORTED");
#else #else
@ -469,81 +523,82 @@ int main(int argc, char *argv[]) {
printf("====== arguments config ======\n"); printf("====== arguments config ======\n");
{ {
printf("host: %s\n", tsArguments.host); printf("host: %s\n", g_args.host);
printf("user: %s\n", tsArguments.user); printf("user: %s\n", g_args.user);
printf("password: %s\n", tsArguments.password); printf("password: %s\n", g_args.password);
printf("port: %u\n", tsArguments.port); printf("port: %u\n", g_args.port);
printf("cversion: %s\n", tsArguments.cversion); printf("cversion: %s\n", g_args.cversion);
printf("mysqlFlag: %d\n", tsArguments.mysqlFlag); printf("mysqlFlag: %d\n", g_args.mysqlFlag);
printf("outpath: %s\n", tsArguments.outpath); printf("outpath: %s\n", g_args.outpath);
printf("inpath: %s\n", tsArguments.inpath); printf("inpath: %s\n", g_args.inpath);
printf("resultFile: %s\n", tsArguments.resultFile); printf("resultFile: %s\n", g_args.resultFile);
printf("encode: %s\n", tsArguments.encode); printf("encode: %s\n", g_args.encode);
printf("all_databases: %d\n", tsArguments.all_databases); printf("all_databases: %d\n", g_args.all_databases);
printf("databases: %d\n", tsArguments.databases); printf("databases: %d\n", g_args.databases);
printf("schemaonly: %d\n", tsArguments.schemaonly); printf("schemaonly: %d\n", g_args.schemaonly);
printf("with_property: %d\n", tsArguments.with_property); printf("with_property: %d\n", g_args.with_property);
printf("start_time: %" PRId64 "\n", tsArguments.start_time); printf("start_time: %" PRId64 "\n", g_args.start_time);
printf("end_time: %" PRId64 "\n", tsArguments.end_time); printf("end_time: %" PRId64 "\n", g_args.end_time);
printf("data_batch: %d\n", tsArguments.data_batch); printf("data_batch: %d\n", g_args.data_batch);
printf("max_sql_len: %d\n", tsArguments.max_sql_len); printf("max_sql_len: %d\n", g_args.max_sql_len);
printf("table_batch: %d\n", tsArguments.table_batch); printf("table_batch: %d\n", g_args.table_batch);
printf("thread_num: %d\n", tsArguments.thread_num); printf("thread_num: %d\n", g_args.thread_num);
printf("allow_sys: %d\n", tsArguments.allow_sys); printf("allow_sys: %d\n", g_args.allow_sys);
printf("abort: %d\n", tsArguments.abort); printf("abort: %d\n", g_args.abort);
printf("isDumpIn: %d\n", tsArguments.isDumpIn); printf("isDumpIn: %d\n", g_args.isDumpIn);
printf("arg_list_len: %d\n", tsArguments.arg_list_len); printf("arg_list_len: %d\n", g_args.arg_list_len);
printf("debug_print: %d\n", g_args.debug_print);
for (int32_t i = 0; i < tsArguments.arg_list_len; i++) { for (int32_t i = 0; i < g_args.arg_list_len; i++) {
printf("arg_list[%d]: %s\n", i, tsArguments.arg_list[i]); printf("arg_list[%d]: %s\n", i, g_args.arg_list[i]);
} }
} }
printf("==============================\n"); printf("==============================\n");
if (tsArguments.cversion[0] != 0){ if (g_args.cversion[0] != 0){
tstrncpy(version, tsArguments.cversion, 11); tstrncpy(version, g_args.cversion, 11);
} }
if (taosCheckParam(&tsArguments) < 0) { if (taosCheckParam(&g_args) < 0) {
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
g_fpOfResult = fopen(tsArguments.resultFile, "a"); g_fpOfResult = fopen(g_args.resultFile, "a");
if (NULL == g_fpOfResult) { if (NULL == g_fpOfResult) {
fprintf(stderr, "Failed to open %s for save result\n", tsArguments.resultFile); fprintf(stderr, "Failed to open %s for save result\n", g_args.resultFile);
return 1; return 1;
}; };
fprintf(g_fpOfResult, "#############################################################################\n"); fprintf(g_fpOfResult, "#############################################################################\n");
fprintf(g_fpOfResult, "============================== arguments config =============================\n"); fprintf(g_fpOfResult, "============================== arguments config =============================\n");
{ {
fprintf(g_fpOfResult, "host: %s\n", tsArguments.host); fprintf(g_fpOfResult, "host: %s\n", g_args.host);
fprintf(g_fpOfResult, "user: %s\n", tsArguments.user); fprintf(g_fpOfResult, "user: %s\n", g_args.user);
fprintf(g_fpOfResult, "password: %s\n", tsArguments.password); fprintf(g_fpOfResult, "password: %s\n", g_args.password);
fprintf(g_fpOfResult, "port: %u\n", tsArguments.port); fprintf(g_fpOfResult, "port: %u\n", g_args.port);
fprintf(g_fpOfResult, "cversion: %s\n", tsArguments.cversion); fprintf(g_fpOfResult, "cversion: %s\n", g_args.cversion);
fprintf(g_fpOfResult, "mysqlFlag: %d\n", tsArguments.mysqlFlag); fprintf(g_fpOfResult, "mysqlFlag: %d\n", g_args.mysqlFlag);
fprintf(g_fpOfResult, "outpath: %s\n", tsArguments.outpath); fprintf(g_fpOfResult, "outpath: %s\n", g_args.outpath);
fprintf(g_fpOfResult, "inpath: %s\n", tsArguments.inpath); fprintf(g_fpOfResult, "inpath: %s\n", g_args.inpath);
fprintf(g_fpOfResult, "resultFile: %s\n", tsArguments.resultFile); fprintf(g_fpOfResult, "resultFile: %s\n", g_args.resultFile);
fprintf(g_fpOfResult, "encode: %s\n", tsArguments.encode); fprintf(g_fpOfResult, "encode: %s\n", g_args.encode);
fprintf(g_fpOfResult, "all_databases: %d\n", tsArguments.all_databases); fprintf(g_fpOfResult, "all_databases: %d\n", g_args.all_databases);
fprintf(g_fpOfResult, "databases: %d\n", tsArguments.databases); fprintf(g_fpOfResult, "databases: %d\n", g_args.databases);
fprintf(g_fpOfResult, "schemaonly: %d\n", tsArguments.schemaonly); fprintf(g_fpOfResult, "schemaonly: %d\n", g_args.schemaonly);
fprintf(g_fpOfResult, "with_property: %d\n", tsArguments.with_property); fprintf(g_fpOfResult, "with_property: %d\n", g_args.with_property);
fprintf(g_fpOfResult, "start_time: %" PRId64 "\n", tsArguments.start_time); fprintf(g_fpOfResult, "start_time: %" PRId64 "\n", g_args.start_time);
fprintf(g_fpOfResult, "end_time: %" PRId64 "\n", tsArguments.end_time); fprintf(g_fpOfResult, "end_time: %" PRId64 "\n", g_args.end_time);
fprintf(g_fpOfResult, "data_batch: %d\n", tsArguments.data_batch); fprintf(g_fpOfResult, "data_batch: %d\n", g_args.data_batch);
fprintf(g_fpOfResult, "max_sql_len: %d\n", tsArguments.max_sql_len); fprintf(g_fpOfResult, "max_sql_len: %d\n", g_args.max_sql_len);
fprintf(g_fpOfResult, "table_batch: %d\n", tsArguments.table_batch); fprintf(g_fpOfResult, "table_batch: %d\n", g_args.table_batch);
fprintf(g_fpOfResult, "thread_num: %d\n", tsArguments.thread_num); fprintf(g_fpOfResult, "thread_num: %d\n", g_args.thread_num);
fprintf(g_fpOfResult, "allow_sys: %d\n", tsArguments.allow_sys); fprintf(g_fpOfResult, "allow_sys: %d\n", g_args.allow_sys);
fprintf(g_fpOfResult, "abort: %d\n", tsArguments.abort); fprintf(g_fpOfResult, "abort: %d\n", g_args.abort);
fprintf(g_fpOfResult, "isDumpIn: %d\n", tsArguments.isDumpIn); fprintf(g_fpOfResult, "isDumpIn: %d\n", g_args.isDumpIn);
fprintf(g_fpOfResult, "arg_list_len: %d\n", tsArguments.arg_list_len); fprintf(g_fpOfResult, "arg_list_len: %d\n", g_args.arg_list_len);
for (int32_t i = 0; i < tsArguments.arg_list_len; i++) { for (int32_t i = 0; i < g_args.arg_list_len; i++) {
fprintf(g_fpOfResult, "arg_list[%d]: %s\n", i, tsArguments.arg_list[i]); fprintf(g_fpOfResult, "arg_list[%d]: %s\n", i, g_args.arg_list[i]);
} }
} }
@ -552,11 +607,11 @@ int main(int argc, char *argv[]) {
time_t tTime = time(NULL); time_t tTime = time(NULL);
struct tm tm = *localtime(&tTime); struct tm tm = *localtime(&tTime);
if (tsArguments.isDumpIn) { if (g_args.isDumpIn) {
fprintf(g_fpOfResult, "============================== DUMP IN ============================== \n"); fprintf(g_fpOfResult, "============================== DUMP IN ============================== \n");
fprintf(g_fpOfResult, "# DumpIn start time: %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1, fprintf(g_fpOfResult, "# DumpIn start time: %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1,
tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec); tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec);
if (taosDumpIn(&tsArguments) < 0) { if (taosDumpIn(&g_args) < 0) {
fprintf(g_fpOfResult, "\n"); fprintf(g_fpOfResult, "\n");
fclose(g_fpOfResult); fclose(g_fpOfResult);
return -1; return -1;
@ -565,7 +620,7 @@ int main(int argc, char *argv[]) {
fprintf(g_fpOfResult, "============================== DUMP OUT ============================== \n"); fprintf(g_fpOfResult, "============================== DUMP OUT ============================== \n");
fprintf(g_fpOfResult, "# DumpOut start time: %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1, fprintf(g_fpOfResult, "# DumpOut start time: %d-%02d-%02d %02d:%02d:%02d\n", tm.tm_year + 1900, tm.tm_mon + 1,
tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec); tm.tm_mday, tm.tm_hour, tm.tm_min, tm.tm_sec);
if (taosDumpOut(&tsArguments) < 0) { if (taosDumpOut(&g_args) < 0) {
fprintf(g_fpOfResult, "\n"); fprintf(g_fpOfResult, "\n");
fclose(g_fpOfResult); fclose(g_fpOfResult);
return -1; return -1;
@ -1236,8 +1291,8 @@ void* taosDumpOutWorkThreadFp(void *arg)
FILE *fp = NULL; FILE *fp = NULL;
memset(tmpBuf, 0, TSDB_FILENAME_LEN + 128); memset(tmpBuf, 0, TSDB_FILENAME_LEN + 128);
if (tsArguments.outpath[0] != 0) { if (g_args.outpath[0] != 0) {
sprintf(tmpBuf, "%s/%s.tables.%d.sql", tsArguments.outpath, pThread->dbName, pThread->threadIndex); sprintf(tmpBuf, "%s/%s.tables.%d.sql", g_args.outpath, pThread->dbName, pThread->threadIndex);
} else { } else {
sprintf(tmpBuf, "%s.tables.%d.sql", pThread->dbName, pThread->threadIndex); sprintf(tmpBuf, "%s.tables.%d.sql", pThread->dbName, pThread->threadIndex);
} }
@ -1270,7 +1325,7 @@ void* taosDumpOutWorkThreadFp(void *arg)
ssize_t readLen = read(fd, &tableRecord, sizeof(STableRecord)); ssize_t readLen = read(fd, &tableRecord, sizeof(STableRecord));
if (readLen <= 0) break; if (readLen <= 0) break;
int ret = taosDumpTable(tableRecord.name, tableRecord.metric, &tsArguments, fp, pThread->taosCon, pThread->dbName); int ret = taosDumpTable(tableRecord.name, tableRecord.metric, &g_args, fp, pThread->taosCon, pThread->dbName);
if (ret >= 0) { if (ret >= 0) {
// TODO: sum table count and table rows by self // TODO: sum table count and table rows by self
pThread->tablesOfDumpOut++; pThread->tablesOfDumpOut++;
@ -1282,13 +1337,13 @@ void* taosDumpOutWorkThreadFp(void *arg)
} }
tablesInOneFile++; tablesInOneFile++;
if (tablesInOneFile >= tsArguments.table_batch) { if (tablesInOneFile >= g_args.table_batch) {
fclose(fp); fclose(fp);
tablesInOneFile = 0; tablesInOneFile = 0;
memset(tmpBuf, 0, TSDB_FILENAME_LEN + 128); memset(tmpBuf, 0, TSDB_FILENAME_LEN + 128);
if (tsArguments.outpath[0] != 0) { if (g_args.outpath[0] != 0) {
sprintf(tmpBuf, "%s/%s.tables.%d-%d.sql", tsArguments.outpath, pThread->dbName, pThread->threadIndex, fileNameIndex); sprintf(tmpBuf, "%s/%s.tables.%d-%d.sql", g_args.outpath, pThread->dbName, pThread->threadIndex, fileNameIndex);
} else { } else {
sprintf(tmpBuf, "%s.tables.%d-%d.sql", pThread->dbName, pThread->threadIndex, fileNameIndex); sprintf(tmpBuf, "%s.tables.%d-%d.sql", pThread->dbName, pThread->threadIndex, fileNameIndex);
} }
@ -1491,14 +1546,14 @@ int taosDumpDb(SDbInfo *dbInfo, struct arguments *arguments, FILE *fp, TAOS *tao
taos_free_result(res); taos_free_result(res);
lseek(fd, 0, SEEK_SET); lseek(fd, 0, SEEK_SET);
int maxThreads = tsArguments.thread_num; int maxThreads = g_args.thread_num;
int tableOfPerFile ; int tableOfPerFile ;
if (numOfTable <= tsArguments.thread_num) { if (numOfTable <= g_args.thread_num) {
tableOfPerFile = 1; tableOfPerFile = 1;
maxThreads = numOfTable; maxThreads = numOfTable;
} else { } else {
tableOfPerFile = numOfTable / tsArguments.thread_num; tableOfPerFile = numOfTable / g_args.thread_num;
if (0 != numOfTable % tsArguments.thread_num) { if (0 != numOfTable % g_args.thread_num) {
tableOfPerFile += 1; tableOfPerFile += 1;
} }
} }
@ -2214,7 +2269,7 @@ void* taosDumpInWorkThreadFp(void *arg)
continue; continue;
} }
fprintf(stderr, "Success Open input file: %s\n", SQLFileName); fprintf(stderr, "Success Open input file: %s\n", SQLFileName);
taosDumpInOneFile(pThread->taosCon, fp, tsfCharset, tsArguments.encode, SQLFileName); taosDumpInOneFile(pThread->taosCon, fp, tsfCharset, g_args.encode, SQLFileName);
} }
} }