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

Merge pull request #4543 from taosdata/hotfix/test 

Hotfix/test
This commit is contained in:
huili 2020-12-11 18:30:50 +08:00 committed by GitHub
parent 070d3aa895 2f5967d5cd
commit 297ec49aa3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 135 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -698,79 +698,97 @@ int32_t taosSaveTableOfMetricToTempFile(TAOS *taosCon, char* metric, struct argu
sprintf(tmpCommand, "select tbname from %s", metric); sprintf(tmpCommand, "select tbname from %s", metric);
TAOS_RES *result = taos_query(taosCon, tmpCommand); TAOS_RES *res = taos_query(taosCon, tmpCommand);
int32_t code = taos_errno(result); int32_t code = taos_errno(res);
if (code != 0) { if (code != 0) {
fprintf(stderr, "failed to run command %s\n", tmpCommand); fprintf(stderr, "failed to run command %s\n", tmpCommand);
free(tmpCommand); free(tmpCommand);
taos_free_result(result); taos_free_result(res);
return -1;
}
free(tmpCommand);
char tmpBuf[TSDB_FILENAME_LEN + 1];
memset(tmpBuf, 0, TSDB_FILENAME_LEN);
sprintf(tmpBuf, ".select-tbname.tmp");
fd = open(tmpBuf, O_RDWR | O_CREAT | O_TRUNC, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH);
if (fd == -1) {
fprintf(stderr, "failed to open temp file: %s\n", tmpBuf);
taos_free_result(res);
return -1; return -1;
} }
int table_batch = arguments->table_batch; TAOS_FIELD *fields = taos_fetch_fields(res);
int affectdRows = taos_affected_rows(result);
if (affectdRows <= 0) {
free(tmpCommand);
taos_free_result(result);
return -1;
}
int maxNumOfThread = affectdRows / table_batch + 1;
if (maxNumOfThread > 2 * g_numOfCores) {
maxNumOfThread = 2 * g_numOfCores;
}
table_batch = affectdRows / maxNumOfThread + 1;
TAOS_FIELD *fields = taos_fetch_fields(result);
int32_t numOfTable = 0;
int32_t numOfThread = *totalNumOfThread;
char tmpFileName[TSDB_FILENAME_LEN + 1];
while ((row = taos_fetch_row(result)) != NULL) {
if (0 == numOfTable) {
memset(tmpFileName, 0, TSDB_FILENAME_LEN);
sprintf(tmpFileName, ".tables.tmp.%d", numOfThread);
fd = open(tmpFileName, O_RDWR | O_CREAT, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH);
if (fd == -1) {
fprintf(stderr, "failed to open temp file: %s\n", tmpFileName);
taos_free_result(result);
for (int32_t loopCnt = 0; loopCnt < numOfThread; loopCnt++) {
sprintf(tmpFileName, ".tables.tmp.%d", loopCnt);
(void)remove(tmpFileName);
}
free(tmpCommand);
return -1;
}
numOfThread++;
}
int32_t numOfTable = 0;
while ((row = taos_fetch_row(res)) != NULL) {
memset(&tableRecord, 0, sizeof(STableRecord)); memset(&tableRecord, 0, sizeof(STableRecord));
tstrncpy(tableRecord.name, (char *)row[0], fields[0].bytes); tstrncpy(tableRecord.name, (char *)row[0], fields[0].bytes);
tstrncpy(tableRecord.metric, metric, TSDB_TABLE_NAME_LEN); tstrncpy(tableRecord.metric, metric, TSDB_TABLE_NAME_LEN);
taosWrite(fd, &tableRecord, sizeof(STableRecord)); taosWrite(fd, &tableRecord, sizeof(STableRecord));
numOfTable++; numOfTable++;
if (numOfTable >= table_batch) {
numOfTable = 0;
close(fd);
fd = -1;
}
} }
taos_free_result(res);
lseek(fd, 0, SEEK_SET);
int maxThreads = arguments->thread_num;
int tableOfPerFile ;
if (numOfTable <= arguments->thread_num) {
tableOfPerFile = 1;
maxThreads = numOfTable;
} else {
tableOfPerFile = numOfTable / arguments->thread_num;
if (0 != numOfTable % arguments->thread_num) {
tableOfPerFile += 1;
}
}
char* tblBuf = (char*)calloc(1, tableOfPerFile * sizeof(STableRecord));
if (NULL == tblBuf){
fprintf(stderr, "failed to calloc %" PRIzu "\n", tableOfPerFile * sizeof(STableRecord));
close(fd);
return -1;
}
int32_t numOfThread = *totalNumOfThread;
int subFd = -1;
for (; numOfThread < maxThreads; numOfThread++) {
memset(tmpBuf, 0, TSDB_FILENAME_LEN);
sprintf(tmpBuf, ".tables.tmp.%d", numOfThread);
subFd = open(tmpBuf, O_RDWR | O_CREAT | O_TRUNC, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH);
if (subFd == -1) {
fprintf(stderr, "failed to open temp file: %s\n", tmpBuf);
for (int32_t loopCnt = 0; loopCnt < numOfThread; loopCnt++) {
sprintf(tmpBuf, ".tables.tmp.%d", loopCnt);
(void)remove(tmpBuf);
}
sprintf(tmpBuf, ".select-tbname.tmp");
(void)remove(tmpBuf);
close(fd);
return -1;
}
// read tableOfPerFile for fd, write to subFd
ssize_t readLen = read(fd, tblBuf, tableOfPerFile * sizeof(STableRecord));
if (readLen <= 0) {
close(subFd);
break;
}
taosWrite(subFd, tblBuf, readLen);
close(subFd);
}
sprintf(tmpBuf, ".select-tbname.tmp");
(void)remove(tmpBuf);
if (fd >= 0) { if (fd >= 0) {
close(fd); close(fd);
fd = -1; fd = -1;
} }
taos_free_result(result);
*totalNumOfThread = numOfThread; *totalNumOfThread = numOfThread;
free(tmpCommand);
return 0; return 0;
} }
@ -1443,57 +1461,80 @@ int taosDumpDb(SDbInfo *dbInfo, struct arguments *arguments, FILE *fp, TAOS *tao
return -1; return -1;
} }
int table_batch = arguments->table_batch; char tmpBuf[TSDB_FILENAME_LEN + 1];
int affectdRows = taos_affected_rows(res); memset(tmpBuf, 0, TSDB_FILENAME_LEN);
if (affectdRows <= 0) { sprintf(tmpBuf, ".show-tables.tmp");
fd = open(tmpBuf, O_RDWR | O_CREAT | O_TRUNC, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH);
if (fd == -1) {
fprintf(stderr, "failed to open temp file: %s\n", tmpBuf);
taos_free_result(res); taos_free_result(res);
return -1; return -1;
} }
int maxNumOfThread = affectdRows / table_batch + 1;
if (maxNumOfThread > 2 * g_numOfCores) {
maxNumOfThread = 2 * g_numOfCores;
}
table_batch = affectdRows / maxNumOfThread + 1;
TAOS_FIELD *fields = taos_fetch_fields(res); TAOS_FIELD *fields = taos_fetch_fields(res);
int32_t numOfTable = 0;
int32_t numOfThread = 0;
char tmpBuf[TSDB_FILENAME_LEN + 1];
while ((row = taos_fetch_row(res)) != NULL) {
if (0 == numOfTable) {
memset(tmpBuf, 0, TSDB_FILENAME_LEN);
sprintf(tmpBuf, ".tables.tmp.%d", numOfThread);
fd = open(tmpBuf, O_RDWR | O_CREAT | O_TRUNC, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH);
if (fd == -1) {
fprintf(stderr, "failed to open temp file: %s\n", tmpBuf);
taos_free_result(res);
for (int32_t loopCnt = 0; loopCnt < numOfThread; loopCnt++) {
sprintf(tmpBuf, ".tables.tmp.%d", loopCnt);
(void)remove(tmpBuf);
}
return -1;
}
numOfThread++;
}
int32_t numOfTable = 0;
while ((row = taos_fetch_row(res)) != NULL) {
memset(&tableRecord, 0, sizeof(STableRecord)); memset(&tableRecord, 0, sizeof(STableRecord));
tstrncpy(tableRecord.name, (char *)row[TSDB_SHOW_TABLES_NAME_INDEX], fields[TSDB_SHOW_TABLES_NAME_INDEX].bytes); tstrncpy(tableRecord.name, (char *)row[TSDB_SHOW_TABLES_NAME_INDEX], fields[TSDB_SHOW_TABLES_NAME_INDEX].bytes);
tstrncpy(tableRecord.metric, (char *)row[TSDB_SHOW_TABLES_METRIC_INDEX], fields[TSDB_SHOW_TABLES_METRIC_INDEX].bytes); tstrncpy(tableRecord.metric, (char *)row[TSDB_SHOW_TABLES_METRIC_INDEX], fields[TSDB_SHOW_TABLES_METRIC_INDEX].bytes);
taosWrite(fd, &tableRecord, sizeof(STableRecord)); taosWrite(fd, &tableRecord, sizeof(STableRecord));
numOfTable++; numOfTable++;
if (numOfTable >= table_batch) {
numOfTable = 0;
close(fd);
fd = -1;
}
} }
taos_free_result(res);
lseek(fd, 0, SEEK_SET);
int maxThreads = tsArguments.thread_num;
int tableOfPerFile ;
if (numOfTable <= tsArguments.thread_num) {
tableOfPerFile = 1;
maxThreads = numOfTable;
} else {
tableOfPerFile = numOfTable / tsArguments.thread_num;
if (0 != numOfTable % tsArguments.thread_num) {
tableOfPerFile += 1;
}
}
char* tblBuf = (char*)calloc(1, tableOfPerFile * sizeof(STableRecord));
if (NULL == tblBuf){
fprintf(stderr, "failed to calloc %" PRIzu "\n", tableOfPerFile * sizeof(STableRecord));
close(fd);
return -1;
}
int32_t numOfThread = 0;
int subFd = -1;
for (numOfThread = 0; numOfThread < maxThreads; numOfThread++) {
memset(tmpBuf, 0, TSDB_FILENAME_LEN);
sprintf(tmpBuf, ".tables.tmp.%d", numOfThread);
subFd = open(tmpBuf, O_RDWR | O_CREAT | O_TRUNC, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH);
if (subFd == -1) {
fprintf(stderr, "failed to open temp file: %s\n", tmpBuf);
for (int32_t loopCnt = 0; loopCnt < numOfThread; loopCnt++) {
sprintf(tmpBuf, ".tables.tmp.%d", loopCnt);
(void)remove(tmpBuf);
}
sprintf(tmpBuf, ".show-tables.tmp");
(void)remove(tmpBuf);
close(fd);
return -1;
}
// read tableOfPerFile for fd, write to subFd
ssize_t readLen = read(fd, tblBuf, tableOfPerFile * sizeof(STableRecord));
if (readLen <= 0) {
close(subFd);
break;
}
taosWrite(subFd, tblBuf, readLen);
close(subFd);
}
sprintf(tmpBuf, ".show-tables.tmp");
(void)remove(tmpBuf);
if (fd >= 0) { if (fd >= 0) {
close(fd); close(fd);