Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
未验证 提交 d342ebcd 编写于 作者: S Shengliang Guan 提交者： GitHub
浏览文件
操作

Merge pull request #18720 from taosdata/refact/submit_req 

Refact/submit req
上级 19e78cd5 b1d4993a
展开全部 隐藏空白更改
内联 并排
Showing with 9068 addition and 6343 deletion
include/common/tcommon.h
浏览文件 @ d342ebcd
......@@ -341,7 +341,7 @@ typedef struct SExprInfo {
typedef struct {
const char* key;
int32_t keyLen;
size_t keyLen;
uint8_t type;
union {
const char* value;
......@@ -350,7 +350,7 @@ typedef struct {
double d;
float f;
};
int32_t length;
size_t length;
} SSmlKv;
#define QUERY_ASC_FORWARD_STEP 1
......
include/common/tdatablock.h
浏览文件 @ d342ebcd
......@@ -265,7 +265,7 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag);
// for debug
char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf);
int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataBlocks, STSchema* pTSchema, int32_t vgId,
int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pDataBlocks, const STSchema* pTSchema, int64_t uid, int32_t vgId,
tb_uid_t suid);
char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);
......
include/common/tdataformat.h
浏览文件 @ d342ebcd
......@@ -44,18 +44,38 @@ typedef struct SColData SColData;
#define HAS_VALUE ((uint8_t)0x4)
// bitmap ================================
const static uint8_t BIT2_MAP[4][4] = {{0b00000000, 0b00000001, 0b00000010, 0},
{0b00000000, 0b00000100, 0b00001000, 2},
{0b00000000, 0b00010000, 0b00100000, 4},
{0b00000000, 0b01000000, 0b10000000, 6}};
#define N1(n) ((((uint8_t)1) << (n)) - 1)
#define BIT1_SIZE(n) ((((n)-1) >> 3) + 1)
#define BIT2_SIZE(n) ((((n)-1) >> 2) + 1)
#define SET_BIT1(p, i, v) ((p)[(i) >> 3] = (p)[(i) >> 3] & N1((i)&7) | (((uint8_t)(v)) << ((i)&7)))
#define GET_BIT1(p, i) (((p)[(i) >> 3] >> ((i)&7)) & ((uint8_t)1))
#define SET_BIT2(p, i, v) ((p)[(i) >> 2] = (p)[(i) >> 2] & N1(BIT2_MAP[(i)&3][3]) | BIT2_MAP[(i)&3][(v)])
#define GET_BIT2(p, i) (((p)[(i) >> 2] >> BIT2_MAP[(i)&3][3]) & ((uint8_t)3))
const static uint8_t BIT1_MAP[8] = {0b11111110, 0b11111101, 0b11111011, 0b11110111,
0b11101111, 0b11011111, 0b10111111, 0b01111111};
const static uint8_t BIT2_MAP[4] = {0b11111100, 0b11110011, 0b11001111, 0b00111111};
#define ONE ((uint8_t)1)
#define THREE ((uint8_t)3)
#define DIV_8(i) ((i) >> 3)
#define MOD_8(i) ((i)&7)
#define DIV_4(i) ((i) >> 2)
#define MOD_4(i) ((i)&3)
#define MOD_4_TIME_2(i) (MOD_4(i) << 1)
#define BIT1_SIZE(n) (DIV_8((n)-1) + 1)
#define BIT2_SIZE(n) (DIV_4((n)-1) + 1)
#define SET_BIT1(p, i, v) ((p)[DIV_8(i)] = (p)[DIV_8(i)] & BIT1_MAP[MOD_8(i)] | ((v) << MOD_8(i)))
#define SET_BIT1_EX(p, i, v) \
do { \
if (MOD_8(i) == 0) { \
(p)[DIV_8(i)] = 0; \
} \
SET_BIT1(p, i, v); \
} while (0)
#define GET_BIT1(p, i) (((p)[DIV_8(i)] >> MOD_8(i)) & ONE)
#define SET_BIT2(p, i, v) ((p)[DIV_4(i)] = (p)[DIV_4(i)] & BIT2_MAP[MOD_4(i)] | ((v) << MOD_4_TIME_2(i)))
#define SET_BIT2_EX(p, i, v) \
do { \
if (MOD_4(i) == 0) { \
(p)[DIV_4(i)] = 0; \
} \
SET_BIT2(p, i, v); \
} while (0)
#define GET_BIT2(p, i) (((p)[DIV_4(i)] >> MOD_4_TIME_2(i)) & THREE)
// SBuffer ================================
struct SBuffer {
......@@ -70,9 +90,6 @@ int32_t tBufferInit(SBuffer *pBuffer, int64_t size);
int32_t tBufferPut(SBuffer *pBuffer, const void *pData, int64_t nData);
int32_t tBufferReserve(SBuffer *pBuffer, int64_t nData, void **ppData);
// STSchema ================================
void tDestroyTSchema(STSchema *pTSchema);
// SColVal ================================
#define CV_FLAG_VALUE ((int8_t)0x0)
#define CV_FLAG_NONE ((int8_t)0x1)
......@@ -87,8 +104,12 @@ void tDestroyTSchema(STSchema *pTSchema);
#define COL_VAL_IS_VALUE(CV) ((CV)->flag == CV_FLAG_VALUE)
// SRow ================================
int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SBuffer *pBuffer);
int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow);
void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
void tRowDestroy(SRow *pRow);
void tRowSort(SArray *aRowP);
int32_t tRowMerge(SArray *aRowP, STSchema *pTSchema, int8_t flag);
int32_t tRowAppendToColData(SRow *pRow, STSchema *pTSchema, SColData *aColData, int32_t nColData);
// SRowIter ================================
int32_t tRowIterOpen(SRow *pRow, STSchema *pTSchema, SRowIter **ppIter);
......@@ -110,15 +131,28 @@ void debugPrintSTag(STag *pTag, const char *tag, int32_t ln); // TODO: remov
int32_t parseJsontoTagData(const char *json, SArray *pTagVals, STag **ppTag, void *pMsgBuf);
// SColData ================================
typedef void *(*xMallocFn)(void *, int32_t);
void tColDataDestroy(void *ph);
void tColDataInit(SColData *pColData, int16_t cid, int8_t type, int8_t smaOn);
void tColDataClear(SColData *pColData);
void tColDataDeepClear(SColData *pColData);
int32_t tColDataAppendValue(SColData *pColData, SColVal *pColVal);
void tColDataGetValue(SColData *pColData, int32_t iVal, SColVal *pColVal);
uint8_t tColDataGetBitValue(const SColData *pColData, int32_t iVal);
int32_t tColDataCopy(SColData *pColDataSrc, SColData *pColDataDest);
int32_t tColDataCopy(SColData *pColDataFrom, SColData *pColData, xMallocFn xMalloc, void *arg);
extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max, int64_t *min, int16_t *numOfNull);
// for stmt bind
int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind);
void tColDataSortMerge(SArray *colDataArr);
//for raw block
int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes,
int32_t nRows, char* lengthOrbitmap, char *data);
// for encode/decode
int32_t tPutColData(uint8_t *pBuf, SColData *pColData);
int32_t tGetColData(uint8_t *pBuf, SColData *pColData);
// STRUCT ================================
struct STColumn {
col_id_t colId;
......@@ -225,23 +259,9 @@ struct STag {
memcpy(varDataVal(x), (str), (_size)); \
} while (0);
// ----------------- SCHEMA BUILDER DEFINITION
typedef struct {
int32_t tCols;
int32_t nCols;
schema_ver_t version;
uint16_t flen;
int32_t tlen;
STColumn *columns;
} STSchemaBuilder;
int32_t tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
void tdDestroyTSchemaBuilder(STSchemaBuilder *pBuilder);
void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags, col_id_t colId, col_bytes_t bytes);
STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
// STSchema ================================
STSchema *tBuildTSchema(SSchema *aSchema, int32_t numOfCols, int32_t version);
void tDestroyTSchema(STSchema *pTSchema);
#endif
......
include/common/tmsg.h
浏览文件 @ d342ebcd
......@@ -482,8 +482,6 @@ static FORCE_INLINE int32_t tDecodeSSchemaWrapperEx(SDecoder* pDecoder, SSchemaW
return 0;
}
STSchema* tdGetSTSChemaFromSSChema(SSchema* pSchema, int32_t nCols, int32_t sver);
typedef struct {
char name[TSDB_TABLE_FNAME_LEN];
int8_t igExists;
......@@ -1734,6 +1732,8 @@ typedef struct {
int32_t execId;
} STaskDropReq;
int32_t tSerializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
int32_t tDeserializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
int32_t tSerializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
int32_t tDeserializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
......@@ -2081,10 +2081,15 @@ typedef struct SVCreateTbReq {
};
} SVCreateTbReq;
int tEncodeSVCreateTbReq(SEncoder* pCoder, const SVCreateTbReq* pReq);
int tDecodeSVCreateTbReq(SDecoder* pCoder, SVCreateTbReq* pReq);
int tEncodeSVCreateTbReq(SEncoder* pCoder, const SVCreateTbReq* pReq);
int tDecodeSVCreateTbReq(SDecoder* pCoder, SVCreateTbReq* pReq);
void tDestroySVCreateTbReq(SVCreateTbReq* pReq, int32_t flags);
static FORCE_INLINE void tdDestroySVCreateTbReq(SVCreateTbReq* req) {
if (NULL == req) {
return;
}
taosMemoryFreeClear(req->name);
taosMemoryFreeClear(req->comment);
if (req->type == TSDB_CHILD_TABLE) {
......@@ -3232,6 +3237,57 @@ int32_t tSerializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
int32_t tDeserializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
int32_t tSerializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
int32_t tDeserializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
int32_t tSerializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
int32_t tDeserializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
int32_t tSerializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
int32_t tDeserializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
#define SUBMIT_REQ_AUTO_CREATE_TABLE 0x1
#define SUBMIT_REQ_COLUMN_DATA_FORMAT 0x2
typedef struct {
int32_t flags;
SVCreateTbReq* pCreateTbReq;
int64_t suid;
int64_t uid;
int32_t sver;
union {
SArray* aRowP;
SArray* aCol;
};
} SSubmitTbData;
typedef struct {
SArray* aSubmitTbData; // SArray<SSubmitTbData>
} SSubmitReq2;
int32_t tEncodeSSubmitReq2(SEncoder* pCoder, const SSubmitReq2* pReq);
int32_t tDecodeSSubmitReq2(SDecoder* pCoder, SSubmitReq2* pReq);
void tDestroySSubmitTbData(SSubmitTbData* pTbData, int32_t flag);
void tDestroySSubmitReq2(SSubmitReq2* pReq, int32_t flag);
typedef struct {
int32_t affectedRows;
SArray* aCreateTbRsp; // SArray<SVCreateTbRsp>
} SSubmitRsp2;
int32_t tEncodeSSubmitRsp2(SEncoder* pCoder, const SSubmitRsp2* pRsp);
int32_t tDecodeSSubmitRsp2(SDecoder* pCoder, SSubmitRsp2* pRsp);
void tDestroySSubmitRsp2(SSubmitRsp2* pRsp, int32_t flag);
#define TSDB_MSG_FLG_ENCODE 0x1
#define TSDB_MSG_FLG_DECODE 0x2
typedef struct {
union {
struct {
void* msgStr;
int32_t msgLen;
int64_t ver;
};
void* pDataBlock;
};
} SPackedData;
#pragma pack(pop)
......
include/libs/executor/executor.h
浏览文件 @ d342ebcd
......@@ -190,7 +190,9 @@ int32_t qStreamPrepareTsdbScan(qTaskInfo_t tinfo, uint64_t uid, int64_t ts);
int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType);
int32_t qStreamScanMemData(qTaskInfo_t tinfo, const SSubmitReq* pReq);
// int32_t qStreamScanMemData(qTaskInfo_t tinfo, const SSubmitReq* pReq, int64_t ver);
//
int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit);
int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);
......
include/libs/nodes/querynodes.h
浏览文件 @ d342ebcd
......@@ -361,33 +361,33 @@ typedef struct SVgDataBlocks {
void* pData; // SSubmitReq + SSubmitBlk + ...
} SVgDataBlocks;
typedef void (*FFreeDataBlockHash)(SHashObj*);
typedef void (*FFreeDataBlockArray)(SArray*);
typedef void (*FFreeTableBlockHash)(SHashObj*);
typedef void (*FFreeVgourpBlockArray)(SArray*);
typedef struct SVnodeModifyOpStmt {
ENodeType nodeType;
ENodeType sqlNodeType;
SArray* pDataBlocks; // data block for each vgroup, SArray<SVgDataBlocks*>.
uint32_t insertType; // insert data from [file|sql statement| bound statement]
const char* pSql; // current sql statement position
int32_t totalRowsNum;
int32_t totalTbNum;
SName targetTableName;
SName usingTableName;
const char* pBoundCols;
struct STableMeta* pTableMeta;
SHashObj* pVgroupsHashObj;
SHashObj* pTableBlockHashObj;
SHashObj* pSubTableHashObj;
SHashObj* pTableNameHashObj;
SHashObj* pDbFNameHashObj;
SArray* pVgDataBlocks;
SVCreateTbReq createTblReq;
TdFilePtr fp;
FFreeDataBlockHash freeHashFunc;
FFreeDataBlockArray freeArrayFunc;
bool usingTableProcessing;
bool fileProcessing;
ENodeType nodeType;
ENodeType sqlNodeType;
SArray* pDataBlocks; // data block for each vgroup, SArray<SVgDataBlocks*>.
uint32_t insertType; // insert data from [file|sql statement| bound statement]
const char* pSql; // current sql statement position
int32_t totalRowsNum;
int32_t totalTbNum;
SName targetTableName;
SName usingTableName;
const char* pBoundCols;
struct STableMeta* pTableMeta;
SHashObj* pVgroupsHashObj;
SHashObj* pTableBlockHashObj; // SHashObj<tuid, STableDataCxt*>
SHashObj* pSubTableHashObj;
SHashObj* pTableNameHashObj;
SHashObj* pDbFNameHashObj;
SArray* pVgDataBlocks; // SArray<SVgroupDataCxt*>
SVCreateTbReq* pCreateTblReq;
TdFilePtr fp;
FFreeTableBlockHash freeHashFunc;
FFreeVgourpBlockArray freeArrayFunc;
bool usingTableProcessing;
bool fileProcessing;
} SVnodeModifyOpStmt;
typedef struct SExplainOptions {
......
include/libs/parser/parser.h
浏览文件 @ d342ebcd
......@@ -58,7 +58,6 @@ typedef struct SParseContext {
bool isSuperUser;
bool enableSysInfo;
bool async;
int8_t schemalessType;
const char* svrVer;
bool nodeOffline;
SArray* pTableMetaPos; // sql table pos => catalog data pos
......@@ -85,12 +84,12 @@ int32_t qSetSTableIdForRsma(SNode* pStmt, int64_t uid);
void qCleanupKeywordsTable();
int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash);
int32_t qResetStmtDataBlock(void* block, bool keepBuf);
int32_t qCloneStmtDataBlock(void** pDst, void* pSrc);
void qFreeStmtDataBlock(void* pDataBlock);
int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t vgId);
void qDestroyStmtDataBlock(void* pBlock);
STableMeta* qGetTableMetaInDataBlock(void* pDataBlock);
int32_t qResetStmtDataBlock(STableDataCxt* block, bool keepBuf);
int32_t qCloneStmtDataBlock(STableDataCxt** pDst, STableDataCxt* pSrc, bool reset);
int32_t qRebuildStmtDataBlock(STableDataCxt** pDst, STableDataCxt* pSrc, uint64_t uid, uint64_t suid, int32_t vgId, bool rebuildCreateTb);
void qDestroyStmtDataBlock(STableDataCxt* pBlock);
STableMeta* qGetTableMetaInDataBlock(STableDataCxt* pDataBlock);
int32_t qCloneCurrentTbData(STableDataCxt* pDataBlock, SSubmitTbData **pData);
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx);
int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery);
......@@ -105,11 +104,18 @@ void destroyBoundColumnInfo(void* pBoundInfo);
int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf,
int32_t msgBufLen);
void* smlInitHandle(SQuery* pQuery);
void smlDestroyHandle(void* pHandle);
int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta,
void qDestroyBoundColInfo(void* pInfo);
SQuery* smlInitHandle();
int32_t smlBuildRow(STableDataCxt* pTableCxt);
int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void *kv, int32_t index);
STableDataCxt* smlInitTableDataCtx(SQuery* query, STableMeta* pTableMeta);
int32_t smlBindData(SQuery* handle, bool dataFormat, SArray* tags, SArray* colsSchema, SArray* cols, STableMeta* pTableMeta,
char* tableName, const char* sTableName, int32_t sTableNameLen, int32_t ttl, char* msgBuf, int16_t msgBufLen);
int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash);
int rawBlockBindData(SQuery *query, STableMeta* pTableMeta, void* data, SVCreateTbReq* pCreateTb, TAOS_FIELD *fields, int numFields);
int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray);
SArray* serializeVgroupsCreateTableBatch(SHashObj* pVgroupHashmap);
......
include/libs/qcom/query.h
浏览文件 @ d342ebcd
......@@ -163,6 +163,23 @@ typedef struct STargetInfo {
int32_t vgId;
} STargetInfo;
typedef struct SBoundColInfo {
int16_t* pColIndex; // bound index => schema index
int32_t numOfCols;
int32_t numOfBound;
} SBoundColInfo;
typedef struct STableDataCxt {
STableMeta* pMeta;
STSchema* pSchema;
SBoundColInfo boundColsInfo;
SArray* pValues;
SSubmitTbData* pData;
TSKEY lastTs;
bool ordered;
bool duplicateTs;
} STableDataCxt;
typedef int32_t (*__async_send_cb_fn_t)(void* param, SDataBuf* pMsg, int32_t code);
typedef int32_t (*__async_exec_fn_t)(void* param);
......@@ -238,6 +255,7 @@ int32_t dataConverToStr(char* str, int type, void* buf, int32_t bufSize, int32_t
char* parseTagDatatoJson(void* p);
int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst);
int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst);
int32_t cloneSVreateTbReq(SVCreateTbReq* pSrc, SVCreateTbReq** pDst);
void freeVgInfo(SDBVgInfo* vgInfo);
extern int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char** msg, int32_t msgSize, int32_t* msgLen,
......
include/libs/stream/tstream.h
浏览文件 @ d342ebcd
......@@ -103,6 +103,7 @@ typedef struct {
int8_t type;
} SStreamQueueItem;
#if 0
typedef struct {
int8_t type;
int64_t ver;
......@@ -116,6 +117,21 @@ typedef struct {
SArray* dataRefs; // SArray<int32_t*>
SArray* reqs; // SArray<SSubmitReq*>
} SStreamMergedSubmit;
#endif
typedef struct {
int8_t type;
int64_t ver;
int32_t* dataRef;
SPackedData submit;
} SStreamDataSubmit2;
typedef struct {
int8_t type;
int64_t ver;
SArray* dataRefs; // SArray<int32_t*>
SArray* submits; // SArray<SPackedSubmit>
} SStreamMergedSubmit2;
typedef struct {
int8_t type;
......@@ -219,11 +235,11 @@ static FORCE_INLINE void* streamQueueNextItem(SStreamQueue* queue) {
}
}
SStreamDataSubmit* streamDataSubmitNew(SSubmitReq* pReq);
SStreamDataSubmit2* streamDataSubmitNew(SPackedData submit);
void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit);
void streamDataSubmitRefDec(SStreamDataSubmit2* pDataSubmit);
SStreamDataSubmit* streamSubmitRefClone(SStreamDataSubmit* pSubmit);
SStreamDataSubmit2* streamSubmitRefClone(SStreamDataSubmit2* pSubmit);
typedef struct {
char* qmsg;
......@@ -355,14 +371,15 @@ void tFreeSStreamTask(SStreamTask* pTask);
static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem* pItem) {
if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
SStreamDataSubmit* pSubmitClone = streamSubmitRefClone((SStreamDataSubmit*)pItem);
SStreamDataSubmit2* pSubmitClone = streamSubmitRefClone((SStreamDataSubmit2*)pItem);
if (pSubmitClone == NULL) {
qDebug("task %d %p submit enqueue failed since out of memory", pTask->taskId, pTask);
terrno = TSDB_CODE_OUT_OF_MEMORY;
atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__FAILED);
return -1;
}
qDebug("task %d %p submit enqueue %p %p %p", pTask->taskId, pTask, pItem, pSubmitClone, pSubmitClone->data);
qDebug("task %d %p submit enqueue %p %p %p %d %" PRId64, pTask->taskId, pTask, pItem, pSubmitClone,
pSubmitClone->submit.msgStr, pSubmitClone->submit.msgLen, pSubmitClone->submit.ver);
taosWriteQitem(pTask->inputQueue->queue, pSubmitClone);
// qStreamInput(pTask->exec.executor, pSubmitClone);
} else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE ||
......
include/util/tRealloc.h
浏览文件 @ d342ebcd
......@@ -52,11 +52,13 @@ _exit:
return code;
}
static FORCE_INLINE void tFree(uint8_t *pBuf) {
if (pBuf) {
taosMemoryFree(pBuf - sizeof(int64_t));
}
}
#define tFree(BUF) \
do { \
if (BUF) { \
taosMemoryFree((uint8_t *)(BUF) - sizeof(int64_t)); \
(BUF) = NULL; \
} \
} while (0)
#ifdef __cplusplus
}
......
include/util/taoserror.h
浏览文件 @ d342ebcd
......@@ -156,6 +156,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_TSC_QUERY_KILLED TAOS_DEF_ERROR_CODE(0, 0X022D)
#define TSDB_CODE_TSC_NO_EXEC_NODE TAOS_DEF_ERROR_CODE(0, 0X022E)
#define TSDB_CODE_TSC_NOT_STABLE_ERROR TAOS_DEF_ERROR_CODE(0, 0X022F)
#define TSDB_CODE_TSC_STMT_CACHE_ERROR TAOS_DEF_ERROR_CODE(0, 0X0230)
// mnode-common
// #define TSDB_CODE_MND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0300) // 2.x
......
include/util/tarray.h
浏览文件 @ d342ebcd
......@@ -22,19 +22,6 @@
extern "C" {
#endif
#if 0
#define TARRAY(TYPE) \
struct { \
int32_t tarray_size_; \
int32_t tarray_neles_; \
struct TYPE* td_array_data_; \
}
#define TARRAY_SIZE(ARRAY) (ARRAY)->tarray_size_
#define TARRAY_NELES(ARRAY) (ARRAY)->tarray_neles_
#define TARRAY_ELE_AT(ARRAY, IDX) ((ARRAY)->td_array_data_ + idx)
#endif
#define TARRAY_MIN_SIZE 8
#define TARRAY_GET_ELEM(array, index) ((void*)((char*)((array)->pData) + (index) * (array)->elemSize))
#define TARRAY_ELEM_IDX(array, ele) (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
......@@ -46,6 +33,9 @@ typedef struct SArray {
void* pData;
} SArray;
#define TARRAY_SIZE(array) ((array)->size)
#define TARRAY_DATA(array) ((array)->pData)
/**
*
* @param size
......@@ -194,6 +184,13 @@ void taosArrayPopTailBatch(SArray* pArray, size_t cnt);
*/
void taosArrayRemove(SArray* pArray, size_t index);
/**
* remove batch entry from the given index
* @param pArray
* @param index
*/
void taosArrayRemoveBatch(SArray* pArray, size_t index, size_t num, FDelete fp);
/**
* copy the whole array from source to destination
* @param pDst
......
include/util/tdef.h
浏览文件 @ d342ebcd
......@@ -189,12 +189,13 @@ typedef enum ELogicConditionType {
#define TSDB_MAX_COLUMNS 4096
#define TSDB_MIN_COLUMNS 2 // PRIMARY COLUMN(timestamp) + other columns
#define TSDB_NODE_NAME_LEN 64
#define TSDB_TABLE_NAME_LEN 193 // it is a null-terminated string
#define TSDB_TOPIC_NAME_LEN 193 // it is a null-terminated string
#define TSDB_CGROUP_LEN 193 // it is a null-terminated string
#define TSDB_DB_NAME_LEN 65
#define TSDB_DB_FNAME_LEN (TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN + TSDB_NAME_DELIMITER_LEN)
#define TSDB_NODE_NAME_LEN 64
#define TSDB_TABLE_NAME_LEN 193 // it is a null-terminated string
#define TSDB_TOPIC_NAME_LEN 193 // it is a null-terminated string
#define TSDB_CGROUP_LEN 193 // it is a null-terminated string
#define TSDB_USER_CGROUP_LEN (TSDB_USER_LEN + TSDB_CGROUP_LEN) // it is a null-terminated string
#define TSDB_DB_NAME_LEN 65
#define TSDB_DB_FNAME_LEN (TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN + TSDB_NAME_DELIMITER_LEN)
#define TSDB_FUNC_NAME_LEN 65
#define TSDB_FUNC_COMMENT_LEN 1024 * 1024
......
include/util/tencode.h
浏览文件 @ d342ebcd
......@@ -116,6 +116,7 @@ static int32_t tEncodeI64v(SEncoder* pCoder, int64_t val);
static int32_t tEncodeFloat(SEncoder* pCoder, float val);
static int32_t tEncodeDouble(SEncoder* pCoder, double val);
static int32_t tEncodeBinary(SEncoder* pCoder, const uint8_t* val, uint32_t len);
static int32_t tEncodeBinaryEx(SEncoder* pCoder, const uint8_t* val, uint32_t len);
static int32_t tEncodeCStrWithLen(SEncoder* pCoder, const char* val, uint32_t len);
static int32_t tEncodeCStr(SEncoder* pCoder, const char* val);
......
source/client/inc/clientInt.h
浏览文件 @ d342ebcd
......@@ -149,7 +149,6 @@ typedef struct STscObj {
int32_t numOfReqs; // number of sqlObj bound to this connection
SAppInstInfo* pAppInfo;
SHashObj* pRequests;
int8_t schemalessType; // todo remove it, this attribute should be move to request
} STscObj;
typedef struct SResultColumn {
......
source/client/inc/clientLog.h
浏览文件 @ d342ebcd
......@@ -30,7 +30,7 @@ extern "C" {
#define tscDebug(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSC ", DEBUG_DEBUG, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscTrace(...) do { if (cDebugFlag & DEBUG_TRACE) { taosPrintLog("TSC ", DEBUG_TRACE, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscDebugL(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLongString("TSC ", DEBUG_DEBUG, cDebugFlag, __VA_ARGS__); }} while(0)
//#define tscPerf(...) do { if (cDebugFlag & DEBUG_INFO) { taosPrintLog("TSC ", DEBUG_INFO, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscPerf(...) do { if (cDebugFlag & DEBUG_INFO) { taosPrintLog("TSC ", 0, cDebugFlag, __VA_ARGS__); }} while(0)
// clang-format on
#ifdef __cplusplus
......
source/client/inc/clientSml.h 0 → 100644
浏览文件 @ d342ebcd
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_CLIENTSML_H
#define TDENGINE_CLIENTSML_H
#ifdef __cplusplus
extern "C" {
#endif
#include "catalog.h"
#include "clientInt.h"
#include "osThread.h"
#include "query.h"
#include "taos.h"
#include "taoserror.h"
#include "tcommon.h"
#include "tdef.h"
#include "tglobal.h"
#include "tlog.h"
#include "tmsg.h"
#include "tname.h"
#include "ttime.h"
#include "ttypes.h"
#include "cJSON.h"
#if (defined(__GNUC__) && (__GNUC__ >= 3)) || (defined(__INTEL_COMPILER) && (__INTEL_COMPILER >= 800)) || defined(__clang__)
# define expect(expr,value) (__builtin_expect ((expr),(value)) )
#else
# define expect(expr,value) (expr)
#endif
#ifndef likely
#define likely(expr) expect((expr) != 0, 1)
#endif
#ifndef unlikely
#define unlikely(expr) expect((expr) != 0, 0)
#endif
#define SPACE ' '
#define COMMA ','
#define EQUAL '='
#define QUOTE '"'
#define SLASH '\\'
#define JUMP_SPACE(sql, sqlEnd) \
while (sql < sqlEnd) { \
if (unlikely(*sql == SPACE)) \
sql++; \
else \
break; \
}
#define IS_INVALID_COL_LEN(len) ((len) <= 0 || (len) >= TSDB_COL_NAME_LEN)
#define IS_INVALID_TABLE_LEN(len) ((len) <= 0 || (len) >= TSDB_TABLE_NAME_LEN)
#define TS "_ts"
#define TS_LEN 3
#define VALUE "_value"
#define VALUE_LEN 6
#define MAX_RETRY_TIMES 5
typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;
typedef enum {
SCHEMA_ACTION_NULL,
SCHEMA_ACTION_CREATE_STABLE,
SCHEMA_ACTION_ADD_COLUMN,
SCHEMA_ACTION_ADD_TAG,
SCHEMA_ACTION_CHANGE_COLUMN_SIZE,
SCHEMA_ACTION_CHANGE_TAG_SIZE,
} ESchemaAction;
typedef struct {
const void *key;
int32_t keyLen;
void *value;
bool used;
}Node;
typedef struct NodeList{
Node data;
struct NodeList* next;
}NodeList;
typedef struct {
char *measure;
char *tags;
char *cols;
char *timestamp;
int32_t measureLen;
int32_t measureTagsLen;
int32_t tagsLen;
int32_t colsLen;
int32_t timestampLen;
SArray *colArray;
} SSmlLineInfo;
typedef struct {
const char *sTableName; // super table name
int32_t sTableNameLen;
char childTableName[TSDB_TABLE_NAME_LEN];
uint64_t uid;
void *key; // for openTsdb
SArray *tags;
// elements are SHashObj<cols key string, SSmlKv*> for find by key quickly
SArray *cols;
STableDataCxt *tableDataCtx;
} SSmlTableInfo;
typedef struct {
SArray *tags; // save the origin order to create table
SHashObj *tagHash; // elements are <key, index in tags>
SArray *cols;
SHashObj *colHash;
STableMeta *tableMeta;
} SSmlSTableMeta;
typedef struct {
int32_t len;
char *buf;
} SSmlMsgBuf;
typedef struct {
int32_t code;
int32_t lineNum;
int32_t numOfSTables;
int32_t numOfCTables;
int32_t numOfCreateSTables;
int32_t numOfAlterColSTables;
int32_t numOfAlterTagSTables;
int64_t parseTime;
int64_t schemaTime;
int64_t insertBindTime;
int64_t insertRpcTime;
int64_t endTime;
} SSmlCostInfo;
typedef struct {
int64_t id;
SMLProtocolType protocol;
int8_t precision;
bool reRun;
bool dataFormat; // true means that the name and order of keys in each line are the same(only for influx protocol)
bool isRawLine;
int32_t ttl;
NodeList *childTables;
NodeList *superTables;
SHashObj *pVgHash;
STscObj *taos;
SCatalog *pCatalog;
SRequestObj *pRequest;
SQuery *pQuery;
SSmlCostInfo cost;
int32_t lineNum;
SSmlMsgBuf msgBuf;
// cJSON *root; // for parse json
int8_t offset[4];
SSmlLineInfo *lines; // element is SSmlLineInfo
//
SArray *preLineTagKV;
SArray *preLineColKV;
SSmlLineInfo preLine;
STableMeta *currSTableMeta;
STableDataCxt *currTableDataCtx;
bool needModifySchema;
} SSmlHandle;
#define IS_SAME_CHILD_TABLE (elements->measureTagsLen == info->preLine.measureTagsLen \
&& memcmp(elements->measure, info->preLine.measure, elements->measureTagsLen) == 0)
#define IS_SAME_SUPER_TABLE (elements->measureLen == info->preLine.measureLen \
&& memcmp(elements->measure, info->preLine.measure, elements->measureLen) == 0)
#define IS_SAME_KEY (preKV->keyLen == kv.keyLen && memcmp(preKV->key, kv.key, kv.keyLen) == 0)
extern int64_t smlFactorNS[3];
extern int64_t smlFactorS[3];
typedef int32_t (*_equal_fn_sml)(const void *, const void *);
SSmlHandle *smlBuildSmlInfo(TAOS *taos);
void smlDestroyInfo(SSmlHandle *info);
void smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset);
void smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset);
SArray *smlJsonParseTags(char *start, char *end);
bool smlParseNumberOld(SSmlKv *kvVal, SSmlMsgBuf *msg);
void* nodeListGet(NodeList* list, const void *key, int32_t len, _equal_fn_sml fn);
int nodeListSet(NodeList** list, const void *key, int32_t len, void* value, _equal_fn_sml fn);
int nodeListSize(NodeList* list);
bool smlDoubleToInt64OverFlow(double num);
int32_t smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2);
bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg);
int64_t smlGetTimeValue(const char *value, int32_t len, uint8_t fromPrecision, uint8_t toPrecision);
int8_t smlGetTsTypeByLen(int32_t len);
SSmlTableInfo* smlBuildTableInfo(int numRows, const char* measure, int32_t measureLen);
SSmlSTableMeta* smlBuildSTableMeta(bool isDataFormat);
int32_t smlSetCTableName(SSmlTableInfo *oneTable);
STableMeta* smlGetMeta(SSmlHandle *info, const void* measure, int32_t measureLen);
int32_t is_same_child_table_telnet(const void *a, const void *b);
int64_t smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len);
int32_t smlClearForRerun(SSmlHandle *info);
int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg);
int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
int32_t smlParseJSON(SSmlHandle *info, char *payload);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_CLIENTSML_H
source/client/inc/clientStmt.h
浏览文件 @ d342ebcd
......@@ -21,8 +21,6 @@ extern "C" {
#endif
#include "catalog.h"
typedef void STableDataBlocks;
typedef enum {
STMT_TYPE_INSERT = 1,
STMT_TYPE_MULTI_INSERT,
......@@ -43,8 +41,8 @@ typedef enum {
} STMT_STATUS;
typedef struct SStmtTableCache {
STableDataBlocks *pDataBlock;
void *boundTags;
STableDataCxt *pDataCtx;
void *boundTags;
} SStmtTableCache;
typedef struct SStmtQueryResInfo {
......@@ -71,10 +69,11 @@ typedef struct SStmtBindInfo {
} SStmtBindInfo;
typedef struct SStmtExecInfo {
int32_t affectedRows;
SRequestObj *pRequest;
SHashObj *pBlockHash;
bool autoCreateTbl;
int32_t affectedRows;
SRequestObj *pRequest;
SHashObj *pBlockHash;
STableDataCxt *pCurrBlock;
SSubmitTbData *pCurrTbData;
} SStmtExecInfo;
typedef struct SStmtSQLInfo {
......
source/client/src/clientEnv.c
浏览文件 @ d342ebcd
......@@ -76,13 +76,19 @@ static void deregisterRequest(SRequestObj *pRequest) {
"current:%d, app current:%d",
pRequest->self, pTscObj->id, pRequest->requestId, duration / 1000.0, num, currentInst);
tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
"us, exec:%" PRId64 "us, stmtType:%d",
duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd - pRequest->metric.ctgEnd,
pRequest->metric.execEnd - pRequest->metric.semanticEnd, pRequest->stmtType);
if (QUERY_NODE_VNODE_MODIFY_STMT == pRequest->stmtType) {
// tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
// "us, exec:%" PRId64 "us",
// duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
// pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
// pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd);
atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
// tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
// "us, exec:%" PRId64 "us",
// duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
// pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
// pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd);
// atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
} else if (QUERY_NODE_SELECT_STMT == pRequest->stmtType) {
// tscPerf("select duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
// "us, planner:%" PRId64 "us, exec:%" PRId64 "us, reqId:0x%" PRIx64,
......@@ -264,7 +270,6 @@ void *createTscObj(const char *user, const char *auth, const char *db, int32_t c
taosThreadMutexInit(&pObj->mutex, NULL);
pObj->id = taosAddRef(clientConnRefPool, pObj);
pObj->schemalessType = 1;
atomic_add_fetch_64(&pObj->pAppInfo->numOfConns, 1);
......
source/client/src/clientImpl.c
浏览文件 @ d342ebcd
......@@ -239,7 +239,6 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
.pTransporter = pTscObj->pAppInfo->pTransporter,
.pStmtCb = pStmtCb,
.pUser = pTscObj->user,
.schemalessType = pTscObj->schemalessType,
.isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
.enableSysInfo = pTscObj->sysInfo,
.svrVer = pTscObj->sVer,
......@@ -741,47 +740,21 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
}
int32_t handleSubmitExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
int32_t code = 0;
SArray* pArray = NULL;
SSubmitRsp* pRsp = (SSubmitRsp*)res;
if (pRsp->nBlocks <= 0) {
taosMemoryFreeClear(pRsp->pBlocks);
SArray* pArray = NULL;
SSubmitRsp2* pRsp = (SSubmitRsp2*)res;
if (NULL == pRsp->aCreateTbRsp) {
return TSDB_CODE_SUCCESS;
}
pArray = taosArrayInit(pRsp->nBlocks, sizeof(STbSVersion));
if (NULL == pArray) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return TSDB_CODE_OUT_OF_MEMORY;
}
for (int32_t i = 0; i < pRsp->nBlocks; ++i) {
SSubmitBlkRsp* blk = pRsp->pBlocks + i;
if (blk->pMeta) {
handleCreateTbExecRes(blk->pMeta, pCatalog);
tFreeSTableMetaRsp(blk->pMeta);
taosMemoryFreeClear(blk->pMeta);
}
if (NULL == blk->tblFName || 0 == blk->tblFName[0]) {
continue;
int32_t tbNum = taosArrayGetSize(pRsp->aCreateTbRsp);
for (int32_t i = 0; i < tbNum; ++i) {
SVCreateTbRsp* pTbRsp = (SVCreateTbRsp*)taosArrayGet(pRsp->aCreateTbRsp, i);
if (pTbRsp->pMeta) {
handleCreateTbExecRes(pTbRsp->pMeta, pCatalog);
}
STbSVersion tbSver = {.tbFName = blk->tblFName, .sver = blk->sver};
taosArrayPush(pArray, &tbSver);
}
SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
.requestId = pRequest->requestId,
.requestObjRefId = pRequest->self,
.mgmtEps = *epset};
code = catalogChkTbMetaVersion(pCatalog, &conn, pArray);
_return:
taosArrayDestroy(pArray);
return code;
return TSDB_CODE_SUCCESS;
}
int32_t handleQueryExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
......
source/client/src/clientMain.c
浏览文件 @ d342ebcd
......@@ -866,7 +866,6 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
.pTransporter = pTscObj->pAppInfo->pTransporter,
.pStmtCb = NULL,
.pUser = pTscObj->user,
.schemalessType = pTscObj->schemalessType,
.isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
.enableSysInfo = pTscObj->sysInfo,
.async = true,
......
source/client/src/clientSml.c
浏览文件 @ d342ebcd
此差异已折叠。
source/client/src/clientSmlJson.c 0 → 100644
浏览文件 @ d342ebcd
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "clientSml.h"
#define JUMP_JSON_SPACE(start) \
while(*(start)){\
if(unlikely(*(start) > 32))\
break;\
else\
(start)++;\
}
SArray *smlJsonParseTags(char *start, char *end){
SArray *tags = taosArrayInit(4, sizeof(SSmlKv));
while(start < end){
SSmlKv kv = {0};
kv.type = TSDB_DATA_TYPE_NCHAR;
bool isInQuote = false;
while(start < end){
if(unlikely(!isInQuote && *start == '"')){
start++;
kv.key = start;
isInQuote = true;
continue;
}
if(unlikely(isInQuote && *start == '"')){
kv.keyLen = start - kv.key;
start++;
break;
}
start++;
}
bool hasColon = false;
while(start < end){
if(unlikely(!hasColon && *start == ':')){
start++;
hasColon = true;
continue;
}
if(unlikely(hasColon && kv.value == NULL && (*start > 32 && *start != '"'))){
kv.value = start;
start++;
continue;
}
if(unlikely(hasColon && kv.value != NULL && (*start == '"' || *start == ',' || *start == '}'))){
kv.length = start - kv.value;
taosArrayPush(tags, &kv);
start++;
break;
}
start++;
}
}
return tags;
}
static int32_t smlParseTagsFromJSON(SSmlHandle *info, SSmlLineInfo *elements) {
int32_t ret = TSDB_CODE_SUCCESS;
if(is_same_child_table_telnet(elements, &info->preLine) == 0){
return TSDB_CODE_SUCCESS;
}
bool isSameMeasure = IS_SAME_SUPER_TABLE;
int cnt = 0;
SArray *preLineKV = info->preLineTagKV;
bool isSuperKVInit = true;
SArray *superKV = NULL;
if(info->dataFormat){
if(unlikely(!isSameMeasure)){
SSmlSTableMeta *sMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
if(unlikely(sMeta == NULL)){
sMeta = smlBuildSTableMeta(info->dataFormat);
STableMeta * pTableMeta = smlGetMeta(info, elements->measure, elements->measureLen);
sMeta->tableMeta = pTableMeta;
if(pTableMeta == NULL){
info->dataFormat = false;
info->reRun = true;
return TSDB_CODE_SUCCESS;
}
nodeListSet(&info->superTables, elements->measure, elements->measureLen, sMeta, NULL);
}
info->currSTableMeta = sMeta->tableMeta;
superKV = sMeta->tags;
if(unlikely(taosArrayGetSize(superKV) == 0)){
isSuperKVInit = false;
}
taosArraySetSize(preLineKV, 0);
}
}else{
taosArraySetSize(preLineKV, 0);
}
SArray *tags = smlJsonParseTags(elements->tags, elements->tags + elements->tagsLen);
int32_t tagNum = taosArrayGetSize(tags);
for (int32_t i = 0; i < tagNum; ++i) {
SSmlKv kv = *(SSmlKv*)taosArrayGet(tags, i);
if(info->dataFormat){
if(unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)){
info->dataFormat = false;
info->reRun = true;
taosArrayDestroy(tags);
return TSDB_CODE_SUCCESS;
}
if(isSameMeasure){
if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
info->dataFormat = false;
info->reRun = true;
taosArrayDestroy(tags);
return TSDB_CODE_SUCCESS;
}
SSmlKv *preKV = (SSmlKv *)taosArrayGet(preLineKV, cnt);
if(unlikely(kv.length > preKV->length)){
preKV->length = kv.length;
SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
ASSERT(tableMeta != NULL);
SSmlKv *oldKV = (SSmlKv *)taosArrayGet(tableMeta->tags, cnt);
oldKV->length = kv.length;
info->needModifySchema = true;
}
if(unlikely(!IS_SAME_KEY)){
info->dataFormat = false;
info->reRun = true;
taosArrayDestroy(tags);
return TSDB_CODE_SUCCESS;
}
}else{
if(isSuperKVInit){
if(unlikely(cnt >= taosArrayGetSize(superKV))) {
info->dataFormat = false;
info->reRun = true;
taosArrayDestroy(tags);
return TSDB_CODE_SUCCESS;
}
SSmlKv *preKV = (SSmlKv *)taosArrayGet(superKV, cnt);
if(unlikely(kv.length > preKV->length)) {
preKV->length = kv.length;
}else{
kv.length = preKV->length;
}
info->needModifySchema = true;
if(unlikely(!IS_SAME_KEY)){
info->dataFormat = false;
info->reRun = true;
taosArrayDestroy(tags);
return TSDB_CODE_SUCCESS;
}
}else{
taosArrayPush(superKV, &kv);
}
taosArrayPush(preLineKV, &kv);
}
}else{
taosArrayPush(preLineKV, &kv);
}
cnt++;
}
taosArrayDestroy(tags);
SSmlTableInfo *tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements, POINTER_BYTES, is_same_child_table_telnet);
if (unlikely(tinfo == NULL)) {
tinfo = smlBuildTableInfo(1, elements->measure, elements->measureLen);
if (unlikely(!tinfo)) {
return TSDB_CODE_OUT_OF_MEMORY;
}
tinfo->tags = taosArrayDup(preLineKV, NULL);
smlSetCTableName(tinfo);
if (info->dataFormat) {
info->currSTableMeta->uid = tinfo->uid;
tinfo->tableDataCtx = smlInitTableDataCtx(info->pQuery, info->currSTableMeta);
if (tinfo->tableDataCtx == NULL) {
smlBuildInvalidDataMsg(&info->msgBuf, "smlInitTableDataCtx error", NULL);
return TSDB_CODE_SML_INVALID_DATA;
}
}
SSmlLineInfo *key = (SSmlLineInfo *)taosMemoryMalloc(sizeof(SSmlLineInfo));
*key = *elements;
tinfo->key = key;
nodeListSet(&info->childTables, key, POINTER_BYTES, tinfo, is_same_child_table_telnet);
}
if (info->dataFormat) info->currTableDataCtx = tinfo->tableDataCtx;
return ret;
}
static char* smlJsonGetObj(char *payload){
int leftBracketCnt = 0;
while(*payload) {
if (unlikely(*payload == '{')) {
leftBracketCnt++;
payload++;
continue;
}
if (unlikely(*payload == '}')) {
leftBracketCnt--;
payload++;
if (leftBracketCnt == 0) {
return payload;
} else if (leftBracketCnt < 0) {
return NULL;
}
continue;
}
payload++;
}
return NULL;
}
void smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset){
int index = 0;
while(*(*start)){
if((*start)[0] != '"'){
(*start)++;
continue;
}
if(unlikely(index >= 4)) {
uError("index >= 4, %s", *start)
break;
}
char *sTmp = *start;
if((*start)[1] == 'm' && (*start)[2] == 'e' && (*start)[3] == 't'
&& (*start)[4] == 'r' && (*start)[5] == 'i' && (*start)[6] == 'c' && (*start)[7] == '"'){
(*start) += 8;
bool isInQuote = false;
while(*(*start)){
if(unlikely(!isInQuote && *(*start) == '"')){
(*start)++;
offset[index++] = *start - sTmp;
element->measure = (*start);
isInQuote = true;
continue;
}
if(unlikely(isInQuote && *(*start) == '"')){
element->measureLen = (*start) - element->measure;
break;
}
(*start)++;
}
}else if((*start)[1] == 't' && (*start)[2] == 'i' && (*start)[3] == 'm'
&& (*start)[4] == 'e' && (*start)[5] == 's' && (*start)[6] == 't'
&& (*start)[7] == 'a' && (*start)[8] == 'm' && (*start)[9] == 'p' && (*start)[10] == '"'){
(*start) += 11;
bool hasColon = false;
while(*(*start)){
if(unlikely(!hasColon && *(*start) == ':')){
(*start)++;
JUMP_JSON_SPACE((*start))
offset[index++] = *start - sTmp;
element->timestamp = (*start);
hasColon = true;
continue;
}
if(unlikely(hasColon && (*(*start) == ',' || *(*start) == '}' || (*(*start)) <= 32))){
element->timestampLen = (*start) - element->timestamp;
break;
}
(*start)++;
}
}else if((*start)[1] == 'v' && (*start)[2] == 'a' && (*start)[3] == 'l'
&& (*start)[4] == 'u' && (*start)[5] == 'e' && (*start)[6] == '"'){
(*start) += 7;
bool hasColon = false;
while(*(*start)){
if(unlikely(!hasColon && *(*start) == ':')){
(*start)++;
JUMP_JSON_SPACE((*start))
offset[index++] = *start - sTmp;
element->cols = (*start);
hasColon = true;
continue;
}
if(unlikely(hasColon && (*(*start) == ',' || *(*start) == '}' || (*(*start)) <= 32))){
element->colsLen = (*start) - element->cols;
break;
}
(*start)++;
}
}else if((*start)[1] == 't' && (*start)[2] == 'a' && (*start)[3] == 'g'
&& (*start)[4] == 's' && (*start)[5] == '"'){
(*start) += 6;
while(*(*start)){
if(unlikely(*(*start) == ':')){
(*start)++;
JUMP_JSON_SPACE((*start))
offset[index++] = *start - sTmp;
element->tags = (*start);
char* tmp = smlJsonGetObj((*start));
if(tmp){
element->tagsLen = tmp - (*start);
*start = tmp;
}
break;
}
(*start)++;
}
}
if(*(*start) == '}'){
(*start)++;
break;
}
(*start)++;
}
}
void smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset){
int index = 0;
while(*(*start)){
if((*start)[0] != '"'){
(*start)++;
continue;
}
if(unlikely(index >= 4)) {
uError("index >= 4, %s", *start)
break;
}
if((*start)[1] == 'm'){
(*start) += offset[index++];
element->measure = *start;
while(*(*start)){
if(unlikely(*(*start) == '"')){
element->measureLen = (*start) - element->measure;
break;
}
(*start)++;
}
}else if((*start)[1] == 't' && (*start)[2] == 'i'){
(*start) += offset[index++];
element->timestamp = *start;
while(*(*start)){
if(unlikely(*(*start) == ',' || *(*start) == '}' || (*(*start)) <= 32)){
element->timestampLen = (*start) - element->timestamp;
break;
}
(*start)++;
}
}else if((*start)[1] == 'v'){
(*start) += offset[index++];
element->cols = *start;
while(*(*start)){
if(unlikely( *(*start) == ',' || *(*start) == '}' || (*(*start)) <= 32)){
element->colsLen = (*start) - element->cols;
break;
}
(*start)++;
}
}else if((*start)[1] == 't' && (*start)[2] == 'a'){
(*start) += offset[index++];
element->tags = (*start);
char* tmp = smlJsonGetObj((*start));
if(tmp){
element->tagsLen = tmp - (*start);
*start = tmp;
}
break;
}
if(*(*start) == '}'){
(*start)++;
break;
}
(*start)++;
}
}
static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo *elements) {
int32_t ret = TSDB_CODE_SUCCESS;
if(info->offset[0] == 0){
smlJsonParseObjFirst(start, elements, info->offset);
}else{
smlJsonParseObj(start, elements, info->offset);
}
if(**start == '\0') return TSDB_CODE_SUCCESS;
SSmlKv kv = {.key = VALUE, .keyLen = VALUE_LEN, .value = elements->cols, .length = (size_t)elements->colsLen};
if (smlParseNumber(&kv, &info->msgBuf)) {
kv.length = (int16_t)tDataTypes[kv.type].bytes;
}else{
return TSDB_CODE_TSC_INVALID_VALUE;
}
// Parse tags
ret = smlParseTagsFromJSON(info, elements);
if (unlikely(ret)) {
uError("OTD:0x%" PRIx64 " Unable to parse tags from JSON payload", info->id);
return ret;
}
if(unlikely(info->reRun)){
return TSDB_CODE_SUCCESS;
}
// Parse timestamp
// notice!!! put ts back to tag to ensure get meta->precision
int64_t ts = smlParseOpenTsdbTime(info, elements->timestamp, elements->timestampLen);
if (unlikely(ts < 0)) {
uError("OTD:0x%" PRIx64 " Unable to parse timestamp from JSON payload", info->id);
return TSDB_CODE_INVALID_TIMESTAMP;
}
SSmlKv kvTs = { .key = TS, .keyLen = TS_LEN, .type = TSDB_DATA_TYPE_TIMESTAMP, .i = ts, .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
if(info->dataFormat){
ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kvTs, 0);
if(ret == TSDB_CODE_SUCCESS){
ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 1);
}
if(ret == TSDB_CODE_SUCCESS){
ret = smlBuildRow(info->currTableDataCtx);
}
if (unlikely(ret != TSDB_CODE_SUCCESS)) {
smlBuildInvalidDataMsg(&info->msgBuf, "smlBuildCol error", NULL);
return ret;
}
}else{
if(elements->colArray == NULL){
elements->colArray = taosArrayInit(16, sizeof(SSmlKv));
}
taosArrayPush(elements->colArray, &kvTs);
taosArrayPush(elements->colArray, &kv);
}
info->preLine = *elements;
return TSDB_CODE_SUCCESS;
}
int32_t smlParseJSON(SSmlHandle *info, char *payload) {
int32_t payloadNum = 1 << 15;
int32_t ret = TSDB_CODE_SUCCESS;
int cnt = 0;
char *dataPointStart = payload;
while (1) {
if(info->dataFormat) {
SSmlLineInfo element = {0};
ret = smlParseJSONString(info, &dataPointStart, &element);
}else{
if(cnt >= payloadNum){
payloadNum = payloadNum << 1;
void* tmp = taosMemoryRealloc(info->lines, payloadNum * sizeof(SSmlLineInfo));
if(tmp != NULL){
info->lines = (SSmlLineInfo*)tmp;
}
}
ret = smlParseJSONString(info, &dataPointStart, info->lines + cnt);
}
if (unlikely(ret != TSDB_CODE_SUCCESS)) {
uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
return ret;
}
if(*dataPointStart == '\0') break;
if(unlikely(info->reRun)){
cnt = 0;
dataPointStart = payload;
info->lineNum = payloadNum;
ret = smlClearForRerun(info);
if(ret != TSDB_CODE_SUCCESS){
return ret;
}
continue;
}
cnt++;
}
info->lineNum = cnt;
return TSDB_CODE_SUCCESS;
}
source/client/src/clientSmlLine.c 0 → 100644
浏览文件 @ d342ebcd
此差异已折叠。
source/client/src/clientSmlTelnet.c 0 → 100644
浏览文件 @ d342ebcd
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "clientSml.h"
int32_t is_same_child_table_telnet(const void *a, const void *b){
SSmlLineInfo *t1 = (SSmlLineInfo *)a;
SSmlLineInfo *t2 = (SSmlLineInfo *)b;
return (((t1->measureLen == t2->measureLen) && memcmp(t1->measure, t2->measure, t1->measureLen) == 0)
&& ((t1->tagsLen == t2->tagsLen) && memcmp(t1->tags, t2->tags, t1->tagsLen) == 0)) ? 0 : 1;
}
int64_t smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len) {
uint8_t toPrecision = info->currSTableMeta ? info->currSTableMeta->tableInfo.precision : TSDB_TIME_PRECISION_NANO;
if (unlikely(!data)) {
smlBuildInvalidDataMsg(&info->msgBuf, "timestamp can not be null", NULL);
return -1;
}
if (unlikely(len == 1 && data[0] == '0')) {
return taosGetTimestampNs()/smlFactorNS[toPrecision];
}
int8_t fromPrecision = smlGetTsTypeByLen(len);
if (unlikely(fromPrecision == -1)) {
smlBuildInvalidDataMsg(&info->msgBuf,
"timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", data);
return -1;
}
int64_t ts = smlGetTimeValue(data, len, fromPrecision, toPrecision);
if (unlikely(ts == -1)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
return -1;
}
return ts;
}
static void smlParseTelnetElement(char **sql, char *sqlEnd, char **data, int32_t *len) {
while (*sql < sqlEnd) {
if (unlikely((**sql != SPACE && !(*data)))) {
*data = *sql;
} else if (unlikely(**sql == SPACE && *data)) {
*len = *sql - *data;
break;
}
(*sql)++;
}
}
static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
if(is_same_child_table_telnet(elements, &info->preLine) == 0){
return TSDB_CODE_SUCCESS;
}
bool isSameMeasure = IS_SAME_SUPER_TABLE;
int cnt = 0;
SArray *preLineKV = info->preLineTagKV;
bool isSuperKVInit = true;
SArray *superKV = NULL;
if(info->dataFormat){
if(!isSameMeasure){
SSmlSTableMeta *sMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
if(unlikely(sMeta == NULL)){
sMeta = smlBuildSTableMeta(info->dataFormat);
STableMeta * pTableMeta = smlGetMeta(info, elements->measure, elements->measureLen);
sMeta->tableMeta = pTableMeta;
if(pTableMeta == NULL){
info->dataFormat = false;
info->reRun = true;
return TSDB_CODE_SUCCESS;
}
nodeListSet(&info->superTables, elements->measure, elements->measureLen, sMeta, NULL);
}
info->currSTableMeta = sMeta->tableMeta;
superKV = sMeta->tags;
if(unlikely(taosArrayGetSize(superKV) == 0)){
isSuperKVInit = false;
}
taosArraySetSize(preLineKV, 0);
}
}else{
taosArraySetSize(preLineKV, 0);
}
const char *sql = data;
while (sql < sqlEnd) {
JUMP_SPACE(sql, sqlEnd)
if (unlikely(*sql == '\0')) break;
const char *key = sql;
size_t keyLen = 0;
// parse key
while (sql < sqlEnd) {
if (unlikely(*sql == SPACE)) {
smlBuildInvalidDataMsg(msg, "invalid data", sql);
return TSDB_CODE_SML_INVALID_DATA;
}
if (unlikely(*sql == EQUAL)) {
keyLen = sql - key;
sql++;
break;
}
sql++;
}
if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
}
// if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
// smlBuildInvalidDataMsg(msg, "dumplicate key", key);
// return TSDB_CODE_TSC_DUP_NAMES;
// }
// parse value
const char *value = sql;
size_t valueLen = 0;
while (sql < sqlEnd) {
// parse value
if (unlikely(*sql == SPACE)) {
break;
}
if (unlikely(*sql == EQUAL)) {
smlBuildInvalidDataMsg(msg, "invalid data", sql);
return TSDB_CODE_SML_INVALID_DATA;
}
sql++;
}
valueLen = sql - value;
if (unlikely(valueLen == 0)) {
smlBuildInvalidDataMsg(msg, "invalid value", value);
return TSDB_CODE_TSC_INVALID_VALUE;
}
if (unlikely(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
}
SSmlKv kv = {.key = key, .keyLen = keyLen, .type = TSDB_DATA_TYPE_NCHAR, .value = value, .length = valueLen};
if(info->dataFormat){
if(unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)){
info->dataFormat = false;
info->reRun = true;
return TSDB_CODE_SUCCESS;
}
if(isSameMeasure){
if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
info->dataFormat = false;
info->reRun = true;
return TSDB_CODE_SUCCESS;
}
SSmlKv *preKV = (SSmlKv *)taosArrayGet(preLineKV, cnt);
if(unlikely(kv.length > preKV->length)){
preKV->length = kv.length;
SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
ASSERT(tableMeta != NULL);
SSmlKv *oldKV = (SSmlKv *)taosArrayGet(tableMeta->tags, cnt);
oldKV->length = kv.length;
info->needModifySchema = true;
}
if(unlikely(!IS_SAME_KEY)){
info->dataFormat = false;
info->reRun = true;
return TSDB_CODE_SUCCESS;
}
}else{
if(isSuperKVInit){
if(unlikely(cnt >= taosArrayGetSize(superKV))) {
info->dataFormat = false;
info->reRun = true;
return TSDB_CODE_SUCCESS;
}
SSmlKv *preKV = (SSmlKv *)taosArrayGet(superKV, cnt);
if(unlikely(kv.length > preKV->length)) {
preKV->length = kv.length;
}else{
kv.length = preKV->length;
}
info->needModifySchema = true;
if(unlikely(!IS_SAME_KEY)){
info->dataFormat = false;
info->reRun = true;
return TSDB_CODE_SUCCESS;
}
}else{
taosArrayPush(superKV, &kv);
}
taosArrayPush(preLineKV, &kv);
}
}else{
taosArrayPush(preLineKV, &kv);
}
cnt++;
}
SSmlTableInfo *tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements, POINTER_BYTES, is_same_child_table_telnet);
if (unlikely(tinfo == NULL)) {
tinfo = smlBuildTableInfo(1, elements->measure, elements->measureLen);
if (!tinfo) {
return TSDB_CODE_OUT_OF_MEMORY;
}
tinfo->tags = taosArrayDup(preLineKV, NULL);
smlSetCTableName(tinfo);
if (info->dataFormat) {
info->currSTableMeta->uid = tinfo->uid;
tinfo->tableDataCtx = smlInitTableDataCtx(info->pQuery, info->currSTableMeta);
if (tinfo->tableDataCtx == NULL) {
smlBuildInvalidDataMsg(&info->msgBuf, "smlInitTableDataCtx error", NULL);
return TSDB_CODE_SML_INVALID_DATA;
}
}
SSmlLineInfo *key = (SSmlLineInfo *)taosMemoryMalloc(sizeof(SSmlLineInfo));
*key = *elements;
tinfo->key = key;
nodeListSet(&info->childTables, key, POINTER_BYTES, tinfo, is_same_child_table_telnet);
}
if (info->dataFormat) info->currTableDataCtx = tinfo->tableDataCtx;
return TSDB_CODE_SUCCESS;
}
// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements) {
if (!sql) return TSDB_CODE_SML_INVALID_DATA;
// parse metric
smlParseTelnetElement(&sql, sqlEnd, &elements->measure, &elements->measureLen);
if (unlikely((!(elements->measure) || IS_INVALID_TABLE_LEN(elements->measureLen)))) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
}
// parse timestamp
smlParseTelnetElement(&sql, sqlEnd, &elements->timestamp, &elements->timestampLen);
if (unlikely(!elements->timestamp || elements->timestampLen == 0)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
return TSDB_CODE_SML_INVALID_DATA;
}
bool needConverTime = false; // get TS before parse tag(get meta), so need conver time
if(info->dataFormat && info->currSTableMeta == NULL){
needConverTime = true;
}
int64_t ts = smlParseOpenTsdbTime(info, elements->timestamp, elements->timestampLen);
if (unlikely(ts < 0)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
return TSDB_CODE_INVALID_TIMESTAMP;
}
SSmlKv kvTs = { .key = TS, .keyLen = TS_LEN, .type = TSDB_DATA_TYPE_TIMESTAMP, .i = ts, .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
// parse value
smlParseTelnetElement(&sql, sqlEnd, &elements->cols, &elements->colsLen);
if (unlikely(!elements->cols || elements->colsLen == 0)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
return TSDB_CODE_TSC_INVALID_VALUE;
}
SSmlKv kv = {.key = VALUE, .keyLen = VALUE_LEN, .value = elements->cols, .length = (size_t)elements->colsLen};
if (smlParseValue(&kv, &info->msgBuf) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_TSC_INVALID_VALUE;
}
JUMP_SPACE(sql, sqlEnd)
elements->tags = sql;
elements->tagsLen = sqlEnd - sql;
if (unlikely(!elements->tags || elements->tagsLen == 0)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
return TSDB_CODE_TSC_INVALID_VALUE;
}
int ret = smlParseTelnetTags(info, sql, sqlEnd, elements, &info->msgBuf);
if (unlikely(ret != TSDB_CODE_SUCCESS)) {
return ret;
}
if(unlikely(info->reRun)){
return TSDB_CODE_SUCCESS;
}
if(info->dataFormat){
if(needConverTime) {
kvTs.i = convertTimePrecision(kvTs.i, TSDB_TIME_PRECISION_NANO, info->currSTableMeta->tableInfo.precision);
}
ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kvTs, 0);
if(ret == TSDB_CODE_SUCCESS){
ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 1);
}
if(ret == TSDB_CODE_SUCCESS){
ret = smlBuildRow(info->currTableDataCtx);
}
if (unlikely(ret != TSDB_CODE_SUCCESS)) {
smlBuildInvalidDataMsg(&info->msgBuf, "smlBuildCol error", NULL);
return ret;
}
}else{
if(elements->colArray == NULL){
elements->colArray = taosArrayInit(16, sizeof(SSmlKv));
}
taosArrayPush(elements->colArray, &kvTs);
taosArrayPush(elements->colArray, &kv);
}
info->preLine = *elements;
return TSDB_CODE_SUCCESS;
}
\ No newline at end of file
source/client/src/clientStmt.c
浏览文件 @ d342ebcd
......@@ -152,9 +152,10 @@ int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
return TSDB_CODE_SUCCESS;
}
int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName, bool autoCreateTbl) {
int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName,
bool autoCreateTbl) {
STscStmt* pStmt = (STscStmt*)stmt;
char tbFName[TSDB_TABLE_FNAME_LEN];
char tbFName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(tbName, tbFName);
memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
......@@ -171,12 +172,11 @@ int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags,
return TSDB_CODE_SUCCESS;
}
int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash, bool autoCreateTbl) {
int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
STscStmt* pStmt = (STscStmt*)stmt;
pStmt->sql.pVgHash = pVgHash;
pStmt->exec.pBlockHash = pBlockHash;
pStmt->exec.autoCreateTbl = autoCreateTbl;
return TSDB_CODE_SUCCESS;
}
......@@ -186,7 +186,7 @@ int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SNam
STscStmt* pStmt = (STscStmt*)stmt;
STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl));
STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash, autoCreateTbl));
STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
pStmt->sql.autoCreateTbl = autoCreateTbl;
......@@ -214,16 +214,16 @@ int32_t stmtCacheBlock(STscStmt* pStmt) {
return TSDB_CODE_SUCCESS;
}
STableDataBlocks** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (!pSrc) {
return TSDB_CODE_OUT_OF_MEMORY;
}
STableDataBlocks* pDst = NULL;
STableDataCxt* pDst = NULL;
STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc));
STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
SStmtTableCache cache = {
.pDataBlock = pDst,
.pDataCtx = pDst,
.boundTags = pStmt->bInfo.boundTags,
};
......@@ -241,6 +241,8 @@ int32_t stmtCacheBlock(STscStmt* pStmt) {
}
int32_t stmtParseSql(STscStmt* pStmt) {
pStmt->exec.pCurrBlock = NULL;
SStmtCallback stmtCb = {
.pStmt = pStmt,
.getTbNameFn = stmtGetTbName,
......@@ -273,7 +275,7 @@ int32_t stmtCleanBindInfo(STscStmt* pStmt) {
pStmt->bInfo.tbName[0] = 0;
pStmt->bInfo.tbFName[0] = 0;
if (!pStmt->bInfo.tagsCached) {
destroyBoundColumnInfo(pStmt->bInfo.boundTags);
qDestroyBoundColInfo(pStmt->bInfo.boundTags);
taosMemoryFreeClear(pStmt->bInfo.boundTags);
}
memset(pStmt->bInfo.stbFName, 0, TSDB_TABLE_FNAME_LEN);
......@@ -289,29 +291,25 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
size_t keyLen = 0;
void* pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
while (pIter) {
STableDataBlocks* pBlocks = *(STableDataBlocks**)pIter;
char* key = taosHashGetKey(pIter, &keyLen);
STableMeta* pMeta = qGetTableMetaInDataBlock(pBlocks);
STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
char* key = taosHashGetKey(pIter, &keyLen);
STableMeta* pMeta = qGetTableMetaInDataBlock(pBlocks);
if (keepTable && (strlen(pStmt->bInfo.tbFName) == keyLen) && strncmp(pStmt->bInfo.tbFName, key, keyLen) == 0) {
STMT_ERR_RET(qResetStmtDataBlock(pBlocks, true));
if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
ASSERT(NULL == pBlocks->pData);
TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
continue;
}
if (STMT_TYPE_MULTI_INSERT == pStmt->sql.type) {
qFreeStmtDataBlock(pBlocks);
} else {
qDestroyStmtDataBlock(pBlocks);
}
qDestroyStmtDataBlock(pBlocks);
taosHashRemove(pStmt->exec.pBlockHash, key, keyLen);
pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
}
pStmt->exec.autoCreateTbl = false;
if (keepTable) {
return TSDB_CODE_SUCCESS;
}
......@@ -319,6 +317,9 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
taosHashCleanup(pStmt->exec.pBlockHash);
pStmt->exec.pBlockHash = NULL;
tDestroySSubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
taosMemoryFreeClear(pStmt->exec.pCurrTbData);
STMT_ERR_RET(stmtCleanBindInfo(pStmt));
return TSDB_CODE_SUCCESS;
......@@ -337,8 +338,8 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
while (pIter) {
SStmtTableCache* pCache = (SStmtTableCache*)pIter;
qDestroyStmtDataBlock(pCache->pDataBlock);
destroyBoundColumnInfo(pCache->boundTags);
qDestroyStmtDataBlock(pCache->pDataCtx);
qDestroyBoundColInfo(pCache->boundTags);
taosMemoryFreeClear(pCache->boundTags);
pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
......@@ -354,7 +355,8 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
return TSDB_CODE_SUCCESS;
}
int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks* pDataBlock, STableDataBlocks** newBlock, uint64_t uid) {
int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
uint64_t suid) {
SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
SVgroupInfo vgInfo = {0};
SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
......@@ -366,7 +368,9 @@ int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks* pDataBlock, STab
STMT_ERR_RET(
taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo)));
STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, vgInfo.vgId));
STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgInfo.vgId, pStmt->sql.autoCreateTbl));
STMT_DLOG("tableDataCxt rebuilt, uid:%" PRId64 ", vgId:%d", uid, vgInfo.vgId);
return TSDB_CODE_SUCCESS;
}
......@@ -375,12 +379,13 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
pStmt->bInfo.needParse = true;
pStmt->bInfo.inExecCache = false;
STableDataBlocks* pBlockInExec =
taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (pBlockInExec) {
STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (pCxtInExec) {
pStmt->bInfo.needParse = false;
pStmt->bInfo.inExecCache = true;
pStmt->exec.pCurrBlock = *pCxtInExec;
if (pStmt->sql.autoCreateTbl) {
tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
return TSDB_CODE_SUCCESS;
......@@ -407,18 +412,18 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
if (pCache) {
pStmt->bInfo.needParse = false;
pStmt->exec.autoCreateTbl = true;
pStmt->bInfo.tbUid = 0;
STableDataBlocks* pNewBlock = NULL;
STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, 0));
STableDataCxt* pNewBlock = NULL;
STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, 0, pStmt->bInfo.tbSuid));
if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
POINTER_BYTES)) {
STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
pStmt->exec.pCurrBlock = pNewBlock;
tscDebug("reuse stmt block for tb %s in sqlBlock, suid:0x%" PRIx64, pStmt->bInfo.tbFName, pStmt->bInfo.tbSuid);
return TSDB_CODE_SUCCESS;
......@@ -489,14 +494,16 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
pStmt->bInfo.boundTags = pCache->boundTags;
pStmt->bInfo.tagsCached = true;
STableDataBlocks* pNewBlock = NULL;
STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, uid));
STableDataCxt* pNewBlock = NULL;
STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid));
if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
POINTER_BYTES)) {
STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
pStmt->exec.pCurrBlock = pNewBlock;
tscDebug("tb %s in sqlBlock list, set to current", pStmt->bInfo.tbFName);
return TSDB_CODE_SUCCESS;
......@@ -614,8 +621,8 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
return TSDB_CODE_SUCCESS;
}
STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataCxt** pDataBlock =
(STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_APP_ERROR);
......@@ -626,8 +633,6 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
pStmt->exec.pRequest->msgBufLen));
pStmt->exec.autoCreateTbl = true;
return TSDB_CODE_SUCCESS;
}
......@@ -637,8 +642,8 @@ int stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
}
STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataCxt** pDataBlock =
(STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_APP_ERROR);
......@@ -655,8 +660,8 @@ int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
}
STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataCxt** pDataBlock =
(STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_APP_ERROR);
......@@ -729,11 +734,18 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
return TSDB_CODE_SUCCESS;
}
STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_APP_ERROR);
STableDataCxt** pDataBlock = NULL;
if (pStmt->exec.pCurrBlock) {
pDataBlock = &pStmt->exec.pCurrBlock;
} else {
pDataBlock =
(STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
}
pStmt->exec.pCurrBlock = *pDataBlock;
}
if (colIdx < 0) {
......@@ -779,10 +791,10 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
int32_t code = 0;
int32_t finalCode = 0;
size_t keyLen = 0;
STableDataBlocks** pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
void* pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
while (pIter) {
STableDataBlocks* pBlock = *pIter;
char* key = taosHashGetKey(pIter, &keyLen);
STableDataCxt* pBlock = *(STableDataCxt**)pIter;
char* key = taosHashGetKey(pIter, &keyLen);
STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
if (pMeta->uid) {
......@@ -848,7 +860,7 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
pMeta->uid = pTableMeta->uid;
pStmt->bInfo.tbUid = pTableMeta->uid;
taosMemoryFree(pTableMeta);
taosMemoryFree(pTableMeta);
}
pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
......@@ -861,7 +873,6 @@ int stmtExec(TAOS_STMT* stmt) {
STscStmt* pStmt = (STscStmt*)stmt;
int32_t code = 0;
SSubmitRsp* pRsp = NULL;
bool autoCreateTbl = pStmt->exec.autoCreateTbl;
STMT_DLOG_E("start to exec");
......@@ -870,8 +881,13 @@ int stmtExec(TAOS_STMT* stmt) {
if (STMT_TYPE_QUERY == pStmt->sql.type) {
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
} else {
tDestroySSubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
taosMemoryFreeClear(pStmt->exec.pCurrTbData);
STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));
STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, (autoCreateTbl ? (void**)&pRsp : NULL));
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
}
if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
......@@ -894,15 +910,6 @@ _return:
stmtCleanExecInfo(pStmt, (code ? false : true), false);
if (TSDB_CODE_SUCCESS == code && autoCreateTbl) {
if (NULL == pRsp) {
tscError("no submit resp got for auto create table");
code = TSDB_CODE_APP_ERROR;
} else {
code = stmtUpdateTableUid(pStmt, pRsp);
}
}
tFreeSSubmitRsp(pRsp);
++pStmt->sql.runTimes;
......
source/client/test/smlTest.cpp
浏览文件 @ d342ebcd
此差异已折叠。
source/common/src/tdatablock.c
浏览文件 @ d342ebcd
此差异已折叠。
source/common/src/tdataformat.c
浏览文件 @ d342ebcd
此差异已折叠。
source/common/src/tmsg.c
浏览文件 @ d342ebcd
此差异已折叠。
source/common/src/tname.c
浏览文件 @ d342ebcd
......@@ -298,8 +298,8 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
}
static int compareKv(const void* p1, const void* p2) {
SSmlKv* kv1 = *(SSmlKv**)p1;
SSmlKv* kv2 = *(SSmlKv**)p2;
SSmlKv* kv1 = (SSmlKv*)p1;
SSmlKv* kv2 = (SSmlKv*)p2;
int32_t kvLen1 = kv1->keyLen;
int32_t kvLen2 = kv2->keyLen;
int32_t res = strncasecmp(kv1->key, kv2->key, TMIN(kvLen1, kvLen2));
......@@ -320,7 +320,7 @@ void buildChildTableName(RandTableName* rName) {
taosArraySort(rName->tags, compareKv);
for (int j = 0; j < taosArrayGetSize(rName->tags); ++j) {
taosStringBuilderAppendChar(&sb, ',');
SSmlKv* tagKv = taosArrayGetP(rName->tags, j);
SSmlKv* tagKv = taosArrayGet(rName->tags, j);
taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
taosStringBuilderAppendChar(&sb, '=');
if (IS_VAR_DATA_TYPE(tagKv->type)) {
......
source/common/test/dataformatTest.cpp
浏览文件 @ d342ebcd
......@@ -117,7 +117,7 @@ STSchema *genSTSchema(int16_t nCols) {
}
STSchema *pResult = NULL;
pResult = tdGetSTSChemaFromSSChema(pSchema, nCols, 1);
pResult = tBuildTSchema(pSchema, nCols, 1);
taosMemoryFree(pSchema);
return pResult;
......
source/dnode/mnode/impl/src/mndConsumer.c
浏览文件 @ d342ebcd
......@@ -558,6 +558,10 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
goto SUBSCRIBE_OVER;
}
if (mndCheckTopicPrivilege(pMnode, pMsg->info.conn.user, MND_OPER_SUBSCRIBE, pTopic) != 0) {
goto SUBSCRIBE_OVER;
}
mndReleaseTopic(pMnode, pTopic);
}
......
source/dnode/mnode/impl/src/mndTopic.c
浏览文件 @ d342ebcd
......@@ -696,14 +696,9 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
sdbRelease(pSdb, pConsumer);
}
#if 0
if (pTopic->refConsumerCnt != 0) {
mndReleaseTopic(pMnode, pTopic);
terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
mError("topic:%s, failed to drop since %s", dropReq.name, terrstr());
if (mndCheckDbPrivilegeByName(pMnode, pReq->info.conn.user, MND_OPER_READ_DB, pTopic->db) != 0) {
return -1;
}
#endif
STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "drop-topic");
if (pTrans == NULL) {
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ d342ebcd
......@@ -222,11 +222,19 @@ typedef struct SSnapContext {
} SSnapContext;
typedef struct STqReader {
int64_t ver;
const SSubmitReq *pMsg;
SSubmitBlk *pBlock;
SSubmitMsgIter msgIter;
SSubmitBlkIter blkIter;
// const SSubmitReq *pMsg;
// SSubmitBlk *pBlock;
// SSubmitMsgIter msgIter;
// SSubmitBlkIter blkIter;
int64_t ver;
SPackedData msg2;
int8_t setMsg;
SSubmitReq2 submit;
int32_t nextBlk;
int64_t lastBlkUid;
SWalReader *pWalReader;
......@@ -251,11 +259,14 @@ int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList);
int32_t tqSeekVer(STqReader *pReader, int64_t ver);
int32_t tqNextBlock(STqReader *pReader, SFetchRet *ret);
int32_t tqReaderSetDataMsg(STqReader *pReader, const SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReader *pReader);
bool tqNextDataBlockFilterOut(STqReader *pReader, SHashObj *filterOutUids);
int32_t tqRetrieveDataBlock(SSDataBlock *pBlock, STqReader *pReader);
int32_t tqRetrieveTaosxBlock(STqReader *pReader, SArray *blocks, SArray *schemas);
int32_t tqReaderSetSubmitReq2(STqReader *pReader, void *msgStr, int32_t msgLen, int64_t ver);
// int32_t tqReaderSetDataMsg(STqReader *pReader, const SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock2(STqReader *pReader);
bool tqNextDataBlockFilterOut2(STqReader *pReader, SHashObj *filterOutUids);
int32_t tqRetrieveDataBlock2(SSDataBlock *pBlock, STqReader *pReader);
int32_t tqRetrieveTaosxBlock2(STqReader *pReader, SArray *blocks, SArray *schemas);
// int32_t tqRetrieveDataBlock(SSDataBlock *pBlock, STqReader *pReader);
// int32_t tqRetrieveTaosxBlock(STqReader *pReader, SArray *blocks, SArray *schemas);
int32_t vnodeEnqueueStreamMsg(SVnode *pVnode, SRpcMsg *pMsg);
......
source/dnode/vnode/src/inc/tq.h
浏览文件 @ d342ebcd
......@@ -154,7 +154,8 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum);
// tqExec
int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp* pRsp);
int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxRsp* pRsp);
// int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp* pRsp);
int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision);
int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp);
int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry);
......@@ -182,7 +183,8 @@ int32_t tqOffsetCommitFile(STqOffsetStore* pStore);
// tqSink
// void tqSinkToTableMerge(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
// void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
// tqOffset
char* tqOffsetBuildFName(const char* path, int32_t fVer);
......
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ d342ebcd
此差异已折叠。
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ d342ebcd
......@@ -90,6 +90,7 @@ typedef struct SCommitInfo SCommitInfo;
#define VND_INFO_FNAME "vnode.json"
// vnd.h
void* vnodeBufPoolMalloc(SVBufPool* pPool, int size);
void* vnodeBufPoolMallocAligned(SVBufPool* pPool, int size);
void vnodeBufPoolFree(SVBufPool* pPool, void* p);
......@@ -162,10 +163,9 @@ int32_t tsdbCommit(STsdb* pTsdb, SCommitInfo* pInfo);
int32_t tsdbFinishCommit(STsdb* pTsdb);
int32_t tsdbRollbackCommit(STsdb* pTsdb);
int32_t tsdbDoRetention(STsdb* pTsdb, int64_t now);
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq* pMsg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock,
SSubmitBlkRsp* pRsp);
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq2* pMsg);
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq2* pMsg, SSubmitRsp2* pRsp);
int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitTbData* pSubmitTbData, int32_t* affectedRows);
int32_t tsdbDeleteTableData(STsdb* pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey);
int32_t tsdbSetKeepCfg(STsdb* pTsdb, STsdbCfg* pCfg);
......@@ -190,7 +190,7 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msg
int32_t tqProcessTaskDropReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
int32_t tqProcessStreamTaskCheckReq(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessStreamTaskCheckRsp(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
int32_t tqProcessSubmitReq(STQ* pTq, SSubmitReq* data, int64_t ver);
int32_t tqProcessSubmitReq(STQ* pTq, SPackedData submit);
int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver);
int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec);
......@@ -203,9 +203,9 @@ int32_t tqProcessTaskRecoverFinishReq(STQ* pTq, SRpcMsg* pMsg);
int32_t tqProcessTaskRecoverFinishRsp(STQ* pTq, SRpcMsg* pMsg);
int32_t tqCheckLogInWal(STQ* pTq, int64_t version);
SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema,
SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, const char* stbFullName,
SBatchDeleteReq* pDeleteReq);
int32_t tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema,
SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, const char* stbFullName,
SBatchDeleteReq* pDeleteReq, void** ppData, int32_t* pLen);
// sma
int32_t smaInit();
......@@ -223,7 +223,7 @@ int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg);
int32_t tdProcessRSmaCreate(SSma* pSma, SVCreateStbReq* pReq);
int32_t tdProcessRSmaSubmit(SSma* pSma, void* pMsg, int32_t inputType);
int32_t tdProcessRSmaSubmit(SSma* pSma, int64_t version, void* pReq, void* pMsg, int32_t len, int32_t inputType);
int32_t tdProcessRSmaDrop(SSma* pSma, SVDropStbReq* pReq);
int32_t tdFetchTbUidList(SSma* pSma, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid);
int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore, bool isAdd);
......
source/dnode/vnode/src/meta/metaQuery.c
浏览文件 @ d342ebcd
......@@ -611,23 +611,14 @@ tb_uid_t metaStbCursorNext(SMStbCursor *pStbCur) {
}
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, int lock) {
// SMetaReader mr = {0};
STSchema *pTSchema = NULL;
SSchemaWrapper *pSW = NULL;
STSchemaBuilder sb = {0};
SSchema *pSchema;
SSchema *pSchema = NULL;
pSW = metaGetTableSchema(pMeta, uid, sver, lock);
if (!pSW) return NULL;
tdInitTSchemaBuilder(&sb, pSW->version);
for (int i = 0; i < pSW->nCols; i++) {
pSchema = pSW->pSchema + i;
tdAddColToSchema(&sb, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes);
}
pTSchema = tdGetSchemaFromBuilder(&sb);
tdDestroyTSchemaBuilder(&sb);
pTSchema = tBuildTSchema(pSW->pSchema, pSW->nCols, pSW->version);
taosMemoryFree(pSW->pSchema);
taosMemoryFree(pSW);
......@@ -708,21 +699,11 @@ int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sv
tdbFree(pData);
// convert
STSchemaBuilder sb = {0};
tdInitTSchemaBuilder(&sb, pSchemaWrapper->version);
for (int i = 0; i < pSchemaWrapper->nCols; i++) {
SSchema *pSchema = pSchemaWrapper->pSchema + i;
tdAddColToSchema(&sb, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes);
}
STSchema *pTSchema = tdGetSchemaFromBuilder(&sb);
STSchema *pTSchema = tBuildTSchema(pSchemaWrapper->pSchema, pSchemaWrapper->nCols, pSchemaWrapper->version);
if (pTSchema == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
tdDestroyTSchemaBuilder(&sb);
*ppTSchema = pTSchema;
taosMemoryFree(pSchemaWrapper->pSchema);
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ d342ebcd
此差异已折叠。
source/dnode/vnode/src/tq/tqExec.c
浏览文件 @ d342ebcd
此差异已折叠。
source/dnode/vnode/src/tq/tqPush.c
浏览文件 @ d342ebcd
此差异已折叠。
source/dnode/vnode/src/tq/tqRead.c
浏览文件 @ d342ebcd
此差异已折叠。
source/dnode/vnode/src/tq/tqSink.c
浏览文件 @ d342ebcd
此差异已折叠。
source/libs/executor/src/executor.c
浏览文件 @ d342ebcd
此差异已折叠。
source/libs/qcom/src/queryUtil.c
浏览文件 @ d342ebcd
此差异已折叠。
source/libs/stream/src/streamData.c
浏览文件 @ d342ebcd
此差异已折叠。
source/libs/stream/src/streamExec.c
浏览文件 @ d342ebcd
此差异已折叠。
source/libs/wal/src/walRead.c
浏览文件 @ d342ebcd
此差异已折叠。
source/util/src/tarray.c
浏览文件 @ d342ebcd
此差异已折叠。
source/util/src/tcompare.c
浏览文件 @ d342ebcd
此差异已折叠。
source/util/src/terror.c
浏览文件 @ d342ebcd
此差异已折叠。
tests/parallel_test/cases.task
浏览文件 @ d342ebcd
此差异已折叠。
tests/script/api/batchprepare.c
浏览文件 @ d342ebcd
此差异已折叠。
tests/system-test/2-query/sml.py
浏览文件 @ d342ebcd
此差异已折叠。
utils/test/c/sml_test.c
浏览文件 @ d342ebcd
此差异已折叠。
utils/test/c/tmq_taosx_ci.c
浏览文件 @ d342ebcd
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或