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

enh: tag refactor for normal tags

This commit is contained in:
Cary Xu 2022-05-28 17:16:04 +08:00
parent 0bbbd24e1d
commit 9b4c9e4802
17 changed files with 439 additions and 418 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

148
include/common/tdataformat.h
View File

@ -59,13 +59,18 @@ void tTSRowBuilderReset(STSRowBuilder *pBuilder);
int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, uint8_t *pData, uint32_t nData); int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, uint8_t *pData, uint32_t nData);
int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow); int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow);
// STagVal
static FORCE_INLINE void tTagValSet(STagVal *pTagVal, void *key, int8_t type, uint8_t *pData, uint32_t nData,
bool isJson);
// STag // STag
int32_t tTagNew(STagVal *pTagVals, int16_t nTag, int32_t version, int8_t isJson, STag **ppTag); int32_t tTagNew(STagVal *pTagVals, int16_t nTag, int32_t version, int8_t isJson, STag **ppTag);
void tTagFree(STag *pTag); void tTagFree(STag *pTag);
void tTagGet(STag *pTag, STagVal *pTagVal); bool tTagGet(const STag *pTag, STagVal *pTagVal);
int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag); int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag);
int32_t tDecodeTag(SDecoder *pDecoder, STag **ppTag); int32_t tDecodeTag(SDecoder *pDecoder, STag **ppTag);
int32_t tTagToValArray(STag *pTag, SArray **ppArray); int32_t tTagToValArray(const STag *pTag, SArray **ppArray);
void debugPrintSTag(STag *pTag, const char *tag, int32_t ln);
// STRUCT ================= // STRUCT =================
struct STColumn { struct STColumn {
@ -128,6 +133,29 @@ struct STagVal {
uint8_t *pData; uint8_t *pData;
}; };
static FORCE_INLINE void tTagValSet(STagVal *pTagVal, void *key, int8_t type, uint8_t *pData, uint32_t nData,
bool isJson) {
if (isJson) {
pTagVal->pKey = (char *)key;
} else {
pTagVal->cid = *(int16_t *)key;
}
pTagVal->type = type;
pTagVal->pData = pData;
pTagVal->nData = nData;
}
#pragma pack(push, 1)
struct STag {
int8_t isJson;
int16_t len;
int16_t nTag;
int32_t ver;
int16_t idx[];
};
#pragma pack(pop)
#if 1 //================================================================================================================================================ #if 1 //================================================================================================================================================
// Imported since 3.0 and use bitmap to demonstrate None/Null/Norm, while use Null/Norm below 3.0 without of bitmap. // Imported since 3.0 and use bitmap to demonstrate None/Null/Norm, while use Null/Norm below 3.0 without of bitmap.
#define TD_SUPPORT_BITMAP #define TD_SUPPORT_BITMAP
@ -186,7 +214,8 @@ struct STagVal {
#define schemaColAt(s, i) ((s)->columns + i) #define schemaColAt(s, i) ((s)->columns + i)
#define tdFreeSchema(s) taosMemoryFreeClear((s)) #define tdFreeSchema(s) taosMemoryFreeClear((s))
STSchema *tdDupSchema(const STSchema *pSchema); STSchema *
tdDupSchema(const STSchema *pSchema);
int32_t tdEncodeSchema(void **buf, STSchema *pSchema); int32_t tdEncodeSchema(void **buf, STSchema *pSchema);
void *tdDecodeSchema(void *buf, STSchema **pRSchema); void *tdDecodeSchema(void *buf, STSchema **pRSchema);
@ -370,109 +399,8 @@ SDataCols *tdFreeDataCols(SDataCols *pCols);
int32_t tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToMerge, int32_t *pOffset, bool update, int32_t tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToMerge, int32_t *pOffset, bool update,
TDRowVerT maxVer); TDRowVerT maxVer);
// ----------------- K-V data row structure
/* |<-------------------------------------- len -------------------------------------------->|
* |<----- header ----->|<--------------------------- body -------------------------------->|
* +----------+----------+---------------------------------+---------------------------------+
* | uint16_t | int16_t | | |
* +----------+----------+---------------------------------+---------------------------------+
* | len | ncols | cols index | data part |
* +----------+----------+---------------------------------+---------------------------------+
*/
typedef void *SKVRow;
typedef struct {
int16_t colId;
uint16_t offset;
} SColIdx;
#define TD_KV_ROW_HEAD_SIZE (sizeof(uint16_t) + sizeof(int16_t))
#define kvRowLen(r) (*(uint16_t *)(r))
#define kvRowNCols(r) (*(int16_t *)POINTER_SHIFT(r, sizeof(uint16_t)))
#define kvRowSetLen(r, len) kvRowLen(r) = (len)
#define kvRowSetNCols(r, n) kvRowNCols(r) = (n)
#define kvRowColIdx(r) (SColIdx *)POINTER_SHIFT(r, TD_KV_ROW_HEAD_SIZE)
#define kvRowValues(r) POINTER_SHIFT(r, TD_KV_ROW_HEAD_SIZE + sizeof(SColIdx) * kvRowNCols(r))
#define kvRowCpy(dst, r) memcpy((dst), (r), kvRowLen(r))
#define kvRowColVal(r, colIdx) POINTER_SHIFT(kvRowValues(r), (colIdx)->offset)
#define kvRowColIdxAt(r, i) (kvRowColIdx(r) + (i))
#define kvRowFree(r) taosMemoryFreeClear(r)
#define kvRowEnd(r) POINTER_SHIFT(r, kvRowLen(r))
#define kvRowValLen(r) (kvRowLen(r) - TD_KV_ROW_HEAD_SIZE - sizeof(SColIdx) * kvRowNCols(r))
#define kvRowTKey(r) (*(TKEY *)(kvRowValues(r)))
#define kvRowKey(r) tdGetKey(kvRowTKey(r))
#define kvRowKeys(r) POINTER_SHIFT(r, *(uint16_t *)POINTER_SHIFT(r, TD_KV_ROW_HEAD_SIZE + sizeof(int16_t)))
#define kvRowDeleted(r) TKEY_IS_DELETED(kvRowTKey(r))
SKVRow tdKVRowDup(SKVRow row);
int32_t tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value);
int32_t tdEncodeKVRow(void **buf, SKVRow row);
void *tdDecodeKVRow(void *buf, SKVRow *row);
void tdSortKVRowByColIdx(SKVRow row);
static FORCE_INLINE int32_t comparTagId(const void *key1, const void *key2) {
if (*(int16_t *)key1 > ((SColIdx *)key2)->colId) {
return 1;
} else if (*(int16_t *)key1 < ((SColIdx *)key2)->colId) {
return -1;
} else {
return 0;
}
}
static FORCE_INLINE void *tdGetKVRowValOfCol(const SKVRow row, int16_t colId) {
void *ret = taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_EQ);
if (ret == NULL) return NULL;
return kvRowColVal(row, (SColIdx *)ret);
}
static FORCE_INLINE void *tdGetKVRowIdxOfCol(SKVRow row, int16_t colId) {
return taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_EQ);
}
// ----------------- K-V data row builder
typedef struct {
int16_t tCols;
int16_t nCols;
SColIdx *pColIdx;
uint16_t alloc;
uint16_t size;
void *buf;
} SKVRowBuilder;
int32_t tdInitKVRowBuilder(SKVRowBuilder *pBuilder);
void tdDestroyKVRowBuilder(SKVRowBuilder *pBuilder);
void tdResetKVRowBuilder(SKVRowBuilder *pBuilder);
SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder);
static FORCE_INLINE int32_t tdAddColToKVRow(SKVRowBuilder *pBuilder, col_id_t colId, const void *value, int32_t tlen) {
if (pBuilder->nCols >= pBuilder->tCols) {
pBuilder->tCols *= 2;
SColIdx *pColIdx = (SColIdx *)taosMemoryRealloc((void *)(pBuilder->pColIdx), sizeof(SColIdx) * pBuilder->tCols);
if (pColIdx == NULL) return -1;
pBuilder->pColIdx = pColIdx;
}
pBuilder->pColIdx[pBuilder->nCols].colId = colId;
pBuilder->pColIdx[pBuilder->nCols].offset = pBuilder->size;
pBuilder->nCols++;
if (tlen > pBuilder->alloc - pBuilder->size) {
while (tlen > pBuilder->alloc - pBuilder->size) {
pBuilder->alloc *= 2;
}
void *buf = taosMemoryRealloc(pBuilder->buf, pBuilder->alloc);
if (buf == NULL) return -1;
pBuilder->buf = buf;
}
memcpy(POINTER_SHIFT(pBuilder->buf, pBuilder->size), value, tlen);
pBuilder->size += tlen;
return 0;
}
#endif #endif
#ifdef __cplusplus #ifdef __cplusplus
@ -480,3 +408,15 @@ static FORCE_INLINE int32_t tdAddColToKVRow(SKVRowBuilder *pBuilder, col_id_t co
#endif #endif
#endif /*_TD_COMMON_DATA_FORMAT_H_*/ #endif /*_TD_COMMON_DATA_FORMAT_H_*/
// SKVRowBuilder;
// int32_t tdInitKVRowBuilder(SKVRowBuilder *pBuilder);
// void tdDestroyKVRowBuilder(SKVRowBuilder *pBuilder);
// void tdResetKVRowBuilder(SKVRowBuilder *pBuilder);
// SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder);
// static FORCE_INLINE int32_t tdAddColToKVRow(SKVRowBuilder *pBuilder, col_id_t colId, const void *value, int32_t tlen)
// #ifdef JSON_TAG_REFACTOR
// TODO: JSON_TAG_TODO

9
include/common/tmsg.h
View File

@ -1747,6 +1747,15 @@ typedef struct SVCreateTbReq {
int tEncodeSVCreateTbReq(SEncoder* pCoder, const SVCreateTbReq* pReq); int tEncodeSVCreateTbReq(SEncoder* pCoder, const SVCreateTbReq* pReq);
int tDecodeSVCreateTbReq(SDecoder* pCoder, SVCreateTbReq* pReq); int tDecodeSVCreateTbReq(SDecoder* pCoder, SVCreateTbReq* pReq);
static FORCE_INLINE void tdDestroySVCreateTbReq(SVCreateTbReq* req) {
taosMemoryFreeClear(req->name);
if (req->type == TSDB_CHILD_TABLE) {
taosMemoryFreeClear(req->ctb.pTag);
} else if (req->type == TSDB_NORMAL_TABLE) {
taosMemoryFreeClear(req->ntb.schemaRow.pSchema);
}
}
typedef struct { typedef struct {
int32_t nReqs; int32_t nReqs;
union { union {

18
source/client/src/clientImpl.c
View File

@ -949,11 +949,23 @@ static char* parseTagDatatoJson(void* p) {
goto end; goto end;
} }
int16_t nCols = kvRowNCols(p); SArray* pTagVals = NULL;
if (tTagToValArray((const STag*)p, &pTagVals) != 0) {
goto end;
}
int16_t nCols = taosArrayGetSize(pTagVals);
char tagJsonKey[256] = {0}; char tagJsonKey[256] = {0};
for (int j = 0; j < nCols; ++j) { for (int j = 0; j < nCols; ++j) {
SColIdx* pColIdx = kvRowColIdxAt(p, j); STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, j);
char* val = (char*)(kvRowColVal(p, pColIdx));
// JSON_TAG_REFACTOR
pTagVal->nData;
pTagVal->pData; // for varChar, len not included
// TODO: adapt below code;
char* val = (char*)pTagVal->pData;
// TODO: adapt below code;
if (j == 0) { if (j == 0) {
if (*val == TSDB_DATA_TYPE_NULL) { if (*val == TSDB_DATA_TYPE_NULL) {
string = taosMemoryCalloc(1, 8); string = taosMemoryCalloc(1, 8);

53
source/common/src/tdatablock.c
View File

@ -127,7 +127,7 @@ int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, con
} else if (*pData == TSDB_DATA_TYPE_BOOL) { } else if (*pData == TSDB_DATA_TYPE_BOOL) {
dataLen = CHAR_BYTES; dataLen = CHAR_BYTES;
} else if (*pData == TSDB_DATA_TYPE_JSON) { } else if (*pData == TSDB_DATA_TYPE_JSON) {
dataLen = kvRowLen(pData + CHAR_BYTES); dataLen = ((STag*)(pData + CHAR_BYTES))->len;
} else { } else {
ASSERT(0); ASSERT(0);
} }
@ -1652,18 +1652,24 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
createTbReq.type = TSDB_CHILD_TABLE; createTbReq.type = TSDB_CHILD_TABLE;
createTbReq.ctb.suid = suid; createTbReq.ctb.suid = suid;
SKVRowBuilder kvRowBuilder = {0}; STagVal tagVal = {.cid = 1,
if (tdInitKVRowBuilder(&kvRowBuilder) < 0) { .type = TSDB_DATA_TYPE_UBIGINT,
ASSERT(0); .pData = (uint8_t*)&pDataBlock->info.groupId,
.nData = sizeof(uint64_t)};
STag* pTag = NULL;
tTagNew(&tagVal, 1, 1, false, &pTag);
if (!pTag) {
tdDestroySVCreateTbReq(&createTbReq);
return NULL;
} }
tdAddColToKVRow(&kvRowBuilder, 1, &pDataBlock->info.groupId, sizeof(uint64_t)); createTbReq.ctb.pTag = (uint8_t*)pTag;
createTbReq.ctb.pTag = tdGetKVRowFromBuilder(&kvRowBuilder);
tdDestroyKVRowBuilder(&kvRowBuilder);
int32_t code; int32_t code;
tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code); tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
tdDestroySVCreateTbReq(&createTbReq);
if (code < 0) return NULL; if (code < 0) return NULL;
taosMemoryFree(cname);
} }
cap += sizeof(SSubmitBlk) + schemaLen + rows * maxLen; cap += sizeof(SSubmitBlk) + schemaLen + rows * maxLen;
@ -1706,22 +1712,37 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
createTbReq.type = TSDB_CHILD_TABLE; createTbReq.type = TSDB_CHILD_TABLE;
createTbReq.ctb.suid = suid; createTbReq.ctb.suid = suid;
SKVRowBuilder kvRowBuilder = {0}; STagVal tagVal = {.cid = 1,
if (tdInitKVRowBuilder(&kvRowBuilder) < 0) { .type = TSDB_DATA_TYPE_UBIGINT,
ASSERT(0); .pData = (uint8_t*)&pDataBlock->info.groupId,
.nData = sizeof(uint64_t)};
STag* pTag = NULL;
tTagNew(&tagVal, 1, 1, false, &pTag);
if (!pTag) {
tdDestroySVCreateTbReq(&createTbReq);
taosMemoryFreeClear(ret);
return NULL;
} }
tdAddColToKVRow(&kvRowBuilder, 1, &pDataBlock->info.groupId, sizeof(uint64_t)); createTbReq.ctb.pTag = (uint8_t*)pTag;
createTbReq.ctb.pTag = tdGetKVRowFromBuilder(&kvRowBuilder);
tdDestroyKVRowBuilder(&kvRowBuilder);
int32_t code; int32_t code;
tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code); tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
if (code < 0) return NULL; if (code < 0) {
tdDestroySVCreateTbReq(&createTbReq);
taosMemoryFreeClear(ret);
return NULL;
}
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, blockData, schemaLen); tEncoderInit(&encoder, blockData, schemaLen);
if (tEncodeSVCreateTbReq(&encoder, &createTbReq) < 0) return NULL; code = tEncodeSVCreateTbReq(&encoder, &createTbReq);
tEncoderClear(&encoder); tEncoderClear(&encoder);
tdDestroySVCreateTbReq(&createTbReq);
if (code < 0) {
taosMemoryFreeClear(ret);
return NULL;
}
} }
blkHead->schemaLen = htonl(schemaLen); blkHead->schemaLen = htonl(schemaLen);

251
source/common/src/tdataformat.c
View File

@ -19,6 +19,8 @@
#include "tdatablock.h" #include "tdatablock.h"
#include "tlog.h" #include "tlog.h"
static int32_t tGetTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson);
typedef struct SKVIdx { typedef struct SKVIdx {
int32_t cid; int32_t cid;
int32_t offset; int32_t offset;
@ -31,16 +33,6 @@ typedef struct {
} STSKVRow; } STSKVRow;
#pragma pack(pop) #pragma pack(pop)
#pragma pack(push, 1)
struct STag {
int8_t isJson;
int16_t len;
int16_t nTag;
int32_t ver;
int16_t idx[];
};
#pragma pack(pop)
#define TSROW_IS_KV_ROW(r) ((r)->flags & TSROW_KV_ROW) #define TSROW_IS_KV_ROW(r) ((r)->flags & TSROW_KV_ROW)
#define BIT1_SIZE(n) (((n)-1) / 8 + 1) #define BIT1_SIZE(n) (((n)-1) / 8 + 1)
#define BIT2_SIZE(n) (((n)-1) / 4 + 1) #define BIT2_SIZE(n) (((n)-1) / 4 + 1)
@ -531,6 +523,77 @@ static int tTagValCmprFn(const void *p1, const void *p2) {
static int tTagValJsonCmprFn(const void *p1, const void *p2) { static int tTagValJsonCmprFn(const void *p1, const void *p2) {
return strcmp(((STagVal *)p1)[0].pKey, ((STagVal *)p2)[0].pKey); return strcmp(((STagVal *)p1)[0].pKey, ((STagVal *)p2)[0].pKey);
} }
static void debugPrintTagVal(int8_t type, const void *val, int32_t vlen, const char *tag, int32_t ln) {
switch (type) {
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_NCHAR: {
char tmpVal[32] = {0};
memcpy(tmpVal, val, 32);
printf("%s:%d type:%d vlen:%d, val:\"%s\"\n", tag, ln, (int32_t)type, vlen, tmpVal);
} break;
case TSDB_DATA_TYPE_FLOAT:
printf("%s:%d type:%d vlen:%d, val:%f\n", tag, ln, (int32_t)type, vlen, *(float *)val);
break;
case TSDB_DATA_TYPE_DOUBLE:
printf("%s:%d type:%d vlen:%d, val:%lf\n", tag, ln, (int32_t)type, vlen, *(double *)val);
break;
case TSDB_DATA_TYPE_BOOL:
printf("%s:%d type:%d vlen:%d, val:%" PRIu8 "\n", tag, ln, (int32_t)type, vlen, *(uint8_t *)val);
break;
case TSDB_DATA_TYPE_TINYINT:
printf("%s:%d type:%d vlen:%d, val:%" PRIi8 "\n", tag, ln, (int32_t)type, vlen, *(int8_t *)val);
break;
case TSDB_DATA_TYPE_SMALLINT:
printf("%s:%d type:%d vlen:%d, val:%" PRIi16 "\n", tag, ln, (int32_t)type, vlen, *(int16_t *)val);
break;
case TSDB_DATA_TYPE_INT:
printf("%s:%d type:%d vlen:%d, val:%" PRIi32 "\n", tag, ln, (int32_t)type, vlen, *(int32_t *)val);
break;
case TSDB_DATA_TYPE_BIGINT:
printf("%s:%d type:%d vlen:%d, val:%" PRIi64 "\n", tag, ln, (int32_t)type, vlen, *(int64_t *)val);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
printf("%s:%d type:%d vlen:%d, val:%" PRIi64 "\n", tag, ln, (int32_t)type, vlen, *(int64_t *)val);
break;
case TSDB_DATA_TYPE_UTINYINT:
printf("%s:%d type:%d vlen:%d, val:%" PRIu8 "\n", tag, ln, (int32_t)type, vlen, *(uint8_t *)val);
break;
case TSDB_DATA_TYPE_USMALLINT:
printf("%s:%d type:%d vlen:%d, val:%" PRIu16 "\n", tag, ln, (int32_t)type, vlen, *(uint16_t *)val);
break;
case TSDB_DATA_TYPE_UINT:
printf("%s:%d type:%d vlen:%d, val:%" PRIu32 "\n", tag, ln, (int32_t)type, vlen, *(uint32_t *)val);
break;
case TSDB_DATA_TYPE_UBIGINT:
printf("%s:%d type:%d vlen:%d, val:%" PRIu64 "\n", tag, ln, (int32_t)type, vlen, *(uint64_t *)val);
break;
default:
ASSERT(0);
break;
}
}
void debugPrintSTag(STag *pTag, const char *tag, int32_t ln) {
printf("%s:%d >>> STAG === isJson:%s, len: %d, nTag: %d, sver:%d\n", tag, ln, pTag->isJson ? "true" : "false",
(int32_t)pTag->len, (int32_t)pTag->nTag, pTag->ver);
char *p = (char *)&pTag->idx[pTag->nTag];
for (uint16_t n = 0; n < pTag->nTag; ++n) {
int16_t *pIdx = pTag->idx + n;
STagVal tagVal = {0};
if (pTag->isJson) {
tagVal.pKey = (char *)POINTER_SHIFT(p, *pIdx);
} else {
tagVal.cid = *(int16_t *)POINTER_SHIFT(p, *pIdx);
}
printf("%s:%d loop[%d-%d] offset=%d\n", __func__, __LINE__, (int32_t)pTag->nTag, (int32_t)n, *pIdx);
tGetTagVal(p, &tagVal, pTag->isJson);
debugPrintTagVal(tagVal.type, tagVal.pData, tagVal.nData, __func__, __LINE__);
}
printf("\n");
}
static int32_t tPutTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson) { static int32_t tPutTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson) {
int32_t n = 0; int32_t n = 0;
@ -544,6 +607,7 @@ static int32_t tPutTagVal(uint8_t *p, STagVal *pTagVal, int8_t isJson) {
// type // type
n += tPutI8(p ? p + n : p, pTagVal->type); n += tPutI8(p ? p + n : p, pTagVal->type);
debugPrintTagVal(pTagVal->type, pTagVal->pData, pTagVal->nData, __func__, __LINE__);
// value // value
if (IS_VAR_DATA_TYPE(pTagVal->type)) { if (IS_VAR_DATA_TYPE(pTagVal->type)) {
n += tPutBinary(p ? p + n : p, pTagVal->pData, pTagVal->nData); n += tPutBinary(p ? p + n : p, pTagVal->pData, pTagVal->nData);
@ -621,6 +685,8 @@ int32_t tTagNew(STagVal *pTagVals, int16_t nTag, int32_t version, int8_t isJson,
n += tPutTagVal(p + n, &pTagVals[iTag], isJson); n += tPutTagVal(p + n, &pTagVals[iTag], isJson);
} }
debugPrintSTag(*ppTag, __func__, __LINE__);
return code; return code;
_err: _err:
@ -631,7 +697,7 @@ void tTagFree(STag *pTag) {
if (pTag) taosMemoryFree(pTag); if (pTag) taosMemoryFree(pTag);
} }
void tTagGet(STag *pTag, STagVal *pTagVal) { bool tTagGet(const STag *pTag, STagVal *pTagVal) {
int16_t lidx = 0; int16_t lidx = 0;
int16_t ridx = pTag->nTag - 1; int16_t ridx = pTag->nTag - 1;
int16_t midx; int16_t midx;
@ -660,9 +726,10 @@ void tTagGet(STag *pTag, STagVal *pTagVal) {
pTagVal->type = tv.type; pTagVal->type = tv.type;
pTagVal->nData = tv.nData; pTagVal->nData = tv.nData;
pTagVal->pData = tv.pData; pTagVal->pData = tv.pData;
break; return true;
} }
} }
return false;
} }
int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag) { int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag) {
@ -671,7 +738,7 @@ int32_t tEncodeTag(SEncoder *pEncoder, const STag *pTag) {
int32_t tDecodeTag(SDecoder *pDecoder, STag **ppTag) { return tDecodeBinary(pDecoder, (uint8_t **)ppTag, NULL); } int32_t tDecodeTag(SDecoder *pDecoder, STag **ppTag) { return tDecodeBinary(pDecoder, (uint8_t **)ppTag, NULL); }
int32_t tTagToValArray(STag *pTag, SArray **ppArray) { int32_t tTagToValArray(const STag *pTag, SArray **ppArray) {
int32_t code = 0; int32_t code = 0;
uint8_t *p = (uint8_t *)&pTag->idx[pTag->nTag]; uint8_t *p = (uint8_t *)&pTag->idx[pTag->nTag];
STagVal tv; STagVal tv;
@ -1019,162 +1086,4 @@ void tdResetDataCols(SDataCols *pCols) {
} }
} }
SKVRow tdKVRowDup(SKVRow row) {
SKVRow trow = taosMemoryMalloc(kvRowLen(row));
if (trow == NULL) return NULL;
kvRowCpy(trow, row);
return trow;
}
static int compareColIdx(const void *a, const void *b) {
const SColIdx *x = (const SColIdx *)a;
const SColIdx *y = (const SColIdx *)b;
if (x->colId > y->colId) {
return 1;
}
if (x->colId < y->colId) {
return -1;
}
return 0;
}
void tdSortKVRowByColIdx(SKVRow row) { qsort(kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), compareColIdx); }
int tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value) {
SColIdx *pColIdx = NULL;
SKVRow row = *orow;
SKVRow nrow = NULL;
void *ptr = taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_GE);
if (ptr == NULL || ((SColIdx *)ptr)->colId > colId) { // need to add a column value to the row
int diff = IS_VAR_DATA_TYPE(type) ? varDataTLen(value) : TYPE_BYTES[type];
int nRowLen = kvRowLen(row) + sizeof(SColIdx) + diff;
int oRowCols = kvRowNCols(row);
ASSERT(diff > 0);
nrow = taosMemoryMalloc(nRowLen);
if (nrow == NULL) return -1;
kvRowSetLen(nrow, nRowLen);
kvRowSetNCols(nrow, oRowCols + 1);
memcpy(kvRowColIdx(nrow), kvRowColIdx(row), sizeof(SColIdx) * oRowCols);
memcpy(kvRowValues(nrow), kvRowValues(row), kvRowValLen(row));
pColIdx = kvRowColIdxAt(nrow, oRowCols);
pColIdx->colId = colId;
pColIdx->offset = kvRowValLen(row);
memcpy(kvRowColVal(nrow, pColIdx), value, diff); // copy new value
tdSortKVRowByColIdx(nrow);
*orow = nrow;
taosMemoryFree(row);
} else {
ASSERT(((SColIdx *)ptr)->colId == colId);
if (IS_VAR_DATA_TYPE(type)) {
void *pOldVal = kvRowColVal(row, (SColIdx *)ptr);
if (varDataTLen(value) == varDataTLen(pOldVal)) { // just update the column value in place
memcpy(pOldVal, value, varDataTLen(value));
} else { // need to reallocate the memory
int16_t nlen = kvRowLen(row) + (varDataTLen(value) - varDataTLen(pOldVal));
ASSERT(nlen > 0);
nrow = taosMemoryMalloc(nlen);
if (nrow == NULL) return -1;
kvRowSetLen(nrow, nlen);
kvRowSetNCols(nrow, kvRowNCols(row));
int zsize = sizeof(SColIdx) * kvRowNCols(row) + ((SColIdx *)ptr)->offset;
memcpy(kvRowColIdx(nrow), kvRowColIdx(row), zsize);
memcpy(kvRowColVal(nrow, ((SColIdx *)ptr)), value, varDataTLen(value));
// Copy left value part
int lsize = kvRowLen(row) - TD_KV_ROW_HEAD_SIZE - zsize - varDataTLen(pOldVal);
if (lsize > 0) {
memcpy(POINTER_SHIFT(nrow, TD_KV_ROW_HEAD_SIZE + zsize + varDataTLen(value)),
POINTER_SHIFT(row, TD_KV_ROW_HEAD_SIZE + zsize + varDataTLen(pOldVal)), lsize);
}
for (int i = 0; i < kvRowNCols(nrow); i++) {
pColIdx = kvRowColIdxAt(nrow, i);
if (pColIdx->offset > ((SColIdx *)ptr)->offset) {
pColIdx->offset = pColIdx->offset - varDataTLen(pOldVal) + varDataTLen(value);
}
}
*orow = nrow;
taosMemoryFree(row);
}
} else {
memcpy(kvRowColVal(row, (SColIdx *)ptr), value, TYPE_BYTES[type]);
}
}
return 0;
}
int tdEncodeKVRow(void **buf, SKVRow row) {
// May change the encode purpose
if (buf != NULL) {
kvRowCpy(*buf, row);
*buf = POINTER_SHIFT(*buf, kvRowLen(row));
}
return kvRowLen(row);
}
void *tdDecodeKVRow(void *buf, SKVRow *row) {
*row = tdKVRowDup(buf);
if (*row == NULL) return NULL;
return POINTER_SHIFT(buf, kvRowLen(*row));
}
int tdInitKVRowBuilder(SKVRowBuilder *pBuilder) {
pBuilder->tCols = 128;
pBuilder->nCols = 0;
pBuilder->pColIdx = (SColIdx *)taosMemoryMalloc(sizeof(SColIdx) * pBuilder->tCols);
if (pBuilder->pColIdx == NULL) return -1;
pBuilder->alloc = 1024;
pBuilder->size = 0;
pBuilder->buf = taosMemoryMalloc(pBuilder->alloc);
if (pBuilder->buf == NULL) {
taosMemoryFree(pBuilder->pColIdx);
return -1;
}
return 0;
}
void tdDestroyKVRowBuilder(SKVRowBuilder *pBuilder) {
taosMemoryFreeClear(pBuilder->pColIdx);
taosMemoryFreeClear(pBuilder->buf);
}
void tdResetKVRowBuilder(SKVRowBuilder *pBuilder) {
pBuilder->nCols = 0;
pBuilder->size = 0;
}
SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder) {
int tlen = sizeof(SColIdx) * pBuilder->nCols + pBuilder->size;
// if (tlen == 0) return NULL; // nCols == 0 means no tags
tlen += TD_KV_ROW_HEAD_SIZE;
SKVRow row = taosMemoryMalloc(tlen);
if (row == NULL) return NULL;
kvRowSetNCols(row, pBuilder->nCols);
kvRowSetLen(row, tlen);
if (pBuilder->nCols > 0) {
memcpy(kvRowColIdx(row), pBuilder->pColIdx, sizeof(SColIdx) * pBuilder->nCols);
memcpy(kvRowValues(row), pBuilder->buf, pBuilder->size);
}
return row;
}
#endif #endif

6
source/common/src/tmsg.c
View File

@ -3846,7 +3846,7 @@ int tEncodeSVCreateTbReq(SEncoder *pCoder, const SVCreateTbReq *pReq) {
if (pReq->type == TSDB_CHILD_TABLE) { if (pReq->type == TSDB_CHILD_TABLE) {
if (tEncodeI64(pCoder, pReq->ctb.suid) < 0) return -1; if (tEncodeI64(pCoder, pReq->ctb.suid) < 0) return -1;
if (tEncodeBinary(pCoder, pReq->ctb.pTag, kvRowLen(pReq->ctb.pTag)) < 0) return -1; if (tEncodeTag(pCoder, (const STag *)pReq->ctb.pTag) < 0) return -1;
} else if (pReq->type == TSDB_NORMAL_TABLE) { } else if (pReq->type == TSDB_NORMAL_TABLE) {
if (tEncodeSSchemaWrapper(pCoder, &pReq->ntb.schemaRow) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pReq->ntb.schemaRow) < 0) return -1;
} else { } else {
@ -3858,8 +3858,6 @@ int tEncodeSVCreateTbReq(SEncoder *pCoder, const SVCreateTbReq *pReq) {
} }
int tDecodeSVCreateTbReq(SDecoder *pCoder, SVCreateTbReq *pReq) { int tDecodeSVCreateTbReq(SDecoder *pCoder, SVCreateTbReq *pReq) {
uint32_t len;
if (tStartDecode(pCoder) < 0) return -1; if (tStartDecode(pCoder) < 0) return -1;
if (tDecodeI32v(pCoder, &pReq->flags) < 0) return -1; if (tDecodeI32v(pCoder, &pReq->flags) < 0) return -1;
@ -3871,7 +3869,7 @@ int tDecodeSVCreateTbReq(SDecoder *pCoder, SVCreateTbReq *pReq) {
if (pReq->type == TSDB_CHILD_TABLE) { if (pReq->type == TSDB_CHILD_TABLE) {
if (tDecodeI64(pCoder, &pReq->ctb.suid) < 0) return -1; if (tDecodeI64(pCoder, &pReq->ctb.suid) < 0) return -1;
if (tDecodeBinary(pCoder, &pReq->ctb.pTag, &len) < 0) return -1; if (tDecodeTag(pCoder, (STag **)&pReq->ctb.pTag) < 0) return -1;
} else if (pReq->type == TSDB_NORMAL_TABLE) { } else if (pReq->type == TSDB_NORMAL_TABLE) {
if (tDecodeSSchemaWrapper(pCoder, &pReq->ntb.schemaRow) < 0) return -1; if (tDecodeSSchemaWrapper(pCoder, &pReq->ntb.schemaRow) < 0) return -1;
} else { } else {

4
source/common/src/trow.c
View File

@ -1191,9 +1191,9 @@ bool tdGetTpRowDataOfCol(STSRowIter *pIter, col_type_t colType, int32_t offset,
} }
static FORCE_INLINE int32_t compareKvRowColId(const void *key1, const void *key2) { static FORCE_INLINE int32_t compareKvRowColId(const void *key1, const void *key2) {
if (*(int16_t *)key1 > ((SColIdx *)key2)->colId) { if (*(col_id_t *)key1 > ((SKvRowIdx *)key2)->colId) {
return 1; return 1;
} else if (*(int16_t *)key1 < ((SColIdx *)key2)->colId) { } else if (*(col_id_t *)key1 < ((SKvRowIdx *)key2)->colId) {
return -1; return -1;
} else { } else {
return 0; return 0;

5
source/dnode/vnode/src/meta/metaEntry.c
View File

@ -30,7 +30,7 @@ int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) {
if (tEncodeI64(pCoder, pME->ctbEntry.ctime) < 0) return -1; if (tEncodeI64(pCoder, pME->ctbEntry.ctime) < 0) return -1;
if (tEncodeI32(pCoder, pME->ctbEntry.ttlDays) < 0) return -1; if (tEncodeI32(pCoder, pME->ctbEntry.ttlDays) < 0) return -1;
if (tEncodeI64(pCoder, pME->ctbEntry.suid) < 0) return -1; if (tEncodeI64(pCoder, pME->ctbEntry.suid) < 0) return -1;
if (tEncodeBinary(pCoder, pME->ctbEntry.pTags, kvRowLen(pME->ctbEntry.pTags)) < 0) return -1; if (tEncodeTag(pCoder, (const STag *)pME->ctbEntry.pTags) < 0) return -1;
} else if (pME->type == TSDB_NORMAL_TABLE) { } else if (pME->type == TSDB_NORMAL_TABLE) {
if (tEncodeI64(pCoder, pME->ntbEntry.ctime) < 0) return -1; if (tEncodeI64(pCoder, pME->ntbEntry.ctime) < 0) return -1;
if (tEncodeI32(pCoder, pME->ntbEntry.ttlDays) < 0) return -1; if (tEncodeI32(pCoder, pME->ntbEntry.ttlDays) < 0) return -1;
@ -47,7 +47,6 @@ int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) {
} }
int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) { int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
uint32_t len;
if (tStartDecode(pCoder) < 0) return -1; if (tStartDecode(pCoder) < 0) return -1;
if (tDecodeI64(pCoder, &pME->version) < 0) return -1; if (tDecodeI64(pCoder, &pME->version) < 0) return -1;
@ -62,7 +61,7 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
if (tDecodeI64(pCoder, &pME->ctbEntry.ctime) < 0) return -1; if (tDecodeI64(pCoder, &pME->ctbEntry.ctime) < 0) return -1;
if (tDecodeI32(pCoder, &pME->ctbEntry.ttlDays) < 0) return -1; if (tDecodeI32(pCoder, &pME->ctbEntry.ttlDays) < 0) return -1;
if (tDecodeI64(pCoder, &pME->ctbEntry.suid) < 0) return -1; if (tDecodeI64(pCoder, &pME->ctbEntry.suid) < 0) return -1;
if (tDecodeBinary(pCoder, &pME->ctbEntry.pTags, &len) < 0) return -1; // (TODO) if (tDecodeTag(pCoder, (STag **)&pME->ctbEntry.pTags) < 0) return -1; // (TODO)
} else if (pME->type == TSDB_NORMAL_TABLE) { } else if (pME->type == TSDB_NORMAL_TABLE) {
if (tDecodeI64(pCoder, &pME->ntbEntry.ctime) < 0) return -1; if (tDecodeI64(pCoder, &pME->ntbEntry.ctime) < 0) return -1;
if (tDecodeI32(pCoder, &pME->ntbEntry.ttlDays) < 0) return -1; if (tDecodeI32(pCoder, &pME->ntbEntry.ttlDays) < 0) return -1;

4
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -566,5 +566,7 @@ SArray *metaGetSmaTbUids(SMeta *pMeta) {
const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t cid) { const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t cid) {
ASSERT(pEntry->type == TSDB_CHILD_TABLE); ASSERT(pEntry->type == TSDB_CHILD_TABLE);
return tdGetKVRowValOfCol((const SKVRow)pEntry->ctbEntry.pTags, cid); STagVal tagVal = {.cid = cid};
tTagGet((const STag *)pEntry->ctbEntry.pTags, &tagVal);
return tagVal.pData;
} }

39
source/dnode/vnode/src/meta/metaTable.c
View File

@ -563,29 +563,39 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
} }
memcpy((void *)ctbEntry.ctbEntry.pTags, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal); memcpy((void *)ctbEntry.ctbEntry.pTags, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal);
} else { } else {
SKVRowBuilder kvrb = {0}; const STag *pOldTag = (const STag *)ctbEntry.ctbEntry.pTags;
const SKVRow pOldTag = (const SKVRow)ctbEntry.ctbEntry.pTags; STag *pNewTag = NULL;
SKVRow pNewTag = NULL; STagVal *pTagVals = taosMemoryCalloc(pTagSchema->nCols, sizeof(STagVal));
tdInitKVRowBuilder(&kvrb); if (!pTagVals) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
int16_t nTags = 0;
for (int32_t i = 0; i < pTagSchema->nCols; i++) { for (int32_t i = 0; i < pTagSchema->nCols; i++) {
SSchema *pCol = &pTagSchema->pSchema[i]; SSchema *pCol = &pTagSchema->pSchema[i];
STagVal *pTagVal = pTagVals + nTags;
if (iCol == i) { if (iCol == i) {
tdAddColToKVRow(&kvrb, pCol->colId, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal); tTagValSet(pTagVal, &pCol->colId, pCol->type, pAlterTbReq->pTagVal, pAlterTbReq->nTagVal, false);
++nTags;
} else { } else {
void *p = tdGetKVRowValOfCol(pOldTag, pCol->colId); STagVal tagVal = {.cid = pCol->colId};
if (p) { if (tTagGet(pOldTag, &tagVal) && tagVal.pData) {
if (IS_VAR_DATA_TYPE(pCol->type)) { if (IS_VAR_DATA_TYPE(pCol->type)) {
tdAddColToKVRow(&kvrb, pCol->colId, p, varDataTLen(p)); tTagValSet(pTagVal, &pCol->colId, pCol->type, tagVal.pData, varDataTLen(tagVal.pData), false);
} else { } else {
tdAddColToKVRow(&kvrb, pCol->colId, p, pCol->bytes); tTagValSet(pTagVal, &pCol->colId, pCol->type, tagVal.pData, pCol->bytes, false);
} }
++nTags;
} }
} }
} }
if ((terrno = tTagNew(pTagVals, nTags, pTagSchema->version, false, &pNewTag)) < 0) {
ctbEntry.ctbEntry.pTags = tdGetKVRowFromBuilder(&kvrb); taosMemoryFreeClear(pTagVals);
tdDestroyKVRowBuilder(&kvrb); goto _err;
}
ctbEntry.ctbEntry.pTags = (uint8_t *)pNewTag;
taosMemoryFreeClear(pTagVals);
} }
// save to table.db // save to table.db
@ -775,7 +785,10 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
metaDecodeEntry(&dc, &stbEntry); metaDecodeEntry(&dc, &stbEntry);
pTagColumn = &stbEntry.stbEntry.schemaTag.pSchema[0]; pTagColumn = &stbEntry.stbEntry.schemaTag.pSchema[0];
pTagData = tdGetKVRowValOfCol((const SKVRow)pCtbEntry->ctbEntry.pTags, pTagColumn->colId);
STagVal tagVal = {.cid = pTagColumn->colId};
tTagGet((const STag *)pCtbEntry->ctbEntry.pTags, &tagVal);
pTagData = tagVal.pData;
// update tag index // update tag index
#ifdef USE_INVERTED_INDEX #ifdef USE_INVERTED_INDEX

20
source/libs/executor/src/scanoperator.c
View File

@ -305,17 +305,17 @@ void addTagPseudoColumnData(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock)
} else { // these are tags } else { // these are tags
const char* p = NULL; const char* p = NULL;
if (pColInfoData->info.type == TSDB_DATA_TYPE_JSON) { if (pColInfoData->info.type == TSDB_DATA_TYPE_JSON) {
const uint8_t* tmp = mr.me.ctbEntry.pTags; const STag* tmp = (const STag*)mr.me.ctbEntry.pTags;
char* data = taosMemoryCalloc(kvRowLen(tmp) + 1, 1); char* data = taosMemoryCalloc(tmp->len + 1, 1);
if (data == NULL) { if (data == NULL) {
metaReaderClear(&mr); metaReaderClear(&mr);
qError("doTagScan calloc error:%d", kvRowLen(tmp) + 1); qError("doTagScan calloc error:%d", tmp->len + 1);
return; return;
} }
*data = TSDB_DATA_TYPE_JSON; *data = TSDB_DATA_TYPE_JSON;
memcpy(data + 1, tmp, kvRowLen(tmp)); memcpy(data + 1, tmp, tmp->len);
p = data; p = data;
} else { } else {
p = metaGetTableTagVal(&mr.me, pExpr->base.pParam[0].pCol->colId); p = metaGetTableTagVal(&mr.me, pExpr->base.pParam[0].pCol->colId);
@ -1633,16 +1633,16 @@ static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
colDataAppend(pDst, count, str, false); colDataAppend(pDst, count, str, false);
} else { // it is a tag value } else { // it is a tag value
if (pDst->info.type == TSDB_DATA_TYPE_JSON) { if (pDst->info.type == TSDB_DATA_TYPE_JSON) {
const uint8_t* tmp = mr.me.ctbEntry.pTags; const STag* tmp = (const STag*)mr.me.ctbEntry.pTags;
// TODO opt perf by realloc memory // TODO opt perf by realloc memory
char* data = taosMemoryCalloc(kvRowLen(tmp) + 1, 1); char* data = taosMemoryCalloc(tmp->len + 1, 1);
if (data == NULL) { if (data == NULL) {
qError("%s failed to malloc memory, size:%d", GET_TASKID(pTaskInfo), kvRowLen(tmp) + 1); qError("%s failed to malloc memory, size:%d", GET_TASKID(pTaskInfo), tmp->len + 1);
longjmp(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY); longjmp(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
} }
*data = TSDB_DATA_TYPE_JSON; *data = TSDB_DATA_TYPE_JSON;
memcpy(data + 1, tmp, kvRowLen(tmp)); memcpy(data + 1, tmp, tmp->len);
colDataAppend(pDst, count, data, false); colDataAppend(pDst, count, data, false);
taosMemoryFree(data); taosMemoryFree(data);
} else { } else {
@ -1677,8 +1677,8 @@ static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
} }
SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, SExprInfo* pExpr, int32_t numOfOutput, SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, SExprInfo* pExpr, int32_t numOfOutput,
SSDataBlock* pResBlock, SArray* pColMatchInfo, SSDataBlock* pResBlock, SArray* pColMatchInfo, STableListInfo* pTableListInfo,
STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) { SExecTaskInfo* pTaskInfo) {
STagScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo)); STagScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {

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

@ -55,7 +55,9 @@ int32_t getNumOfColumns(const STableMeta* pTableMeta);
int32_t getNumOfTags(const STableMeta* pTableMeta); int32_t getNumOfTags(const STableMeta* pTableMeta);
STableComInfo getTableInfo(const STableMeta* pTableMeta); STableComInfo getTableInfo(const STableMeta* pTableMeta);
STableMeta* tableMetaDup(const STableMeta* pTableMeta); STableMeta* tableMetaDup(const STableMeta* pTableMeta);
#ifdef JSON_TAG_REFACTOR
int32_t parseJsontoTagData(const char* json, SKVRowBuilder* kvRowBuilder, SMsgBuf* errMsg, int16_t startColId); int32_t parseJsontoTagData(const char* json, SKVRowBuilder* kvRowBuilder, SMsgBuf* errMsg, int16_t startColId);
#endif
int32_t trimString(const char* src, int32_t len, char* dst, int32_t dlen); int32_t trimString(const char* src, int32_t len, char* dst, int32_t dlen);

129
source/libs/parser/src/parInsert.c
View File

@ -54,7 +54,7 @@ typedef struct SInsertParseContext {
SMsgBuf msg; // input SMsgBuf msg; // input
STableMeta* pTableMeta; // each table STableMeta* pTableMeta; // each table
SParsedDataColInfo tags; // each table SParsedDataColInfo tags; // each table
SKVRowBuilder tagsBuilder; // each table STagVal* pTagVals; // each table
SVCreateTbReq createTblReq; // each table SVCreateTbReq createTblReq; // each table
SHashObj* pVgroupsHashObj; // global SHashObj* pVgroupsHashObj; // global
SHashObj* pTableBlockHashObj; // global SHashObj* pTableBlockHashObj; // global
@ -72,9 +72,11 @@ static uint8_t TRUE_VALUE = (uint8_t)TSDB_TRUE;
static uint8_t FALSE_VALUE = (uint8_t)TSDB_FALSE; static uint8_t FALSE_VALUE = (uint8_t)TSDB_FALSE;
typedef struct SKvParam { typedef struct SKvParam {
SKVRowBuilder* builder; int16_t nTag;
SSchema* schema; int16_t pos;
char buf[TSDB_MAX_TAGS_LEN]; STagVal* pTagVals;
SSchema* schema;
char buf[TSDB_MAX_TAGS_LEN];
} SKvParam; } SKvParam;
typedef struct SMemParam { typedef struct SMemParam {
@ -212,7 +214,7 @@ static int32_t createSName(SName* pName, SToken* pTableName, int32_t acctId, con
return buildInvalidOperationMsg(pMsgBuf, msg4); return buildInvalidOperationMsg(pMsgBuf, msg4);
} }
char tbname[TSDB_TABLE_FNAME_LEN] = {0}; char tbname[TSDB_TABLE_FNAME_LEN] = {0};
strncpy(tbname, p + 1, tbLen); strncpy(tbname, p + 1, tbLen);
/*tbLen = */ strdequote(tbname); /*tbLen = */ strdequote(tbname);
@ -619,14 +621,14 @@ static int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int
case TSDB_DATA_TYPE_NCHAR: { case TSDB_DATA_TYPE_NCHAR: {
return func(pMsgBuf, pToken->z, pToken->n, param); return func(pMsgBuf, pToken->z, pToken->n, param);
} }
#ifdef JSON_TAG_REFACTOR
case TSDB_DATA_TYPE_JSON: { case TSDB_DATA_TYPE_JSON: {
if (pToken->n > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) { if (pToken->n > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
return buildSyntaxErrMsg(pMsgBuf, "json string too long than 4095", pToken->z); return buildSyntaxErrMsg(pMsgBuf, "json string too long than 4095", pToken->z);
} }
return func(pMsgBuf, pToken->z, pToken->n, param); return func(pMsgBuf, pToken->z, pToken->n, param);
} }
#endif
case TSDB_DATA_TYPE_TIMESTAMP: { case TSDB_DATA_TYPE_TIMESTAMP: {
int64_t tmpVal; int64_t tmpVal;
if (parseTime(end, pToken, timePrec, &tmpVal, pMsgBuf) != TSDB_CODE_SUCCESS) { if (parseTime(end, pToken, timePrec, &tmpVal, pMsgBuf) != TSDB_CODE_SUCCESS) {
@ -757,9 +759,11 @@ static int32_t KvRowAppend(SMsgBuf* pMsgBuf, const void* value, int32_t len, voi
int8_t type = pa->schema->type; int8_t type = pa->schema->type;
int16_t colId = pa->schema->colId; int16_t colId = pa->schema->colId;
#ifdef JSON_TAG_REFACTOR
if (TSDB_DATA_TYPE_JSON == type) { if (TSDB_DATA_TYPE_JSON == type) {
return parseJsontoTagData(value, pa->builder, pMsgBuf, colId); return parseJsontoTagData(value, pa->builder, pMsgBuf, colId);
} }
#endif
if (value == NULL) { // it is a null data if (value == NULL) { // it is a null data
// tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NULL, value, false, pa->toffset, // tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NULL, value, false, pa->toffset,
@ -768,55 +772,63 @@ static int32_t KvRowAppend(SMsgBuf* pMsgBuf, const void* value, int32_t len, voi
} }
if (TSDB_DATA_TYPE_BINARY == type) { if (TSDB_DATA_TYPE_BINARY == type) {
STR_WITH_SIZE_TO_VARSTR(pa->buf, value, len); memcpy(pa->buf + pa->pos, value, len);
tdAddColToKVRow(pa->builder, colId, pa->buf, varDataTLen(pa->buf)); tTagValSet(pa->pTagVals + pa->nTag++, &colId, type, (uint8_t*)(pa->buf + pa->pos), len, false);
pa->pos += len;
} else if (TSDB_DATA_TYPE_NCHAR == type) { } else if (TSDB_DATA_TYPE_NCHAR == type) {
// if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long' // if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
ASSERT((pa->pos + pa->schema->bytes - VARSTR_HEADER_SIZE) <= TSDB_MAX_TAGS_LEN);
int32_t output = 0; int32_t output = 0;
if (!taosMbsToUcs4(value, len, (TdUcs4*)varDataVal(pa->buf), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) { if (!taosMbsToUcs4(value, len, (TdUcs4*)(pa->buf + pa->pos), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
if (errno == E2BIG) { if (errno == E2BIG) {
return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pa->schema->name); return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pa->schema->name);
} }
char buf[512] = {0}; char buf[512] = {0};
snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno)); snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
return buildSyntaxErrMsg(pMsgBuf, buf, value); return buildSyntaxErrMsg(pMsgBuf, buf, value);
} }
tTagValSet(pa->pTagVals + pa->nTag++, &colId, type, (uint8_t*)(pa->buf + pa->pos), output, false);
varDataSetLen(pa->buf, output); pa->pos += output;
tdAddColToKVRow(pa->builder, colId, pa->buf, varDataTLen(pa->buf));
} else { } else {
tdAddColToKVRow(pa->builder, colId, value, TYPE_BYTES[type]); memcpy(pa->buf + pa->pos, value, TYPE_BYTES[type]);
tTagValSet(pa->pTagVals + pa->nTag++, &colId, type, (uint8_t*)(pa->buf + pa->pos), TYPE_BYTES[type], false);
pa->pos + TYPE_BYTES[type];
} }
ASSERT(pa->pos <= TSDB_MAX_TAGS_LEN);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, SKVRow row, int64_t suid) { static int32_t buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTag, int64_t suid) {
pTbReq->type = TD_CHILD_TABLE; pTbReq->type = TD_CHILD_TABLE;
pTbReq->name = strdup(tname); pTbReq->name = strdup(tname);
pTbReq->ctb.suid = suid; pTbReq->ctb.suid = suid;
pTbReq->ctb.pTag = row; pTbReq->ctb.pTag = (uint8_t*)pTag;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
// pSql -> tag1_value, ...) // pSql -> tag1_value, ...)
static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const char* tName) { static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const char* tName) {
if (tdInitKVRowBuilder(&pCxt->tagsBuilder) < 0) { ASSERT(!pCxt->pTagVals);
if (!(pCxt->pTagVals = taosMemoryCalloc(pCxt->tags.numOfBound, sizeof(STagVal)))) {
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
SKvParam param = {.builder = &pCxt->tagsBuilder}; SKvParam param = {.pTagVals = pCxt->pTagVals, .nTag = 0, .pos = 0};
SToken sToken; SToken sToken;
bool isParseBindParam = false; bool isParseBindParam = false;
char tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0}; // used for deleting Escape character: \\, \', \" char tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0}; // used for deleting Escape character: \\, \', \"
// TODO: JSON_TAG_REFACTOR => here would have json tag?
for (int i = 0; i < pCxt->tags.numOfBound; ++i) { for (int i = 0; i < pCxt->tags.numOfBound; ++i) {
NEXT_TOKEN_WITH_PREV(pCxt->pSql, sToken); NEXT_TOKEN_WITH_PREV(pCxt->pSql, sToken);
if (sToken.type == TK_NK_QUESTION) { if (sToken.type == TK_NK_QUESTION) {
isParseBindParam = true; isParseBindParam = true;
if (NULL == pCxt->pStmtCb) { if (NULL == pCxt->pStmtCb) {
taosMemoryFreeClear(pCxt->pTagVals);
return buildSyntaxErrMsg(&pCxt->msg, "? only used in stmt", sToken.z); return buildSyntaxErrMsg(&pCxt->msg, "? only used in stmt", sToken.z);
} }
@ -824,6 +836,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
} }
if (isParseBindParam) { if (isParseBindParam) {
taosMemoryFreeClear(pCxt->pTagVals);
return buildInvalidOperationMsg(&pCxt->msg, "no mix usage for ? and tag values"); return buildInvalidOperationMsg(&pCxt->msg, "no mix usage for ? and tag values");
} }
@ -834,16 +847,19 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
} }
if (isParseBindParam) { if (isParseBindParam) {
taosMemoryFreeClear(pCxt->pTagVals);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
SKVRow row = tdGetKVRowFromBuilder(&pCxt->tagsBuilder); // TODO: JSON_TAG_REFACTOR (would be JSON tag or normal tag)
if (NULL == row) { STag* pTag = NULL;
if (tTagNew(param.pTagVals, param.nTag, 1, false, &pTag) != 0) {
taosMemoryFreeClear(pCxt->pTagVals);
return buildInvalidOperationMsg(&pCxt->msg, "out of memory"); return buildInvalidOperationMsg(&pCxt->msg, "out of memory");
} }
tdSortKVRowByColIdx(row);
return buildCreateTbReq(&pCxt->createTblReq, tName, row, pCxt->pTableMeta->suid); taosMemoryFreeClear(pCxt->pTagVals);
return buildCreateTbReq(&pCxt->createTblReq, tName, pTag, pCxt->pTableMeta->suid);
} }
static int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) { static int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) {
@ -1062,7 +1078,7 @@ void destroyCreateSubTbReq(SVCreateTbReq* pReq) {
static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) { static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) {
taosMemoryFreeClear(pCxt->pTableMeta); taosMemoryFreeClear(pCxt->pTableMeta);
destroyBoundColumnInfo(&pCxt->tags); destroyBoundColumnInfo(&pCxt->tags);
tdDestroyKVRowBuilder(&pCxt->tagsBuilder); taosMemoryFreeClear(pCxt->pTagVals);
destroyCreateSubTbReq(&pCxt->createTblReq); destroyCreateSubTbReq(&pCxt->createTblReq);
} }
@ -1082,9 +1098,9 @@ static void destroyInsertParseContext(SInsertParseContext* pCxt) {
// VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path // VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
// [...]; // [...];
static int32_t parseInsertBody(SInsertParseContext* pCxt) { static int32_t parseInsertBody(SInsertParseContext* pCxt) {
int32_t tbNum = 0; int32_t tbNum = 0;
char tbFName[TSDB_TABLE_FNAME_LEN]; char tbFName[TSDB_TABLE_FNAME_LEN];
bool autoCreateTbl = false; bool autoCreateTbl = false;
// for each table // for each table
while (1) { while (1) {
@ -1186,8 +1202,8 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
memcpy(tags, &pCxt->tags, sizeof(pCxt->tags)); memcpy(tags, &pCxt->tags, sizeof(pCxt->tags));
(*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pTableMeta, tags, tbFName, autoCreateTbl, pCxt->pVgroupsHashObj, (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pTableMeta, tags, tbFName, autoCreateTbl,
pCxt->pTableBlockHashObj); pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj);
memset(&pCxt->tags, 0, sizeof(pCxt->tags)); memset(&pCxt->tags, 0, sizeof(pCxt->tags));
pCxt->pVgroupsHashObj = NULL; pCxt->pVgroupsHashObj = NULL;
@ -1219,6 +1235,7 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
.pSubTableHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK), .pSubTableHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
.pTableNameHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK), .pTableNameHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
.totalNum = 0, .totalNum = 0,
.pTagVals = NULL,
.pOutput = (SVnodeModifOpStmt*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT), .pOutput = (SVnodeModifOpStmt*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT),
.pStmtCb = pContext->pStmtCb}; .pStmtCb = pContext->pStmtCb};
@ -1332,13 +1349,13 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN
return TSDB_CODE_QRY_APP_ERROR; return TSDB_CODE_QRY_APP_ERROR;
} }
SKVRowBuilder tagBuilder; STagVal* pTagVals = taosMemoryCalloc(tags->numOfBound, sizeof(STagVal));
if (tdInitKVRowBuilder(&tagBuilder) < 0) { if (!pTagVals) {
return TSDB_CODE_TSC_OUT_OF_MEMORY; return buildInvalidOperationMsg(&pBuf, "out of memory");
} }
SSchema* pSchema = pDataBlock->pTableMeta->schema; SSchema* pSchema = pDataBlock->pTableMeta->schema;
SKvParam param = {.builder = &tagBuilder}; SKvParam param = {.pTagVals = pTagVals, .nTag = 0, .pos = 0};
for (int c = 0; c < tags->numOfBound; ++c) { for (int c = 0; c < tags->numOfBound; ++c) {
if (bind[c].is_null && bind[c].is_null[0]) { if (bind[c].is_null && bind[c].is_null[0]) {
@ -1357,19 +1374,19 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN
CHECK_CODE(KvRowAppend(&pBuf, (char*)bind[c].buffer, colLen, &param)); CHECK_CODE(KvRowAppend(&pBuf, (char*)bind[c].buffer, colLen, &param));
} }
SKVRow row = tdGetKVRowFromBuilder(&tagBuilder); STag* pTag = NULL;
if (NULL == row) {
tdDestroyKVRowBuilder(&tagBuilder); // TODO: JSON_TAG_REFACTOR (if is json or not)?
if (0 != tTagNew(pTagVals, param.nTag, 1, false, &pTag)) {
return buildInvalidOperationMsg(&pBuf, "out of memory"); return buildInvalidOperationMsg(&pBuf, "out of memory");
} }
tdSortKVRowByColIdx(row);
SVCreateTbReq tbReq = {0}; SVCreateTbReq tbReq = {0};
CHECK_CODE(buildCreateTbReq(&tbReq, tName, row, suid)); CHECK_CODE(buildCreateTbReq(&tbReq, tName, pTag, suid));
CHECK_CODE(buildCreateTbMsg(pDataBlock, &tbReq)); CHECK_CODE(buildCreateTbMsg(pDataBlock, &tbReq));
destroyCreateSubTbReq(&tbReq); destroyCreateSubTbReq(&tbReq);
tdDestroyKVRowBuilder(&tagBuilder); taosMemoryFreeClear(pTagVals);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1601,7 +1618,6 @@ int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD** fields
typedef struct SmlExecTableHandle { typedef struct SmlExecTableHandle {
SParsedDataColInfo tags; // each table SParsedDataColInfo tags; // each table
SKVRowBuilder tagsBuilder; // each table
SVCreateTbReq createTblReq; // each table SVCreateTbReq createTblReq; // each table
} SmlExecTableHandle; } SmlExecTableHandle;
@ -1613,7 +1629,6 @@ typedef struct SmlExecHandle {
static void smlDestroyTableHandle(void* pHandle) { static void smlDestroyTableHandle(void* pHandle) {
SmlExecTableHandle* handle = (SmlExecTableHandle*)pHandle; SmlExecTableHandle* handle = (SmlExecTableHandle*)pHandle;
tdDestroyKVRowBuilder(&handle->tagsBuilder);
destroyBoundColumnInfo(&handle->tags); destroyBoundColumnInfo(&handle->tags);
destroyCreateSubTbReq(&handle->createTblReq); destroyCreateSubTbReq(&handle->createTblReq);
} }
@ -1689,13 +1704,23 @@ static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SS
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t smlBuildTagRow(SArray* cols, SKVRowBuilder* tagsBuilder, SParsedDataColInfo* tags, SSchema* pSchema, /**
SKVRow* row, SMsgBuf* msg) { * @brief No json tag for schemaless
if (tdInitKVRowBuilder(tagsBuilder) < 0) { *
* @param cols
* @param tags
* @param pSchema
* @param ppTag
* @param msg
* @return int32_t
*/
static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* pSchema, STag** ppTag, SMsgBuf* msg) {
STagVal* pTagVals = taosMemoryCalloc(tags->numOfBound, sizeof(STagVal));
if (!pTagVals) {
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
SKvParam param = {.builder = tagsBuilder}; SKvParam param = {.pTagVals = pTagVals, .nTag = 0, .pos = 0};
for (int i = 0; i < tags->numOfBound; ++i) { for (int i = 0; i < tags->numOfBound; ++i) {
SSchema* pTagSchema = &pSchema[tags->boundColumns[i]]; SSchema* pTagSchema = &pSchema[tags->boundColumns[i]];
param.schema = pTagSchema; param.schema = pTagSchema;
@ -1707,11 +1732,12 @@ static int32_t smlBuildTagRow(SArray* cols, SKVRowBuilder* tagsBuilder, SParsedD
} }
} }
*row = tdGetKVRowFromBuilder(tagsBuilder); if (tTagNew(pTagVals, param.nTag, 1, false, ppTag) != 0) {
if (*row == NULL) { taosMemoryFree(pTagVals);
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
tdSortKVRowByColIdx(*row);
taosMemoryFree(pTagVals);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1728,14 +1754,13 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols
buildInvalidOperationMsg(&pBuf, "bound tags error"); buildInvalidOperationMsg(&pBuf, "bound tags error");
return ret; return ret;
} }
SKVRow row = NULL; STag* pTag = NULL;
ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tagsBuilder, &smlHandle->tableExecHandle.tags, pTagsSchema, ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tags, pTagsSchema, &pTag, &pBuf);
&row, &pBuf);
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
return ret; return ret;
} }
buildCreateTbReq(&smlHandle->tableExecHandle.createTblReq, tableName, row, pTableMeta->suid); buildCreateTbReq(&smlHandle->tableExecHandle.createTblReq, tableName, pTag, pTableMeta->suid);
STableDataBlocks* pDataBlock = NULL; STableDataBlocks* pDataBlock = NULL;
ret = getDataBlockFromList(smlHandle->pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid), ret = getDataBlockFromList(smlHandle->pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid),

149
source/libs/parser/src/parTranslater.c
View File

@ -4113,7 +4113,7 @@ static int32_t rewriteCreateTable(STranslateContext* pCxt, SQuery* pQuery) {
return code; return code;
} }
static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap, SCreateSubTableClause* pStmt, SKVRow row, static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap, SCreateSubTableClause* pStmt, const STag *pTag,
uint64_t suid, SVgroupInfo* pVgInfo) { uint64_t suid, SVgroupInfo* pVgInfo) {
char dbFName[TSDB_DB_FNAME_LEN] = {0}; char dbFName[TSDB_DB_FNAME_LEN] = {0};
SName name = {.type = TSDB_DB_NAME_T, .acctId = acctId}; SName name = {.type = TSDB_DB_NAME_T, .acctId = acctId};
@ -4124,7 +4124,7 @@ static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap, S
req.type = TD_CHILD_TABLE; req.type = TD_CHILD_TABLE;
req.name = strdup(pStmt->tableName); req.name = strdup(pStmt->tableName);
req.ctb.suid = suid; req.ctb.suid = suid;
req.ctb.pTag = row; req.ctb.pTag = (uint8_t*)pTag;
if (pStmt->ignoreExists) { if (pStmt->ignoreExists) {
req.flags |= TD_CREATE_IF_NOT_EXISTS; req.flags |= TD_CREATE_IF_NOT_EXISTS;
} }
@ -4144,8 +4144,9 @@ static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap, S
} }
} }
static int32_t addValToKVRow(STranslateContext* pCxt, SValueNode* pVal, const SSchema* pSchema, // static int32_t addValToKVRow(STranslateContext* pCxt, SValueNode* pVal, const SSchema* pSchema,
SKVRowBuilder* pBuilder) { // SKVRowBuilder* pBuilder) {
#ifdef JSON_TAG_REFACTOR
if (pSchema->type == TSDB_DATA_TYPE_JSON) { if (pSchema->type == TSDB_DATA_TYPE_JSON) {
if (pVal->literal && strlen(pVal->literal) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) { if (pVal->literal && strlen(pVal->literal) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
return buildSyntaxErrMsg(&pCxt->msgBuf, "json string too long than 4095", pVal->literal); return buildSyntaxErrMsg(&pCxt->msgBuf, "json string too long than 4095", pVal->literal);
@ -4153,14 +4154,15 @@ static int32_t addValToKVRow(STranslateContext* pCxt, SValueNode* pVal, const SS
return parseJsontoTagData(pVal->literal, pBuilder, &pCxt->msgBuf, pSchema->colId); return parseJsontoTagData(pVal->literal, pBuilder, &pCxt->msgBuf, pSchema->colId);
} }
#endif
if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) { // if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
tdAddColToKVRow(pBuilder, pSchema->colId, nodesGetValueFromNode(pVal), // tdAddColToKVRow(pBuilder, pSchema->colId, nodesGetValueFromNode(pVal),
IS_VAR_DATA_TYPE(pSchema->type) ? varDataTLen(pVal->datum.p) : TYPE_BYTES[pSchema->type]); // IS_VAR_DATA_TYPE(pSchema->type) ? varDataTLen(pVal->datum.p) : TYPE_BYTES[pSchema->type]);
} // }
return TSDB_CODE_SUCCESS; // return TSDB_CODE_SUCCESS;
} // }
static int32_t createValueFromFunction(STranslateContext* pCxt, SFunctionNode* pFunc, SValueNode** pVal) { static int32_t createValueFromFunction(STranslateContext* pCxt, SFunctionNode* pFunc, SValueNode** pVal) {
int32_t code = getFuncInfo(pCxt, pFunc); int32_t code = getFuncInfo(pCxt, pFunc);
@ -4189,18 +4191,28 @@ static int32_t translateTagVal(STranslateContext* pCxt, uint8_t precision, SSche
} }
static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableClause* pStmt, STableMeta* pSuperTableMeta, static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableClause* pStmt, STableMeta* pSuperTableMeta,
SKVRowBuilder* pBuilder) { STag** ppTag) {
int32_t numOfTags = getNumOfTags(pSuperTableMeta); int32_t numOfTags = getNumOfTags(pSuperTableMeta);
if (LIST_LENGTH(pStmt->pValsOfTags) != LIST_LENGTH(pStmt->pSpecificTags) || if (LIST_LENGTH(pStmt->pValsOfTags) != LIST_LENGTH(pStmt->pSpecificTags) ||
numOfTags < LIST_LENGTH(pStmt->pValsOfTags)) { numOfTags < LIST_LENGTH(pStmt->pValsOfTags)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_TAGS_NOT_MATCHED); return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_TAGS_NOT_MATCHED);
} }
STagVal* pTagVals = (STagVal*)taosMemoryCalloc(LIST_LENGTH(pStmt->pValsOfTags), sizeof(STagVal));
char* pTagBuf = taosMemoryCalloc(1, TSDB_MAX_TAGS_LEN);
if (!pTagVals || !pTagBuf) {
taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(pTagBuf);
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_TSC_OUT_OF_MEMORY);
}
int32_t code = 0;
int16_t nTags = 0, nBufPos = 0;
SSchema* pTagSchema = getTableTagSchema(pSuperTableMeta); SSchema* pTagSchema = getTableTagSchema(pSuperTableMeta);
SNode * pTag, *pNode; SNode * pTag, *pNode;
FORBOTH(pTag, pStmt->pSpecificTags, pNode, pStmt->pValsOfTags) { FORBOTH(pTag, pStmt->pSpecificTags, pNode, pStmt->pValsOfTags) {
SColumnNode* pCol = (SColumnNode*)pTag; SColumnNode* pCol = (SColumnNode*)pTag;
SSchema* pSchema = NULL; SSchema* pSchema = NULL;
STagVal* pTagVal = pTagVals + nTags;
for (int32_t i = 0; i < numOfTags; ++i) { for (int32_t i = 0; i < numOfTags; ++i) {
if (0 == strcmp(pCol->colName, pTagSchema[i].name)) { if (0 == strcmp(pCol->colName, pTagSchema[i].name)) {
pSchema = pTagSchema + i; pSchema = pTagSchema + i;
@ -4208,10 +4220,12 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla
} }
} }
if (NULL == pSchema) { if (NULL == pSchema) {
taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(pTagBuf);
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_TAG_NAME, pCol->colName); return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_TAG_NAME, pCol->colName);
} }
SValueNode* pVal = NULL; SValueNode* pVal = NULL;
int32_t code = translateTagVal(pCxt, pSuperTableMeta->tableInfo.precision, pSchema, pNode, &pVal); code = translateTagVal(pCxt, pSuperTableMeta->tableInfo.precision, pSchema, pNode, &pVal);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
if (NULL == pVal) { if (NULL == pVal) {
pVal = (SValueNode*)pNode; pVal = (SValueNode*)pNode;
@ -4219,29 +4233,74 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla
REPLACE_LIST2_NODE(pVal); REPLACE_LIST2_NODE(pVal);
} }
} }
#ifdef JSON_TAG_REFACTOR
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = addValToKVRow(pCxt, pVal, pSchema, pBuilder); code = addValToKVRow(pCxt, pVal, pSchema, pBuilder);
} }
#endif
if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
// TODO: JSON_TAG_TODO: is copy is a must?
void* nodeVal = nodesGetValueFromNode(pVal);
if (IS_VAR_DATA_TYPE(pSchema->type)) {
memcpy(pTagBuf + nBufPos, varDataVal(nodeVal), varDataLen(nodeVal));
tTagValSet(pTagVal, &pSchema->colId, pSchema->type, (uint8_t*)pTagBuf + nBufPos, varDataLen(nodeVal), false);
nBufPos += varDataLen(pVal->datum.p);
} else {
memcpy(pTagBuf + nBufPos, varDataVal(nodeVal), TYPE_BYTES[pSchema->type]);
tTagValSet(pTagVal, &pSchema->colId, pSchema->type, (uint8_t*)pTagBuf + nBufPos, TYPE_BYTES[pSchema->type],
false);
nBufPos += TYPE_BYTES[pSchema->type];
}
}
if (TSDB_CODE_SUCCESS != code) { if (TSDB_CODE_SUCCESS != code) {
taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(pTagBuf);
return code; return code;
} }
} }
// TODO: JSON_TAG_TODO: version
code = tTagNew(pTagVals, nTags, 1, false, ppTag);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(pTagBuf);
return code;
}
taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(pTagBuf);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t buildKVRowForAllTags(STranslateContext* pCxt, SCreateSubTableClause* pStmt, STableMeta* pSuperTableMeta, static int32_t buildKVRowForAllTags(STranslateContext* pCxt, SCreateSubTableClause* pStmt, STableMeta* pSuperTableMeta,
SKVRowBuilder* pBuilder) { STag** ppTag) {
if (getNumOfTags(pSuperTableMeta) != LIST_LENGTH(pStmt->pValsOfTags)) { if (getNumOfTags(pSuperTableMeta) != LIST_LENGTH(pStmt->pValsOfTags)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_TAGS_NOT_MATCHED); return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_TAGS_NOT_MATCHED);
} }
SSchema* pTagSchema = getTableTagSchema(pSuperTableMeta); SSchema* pTagSchemas = getTableTagSchema(pSuperTableMeta);
SNode* pNode; SNode* pNode;
int32_t code = 0;
int32_t index = 0; int32_t index = 0;
int16_t nTag = 0;
STagVal* pTagVals = taosMemoryCalloc(LIST_LENGTH(pStmt->pValsOfTags), sizeof(STagVal));
char* pTagBuf = taosMemoryCalloc(1, TSDB_MAX_TAGS_LEN);
const char* qTagBuf = pTagBuf;
if (!pTagVals || !pTagBuf) {
taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(qTagBuf);
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_TSC_OUT_OF_MEMORY);
}
FOREACH(pNode, pStmt->pValsOfTags) { FOREACH(pNode, pStmt->pValsOfTags) {
SValueNode* pVal = NULL; SValueNode* pVal = NULL;
int32_t code = translateTagVal(pCxt, pSuperTableMeta->tableInfo.precision, pTagSchema + index, pNode, &pVal); STagVal* pTagVal = pTagVals + nTag;
SSchema* pTagSchema = pTagSchemas + index;
code = translateTagVal(pCxt, pSuperTableMeta->tableInfo.precision, pTagSchema, pNode, &pVal);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
if (NULL == pVal) { if (NULL == pVal) {
pVal = (SValueNode*)pNode; pVal = (SValueNode*)pNode;
@ -4249,14 +4308,48 @@ static int32_t buildKVRowForAllTags(STranslateContext* pCxt, SCreateSubTableClau
REPLACE_NODE(pVal); REPLACE_NODE(pVal);
} }
} }
#ifdef JSON_TAG_REFACTOR
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = addValToKVRow(pCxt, pVal, pTagSchema + index++, pBuilder); code = addValToKVRow(pCxt, pVal, pTagSchema + index++, pBuilder);
} }
#endif
if (TSDB_CODE_SUCCESS == code) {
if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
char* tmpVal = nodesGetValueFromNode(pVal);
if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
memcpy(pTagBuf, varDataVal(tmpVal), varDataLen(tmpVal));
tTagValSet(pTagVal, &pTagSchema->colId, pTagSchema->type, (uint8_t*)pTagBuf, varDataLen(tmpVal),
false);
pTagBuf += varDataLen(tmpVal);
} else {
memcpy(pTagBuf, tmpVal, TYPE_BYTES[pTagSchema->type]);
tTagValSet(pTagVal, &pTagSchema->colId, pTagSchema->type, (uint8_t*)pTagBuf,
TYPE_BYTES[pTagSchema->type], false);
pTagBuf += TYPE_BYTES[pTagSchema->type];
}
++nTag;
}
++index;
}
// TODO: buf is need to store the tags
// TODO: JSON_TAG_TODO remove below codes if code is 0 all the time.
if (TSDB_CODE_SUCCESS != code) { if (TSDB_CODE_SUCCESS != code) {
return code; taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(qTagBuf);
return generateSyntaxErrMsg(&pCxt->msgBuf, code);
} }
} }
// TODO: JSON_TAG_TODO: version?
// TODO: JSON_TAG_REFACTOR: json or not
if (TSDB_CODE_SUCCESS != (code = tTagNew(pTagVals, nTag, 1, false, ppTag))) {
taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(qTagBuf);
return generateSyntaxErrMsg(&pCxt->msgBuf, code);
}
taosMemoryFreeClear(pTagVals);
taosMemoryFreeClear(qTagBuf);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -4274,26 +4367,13 @@ static int32_t rewriteCreateSubTable(STranslateContext* pCxt, SCreateSubTableCla
code = getTableMeta(pCxt, pStmt->useDbName, pStmt->useTableName, &pSuperTableMeta); code = getTableMeta(pCxt, pStmt->useDbName, pStmt->useTableName, &pSuperTableMeta);
} }
SKVRowBuilder kvRowBuilder = {0}; STag* pTag = NULL;
if (TSDB_CODE_SUCCESS == code) {
code = tdInitKVRowBuilder(&kvRowBuilder);
}
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
if (NULL != pStmt->pSpecificTags) { if (NULL != pStmt->pSpecificTags) {
code = buildKVRowForBindTags(pCxt, pStmt, pSuperTableMeta, &kvRowBuilder); code = buildKVRowForBindTags(pCxt, pStmt, pSuperTableMeta, &pTag);
} else { } else {
code = buildKVRowForAllTags(pCxt, pStmt, pSuperTableMeta, &kvRowBuilder); code = buildKVRowForAllTags(pCxt, pStmt, pSuperTableMeta, &pTag);
}
}
SKVRow row = NULL;
if (TSDB_CODE_SUCCESS == code) {
row = tdGetKVRowFromBuilder(&kvRowBuilder);
if (NULL == row) {
code = TSDB_CODE_OUT_OF_MEMORY;
} else {
tdSortKVRowByColIdx(row);
} }
} }
@ -4302,11 +4382,10 @@ static int32_t rewriteCreateSubTable(STranslateContext* pCxt, SCreateSubTableCla
code = getTableHashVgroup(pCxt, pStmt->dbName, pStmt->tableName, &info); code = getTableHashVgroup(pCxt, pStmt->dbName, pStmt->tableName, &info);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
addCreateTbReqIntoVgroup(pCxt->pParseCxt->acctId, pVgroupHashmap, pStmt, row, pSuperTableMeta->uid, &info); addCreateTbReqIntoVgroup(pCxt->pParseCxt->acctId, pVgroupHashmap, pStmt, pTag, pSuperTableMeta->uid, &info);
} }
taosMemoryFreeClear(pSuperTableMeta); taosMemoryFreeClear(pSuperTableMeta);
tdDestroyKVRowBuilder(&kvRowBuilder);
return code; return code;
} }
@ -4528,6 +4607,7 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS
pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type); pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type);
if (pStmt->pVal->node.resType.type == TSDB_DATA_TYPE_JSON) { if (pStmt->pVal->node.resType.type == TSDB_DATA_TYPE_JSON) {
#ifdef JSON_TAG_REFACTOR
SKVRowBuilder kvRowBuilder = {0}; SKVRowBuilder kvRowBuilder = {0};
int32_t code = tdInitKVRowBuilder(&kvRowBuilder); int32_t code = tdInitKVRowBuilder(&kvRowBuilder);
@ -4553,6 +4633,7 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS
pReq->pTagVal = row; pReq->pTagVal = row;
pStmt->pVal->datum.p = row; // for free pStmt->pVal->datum.p = row; // for free
tdDestroyKVRowBuilder(&kvRowBuilder); tdDestroyKVRowBuilder(&kvRowBuilder);
#endif
} else { } else {
pReq->nTagVal = pStmt->pVal->node.resType.bytes; pReq->nTagVal = pStmt->pVal->node.resType.bytes;
if (TSDB_DATA_TYPE_NCHAR == pStmt->pVal->node.resType.type) { if (TSDB_DATA_TYPE_NCHAR == pStmt->pVal->node.resType.type) {

2
source/libs/parser/src/parUtil.c
View File

@ -328,6 +328,7 @@ static bool isValidateTag(char* input) {
return true; return true;
} }
#ifdef JSON_TAG_REFACTOR
int32_t parseJsontoTagData(const char* json, SKVRowBuilder* kvRowBuilder, SMsgBuf* pMsgBuf, int16_t startColId) { int32_t parseJsontoTagData(const char* json, SKVRowBuilder* kvRowBuilder, SMsgBuf* pMsgBuf, int16_t startColId) {
// set json NULL data // set json NULL data
uint8_t jsonNULL = TSDB_DATA_TYPE_NULL; uint8_t jsonNULL = TSDB_DATA_TYPE_NULL;
@ -448,6 +449,7 @@ end:
cJSON_Delete(root); cJSON_Delete(root);
return retCode; return retCode;
} }
#endif
static int32_t buildTableReq(SHashObj* pTablesHash, SArray** pTables) { static int32_t buildTableReq(SHashObj* pTablesHash, SArray** pTables) {
if (NULL != pTablesHash) { if (NULL != pTablesHash) {

10
source/libs/scalar/src/sclvector.c
View File

@ -922,8 +922,13 @@ static void doReleaseVec(SColumnInfoData* pCol, int32_t type) {
} }
} }
char *getJsonValue(char *json, char *key){ //todo char *getJsonValue(char *json, char *key) { // todo
json++; // jump type json++; // jump type
STagVal tagVal = {.pKey = key};
tTagGet(((const STag *)json), &tagVal);
return (char *)tagVal.pData;
#if 0
int16_t cols = kvRowNCols(json); int16_t cols = kvRowNCols(json);
for (int i = 0; i < cols; ++i) { for (int i = 0; i < cols; ++i) {
SColIdx *pColIdx = kvRowColIdxAt(json, i); SColIdx *pColIdx = kvRowColIdxAt(json, i);
@ -939,6 +944,7 @@ char *getJsonValue(char *json, char *key){ //todo
} }
} }
return NULL; return NULL;
#endif
} }
void vectorJsonArrow(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t _ord) { void vectorJsonArrow(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t _ord) {

8
source/libs/scalar/test/scalar/scalarTests.cpp
View File

@ -1035,7 +1035,7 @@ void makeJsonArrow(SSDataBlock **src, SNode **opNode, void *json, char *key){
SNode *pLeft = NULL, *pRight = NULL; SNode *pLeft = NULL, *pRight = NULL;
scltMakeValueNode(&pRight, TSDB_DATA_TYPE_BINARY, keyVar); scltMakeValueNode(&pRight, TSDB_DATA_TYPE_BINARY, keyVar);
scltMakeColumnNode(&pLeft, src, TSDB_DATA_TYPE_JSON, kvRowLen(json), 1, json); scltMakeColumnNode(&pLeft, src, TSDB_DATA_TYPE_JSON, ((STag*)json)->len, 1, json);
scltMakeOpNode(opNode, OP_TYPE_JSON_GET_VALUE, TSDB_DATA_TYPE_JSON, pLeft, pRight); scltMakeOpNode(opNode, OP_TYPE_JSON_GET_VALUE, TSDB_DATA_TYPE_JSON, pLeft, pRight);
} }
@ -1105,6 +1105,7 @@ void makeCalculate(void *json, void *key, int32_t rightType, void *rightData, do
nodesDestroyNode(opNode); nodesDestroyNode(opNode);
} }
#if 0
TEST(columnTest, json_column_arith_op) { TEST(columnTest, json_column_arith_op) {
scltInitLogFile(); scltInitLogFile();
char *rightvTmp= "{\"k1\":4,\"k2\":\"hello\",\"k3\":null,\"k4\":true,\"k5\":5.44}"; char *rightvTmp= "{\"k1\":4,\"k2\":\"hello\",\"k3\":null,\"k4\":true,\"k5\":5.44}";
@ -1178,7 +1179,7 @@ TEST(columnTest, json_column_arith_op) {
tdDestroyKVRowBuilder(&kvRowBuilder); tdDestroyKVRowBuilder(&kvRowBuilder);
taosMemoryFree(row); taosMemoryFree(row);
} }
#endif
void *prepareNchar(char* rightData){ void *prepareNchar(char* rightData){
int32_t len = 0; int32_t len = 0;
int32_t inputLen = strlen(rightData); int32_t inputLen = strlen(rightData);
@ -1188,7 +1189,7 @@ void *prepareNchar(char* rightData){
varDataSetLen(t, len); varDataSetLen(t, len);
return t; return t;
} }
#if 0
TEST(columnTest, json_column_logic_op) { TEST(columnTest, json_column_logic_op) {
scltInitLogFile(); scltInitLogFile();
char *rightvTmp= "{\"k1\":4,\"k2\":\"hello\",\"k3\":null,\"k4\":true,\"k5\":5.44,\"k6\":\"6.6hello\"}"; char *rightvTmp= "{\"k1\":4,\"k2\":\"hello\",\"k3\":null,\"k4\":true,\"k5\":5.44,\"k6\":\"6.6hello\"}";
@ -1308,6 +1309,7 @@ TEST(columnTest, json_column_logic_op) {
tdDestroyKVRowBuilder(&kvRowBuilder); tdDestroyKVRowBuilder(&kvRowBuilder);
taosMemoryFree(row); taosMemoryFree(row);
} }
#endif
TEST(columnTest, smallint_value_add_int_column) { TEST(columnTest, smallint_value_add_int_column) {
scltInitLogFile(); scltInitLogFile();