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

fix:[TS-2828] retry if ver is old

This commit is contained in:
wangmm0220 2023-03-09 19:48:21 +08:00
parent fa8d9a62f2
commit 88c3d0d4b3
2 changed files with 67 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1422,6 +1422,7 @@ static int smlProcess(SSmlHandle *info, char *lines[], char *rawLine, char *rawL
do { do {
code = smlModifyDBSchemas(info); code = smlModifyDBSchemas(info);
if (code == 0) break; if (code == 0) break;
taosMsleep(200);
} while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES); } while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES);
if (code != 0) { if (code != 0) {
@ -1446,62 +1447,74 @@ TAOS_RES *taos_schemaless_insert_inner(TAOS *taos, char *lines[], char *rawLine,
terrno = TSDB_CODE_TSC_DISCONNECTED; terrno = TSDB_CODE_TSC_DISCONNECTED;
return NULL; return NULL;
} }
SRequestObj *request = NULL;
SSmlHandle *info = NULL;
while(1){
request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
if (request == NULL) {
uError("SML:taos_schemaless_insert error request is null");
return NULL;
}
SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid); info = smlBuildSmlInfo(taos);
if (request == NULL) { if (info == NULL) {
uError("SML:taos_schemaless_insert error request is null"); request->code = TSDB_CODE_OUT_OF_MEMORY;
return NULL; uError("SML:taos_schemaless_insert error SSmlHandle is null");
return (TAOS_RES *)request;
}
info->pRequest = request;
info->isRawLine = rawLine != NULL;
info->ttl = ttl;
info->precision = precision;
info->protocol = (TSDB_SML_PROTOCOL_TYPE)protocol;
info->msgBuf.buf = info->pRequest->msgBuf;
info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
info->lineNum = numLines;
SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
if (request->pDb == NULL) {
request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
goto end;
}
if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
goto end;
}
if (protocol == TSDB_SML_LINE_PROTOCOL &&
(precision < TSDB_SML_TIMESTAMP_NOT_CONFIGURED || precision > TSDB_SML_TIMESTAMP_NANO_SECONDS)) {
request->code = TSDB_CODE_SML_INVALID_PRECISION_TYPE;
smlBuildInvalidDataMsg(&msg, "precision invalidate for line protocol", NULL);
goto end;
}
if (protocol == TSDB_SML_JSON_PROTOCOL) {
numLines = 1;
} else if (numLines <= 0) {
request->code = TSDB_CODE_SML_INVALID_DATA;
smlBuildInvalidDataMsg(&msg, "line num is invalid", NULL);
goto end;
}
code = smlProcess(info, lines, rawLine, rawLineEnd, numLines);
request->code = code;
info->cost.endTime = taosGetTimestampUs();
info->cost.code = code;
smlPrintStatisticInfo(info);
if(code == TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER || code == TSDB_CODE_SDB_OBJ_CREATING){
uInfo("SML:%"PRIx64" ver is old retry", info->id);
smlDestroyInfo(info);
info = NULL;
taos_free_result(request);
request = NULL;
continue;
}
break;
} }
SSmlHandle *info = smlBuildSmlInfo(taos);
if (info == NULL) {
request->code = TSDB_CODE_OUT_OF_MEMORY;
uError("SML:taos_schemaless_insert error SSmlHandle is null");
return (TAOS_RES *)request;
}
info->pRequest = request;
info->isRawLine = rawLine != NULL;
info->ttl = ttl;
info->precision = precision;
info->protocol = (TSDB_SML_PROTOCOL_TYPE)protocol;
info->msgBuf.buf = info->pRequest->msgBuf;
info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
info->lineNum = numLines;
SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
if (request->pDb == NULL) {
request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
goto end;
}
if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
goto end;
}
if (protocol == TSDB_SML_LINE_PROTOCOL &&
(precision < TSDB_SML_TIMESTAMP_NOT_CONFIGURED || precision > TSDB_SML_TIMESTAMP_NANO_SECONDS)) {
request->code = TSDB_CODE_SML_INVALID_PRECISION_TYPE;
smlBuildInvalidDataMsg(&msg, "precision invalidate for line protocol", NULL);
goto end;
}
if (protocol == TSDB_SML_JSON_PROTOCOL) {
numLines = 1;
} else if (numLines <= 0) {
request->code = TSDB_CODE_SML_INVALID_DATA;
smlBuildInvalidDataMsg(&msg, "line num is invalid", NULL);
goto end;
}
code = smlProcess(info, lines, rawLine, rawLineEnd, numLines);
request->code = code;
info->cost.endTime = taosGetTimestampUs();
info->cost.code = code;
smlPrintStatisticInfo(info);
end: end:
smlDestroyInfo(info); smlDestroyInfo(info);
return (TAOS_RES *)request; return (TAOS_RES *)request;

2
source/client/src/clientSmlLine.c
View File

@ -436,7 +436,7 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
// bind data // bind data
ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, cnt + 1); ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, cnt + 1);
if (unlikely(ret != TSDB_CODE_SUCCESS)) { if (unlikely(ret != TSDB_CODE_SUCCESS)) {
uError("smlBuildCol error, retry"); uDebug("smlBuildCol error, retry");
info->dataFormat = false; info->dataFormat = false;
info->reRun = true; info->reRun = true;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;