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

Merge remote-tracking branch 'origin/feat/TS-4243-3.0' into feat/TS-4243-3.0

This commit is contained in:
Haojun Liao 2024-03-26 17:34:43 +08:00
parent f61596706d 595ae2c5b8
commit fd57ded11e
8 changed files with 178 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/qcom/query.h
View File

@ -197,7 +197,7 @@ typedef struct STableDataCxt {
SBoundColInfo boundColsInfo;
SArray* pValues;
SSubmitTbData* pData;
TSKEY lastTs;
SRowKey lastKey;
bool ordered;
bool duplicateTs;
} STableDataCxt;

2
source/libs/function/src/builtins.c
View File

@ -1632,7 +1632,7 @@ static int32_t translateIrate(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
static int32_t translateIrateImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, bool isPartial) {
uint8_t colType = getSDataTypeFromNode(nodesListGetNode(pFunc->pParameterList, 0))->type;
if (isPartial) {
if (3 != LIST_LENGTH(pFunc->pParameterList)) {
if (3 != LIST_LENGTH(pFunc->pParameterList) && 4 != LIST_LENGTH(pFunc->pParameterList)) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
}
if (!IS_NUMERIC_TYPE(colType)) {

23
source/libs/function/src/builtinsimpl.c
View File

@ -2343,6 +2343,11 @@ EFuncDataRequired firstDynDataReq(void* pRes, SDataBlockInfo* pBlockInfo) {
SFirstLastRes* pResult = GET_ROWCELL_INTERBUF(pEntry);
if (pResult->hasResult) {
if (pResult->pkBytes > 0) {
pResult->pkData = pResult->buf + pResult->bytes;
} else {
pResult->pkData = NULL;
}
if (pResult->ts < pBlockInfo->window.skey) {
return FUNC_DATA_REQUIRED_NOT_LOAD;
} else if (pResult->ts == pBlockInfo->window.skey && pResult->pkData) {
@ -2366,6 +2371,11 @@ EFuncDataRequired lastDynDataReq(void* pRes, SDataBlockInfo* pBlockInfo) {
SFirstLastRes* pResult = GET_ROWCELL_INTERBUF(pEntry);
if (pResult->hasResult) {
if (pResult->pkBytes > 0) {
pResult->pkData = pResult->buf + pResult->bytes;
} else {
pResult->pkData = NULL;
}
if (pResult->ts > pBlockInfo->window.ekey) {
return FUNC_DATA_REQUIRED_NOT_LOAD;
} else if (pResult->ts == pBlockInfo->window.ekey && pResult->pkData) {
@ -6213,12 +6223,17 @@ static void doSaveRateInfo(SRateInfo* pRateInfo, bool isFirst, int64_t ts, char*
}
}
static void initializeRateInfo(SqlFunctionCtx* pCtx, SRateInfo* pRateInfo) {
static void initializeRateInfo(SqlFunctionCtx* pCtx, SRateInfo* pRateInfo, bool isMerge) {
if (pCtx->hasPrimaryKey) {
if (!isMerge) {
pRateInfo->pkType = pCtx->input.pPrimaryKey->info.type;
pRateInfo->pkBytes = pCtx->input.pPrimaryKey->info.bytes;
pRateInfo->firstPk = pRateInfo->pkData;
pRateInfo->lastPk = pRateInfo->pkData + pRateInfo->pkBytes;
} else {
pRateInfo->firstPk = pRateInfo->pkData;
pRateInfo->lastPk = pRateInfo->pkData + pRateInfo->pkBytes;
}
} else {
pRateInfo->firstPk = NULL;
pRateInfo->lastPk = NULL;
@ -6236,7 +6251,7 @@ int32_t irateFunction(SqlFunctionCtx* pCtx) {
funcInputUpdate(pCtx);
initializeRateInfo(pCtx, pRateInfo);
initializeRateInfo(pCtx, pRateInfo, false);
int32_t numOfElems = 0;
int32_t type = pInputCol->info.type;
@ -6358,13 +6373,13 @@ int32_t irateFunctionMerge(SqlFunctionCtx* pCtx) {
}
SRateInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
initializeRateInfo(pCtx, pInfo);
initializeRateInfo(pCtx, pInfo, true);
int32_t start = pInput->startRowIndex;
for (int32_t i = start; i < start + pInput->numOfRows; ++i) {
char* data = colDataGetData(pCol, i);
SRateInfo* pInputInfo = (SRateInfo*)varDataVal(data);
initializeRateInfo(pCtx, pInfo);
initializeRateInfo(pCtx, pInfo, true);
if (pInputInfo->hasResult) {
int32_t code = irateTransferInfo(pInputInfo, pInfo);
if (code != TSDB_CODE_SUCCESS) {

8
source/libs/function/src/functionMgt.c
View File

@ -415,6 +415,8 @@ static int32_t createPartialFunction(const SFunctionNode* pSrcFunc, SFunctionNod
int32_t len = snprintf(name, sizeof(name) - 1, "%s.%p", (*pPartialFunc)->functionName, pSrcFunc);
taosCreateMD5Hash(name, len);
strncpy((*pPartialFunc)->node.aliasName, name, TSDB_COL_NAME_LEN - 1);
(*pPartialFunc)->hasPk = pSrcFunc->hasPk;
(*pPartialFunc)->pkBytes = pSrcFunc->pkBytes;
return TSDB_CODE_SUCCESS;
}
@ -453,7 +455,8 @@ static int32_t createMidFunction(const SFunctionNode* pSrcFunc, const SFunctionN
} else {
nodesDestroyList(pParameterList);
}
(*pMidFunc)->hasPk = pPartialFunc->hasPk;
(*pMidFunc)->pkBytes = pPartialFunc->pkBytes;
return code;
}
@ -482,7 +485,8 @@ static int32_t createMergeFunction(const SFunctionNode* pSrcFunc, const SFunctio
} else {
nodesDestroyList(pParameterList);
}
(*pMergeFunc)->hasPk = pPartialFunc->hasPk;
(*pMergeFunc)->pkBytes = pPartialFunc->pkBytes;
return code;
}

2
source/libs/parser/inc/parInsertUtil.h
View File

@ -46,7 +46,7 @@ void insBuildCreateTbReq(SVCreateTbReq *pTbReq, const char *tname, STag *pTag
SArray *tagName, uint8_t tagNum, int32_t ttl);
int32_t insInitBoundColsInfo(int32_t numOfBound, SBoundColInfo *pInfo);
void insInitColValues(STableMeta *pTableMeta, SArray *aColValues);
void insCheckTableDataOrder(STableDataCxt *pTableCxt, TSKEY tsKey);
void insCheckTableDataOrder(STableDataCxt *pTableCxt, SRowKey *rowKey);
int32_t insGetTableDataCxt(SHashObj *pHash, void *id, int32_t idLen, STableMeta *pTableMeta,
SVCreateTbReq **pCreateTbReq, STableDataCxt **pTableCxt, bool colMode, bool ignoreColVals);
int32_t initTableColSubmitData(STableDataCxt *pTableCxt);

21
source/libs/parser/src/parInsertSml.c
View File

@ -113,7 +113,8 @@ static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchem
SSchema* pTagSchema = &pSchema[tags->pColIndex[i]];
SSmlKv* kv = taosArrayGet(cols, i);
if(kv->keyLen != strlen(pTagSchema->name) || memcmp(kv->key, pTagSchema->name, kv->keyLen) != 0 || kv->type != pTagSchema->type){
if (kv->keyLen != strlen(pTagSchema->name) || memcmp(kv->key, pTagSchema->name, kv->keyLen) != 0 ||
kv->type != pTagSchema->type) {
code = TSDB_CODE_SML_INVALID_DATA;
uError("SML smlBuildTagRow error col not same %s", pTagSchema->name);
goto end;
@ -200,7 +201,9 @@ int32_t smlBuildRow(STableDataCxt* pTableCxt) {
if (TSDB_CODE_SUCCESS != ret) {
return ret;
}
insCheckTableDataOrder(pTableCxt, TD_ROW_KEY(*pRow));
SRowKey key;
tRowGetKey(*pRow, &key);
insCheckTableDataOrder(pTableCxt, &key);
return TSDB_CODE_SUCCESS;
}
@ -209,13 +212,14 @@ int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32
SSchema* pColSchema = schema + index;
SColVal* pVal = taosArrayGet(pTableCxt->pValues, index);
SSmlKv* kv = (SSmlKv*)data;
if(kv->keyLen != strlen(pColSchema->name) || memcmp(kv->key, pColSchema->name, kv->keyLen) != 0 || kv->type != pColSchema->type){
if (kv->keyLen != strlen(pColSchema->name) || memcmp(kv->key, pColSchema->name, kv->keyLen) != 0 ||
kv->type != pColSchema->type) {
ret = TSDB_CODE_SML_INVALID_DATA;
char* tmp = taosMemoryCalloc(kv->keyLen + 1, 1);
if (tmp) {
memcpy(tmp, kv->key, kv->keyLen);
uInfo("SML data(name:%s type:%s) is not same like the db data(name:%s type:%s)",
tmp, tDataTypes[kv->type].name, pColSchema->name, tDataTypes[pColSchema->type].name);
uInfo("SML data(name:%s type:%s) is not same like the db data(name:%s type:%s)", tmp, tDataTypes[kv->type].name,
pColSchema->name, tDataTypes[pColSchema->type].name);
taosMemoryFree(tmp);
} else {
uError("SML smlBuildCol out of memory");
@ -367,7 +371,8 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
}
if (!taosMbsToUcs4(kv->value, kv->length, (TdUcs4*)pUcs4, pColSchema->bytes - VARSTR_HEADER_SIZE, &len)) {
if (errno == E2BIG) {
uError("sml bind taosMbsToUcs4 error, kv length:%d, bytes:%d, kv->value:%s", (int)kv->length, pColSchema->bytes, kv->value);
uError("sml bind taosMbsToUcs4 error, kv length:%d, bytes:%d, kv->value:%s", (int)kv->length,
pColSchema->bytes, kv->value);
buildInvalidOperationMsg(&pBuf, "value too long");
ret = TSDB_CODE_PAR_VALUE_TOO_LONG;
goto end;
@ -396,7 +401,9 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
buildInvalidOperationMsg(&pBuf, "tRowBuild error");
goto end;
}
insCheckTableDataOrder(pTableCxt, TD_ROW_KEY(*pRow));
SRowKey key;
tRowGetKey(*pRow, &key);
insCheckTableDataOrder(pTableCxt, &key);
clearColValArraySml(pTableCxt->pValues);
}

102
source/libs/parser/src/parInsertSql.c
View File

@ -13,12 +13,12 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "geosWrapper.h"
#include "parInsertUtil.h"
#include "parToken.h"
#include "scalar.h"
#include "tglobal.h"
#include "ttime.h"
#include "geosWrapper.h"
typedef struct SInsertParseContext {
SParseContext* pComCxt;
@ -154,19 +154,15 @@ static int32_t parseDuplicateUsingClause(SInsertParseContext* pCxt, SVnodeModify
return code;
}
typedef enum {
BOUND_TAGS,
BOUND_COLUMNS,
BOUND_ALL_AND_TBNAME
} EBoundColumnsType;
typedef enum { BOUND_TAGS, BOUND_COLUMNS, BOUND_ALL_AND_TBNAME } EBoundColumnsType;
static int32_t getTbnameSchemaIndex(STableMeta* pTableMeta) {
return pTableMeta->tableInfo.numOfTags + pTableMeta->tableInfo.numOfColumns;
}
// pStmt->pSql -> field1_name, ...)
static int32_t parseBoundColumns(SInsertParseContext* pCxt, const char** pSql, EBoundColumnsType boundColsType, STableMeta* pTableMeta,
SBoundColInfo* pBoundInfo) {
static int32_t parseBoundColumns(SInsertParseContext* pCxt, const char** pSql, EBoundColumnsType boundColsType,
STableMeta* pTableMeta, SBoundColInfo* pBoundInfo) {
SSchema* pSchema = NULL;
if (boundColsType == BOUND_TAGS) {
pSchema = getTableTagSchema(pTableMeta);
@ -202,8 +198,7 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, const char** pSql, E
token.z = tmpTokenBuf;
token.n = strdequote(token.z);
if (boundColsType == BOUND_ALL_AND_TBNAME &&
token.n == strlen("tbname") && (strcasecmp(token.z, "tbname") == 0)) {
if (boundColsType == BOUND_ALL_AND_TBNAME && token.n == strlen("tbname") && (strcasecmp(token.z, "tbname") == 0)) {
pBoundInfo->pColIndex[pBoundInfo->numOfBound] = tbnameSchemaIndex;
pUseCols[tbnameSchemaIndex] = true;
++pBoundInfo->numOfBound;
@ -288,7 +283,8 @@ static int parseTime(const char** end, SToken* pToken, int16_t timePrec, int64_t
bool firstIsTS = false, secondIsTs = false;
const char* pTokenEnd = *end;
if (TSDB_CODE_SUCCESS != parseTimestampOrInterval(&pTokenEnd, pToken, timePrec, &ts, &interval, pMsgBuf, &firstIsTS)) {
if (TSDB_CODE_SUCCESS !=
parseTimestampOrInterval(&pTokenEnd, pToken, timePrec, &ts, &interval, pMsgBuf, &firstIsTS)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp format", pToken->z);
}
@ -330,8 +326,7 @@ static int parseTime(const char** end, SToken* pToken, int16_t timePrec, int64_t
if (pTokenEnd[i] == ' ' || pTokenEnd[i] == '\t') {
i++;
continue;
}
else if (pTokenEnd[i] == ',' || pTokenEnd[i] == ')') {
} else if (pTokenEnd[i] == ',' || pTokenEnd[i] == ')') {
*end = pTokenEnd + i;
if (!firstIsTS) {
return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp format", pToken->z);
@ -362,7 +357,8 @@ static int parseTime(const char** end, SToken* pToken, int16_t timePrec, int64_t
valueToken.n = len;
}
if (TSDB_CODE_SUCCESS != parseTimestampOrInterval(&pTokenEnd, &valueToken, timePrec, &tempTs, &tempInterval, pMsgBuf, &secondIsTs)) {
if (TSDB_CODE_SUCCESS !=
parseTimestampOrInterval(&pTokenEnd, &valueToken, timePrec, &tempTs, &tempInterval, pMsgBuf, &secondIsTs)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp format", pToken->z);
}
@ -759,8 +755,8 @@ static int32_t buildCreateTbReq(SVnodeModifyOpStmt* pStmt, STag* pTag, SArray* p
int32_t checkAndTrimValue(SToken* pToken, char* tmpTokenBuf, SMsgBuf* pMsgBuf, int8_t type) {
if ((pToken->type != TK_NOW && pToken->type != TK_TODAY && pToken->type != TK_NK_INTEGER &&
pToken->type != TK_NK_STRING && pToken->type != TK_NK_FLOAT && pToken->type != TK_NK_BOOL &&
pToken->type != TK_NULL && pToken->type != TK_NK_HEX && pToken->type != TK_NK_OCT &&
pToken->type != TK_NK_BIN && pToken->type != TK_NK_VARIABLE) ||
pToken->type != TK_NULL && pToken->type != TK_NK_HEX && pToken->type != TK_NK_OCT && pToken->type != TK_NK_BIN &&
pToken->type != TK_NK_VARIABLE) ||
(pToken->n == 0) || (pToken->type == TK_NK_RP)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid data or symbol", pToken->z);
}
@ -1114,8 +1110,8 @@ static int32_t checkAuth(SParseContext* pCxt, SName* pTbName, bool* pMissCache,
return code;
}
static int32_t getTableMeta(SInsertParseContext* pCxt, SName* pTbName, STableMeta** pTableMeta,
bool* pMissCache, bool bUsingTable) {
static int32_t getTableMeta(SInsertParseContext* pCxt, SName* pTbName, STableMeta** pTableMeta, bool* pMissCache,
bool bUsingTable) {
SParseContext* pComCxt = pCxt->pComCxt;
int32_t code = TSDB_CODE_SUCCESS;
if (pComCxt->async) {
@ -1365,13 +1361,11 @@ static int32_t parseBoundColumnsClause(SInsertParseContext* pCxt, SVnodeModifyOp
return buildSyntaxErrMsg(&pCxt->msg, "keyword VALUES or FILE is expected", token.z);
}
// pStmt->pSql -> field1_name, ...)
return parseBoundColumns(pCxt, &pStmt->pSql, BOUND_COLUMNS, pStmt->pTableMeta,
&pTableCxt->boundColsInfo);
return parseBoundColumns(pCxt, &pStmt->pSql, BOUND_COLUMNS, pStmt->pTableMeta, &pTableCxt->boundColsInfo);
}
if (NULL != pStmt->pBoundCols) {
return parseBoundColumns(pCxt, &pStmt->pBoundCols, BOUND_COLUMNS, pStmt->pTableMeta,
&pTableCxt->boundColsInfo);
return parseBoundColumns(pCxt, &pStmt->pBoundCols, BOUND_COLUMNS, pStmt->pTableMeta, &pTableCxt->boundColsInfo);
}
return TSDB_CODE_SUCCESS;
@ -1700,7 +1694,8 @@ typedef union SRowsDataContext{
SStbRowsDataContext* pStbRowsCxt;
} SRowsDataContext;
static int32_t parseTbnameToken(SInsertParseContext* pCxt, SStbRowsDataContext* pStbRowsCxt, SToken* pToken, bool* pFoundCtbName) {
static int32_t parseTbnameToken(SInsertParseContext* pCxt, SStbRowsDataContext* pStbRowsCxt, SToken* pToken,
bool* pFoundCtbName) {
*pFoundCtbName = false;
int32_t code = checkAndTrimValue(pToken, pCxt->tmpTokenBuf, &pCxt->msg, TSDB_DATA_TYPE_BINARY);
if (TK_NK_VARIABLE == pToken->type) {
@ -1733,9 +1728,8 @@ static int32_t parseTbnameToken(SInsertParseContext* pCxt, SStbRowsDataContext*
}
static int32_t processCtbTagsAfterCtbName(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt,
SStbRowsDataContext* pStbRowsCxt, bool ctbFirst,
const SToken* tagTokens, SSchema* const* tagSchemas,
int numOfTagTokens) {
SStbRowsDataContext* pStbRowsCxt, bool ctbFirst, const SToken* tagTokens,
SSchema* const* tagSchemas, int numOfTagTokens) {
int32_t code = TSDB_CODE_SUCCESS;
uint8_t precision = pStmt->pTableMeta->tableInfo.precision;
@ -1749,8 +1743,8 @@ static int32_t processCtbTagsAfterCtbName(SInsertParseContext* pCxt, SVnodeModif
}
if (code == TSDB_CODE_SUCCESS) {
code = parseTagValue(&pCxt->msg, NULL, precision, pTagSchema, pTagToken, pStbRowsCxt->aTagNames, pStbRowsCxt->aTagVals,
&pStbRowsCxt->pTag);
code = parseTagValue(&pCxt->msg, NULL, precision, pTagSchema, pTagToken, pStbRowsCxt->aTagNames,
pStbRowsCxt->aTagVals, &pStbRowsCxt->pTag);
}
}
if (code == TSDB_CODE_SUCCESS && !pStbRowsCxt->isJsonTag) {
@ -1765,9 +1759,9 @@ static int32_t processCtbTagsAfterCtbName(SInsertParseContext* pCxt, SVnodeModif
}
static int32_t doGetStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, const char** ppSql,
SStbRowsDataContext* pStbRowsCxt, SToken* pToken,
const SBoundColInfo* pCols, const SSchema* pSchemas,
SToken* tagTokens, SSchema** tagSchemas, int* pNumOfTagTokens, bool* bFoundTbName) {
SStbRowsDataContext* pStbRowsCxt, SToken* pToken, const SBoundColInfo* pCols,
const SSchema* pSchemas, SToken* tagTokens, SSchema** tagSchemas, int* pNumOfTagTokens,
bool* bFoundTbName) {
int32_t code = TSDB_CODE_SUCCESS;
SArray* pTagNames = pStbRowsCxt->aTagNames;
SArray* pTagVals = pStbRowsCxt->aTagVals;
@ -1808,11 +1802,11 @@ static int32_t doGetStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt*
code = buildInvalidOperationMsg(&pCxt->msg, "not expected row value");
}
if (code == TSDB_CODE_SUCCESS) {
code = parseTagValue(&pCxt->msg, ppSql, precision, (SSchema*)pTagSchema, pToken, pTagNames, pTagVals, &pStbRowsCxt->pTag);
code = parseTagValue(&pCxt->msg, ppSql, precision, (SSchema*)pTagSchema, pToken, pTagNames, pTagVals,
&pStbRowsCxt->pTag);
}
}
}
else if (pCols->pColIndex[i] == tbnameIdx) {
} else if (pCols->pColIndex[i] == tbnameIdx) {
code = parseTbnameToken(pCxt, pStbRowsCxt, pToken, bFoundTbName);
}
@ -1827,8 +1821,7 @@ static int32_t doGetStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt*
}
static int32_t getStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, const char** ppSql,
SStbRowsDataContext* pStbRowsCxt, bool* pGotRow,
SToken* pToken, bool *pCtbFirst) {
SStbRowsDataContext* pStbRowsCxt, bool* pGotRow, SToken* pToken, bool* pCtbFirst) {
SBoundColInfo* pCols = &pStbRowsCxt->boundColsInfo;
SSchema* pSchemas = getTableColumnSchema(pStbRowsCxt->pStbMeta);
@ -1840,8 +1833,8 @@ static int32_t getStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pS
SSchema* tagSchemas[TSDB_MAX_TAGS] = {0};
int numOfTagTokens = 0;
code = doGetStbRowValues(pCxt, pStmt, ppSql, pStbRowsCxt, pToken, pCols, pSchemas, tagTokens,
tagSchemas, &numOfTagTokens, &bFoundTbName);
code = doGetStbRowValues(pCxt, pStmt, ppSql, pStbRowsCxt, pToken, pCols, pSchemas, tagTokens, tagSchemas,
&numOfTagTokens, &bFoundTbName);
if (code == TSDB_CODE_SUCCESS && !bFoundTbName) {
code = buildSyntaxErrMsg(&pCxt->msg, "tbname value expected", pOrigSql);
@ -1870,7 +1863,8 @@ static int32_t getStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pS
return code;
}
static int32_t processCtbAutoCreationAndCtbMeta(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, SStbRowsDataContext* pStbRowsCxt) {
static int32_t processCtbAutoCreationAndCtbMeta(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt,
SStbRowsDataContext* pStbRowsCxt) {
int32_t code = TSDB_CODE_SUCCESS;
pStbRowsCxt->pCreateCtbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
@ -1878,9 +1872,9 @@ static int32_t processCtbAutoCreationAndCtbMeta(SInsertParseContext* pCxt, SVnod
code = TSDB_CODE_OUT_OF_MEMORY;
}
if (code == TSDB_CODE_SUCCESS) {
insBuildCreateTbReq(pStbRowsCxt->pCreateCtbReq, pStbRowsCxt->ctbName.tname, pStbRowsCxt->pTag, pStbRowsCxt->pStbMeta->uid,
pStbRowsCxt->stbName.tname, pStbRowsCxt->aTagNames, getNumOfTags(pStbRowsCxt->pStbMeta),
TSDB_DEFAULT_TABLE_TTL);
insBuildCreateTbReq(pStbRowsCxt->pCreateCtbReq, pStbRowsCxt->ctbName.tname, pStbRowsCxt->pTag,
pStbRowsCxt->pStbMeta->uid, pStbRowsCxt->stbName.tname, pStbRowsCxt->aTagNames,
getNumOfTags(pStbRowsCxt->pStbMeta), TSDB_DEFAULT_TABLE_TTL);
pStbRowsCxt->pTag = NULL;
}
@ -1907,7 +1901,6 @@ static int32_t processCtbAutoCreationAndCtbMeta(SInsertParseContext* pCxt, SVnod
return code;
}
static void clearStbRowsDataContext(SStbRowsDataContext* pStbRowsCxt) {
if (pStbRowsCxt == NULL) return;
@ -1948,7 +1941,9 @@ static int32_t parseOneStbRow(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pSt
SRow** pRow = taosArrayReserve((*ppTableDataCxt)->pData->aRowP, 1);
code = tRowBuild(pStbRowsCxt->aColVals, (*ppTableDataCxt)->pSchema, pRow);
if (TSDB_CODE_SUCCESS == code) {
insCheckTableDataOrder(*ppTableDataCxt, TD_ROW_KEY(*pRow));
SRowKey key;
tRowGetKey(*pRow, &key);
insCheckTableDataOrder(*ppTableDataCxt, &key);
}
}
@ -1961,7 +1956,8 @@ static int32_t parseOneStbRow(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pSt
return TSDB_CODE_SUCCESS;
}
static int parseOneRow(SInsertParseContext* pCxt, const char** pSql, STableDataCxt* pTableCxt, bool* pGotRow, SToken* pToken) {
static int parseOneRow(SInsertParseContext* pCxt, const char** pSql, STableDataCxt* pTableCxt, bool* pGotRow,
SToken* pToken) {
SBoundColInfo* pCols = &pTableCxt->boundColsInfo;
bool isParseBindParam = false;
SSchema* pSchemas = getTableColumnSchema(pTableCxt->pMeta);
@ -2014,7 +2010,9 @@ static int parseOneRow(SInsertParseContext* pCxt, const char** pSql, STableDataC
SRow** pRow = taosArrayReserve(pTableCxt->pData->aRowP, 1);
code = tRowBuild(pTableCxt->pValues, pTableCxt->pSchema, pRow);
if (TSDB_CODE_SUCCESS == code) {
insCheckTableDataOrder(pTableCxt, TD_ROW_KEY(*pRow));
SRowKey key;
tRowGetKey(*pRow, &key);
insCheckTableDataOrder(pTableCxt, &key);
}
}
@ -2112,7 +2110,8 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt
code = parseOneRow(pCxt, (const char**)&pRow, rowsDataCxt.pTableDataCxt, &gotRow, &token);
} else {
STableDataCxt* pTableDataCxt = NULL;
code = parseOneStbRow(pCxt, pStmt, (const char**)&pRow, rowsDataCxt.pStbRowsCxt, &gotRow, &token, &pTableDataCxt);
code =
parseOneStbRow(pCxt, pStmt, (const char**)&pRow, rowsDataCxt.pStbRowsCxt, &gotRow, &token, &pTableDataCxt);
if (code == TSDB_CODE_SUCCESS) {
SStbRowsDataContext* pStbRowsCxt = rowsDataCxt.pStbRowsCxt;
void* pData = pTableDataCxt;
@ -2149,11 +2148,11 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt
return code;
}
static int32_t parseDataFromFileImpl(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, SRowsDataContext rowsDataCxt) {
static int32_t parseDataFromFileImpl(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt,
SRowsDataContext rowsDataCxt) {
// init only for file
if (NULL == pStmt->pTableCxtHashObj) {
pStmt->pTableCxtHashObj =
taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
pStmt->pTableCxtHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
}
int32_t numOfRows = 0;
int32_t code = parseCsvFile(pCxt, pStmt, rowsDataCxt, &numOfRows);
@ -2285,7 +2284,8 @@ static int32_t constructStbRowsDataContext(SVnodeModifyOpStmt* pStmt, SStbRowsDa
static int32_t parseInsertStbClauseBottom(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt) {
int32_t code = TSDB_CODE_SUCCESS;
if (!pStmt->pBoundCols) {
return buildSyntaxErrMsg(&pCxt->msg, "(...tbname, ts...) bounded cols is expected for supertable insertion", pStmt->pSql);
return buildSyntaxErrMsg(&pCxt->msg, "(...tbname, ts...) bounded cols is expected for supertable insertion",
pStmt->pSql);
}
SStbRowsDataContext* pStbRowsCxt = NULL;

24
source/libs/parser/src/parInsertUtil.c
View File

@ -170,9 +170,7 @@ static void initColValues(STableMeta* pTableMeta, SArray* pValues) {
}
}
void insInitColValues(STableMeta* pTableMeta, SArray* aColValues) {
initColValues(pTableMeta, aColValues);
}
void insInitColValues(STableMeta* pTableMeta, SArray* aColValues) { initColValues(pTableMeta, aColValues); }
int32_t insInitBoundColsInfo(int32_t numOfBound, SBoundColInfo* pInfo) {
pInfo->numOfCols = numOfBound;
@ -187,21 +185,22 @@ int32_t insInitBoundColsInfo(int32_t numOfBound, SBoundColInfo* pInfo) {
return TSDB_CODE_SUCCESS;
}
void insCheckTableDataOrder(STableDataCxt* pTableCxt, TSKEY tsKey) {
void insCheckTableDataOrder(STableDataCxt* pTableCxt, SRowKey* rowKey) {
// once the data block is disordered, we do NOT keep last timestamp any more
if (!pTableCxt->ordered) {
return;
}
if (tsKey < pTableCxt->lastTs) {
if (tRowKeyCompare(rowKey, &pTableCxt->lastKey) < 0) {
pTableCxt->ordered = false;
}
if (tsKey == pTableCxt->lastTs) {
if (tRowKeyCompare(rowKey, &pTableCxt->lastKey) == 0) {
pTableCxt->duplicateTs = true;
}
pTableCxt->lastTs = tsKey;
// TODO: for variable length data type, we need to copy it out
pTableCxt->lastKey = *rowKey;
return;
}
@ -217,7 +216,7 @@ static int32_t createTableDataCxt(STableMeta* pTableMeta, SVCreateTbReq** pCreat
int32_t code = TSDB_CODE_SUCCESS;
pTableCxt->lastTs = 0;
pTableCxt->lastKey = (SRowKey){0};
pTableCxt->ordered = true;
pTableCxt->duplicateTs = false;
@ -317,7 +316,6 @@ static int32_t rebuildTableData(SSubmitTbData* pSrc, SSubmitTbData** pDst) {
return code;
}
static void resetColValues(SArray* pValues) {
int32_t num = taosArrayGetSize(pValues);
for (int32_t i = 0; i < num; ++i) {
@ -644,7 +642,8 @@ static bool findFileds(SSchema* pSchema, TAOS_FIELD* fields, int numFields) {
int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreateTbReq** pCreateTb, TAOS_FIELD* tFields,
int numFields, bool needChangeLength) {
void* tmp = taosHashGet(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid, sizeof(pTableMeta->uid));
void* tmp =
taosHashGet(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid, sizeof(pTableMeta->uid));
STableDataCxt* pTableCxt = NULL;
int ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid,
sizeof(pTableMeta->uid), pTableMeta, pCreateTb, &pTableCxt, true, false);
@ -663,8 +662,8 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
}
char* p = (char*)data;
// | version | total length | total rows | blankFill | total columns | flag seg| block group id | column schema | each column
// length |
// | version | total length | total rows | blankFill | total columns | flag seg| block group id | column schema | each
// column length |
int32_t version = *(int32_t*)data;
p += sizeof(int32_t);
p += sizeof(int32_t);
@ -761,7 +760,6 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
break;
}
}
}
for (int c = 0; c < boundInfo->numOfBound; ++c) {