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

opti:get meta cost for schemaless & add config for write batch in schemaless

This commit is contained in:
wangmm0220 2022-11-30 14:30:44 +08:00
parent e07823c55a
commit 162575ea66
3 changed files with 8 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/common/tglobal.h
View File

@ -125,6 +125,7 @@ extern char tsUdfdLdLibPath[];
extern char tsSmlChildTableName[]; extern char tsSmlChildTableName[];
extern char tsSmlTagName[]; extern char tsSmlTagName[];
extern bool tsSmlDataFormat; extern bool tsSmlDataFormat;
extern int32_t tsSmlBatchSize;
// wal // wal
extern int64_t tsWalFsyncDataSizeLimit; extern int64_t tsWalFsyncDataSizeLimit;

5
source/client/src/clientSml.c
View File

@ -79,7 +79,6 @@
#define NCHAR_ADD_LEN 3 // L"nchar" 3 means L" " #define NCHAR_ADD_LEN 3 // L"nchar" 3 means L" "
#define MAX_RETRY_TIMES 5 #define MAX_RETRY_TIMES 5
#define LINE_BATCH 2000
//================================================================================================= //=================================================================================================
typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType; typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;
@ -2563,7 +2562,7 @@ TAOS_RES *taos_schemaless_insert_inner(SRequestObj *request, char *lines[], char
goto end; goto end;
} }
batchs = ceil(((double)numLines) / LINE_BATCH); batchs = ceil(((double)numLines) / tsSmlBatchSize);
params.total = batchs; params.total = batchs;
for (int i = 0; i < batchs; ++i) { for (int i = 0; i < batchs; ++i) {
SRequestObj *req = (SRequestObj *)createRequest(pTscObj->id, TSDB_SQL_INSERT, 0); SRequestObj *req = (SRequestObj *)createRequest(pTscObj->id, TSDB_SQL_INSERT, 0);
@ -2582,7 +2581,7 @@ TAOS_RES *taos_schemaless_insert_inner(SRequestObj *request, char *lines[], char
info->isRawLine = (rawLine == NULL); info->isRawLine = (rawLine == NULL);
info->ttl = ttl; info->ttl = ttl;
int32_t perBatch = LINE_BATCH; int32_t perBatch = tsSmlBatchSize;
if (numLines > perBatch) { if (numLines > perBatch) {
numLines -= perBatch; numLines -= perBatch;

5
source/common/src/tglobal.c
View File

@ -75,6 +75,7 @@ char tsSmlChildTableName[TSDB_TABLE_NAME_LEN] = ""; // user defined child table
// If set to empty system will generate table name using MD5 hash. // If set to empty system will generate table name using MD5 hash.
// true means that the name and order of cols in each line are the same(only for influx protocol) // true means that the name and order of cols in each line are the same(only for influx protocol)
bool tsSmlDataFormat = false; bool tsSmlDataFormat = false;
int32_t tsSmlBatchSize = 10000;
// query // query
int32_t tsQueryPolicy = 1; int32_t tsQueryPolicy = 1;
@ -302,6 +303,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
if (cfgAddString(pCfg, "smlChildTableName", "", 1) != 0) return -1; if (cfgAddString(pCfg, "smlChildTableName", "", 1) != 0) return -1;
if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, 1) != 0) return -1; if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, 1) != 0) return -1;
if (cfgAddBool(pCfg, "smlDataFormat", tsSmlDataFormat, 1) != 0) return -1; if (cfgAddBool(pCfg, "smlDataFormat", tsSmlDataFormat, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "smlBatchSize", tsSmlBatchSize, 1, INT32_MAX, true) != 0) return -1;
if (cfgAddInt32(pCfg, "maxMemUsedByInsert", tsMaxMemUsedByInsert, 1, INT32_MAX, true) != 0) return -1; if (cfgAddInt32(pCfg, "maxMemUsedByInsert", tsMaxMemUsedByInsert, 1, INT32_MAX, true) != 0) return -1;
if (cfgAddInt32(pCfg, "rpcRetryLimit", tsRpcRetryLimit, 1, 100000, 0) != 0) return -1; if (cfgAddInt32(pCfg, "rpcRetryLimit", tsRpcRetryLimit, 1, 100000, 0) != 0) return -1;
if (cfgAddInt32(pCfg, "rpcRetryInterval", tsRpcRetryInterval, 1, 100000, 0) != 0) return -1; if (cfgAddInt32(pCfg, "rpcRetryInterval", tsRpcRetryInterval, 1, 100000, 0) != 0) return -1;
@ -643,6 +645,7 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN); tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN);
tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval; tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval;
tsSmlBatchSize = cfgGetItem(pCfg, "smlBatchSize")->i32;
tsMaxMemUsedByInsert = cfgGetItem(pCfg, "maxMemUsedByInsert")->i32; tsMaxMemUsedByInsert = cfgGetItem(pCfg, "maxMemUsedByInsert")->i32;
tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32; tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32;
@ -1013,6 +1016,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN); tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN);
} else if (strcasecmp("smlDataFormat", name) == 0) { } else if (strcasecmp("smlDataFormat", name) == 0) {
tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval; tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval;
} else if (strcasecmp("smlBatchSize", name) == 0) {
tsSmlBatchSize = cfgGetItem(pCfg, "smlBatchSize")->i32;
} else if (strcasecmp("shellActivityTimer", name) == 0) { } else if (strcasecmp("shellActivityTimer", name) == 0) {
tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32; tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32;
} else if (strcasecmp("supportVnodes", name) == 0) { } else if (strcasecmp("supportVnodes", name) == 0) {