more code change
This commit is contained in:
Hongze Cheng
parent
3ccab8a5b8
commit
93ca55954f
|
|
@ -197,7 +197,7 @@ typedef struct STableDataCxt {
|
|||
SBoundColInfo boundColsInfo;
|
||||
SArray* pValues;
|
||||
SSubmitTbData* pData;
|
||||
TSKEY lastTs;
|
||||
SRowKey lastKey;
|
||||
bool ordered;
|
||||
bool duplicateTs;
|
||||
} STableDataCxt;
|
||||
|
|
|
|||
|
|
@ -45,8 +45,8 @@ int16_t insFindCol(struct SToken *pColname, int16_t start, int16_t end, SSchema
|
|||
void insBuildCreateTbReq(SVCreateTbReq *pTbReq, const char *tname, STag *pTag, int64_t suid, const char *sname,
|
||||
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 insInitColValues(STableMeta *pTableMeta, SArray *aColValues);
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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,15 +212,16 @@ 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){
|
||||
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{
|
||||
} else {
|
||||
uError("SML smlBuildCol out of memory");
|
||||
}
|
||||
goto end;
|
||||
|
|
@ -225,11 +229,11 @@ int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32
|
|||
if (kv->type == TSDB_DATA_TYPE_NCHAR) {
|
||||
int32_t len = 0;
|
||||
int64_t size = pColSchema->bytes - VARSTR_HEADER_SIZE;
|
||||
if(size <= 0){
|
||||
if (size <= 0) {
|
||||
ret = TSDB_CODE_SML_INVALID_DATA;
|
||||
goto end;
|
||||
}
|
||||
char* pUcs4 = taosMemoryCalloc(1, size);
|
||||
char* pUcs4 = taosMemoryCalloc(1, size);
|
||||
if (NULL == pUcs4) {
|
||||
ret = TSDB_CODE_OUT_OF_MEMORY;
|
||||
goto end;
|
||||
|
|
@ -351,7 +355,7 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
|
|||
continue;
|
||||
}
|
||||
SSmlKv* kv = *(SSmlKv**)p;
|
||||
if(kv->type != pColSchema->type){
|
||||
if (kv->type != pColSchema->type) {
|
||||
ret = buildInvalidOperationMsg(&pBuf, "kv type not equal to col type");
|
||||
goto end;
|
||||
}
|
||||
|
|
@ -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);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
@ -29,7 +29,7 @@ typedef struct SInsertParseContext {
|
|||
bool usingDuplicateTable;
|
||||
bool forceUpdate;
|
||||
bool needTableTagVal;
|
||||
bool needRequest; // whether or not request server
|
||||
bool needRequest; // whether or not request server
|
||||
} SInsertParseContext;
|
||||
|
||||
typedef int32_t (*_row_append_fn_t)(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param);
|
||||
|
|
@ -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;
|
||||
|
|
@ -230,7 +225,7 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, const char** pSql, E
|
|||
if (TSDB_CODE_SUCCESS == code && (BOUND_TAGS != boundColsType) && !pUseCols[0]) {
|
||||
code = buildInvalidOperationMsg(&pCxt->msg, "primary timestamp column can not be null");
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code && (BOUND_ALL_AND_TBNAME == boundColsType) &&!pUseCols[tbnameSchemaIndex]) {
|
||||
if (TSDB_CODE_SUCCESS == code && (BOUND_ALL_AND_TBNAME == boundColsType) && !pUseCols[tbnameSchemaIndex]) {
|
||||
code = buildInvalidOperationMsg(&pCxt->msg, "tbname column can not be null");
|
||||
}
|
||||
taosMemoryFree(pUseCols);
|
||||
|
|
@ -285,10 +280,11 @@ static int parseTime(const char** end, SToken* pToken, int16_t timePrec, int64_t
|
|||
int32_t index = 0, i = 0;
|
||||
int64_t interval = 0, tempInterval = 0;
|
||||
int64_t ts = 0, tempTs = 0;
|
||||
bool firstIsTS = false, secondIsTs = false;
|
||||
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);
|
||||
}
|
||||
|
||||
|
|
@ -376,7 +372,7 @@ static int parseTime(const char** end, SToken* pToken, int16_t timePrec, int64_t
|
|||
return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp format", pToken->z);
|
||||
}
|
||||
ts = tempTs;
|
||||
}else {
|
||||
} else {
|
||||
// not support operator between tow interval, such as 2h + 3s
|
||||
if (!firstIsTS) {
|
||||
return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp format", pToken->z);
|
||||
|
|
@ -413,7 +409,7 @@ static int parseTime(const char** end, SToken* pToken, int16_t timePrec, int64_t
|
|||
}
|
||||
|
||||
// need to call geosFreeBuffer(*output) later
|
||||
static int parseGeometry(SToken *pToken, unsigned char **output, size_t *size) {
|
||||
static int parseGeometry(SToken* pToken, unsigned char** output, size_t* size) {
|
||||
int32_t code = TSDB_CODE_FAILED;
|
||||
|
||||
//[ToDo] support to parse WKB as well as WKT
|
||||
|
|
@ -432,19 +428,19 @@ static int parseGeometry(SToken *pToken, unsigned char **output, size_t *size) {
|
|||
return code;
|
||||
}
|
||||
|
||||
static int32_t parseVarbinary(SToken* pToken, uint8_t **pData, uint32_t *nData, int32_t bytes){
|
||||
if(pToken->type != TK_NK_STRING){
|
||||
static int32_t parseVarbinary(SToken* pToken, uint8_t** pData, uint32_t* nData, int32_t bytes) {
|
||||
if (pToken->type != TK_NK_STRING) {
|
||||
return TSDB_CODE_PAR_INVALID_VARBINARY;
|
||||
}
|
||||
|
||||
if(isHex(pToken->z + 1, pToken->n - 2)){
|
||||
if(!isValidateHex(pToken->z + 1, pToken->n - 2)){
|
||||
if (isHex(pToken->z + 1, pToken->n - 2)) {
|
||||
if (!isValidateHex(pToken->z + 1, pToken->n - 2)) {
|
||||
return TSDB_CODE_PAR_INVALID_VARBINARY;
|
||||
}
|
||||
|
||||
void* data = NULL;
|
||||
void* data = NULL;
|
||||
uint32_t size = 0;
|
||||
if(taosHex2Ascii(pToken->z + 1, pToken->n - 2, &data, &size) < 0){
|
||||
if (taosHex2Ascii(pToken->z + 1, pToken->n - 2, &data, &size) < 0) {
|
||||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
|
|
@ -454,7 +450,7 @@ static int32_t parseVarbinary(SToken* pToken, uint8_t **pData, uint32_t *nData,
|
|||
}
|
||||
*pData = data;
|
||||
*nData = size;
|
||||
}else{
|
||||
} else {
|
||||
*pData = taosMemoryCalloc(1, pToken->n);
|
||||
int32_t len = trimString(pToken->z, pToken->n, *pData, pToken->n);
|
||||
*nData = len;
|
||||
|
|
@ -633,7 +629,7 @@ static int32_t parseTagToken(const char** end, SToken* pToken, SSchema* pSchema,
|
|||
}
|
||||
case TSDB_DATA_TYPE_VARBINARY: {
|
||||
code = parseVarbinary(pToken, &val->pData, &val->nData, pSchema->bytes);
|
||||
if(code != TSDB_CODE_SUCCESS){
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
return generateSyntaxErrMsg(pMsgBuf, code, pSchema->name);
|
||||
}
|
||||
break;
|
||||
|
|
@ -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) {
|
||||
|
|
@ -1220,7 +1216,7 @@ static int32_t getTargetTableSchema(SInsertParseContext* pCxt, SVnodeModifyOpStm
|
|||
pCxt->missCache = true;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
SNode* pTagCond = NULL;
|
||||
SNode* pTagCond = NULL;
|
||||
int32_t code = checkAuth(pCxt->pComCxt, &pStmt->targetTableName, &pCxt->missCache, &pTagCond);
|
||||
if (TSDB_CODE_SUCCESS == code && !pCxt->missCache) {
|
||||
code = getTargetTableMetaAndVgroup(pCxt, pStmt, &pCxt->missCache);
|
||||
|
|
@ -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;
|
||||
|
|
@ -1679,34 +1673,35 @@ static void clearColValArray(SArray* pCols) {
|
|||
typedef struct SStbRowsDataContext {
|
||||
SName stbName;
|
||||
|
||||
STableMeta* pStbMeta;
|
||||
SNode* pTagCond;
|
||||
STableMeta* pStbMeta;
|
||||
SNode* pTagCond;
|
||||
SBoundColInfo boundColsInfo;
|
||||
|
||||
// the following fields are for each stb row
|
||||
SArray* aTagVals;
|
||||
SArray* aColVals;
|
||||
SArray* aTagNames;
|
||||
SName ctbName;
|
||||
STag* pTag;
|
||||
STableMeta* pCtbMeta;
|
||||
SArray* aTagVals;
|
||||
SArray* aColVals;
|
||||
SArray* aTagNames;
|
||||
SName ctbName;
|
||||
STag* pTag;
|
||||
STableMeta* pCtbMeta;
|
||||
SVCreateTbReq* pCreateCtbReq;
|
||||
bool hasTimestampTag;
|
||||
bool isJsonTag;
|
||||
} SStbRowsDataContext;
|
||||
|
||||
typedef union SRowsDataContext{
|
||||
STableDataCxt* pTableDataCxt;
|
||||
typedef union SRowsDataContext {
|
||||
STableDataCxt* pTableDataCxt;
|
||||
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) {
|
||||
code = buildInvalidOperationMsg(&pCxt->msg, "not expected tbname");
|
||||
}
|
||||
if (code == TSDB_CODE_SUCCESS){
|
||||
if (code == TSDB_CODE_SUCCESS) {
|
||||
if (isNullValue(TSDB_DATA_TYPE_BINARY, pToken)) {
|
||||
return buildInvalidOperationMsg(&pCxt->msg, "tbname can not be null value");
|
||||
}
|
||||
|
|
@ -1717,7 +1712,7 @@ static int32_t parseTbnameToken(SInsertParseContext* pCxt, SStbRowsDataContext*
|
|||
if (pToken->z[i] == '.') {
|
||||
return buildInvalidOperationMsg(&pCxt->msg, "tbname can not contain '.'");
|
||||
} else {
|
||||
pStbRowsCxt->ctbName.tname[i] = pToken->z[i];
|
||||
pStbRowsCxt->ctbName.tname[i] = pToken->z[i];
|
||||
}
|
||||
}
|
||||
pStbRowsCxt->ctbName.tname[pToken->n] = '\0';
|
||||
|
|
@ -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;
|
||||
|
|
@ -1791,10 +1785,10 @@ static int32_t doGetStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt*
|
|||
}
|
||||
if (pCols->pColIndex[i] < numOfCols) {
|
||||
const SSchema* pSchema = &pSchemas[pCols->pColIndex[i]];
|
||||
SColVal* pVal = taosArrayGet(pStbRowsCxt->aColVals, pCols->pColIndex[i]);
|
||||
SColVal* pVal = taosArrayGet(pStbRowsCxt->aColVals, pCols->pColIndex[i]);
|
||||
code = parseValueToken(pCxt, ppSql, pToken, (SSchema*)pSchema, precision, pVal);
|
||||
if (TK_NK_VARIABLE == pToken->type) {
|
||||
code = buildInvalidOperationMsg(&pCxt->msg, "not expected row value");
|
||||
code = buildInvalidOperationMsg(&pCxt->msg, "not expected row value");
|
||||
}
|
||||
} else if (pCols->pColIndex[i] < tbnameIdx) {
|
||||
const SSchema* pTagSchema = &pSchemas[pCols->pColIndex[i]];
|
||||
|
|
@ -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);
|
||||
}
|
||||
|
||||
|
|
@ -1826,22 +1820,21 @@ static int32_t doGetStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt*
|
|||
return code;
|
||||
}
|
||||
|
||||
static int32_t getStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, const char** ppSql,
|
||||
SStbRowsDataContext* pStbRowsCxt, bool* pGotRow,
|
||||
SToken* pToken, bool *pCtbFirst) {
|
||||
static int32_t getStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt, const char** ppSql,
|
||||
SStbRowsDataContext* pStbRowsCxt, bool* pGotRow, SToken* pToken, bool* pCtbFirst) {
|
||||
SBoundColInfo* pCols = &pStbRowsCxt->boundColsInfo;
|
||||
SSchema* pSchemas = getTableColumnSchema(pStbRowsCxt->pStbMeta);
|
||||
|
||||
bool bFoundTbName = false;
|
||||
bool bFoundTbName = false;
|
||||
const char* pOrigSql = *ppSql;
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
SToken tagTokens[TSDB_MAX_TAGS] = {0};
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
SToken tagTokens[TSDB_MAX_TAGS] = {0};
|
||||
SSchema* tagSchemas[TSDB_MAX_TAGS] = {0};
|
||||
int numOfTagTokens = 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,16 +1872,16 @@ 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;
|
||||
}
|
||||
|
||||
if (code == TSDB_CODE_SUCCESS) {
|
||||
char ctbFName[TSDB_TABLE_FNAME_LEN];
|
||||
tNameExtractFullName(&pStbRowsCxt->ctbName, ctbFName);
|
||||
SVgroupInfo vg;
|
||||
SVgroupInfo vg;
|
||||
SRequestConnInfo conn = {.pTrans = pCxt->pComCxt->pTransporter,
|
||||
.requestId = pCxt->pComCxt->requestId,
|
||||
.requestObjRefId = pCxt->pComCxt->requestRid,
|
||||
|
|
@ -1907,7 +1901,6 @@ static int32_t processCtbAutoCreationAndCtbMeta(SInsertParseContext* pCxt, SVnod
|
|||
return code;
|
||||
}
|
||||
|
||||
|
||||
static void clearStbRowsDataContext(SStbRowsDataContext* pStbRowsCxt) {
|
||||
if (pStbRowsCxt == NULL) return;
|
||||
|
||||
|
|
@ -1938,7 +1931,7 @@ static int32_t parseOneStbRow(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pSt
|
|||
}
|
||||
|
||||
if (code == TSDB_CODE_SUCCESS && bFirstTable) {
|
||||
code = processCtbAutoCreationAndCtbMeta(pCxt, pStmt, pStbRowsCxt);
|
||||
code = processCtbAutoCreationAndCtbMeta(pCxt, pStmt, pStbRowsCxt);
|
||||
}
|
||||
|
||||
code = insGetTableDataCxt(pStmt->pTableBlockHashObj, &pStbRowsCxt->pCtbMeta->uid, sizeof(pStbRowsCxt->pCtbMeta->uid),
|
||||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1957,11 +1952,12 @@ static int32_t parseOneStbRow(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pSt
|
|||
}
|
||||
|
||||
clearStbRowsDataContext(pStbRowsCxt);
|
||||
|
||||
|
||||
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,10 +2110,11 @@ 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;
|
||||
void* pData = pTableDataCxt;
|
||||
taosHashPut(pStmt->pTableCxtHashObj, &pStbRowsCxt->pCtbMeta->uid, sizeof(pStbRowsCxt->pCtbMeta->uid), &pData,
|
||||
POINTER_BYTES);
|
||||
}
|
||||
|
|
@ -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);
|
||||
|
|
@ -2267,7 +2266,7 @@ static int32_t constructStbRowsDataContext(SVnodeModifyOpStmt* pStmt, SStbRowsDa
|
|||
cloneTableMeta(pStbRowsCxt->pStbMeta, &pStbRowsCxt->pCtbMeta);
|
||||
pStbRowsCxt->pCtbMeta->tableType = TSDB_CHILD_TABLE;
|
||||
pStbRowsCxt->pCtbMeta->suid = pStbRowsCxt->pStbMeta->uid;
|
||||
|
||||
|
||||
pStbRowsCxt->aTagNames = taosArrayInit(8, TSDB_COL_NAME_LEN);
|
||||
pStbRowsCxt->aTagVals = taosArrayInit(8, sizeof(STagVal));
|
||||
|
||||
|
|
@ -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;
|
||||
|
|
@ -2297,7 +2297,7 @@ static int32_t parseInsertStbClauseBottom(SInsertParseContext* pCxt, SVnodeModif
|
|||
pStbRowsCxt->hasTimestampTag = false;
|
||||
for (int32_t i = 0; i < pStbRowsCxt->boundColsInfo.numOfBound; ++i) {
|
||||
int16_t schemaIndex = pStbRowsCxt->boundColsInfo.pColIndex[i];
|
||||
if (schemaIndex != getTbnameSchemaIndex(pStmt->pTableMeta) && schemaIndex >= getNumOfColumns(pStmt->pTableMeta) ) {
|
||||
if (schemaIndex != getTbnameSchemaIndex(pStmt->pTableMeta) && schemaIndex >= getNumOfColumns(pStmt->pTableMeta)) {
|
||||
if (pStmt->pTableMeta->schema[schemaIndex].type == TSDB_DATA_TYPE_TIMESTAMP) {
|
||||
pStbRowsCxt->hasTimestampTag = true;
|
||||
}
|
||||
|
|
@ -2326,8 +2326,8 @@ static int32_t parseInsertTableClauseBottom(SInsertParseContext* pCxt, SVnodeMod
|
|||
return buildInvalidOperationMsg(&pCxt->msg, "insert into super table syntax is not supported for stmt");
|
||||
}
|
||||
if (!pStmt->stbSyntax) {
|
||||
STableDataCxt* pTableCxt = NULL;
|
||||
int32_t code = parseSchemaClauseBottom(pCxt, pStmt, &pTableCxt);
|
||||
STableDataCxt* pTableCxt = NULL;
|
||||
int32_t code = parseSchemaClauseBottom(pCxt, pStmt, &pTableCxt);
|
||||
SRowsDataContext rowsDataCxt;
|
||||
rowsDataCxt.pTableDataCxt = pTableCxt;
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
|
|
@ -2442,7 +2442,7 @@ static int32_t parseInsertBodyBottom(SInsertParseContext* pCxt, SVnodeModifyOpSt
|
|||
|
||||
bool fileOnly = (pStmt->insertType == TSDB_QUERY_TYPE_FILE_INSERT);
|
||||
if (fileOnly) {
|
||||
// none data, skip merge & buildvgdata
|
||||
// none data, skip merge & buildvgdata
|
||||
if (0 == taosHashGetSize(pStmt->pTableCxtHashObj)) {
|
||||
pCxt->needRequest = false;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
|
|
@ -2503,7 +2503,7 @@ static int32_t createVnodeModifOpStmt(SInsertParseContext* pCxt, bool reentry, S
|
|||
pStmt->freeHashFunc = insDestroyTableDataCxtHashMap;
|
||||
pStmt->freeArrayFunc = insDestroyVgroupDataCxtList;
|
||||
pStmt->freeStbRowsCxtFunc = destroyStbRowsDataContext;
|
||||
|
||||
|
||||
if (!reentry) {
|
||||
pStmt->pVgroupsHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
|
||||
pStmt->pTableBlockHashObj =
|
||||
|
|
@ -2625,7 +2625,7 @@ static int32_t checkSubtablePrivilegeForTable(const SArray* pTables, SVnodeModif
|
|||
}
|
||||
|
||||
static int32_t processTableSchemaFromMetaData(SInsertParseContext* pCxt, const SMetaData* pMetaData,
|
||||
SVnodeModifyOpStmt* pStmt, bool isStb) {
|
||||
SVnodeModifyOpStmt* pStmt, bool isStb) {
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
if (!isStb && TSDB_SUPER_TABLE == pStmt->pTableMeta->tableType) {
|
||||
code = buildInvalidOperationMsg(&pCxt->msg, "insert data into super table is not supported");
|
||||
|
|
@ -2767,11 +2767,11 @@ static int32_t parseInsertSqlFromStart(SInsertParseContext* pCxt, SVnodeModifyOp
|
|||
}
|
||||
|
||||
static int32_t parseInsertSqlFromCsv(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt) {
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
SRowsDataContext rowsDataCxt;
|
||||
|
||||
if (!pStmt->stbSyntax) {
|
||||
STableDataCxt* pTableCxt = NULL;
|
||||
STableDataCxt* pTableCxt = NULL;
|
||||
code = getTableDataCxt(pCxt, pStmt, &pTableCxt);
|
||||
rowsDataCxt.pTableDataCxt = pTableCxt;
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
|
@ -254,7 +253,7 @@ static int32_t createTableDataCxt(STableMeta* pTableMeta, SVCreateTbReq** pCreat
|
|||
pTableCxt->pData->uid = pTableMeta->uid;
|
||||
pTableCxt->pData->sver = pTableMeta->sversion;
|
||||
pTableCxt->pData->pCreateTbReq = pCreateTbReq != NULL ? *pCreateTbReq : NULL;
|
||||
if(pCreateTbReq != NULL) *pCreateTbReq = NULL;
|
||||
if (pCreateTbReq != NULL) *pCreateTbReq = NULL;
|
||||
if (pTableCxt->pData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
|
||||
pTableCxt->pData->aCol = taosArrayInit(128, sizeof(SColData));
|
||||
if (NULL == pTableCxt->pData->aCol) {
|
||||
|
|
@ -276,7 +275,7 @@ static int32_t createTableDataCxt(STableMeta* pTableMeta, SVCreateTbReq** pCreat
|
|||
}
|
||||
|
||||
static int32_t rebuildTableData(SSubmitTbData* pSrc, SSubmitTbData** pDst) {
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
SSubmitTbData* pTmp = taosMemoryCalloc(1, sizeof(SSubmitTbData));
|
||||
if (NULL == pTmp) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
|
|
@ -313,11 +312,10 @@ static int32_t rebuildTableData(SSubmitTbData* pSrc, SSubmitTbData** pDst) {
|
|||
if (TSDB_CODE_SUCCESS == code) {
|
||||
*pDst = pTmp;
|
||||
}
|
||||
|
||||
|
||||
return code;
|
||||
}
|
||||
|
||||
|
||||
static void resetColValues(SArray* pValues) {
|
||||
int32_t num = taosArrayGetSize(pValues);
|
||||
for (int32_t i = 0; i < num; ++i) {
|
||||
|
|
@ -463,7 +461,7 @@ static int32_t createVgroupDataCxt(STableDataCxt* pTableCxt, SHashObj* pVgroupHa
|
|||
int32_t code = taosHashPut(pVgroupHash, &pVgCxt->vgId, sizeof(pVgCxt->vgId), &pVgCxt, POINTER_BYTES);
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
taosArrayPush(pVgroupList, &pVgCxt);
|
||||
// uDebug("td23101 2vgId:%d, uid:%" PRIu64, pVgCxt->vgId, pTableCxt->pMeta->uid);
|
||||
// uDebug("td23101 2vgId:%d, uid:%" PRIu64, pVgCxt->vgId, pTableCxt->pMeta->uid);
|
||||
*pOutput = pVgCxt;
|
||||
} else {
|
||||
insDestroyVgroupDataCxt(pVgCxt);
|
||||
|
|
@ -613,7 +611,7 @@ int32_t insBuildVgDataBlocks(SHashObj* pVgroupsHashObj, SArray* pVgDataCxtList,
|
|||
if (TSDB_CODE_SUCCESS == code) {
|
||||
dst->numOfTables = taosArrayGetSize(src->pData->aSubmitTbData);
|
||||
code = taosHashGetDup(pVgroupsHashObj, (const char*)&src->vgId, sizeof(src->vgId), &dst->vg);
|
||||
// uError("td23101 3vgId:%d, numEps:%d", src->vgId, dst->vg.epSet.numOfEps);
|
||||
// uError("td23101 3vgId:%d, numEps:%d", src->vgId, dst->vg.epSet.numOfEps);
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = buildSubmitReq(src->vgId, src->pData, &dst->pData, &dst->size);
|
||||
|
|
@ -634,7 +632,7 @@ int32_t insBuildVgDataBlocks(SHashObj* pVgroupsHashObj, SArray* pVgDataCxtList,
|
|||
|
||||
static bool findFileds(SSchema* pSchema, TAOS_FIELD* fields, int numFields) {
|
||||
for (int i = 0; i < numFields; i++) {
|
||||
if(strcmp(pSchema->name, fields[i].name) == 0){
|
||||
if (strcmp(pSchema->name, fields[i].name) == 0) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
|
@ -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);
|
||||
|
|
@ -654,7 +653,7 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
|
|||
}
|
||||
|
||||
pTableCxt->pData->flags |= TD_REQ_FROM_TAOX;
|
||||
if(tmp == NULL){
|
||||
if (tmp == NULL) {
|
||||
ret = initTableColSubmitData(pTableCxt);
|
||||
if (ret != TSDB_CODE_SUCCESS) {
|
||||
uError("initTableColSubmitData error");
|
||||
|
|
@ -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);
|
||||
|
|
@ -699,8 +698,8 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
|
|||
ret = TSDB_CODE_INVALID_PARA;
|
||||
goto end;
|
||||
}
|
||||
if(tFields == NULL){
|
||||
for (int j = 0; j < boundInfo->numOfBound; j++){
|
||||
if (tFields == NULL) {
|
||||
for (int j = 0; j < boundInfo->numOfBound; j++) {
|
||||
SSchema* pColSchema = &pSchema[j];
|
||||
SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, j);
|
||||
if (*fields != pColSchema->type && *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) {
|
||||
|
|
@ -717,7 +716,7 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
|
|||
}
|
||||
char* pData = pStart;
|
||||
ret = tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, offset, pData);
|
||||
if(ret != 0){
|
||||
if (ret != 0) {
|
||||
goto end;
|
||||
}
|
||||
fields += sizeof(int8_t) + sizeof(int32_t);
|
||||
|
|
@ -727,11 +726,11 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
|
|||
pStart += colLength[j];
|
||||
}
|
||||
}
|
||||
}else{
|
||||
} else {
|
||||
for (int i = 0; i < numFields; i++) {
|
||||
for (int j = 0; j < boundInfo->numOfBound; j++){
|
||||
SSchema* pColSchema = &pSchema[j];
|
||||
if(strcmp(pColSchema->name, tFields[i].name) == 0){
|
||||
for (int j = 0; j < boundInfo->numOfBound; j++) {
|
||||
SSchema* pColSchema = &pSchema[j];
|
||||
if (strcmp(pColSchema->name, tFields[i].name) == 0) {
|
||||
if (*fields != pColSchema->type && *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) {
|
||||
uError("type or bytes not equal");
|
||||
ret = TSDB_CODE_INVALID_PARA;
|
||||
|
|
@ -748,7 +747,7 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
|
|||
|
||||
SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, j);
|
||||
ret = tColDataAddValueByDataBlock(pCol, pColSchema->type, pColSchema->bytes, numOfRows, offset, pData);
|
||||
if(ret != 0){
|
||||
if (ret != 0) {
|
||||
goto end;
|
||||
}
|
||||
fields += sizeof(int8_t) + sizeof(int32_t);
|
||||
|
|
@ -761,17 +760,16 @@ int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreate
|
|||
break;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
for (int c = 0; c < boundInfo->numOfBound; ++c) {
|
||||
if( boundInfo->pColIndex[c] != -1){
|
||||
if (boundInfo->pColIndex[c] != -1) {
|
||||
SColData* pCol = taosArrayGet(pTableCxt->pData->aCol, c);
|
||||
ret = tColDataAddValueByDataBlock(pCol, 0, 0, numOfRows, NULL, NULL);
|
||||
if(ret != 0){
|
||||
if (ret != 0) {
|
||||
goto end;
|
||||
}
|
||||
}else{
|
||||
} else {
|
||||
boundInfo->pColIndex[c] = c; // restore for next block
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue