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

Merge branch '3.0' into feature/TD-16011

This commit is contained in:
liuyao 2022-05-26 17:33:59 +08:00 committed by GitHub
parent 270771e39e b911b5ba1c
commit ed31fc754e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 225 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
include/common/tmsg.h
View File

@ -300,9 +300,7 @@ typedef struct SSchema {
typedef struct { typedef struct {
int32_t nCols; int32_t nCols;
int32_t sver; int32_t version;
int32_t tagVer;
int32_t colVer;
SSchema* pSchema; SSchema* pSchema;
} SSchemaWrapper; } SSchemaWrapper;
@ -310,9 +308,7 @@ static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* p
SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper)); SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
if (pSW == NULL) return pSW; if (pSW == NULL) return pSW;
pSW->nCols = pSchemaWrapper->nCols; pSW->nCols = pSchemaWrapper->nCols;
pSW->sver = pSchemaWrapper->sver; pSW->version = pSchemaWrapper->version;
pSW->tagVer = pSchemaWrapper->tagVer;
pSW->colVer = pSchemaWrapper->colVer;
pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema)); pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
if (pSW->pSchema == NULL) { if (pSW->pSchema == NULL) {
taosMemoryFree(pSW); taosMemoryFree(pSW);
@ -367,9 +363,7 @@ static FORCE_INLINE int32_t tDecodeSSchema(SDecoder* pDecoder, SSchema* pSchema)
static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWrapper* pSW) { static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWrapper* pSW) {
int32_t tlen = 0; int32_t tlen = 0;
tlen += taosEncodeVariantI32(buf, pSW->nCols); tlen += taosEncodeVariantI32(buf, pSW->nCols);
tlen += taosEncodeVariantI32(buf, pSW->sver); tlen += taosEncodeVariantI32(buf, pSW->version);
tlen += taosEncodeVariantI32(buf, pSW->tagVer);
tlen += taosEncodeVariantI32(buf, pSW->colVer);
for (int32_t i = 0; i < pSW->nCols; i++) { for (int32_t i = 0; i < pSW->nCols; i++) {
tlen += taosEncodeSSchema(buf, &pSW->pSchema[i]); tlen += taosEncodeSSchema(buf, &pSW->pSchema[i]);
} }
@ -378,9 +372,7 @@ static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWr
static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapper* pSW) { static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapper* pSW) {
buf = taosDecodeVariantI32(buf, &pSW->nCols); buf = taosDecodeVariantI32(buf, &pSW->nCols);
buf = taosDecodeVariantI32(buf, &pSW->sver); buf = taosDecodeVariantI32(buf, &pSW->version);
buf = taosDecodeVariantI32(buf, &pSW->tagVer);
buf = taosDecodeVariantI32(buf, &pSW->colVer);
pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema)); pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
if (pSW->pSchema == NULL) { if (pSW->pSchema == NULL) {
return NULL; return NULL;
@ -394,9 +386,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapp
static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SEncoder* pEncoder, const SSchemaWrapper* pSW) { static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SEncoder* pEncoder, const SSchemaWrapper* pSW) {
if (tEncodeI32v(pEncoder, pSW->nCols) < 0) return -1; if (tEncodeI32v(pEncoder, pSW->nCols) < 0) return -1;
if (tEncodeI32v(pEncoder, pSW->sver) < 0) return -1; if (tEncodeI32v(pEncoder, pSW->version) < 0) return -1;
if (tEncodeI32v(pEncoder, pSW->tagVer) < 0) return -1;
if (tEncodeI32v(pEncoder, pSW->colVer) < 0) return -1;
for (int32_t i = 0; i < pSW->nCols; i++) { for (int32_t i = 0; i < pSW->nCols; i++) {
if (tEncodeSSchema(pEncoder, &pSW->pSchema[i]) < 0) return -1; if (tEncodeSSchema(pEncoder, &pSW->pSchema[i]) < 0) return -1;
} }
@ -406,9 +396,7 @@ static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SEncoder* pEncoder, const SSch
static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SDecoder* pDecoder, SSchemaWrapper* pSW) { static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SDecoder* pDecoder, SSchemaWrapper* pSW) {
if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1; if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1;
if (tDecodeI32v(pDecoder, &pSW->sver) < 0) return -1; if (tDecodeI32v(pDecoder, &pSW->version) < 0) return -1;
if (tDecodeI32v(pDecoder, &pSW->tagVer) < 0) return -1;
if (tDecodeI32v(pDecoder, &pSW->colVer) < 0) return -1;
pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema)); pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
if (pSW->pSchema == NULL) return -1; if (pSW->pSchema == NULL) return -1;
@ -421,9 +409,7 @@ static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SDecoder* pDecoder, SSchemaWra
static FORCE_INLINE int32_t tDecodeSSchemaWrapperEx(SDecoder* pDecoder, SSchemaWrapper* pSW) { static FORCE_INLINE int32_t tDecodeSSchemaWrapperEx(SDecoder* pDecoder, SSchemaWrapper* pSW) {
if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1; if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1;
if (tDecodeI32v(pDecoder, &pSW->sver) < 0) return -1; if (tDecodeI32v(pDecoder, &pSW->version) < 0) return -1;
if (tDecodeI32v(pDecoder, &pSW->tagVer) < 0) return -1;
if (tDecodeI32v(pDecoder, &pSW->colVer) < 0) return -1;
pSW->pSchema = (SSchema*)tDecoderMalloc(pDecoder, pSW->nCols * sizeof(SSchema)); pSW->pSchema = (SSchema*)tDecoderMalloc(pDecoder, pSW->nCols * sizeof(SSchema));
if (pSW->pSchema == NULL) return -1; if (pSW->pSchema == NULL) return -1;
@ -1713,7 +1699,7 @@ typedef struct SVCreateStbReq {
char* name; char* name;
tb_uid_t suid; tb_uid_t suid;
int8_t rollup; int8_t rollup;
SSchemaWrapper schema; SSchemaWrapper schemaRow;
SSchemaWrapper schemaTag; SSchemaWrapper schemaTag;
SRSmaParam pRSmaParam; SRSmaParam pRSmaParam;
} SVCreateStbReq; } SVCreateStbReq;
@ -1745,7 +1731,7 @@ typedef struct SVCreateTbReq {
uint8_t* pTag; uint8_t* pTag;
} ctb; } ctb;
struct { struct {
SSchemaWrapper schema; SSchemaWrapper schemaRow;
} ntb; } ntb;
}; };
} SVCreateTbReq; } SVCreateTbReq;

3
include/libs/parser/parser.h
View File

@ -48,11 +48,12 @@ typedef struct SParseContext {
} SParseContext; } SParseContext;
int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery); int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery);
bool isInsertSql(const char* pStr, size_t length); bool qIsInsertSql(const char* pStr, size_t length);
void qDestroyQuery(SQuery* pQueryNode); void qDestroyQuery(SQuery* pQueryNode);
int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema); int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
int32_t qSetSTableIdForRSma(SNode* pStmt, int64_t uid);
int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash); int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash);
int32_t qResetStmtDataBlock(void* block, bool keepBuf); int32_t qResetStmtDataBlock(void* block, bool keepBuf);

21
source/client/src/clientStmt.c
View File

@ -48,7 +48,8 @@ int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
break; break;
case STMT_EXECUTE: case STMT_EXECUTE:
if (STMT_TYPE_QUERY == pStmt->sql.type) { if (STMT_TYPE_QUERY == pStmt->sql.type) {
if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) && STMT_STATUS_NE(BIND_COL)) { if (STMT_STATUS_NE(ADD_BATCH) && STMT_STATUS_NE(FETCH_FIELDS) && STMT_STATUS_NE(BIND) &&
STMT_STATUS_NE(BIND_COL)) {
code = TSDB_CODE_TSC_STMT_API_ERROR; code = TSDB_CODE_TSC_STMT_API_ERROR;
} }
} else { } else {
@ -230,22 +231,6 @@ int32_t stmtParseSql(STscStmt* pStmt) {
pStmt->sql.type = STMT_TYPE_QUERY; pStmt->sql.type = STMT_TYPE_QUERY;
} }
/*
switch (nodeType(pStmt->sql.pQuery->pRoot)) {
case QUERY_NODE_VNODE_MODIF_STMT:
if (0 == pStmt->sql.type) {
pStmt->sql.type = STMT_TYPE_INSERT;
}
break;
case QUERY_NODE_SELECT_STMT:
pStmt->sql.type = STMT_TYPE_QUERY;
break;
default:
tscError("not supported stmt type %d", nodeType(pStmt->sql.pQuery->pRoot));
STMT_ERR_RET(TSDB_CODE_TSC_STMT_CLAUSE_ERROR);
}
*/
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -861,7 +846,7 @@ int stmtIsInsert(TAOS_STMT* stmt, int* insert) {
if (pStmt->sql.type) { if (pStmt->sql.type) {
*insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type); *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
} else { } else {
*insert = isInsertSql(pStmt->sql.sqlStr, 0); *insert = qIsInsertSql(pStmt->sql.sqlStr, 0);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;

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

@ -3801,7 +3801,7 @@ int tEncodeSVCreateStbReq(SEncoder *pCoder, const SVCreateStbReq *pReq) {
if (tEncodeCStr(pCoder, pReq->name) < 0) return -1; if (tEncodeCStr(pCoder, pReq->name) < 0) return -1;
if (tEncodeI64(pCoder, pReq->suid) < 0) return -1; if (tEncodeI64(pCoder, pReq->suid) < 0) return -1;
if (tEncodeI8(pCoder, pReq->rollup) < 0) return -1; if (tEncodeI8(pCoder, pReq->rollup) < 0) return -1;
if (tEncodeSSchemaWrapper(pCoder, &pReq->schema) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pReq->schemaRow) < 0) return -1;
if (tEncodeSSchemaWrapper(pCoder, &pReq->schemaTag) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pReq->schemaTag) < 0) return -1;
if (pReq->rollup) { if (pReq->rollup) {
if (tEncodeSRSmaParam(pCoder, &pReq->pRSmaParam) < 0) return -1; if (tEncodeSRSmaParam(pCoder, &pReq->pRSmaParam) < 0) return -1;
@ -3817,7 +3817,7 @@ int tDecodeSVCreateStbReq(SDecoder *pCoder, SVCreateStbReq *pReq) {
if (tDecodeCStr(pCoder, &pReq->name) < 0) return -1; if (tDecodeCStr(pCoder, &pReq->name) < 0) return -1;
if (tDecodeI64(pCoder, &pReq->suid) < 0) return -1; if (tDecodeI64(pCoder, &pReq->suid) < 0) return -1;
if (tDecodeI8(pCoder, &pReq->rollup) < 0) return -1; if (tDecodeI8(pCoder, &pReq->rollup) < 0) return -1;
if (tDecodeSSchemaWrapper(pCoder, &pReq->schema) < 0) return -1; if (tDecodeSSchemaWrapper(pCoder, &pReq->schemaRow) < 0) return -1;
if (tDecodeSSchemaWrapper(pCoder, &pReq->schemaTag) < 0) return -1; if (tDecodeSSchemaWrapper(pCoder, &pReq->schemaTag) < 0) return -1;
if (pReq->rollup) { if (pReq->rollup) {
if (tDecodeSRSmaParam(pCoder, &pReq->pRSmaParam) < 0) return -1; if (tDecodeSRSmaParam(pCoder, &pReq->pRSmaParam) < 0) return -1;
@ -3866,7 +3866,7 @@ int tEncodeSVCreateTbReq(SEncoder *pCoder, const SVCreateTbReq *pReq) {
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 (tEncodeBinary(pCoder, pReq->ctb.pTag, kvRowLen(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.schema) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pReq->ntb.schemaRow) < 0) return -1;
} else { } else {
ASSERT(0); ASSERT(0);
} }
@ -3892,7 +3892,7 @@ int tDecodeSVCreateTbReq(SDecoder *pCoder, SVCreateTbReq *pReq) {
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 (tDecodeBinary(pCoder, &pReq->ctb.pTag, &len) < 0) return -1;
} else if (pReq->type == TSDB_NORMAL_TABLE) { } else if (pReq->type == TSDB_NORMAL_TABLE) {
if (tDecodeSSchemaWrapper(pCoder, &pReq->ntb.schema) < 0) return -1; if (tDecodeSSchemaWrapper(pCoder, &pReq->ntb.schemaRow) < 0) return -1;
} else { } else {
ASSERT(0); ASSERT(0);
} }

3
source/dnode/mnode/impl/inc/mndScheduler.h
View File

@ -29,7 +29,8 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream); int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream);
int32_t mndConvertRSmaTask(const char* ast, int8_t triggerType, int64_t watermark, char** pStr, int32_t* pLen); int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int64_t watermark, char** pStr,
int32_t* pLen);
#ifdef __cplusplus #ifdef __cplusplus
} }

9
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -28,13 +28,15 @@
#include "mndTrans.h" #include "mndTrans.h"
#include "mndUser.h" #include "mndUser.h"
#include "mndVgroup.h" #include "mndVgroup.h"
#include "parser.h"
#include "tcompare.h" #include "tcompare.h"
#include "tname.h" #include "tname.h"
#include "tuuid.h" #include "tuuid.h"
extern bool tsStreamSchedV; extern bool tsStreamSchedV;
int32_t mndConvertRSmaTask(const char* ast, int8_t triggerType, int64_t watermark, char** pStr, int32_t* pLen) { int32_t mndConvertRSmaTask(const char* ast, int64_t uid, int8_t triggerType, int64_t watermark, char** pStr,
int32_t* pLen) {
SNode* pAst = NULL; SNode* pAst = NULL;
SQueryPlan* pPlan = NULL; SQueryPlan* pPlan = NULL;
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
@ -44,6 +46,11 @@ int32_t mndConvertRSmaTask(const char* ast, int8_t triggerType, int64_t watermar
goto END; goto END;
} }
if (qSetSTableIdForRSma(pAst, uid) < 0) {
terrno = TSDB_CODE_QRY_INVALID_INPUT;
goto END;
}
SPlanContext cxt = { SPlanContext cxt = {
.pAstRoot = pAst, .pAstRoot = pAst,
.topicQuery = false, .topicQuery = false,

17
source/dnode/mnode/impl/src/mndStb.c
View File

@ -388,25 +388,26 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
req.name = (char *)tNameGetTableName(&name); req.name = (char *)tNameGetTableName(&name);
req.suid = pStb->uid; req.suid = pStb->uid;
req.rollup = pStb->ast1Len > 0 ? 1 : 0; req.rollup = pStb->ast1Len > 0 ? 1 : 0;
req.schema.nCols = pStb->numOfColumns; // todo
req.schema.sver = pStb->version; req.schemaRow.nCols = pStb->numOfColumns;
req.schema.tagVer = pStb->tagVer; req.schemaRow.version = pStb->version;
req.schema.colVer = pStb->colVer; req.schemaRow.pSchema = pStb->pColumns;
req.schema.pSchema = pStb->pColumns;
req.schemaTag.nCols = pStb->numOfTags; req.schemaTag.nCols = pStb->numOfTags;
req.schemaTag.sver = 1; req.schemaTag.version = pStb->tagVer;
req.schemaTag.pSchema = pStb->pTags; req.schemaTag.pSchema = pStb->pTags;
if (req.rollup) { if (req.rollup) {
req.pRSmaParam.xFilesFactor = pStb->xFilesFactor; req.pRSmaParam.xFilesFactor = pStb->xFilesFactor;
req.pRSmaParam.delay = pStb->delay; req.pRSmaParam.delay = pStb->delay;
if (pStb->ast1Len > 0) { if (pStb->ast1Len > 0) {
if (mndConvertRSmaTask(pStb->pAst1, 0, 0, &req.pRSmaParam.qmsg1, &req.pRSmaParam.qmsg1Len) != TSDB_CODE_SUCCESS) { if (mndConvertRSmaTask(pStb->pAst1, pStb->uid, 0, 0, &req.pRSmaParam.qmsg1, &req.pRSmaParam.qmsg1Len) !=
TSDB_CODE_SUCCESS) {
return NULL; return NULL;
} }
} }
if (pStb->ast2Len > 0) { if (pStb->ast2Len > 0) {
if (mndConvertRSmaTask(pStb->pAst2, 0, 0, &req.pRSmaParam.qmsg2, &req.pRSmaParam.qmsg2Len) != TSDB_CODE_SUCCESS) { if (mndConvertRSmaTask(pStb->pAst2, pStb->uid, 0, 0, &req.pRSmaParam.qmsg2, &req.pRSmaParam.qmsg2Len) !=
TSDB_CODE_SUCCESS) {
return NULL; return NULL;
} }
} }

2
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -217,7 +217,7 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
} }
} else { } else {
pTopic->schema.nCols = 0; pTopic->schema.nCols = 0;
pTopic->schema.sver = 0; pTopic->schema.version = 0;
pTopic->schema.pSchema = NULL; pTopic->schema.pSchema = NULL;
} }

4
source/dnode/vnode/inc/vnode.h
View File

@ -182,7 +182,7 @@ struct SMetaEntry {
char *name; char *name;
union { union {
struct { struct {
SSchemaWrapper schema; SSchemaWrapper schemaRow;
SSchemaWrapper schemaTag; SSchemaWrapper schemaTag;
} stbEntry; } stbEntry;
struct { struct {
@ -195,7 +195,7 @@ struct SMetaEntry {
int64_t ctime; int64_t ctime;
int32_t ttlDays; int32_t ttlDays;
int32_t ncid; // next column id int32_t ncid; // next column id
SSchemaWrapper schema; SSchemaWrapper schemaRow;
} ntbEntry; } ntbEntry;
struct { struct {
STSma *tsma; STSma *tsma;

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

@ -24,7 +24,7 @@ int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) {
if (tEncodeCStr(pCoder, pME->name) < 0) return -1; if (tEncodeCStr(pCoder, pME->name) < 0) return -1;
if (pME->type == TSDB_SUPER_TABLE) { if (pME->type == TSDB_SUPER_TABLE) {
if (tEncodeSSchemaWrapper(pCoder, &pME->stbEntry.schema) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pME->stbEntry.schemaRow) < 0) return -1;
if (tEncodeSSchemaWrapper(pCoder, &pME->stbEntry.schemaTag) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pME->stbEntry.schemaTag) < 0) return -1;
} else if (pME->type == TSDB_CHILD_TABLE) { } else if (pME->type == TSDB_CHILD_TABLE) {
if (tEncodeI64(pCoder, pME->ctbEntry.ctime) < 0) return -1; if (tEncodeI64(pCoder, pME->ctbEntry.ctime) < 0) return -1;
@ -35,7 +35,7 @@ int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) {
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;
if (tEncodeI32v(pCoder, pME->ntbEntry.ncid) < 0) return -1; if (tEncodeI32v(pCoder, pME->ntbEntry.ncid) < 0) return -1;
if (tEncodeSSchemaWrapper(pCoder, &pME->ntbEntry.schema) < 0) return -1; if (tEncodeSSchemaWrapper(pCoder, &pME->ntbEntry.schemaRow) < 0) return -1;
} else if (pME->type == TSDB_TSMA_TABLE) { } else if (pME->type == TSDB_TSMA_TABLE) {
if (tEncodeTSma(pCoder, pME->smaEntry.tsma) < 0) return -1; if (tEncodeTSma(pCoder, pME->smaEntry.tsma) < 0) return -1;
} else { } else {
@ -56,7 +56,7 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
if (tDecodeCStr(pCoder, &pME->name) < 0) return -1; if (tDecodeCStr(pCoder, &pME->name) < 0) return -1;
if (pME->type == TSDB_SUPER_TABLE) { if (pME->type == TSDB_SUPER_TABLE) {
if (tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schema) < 0) return -1; if (tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schemaRow) < 0) return -1;
if (tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schemaTag) < 0) return -1; if (tDecodeSSchemaWrapperEx(pCoder, &pME->stbEntry.schemaTag) < 0) return -1;
} else if (pME->type == TSDB_CHILD_TABLE) { } else if (pME->type == TSDB_CHILD_TABLE) {
if (tDecodeI64(pCoder, &pME->ctbEntry.ctime) < 0) return -1; if (tDecodeI64(pCoder, &pME->ctbEntry.ctime) < 0) return -1;
@ -67,7 +67,7 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) {
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;
if (tDecodeI32v(pCoder, &pME->ntbEntry.ncid) < 0) return -1; if (tDecodeI32v(pCoder, &pME->ntbEntry.ncid) < 0) return -1;
if (tDecodeSSchemaWrapperEx(pCoder, &pME->ntbEntry.schema) < 0) return -1; if (tDecodeSSchemaWrapperEx(pCoder, &pME->ntbEntry.schemaRow) < 0) return -1;
} else if (pME->type == TSDB_TSMA_TABLE) { } else if (pME->type == TSDB_TSMA_TABLE) {
pME->smaEntry.tsma = tDecoderMalloc(pCoder, sizeof(STSma)); pME->smaEntry.tsma = tDecoderMalloc(pCoder, sizeof(STSma));
if (!pME->smaEntry.tsma) { if (!pME->smaEntry.tsma) {

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

@ -56,7 +56,7 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
me.type = TSDB_SUPER_TABLE; me.type = TSDB_SUPER_TABLE;
me.uid = pReq->suid; me.uid = pReq->suid;
me.name = pReq->name; me.name = pReq->name;
me.stbEntry.schema = pReq->schema; me.stbEntry.schemaRow = pReq->schemaRow;
me.stbEntry.schemaTag = pReq->schemaTag; me.stbEntry.schemaTag = pReq->schemaTag;
if (metaHandleEntry(pMeta, &me) < 0) goto _err; if (metaHandleEntry(pMeta, &me) < 0) goto _err;
@ -182,16 +182,14 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
nStbEntry.type = TSDB_SUPER_TABLE; nStbEntry.type = TSDB_SUPER_TABLE;
nStbEntry.uid = pReq->suid; nStbEntry.uid = pReq->suid;
nStbEntry.name = pReq->name; nStbEntry.name = pReq->name;
nStbEntry.stbEntry.schema = pReq->schema; nStbEntry.stbEntry.schemaRow = pReq->schemaRow;
nStbEntry.stbEntry.schemaTag = pReq->schemaTag; nStbEntry.stbEntry.schemaTag = pReq->schemaTag;
metaWLock(pMeta); metaWLock(pMeta);
// compare two entry // compare two entry
if (oStbEntry.stbEntry.schema.sver != pReq->schema.sver) { if (oStbEntry.stbEntry.schemaRow.version != pReq->schemaRow.version) {
if (oStbEntry.stbEntry.schema.nCols != pReq->schema.nCols) {
metaSaveToSkmDb(pMeta, &nStbEntry); metaSaveToSkmDb(pMeta, &nStbEntry);
} }
}
// if (oStbEntry.stbEntry.schemaTag.sver != pReq->schemaTag.sver) { // if (oStbEntry.stbEntry.schemaTag.sver != pReq->schemaTag.sver) {
// // change tag schema // // change tag schema
@ -247,8 +245,8 @@ int metaCreateTable(SMeta *pMeta, int64_t version, SVCreateTbReq *pReq) {
} else { } else {
me.ntbEntry.ctime = pReq->ctime; me.ntbEntry.ctime = pReq->ctime;
me.ntbEntry.ttlDays = pReq->ttl; me.ntbEntry.ttlDays = pReq->ttl;
me.ntbEntry.schema = pReq->ntb.schema; me.ntbEntry.schemaRow = pReq->ntb.schemaRow;
me.ntbEntry.ncid = me.ntbEntry.schema.pSchema[me.ntbEntry.schema.nCols - 1].colId + 1; me.ntbEntry.ncid = me.ntbEntry.schemaRow.pSchema[me.ntbEntry.schemaRow.nCols - 1].colId + 1;
} }
if (metaHandleEntry(pMeta, &me) < 0) goto _err; if (metaHandleEntry(pMeta, &me) < 0) goto _err;
@ -381,7 +379,7 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
} }
// search the column to add/drop/update // search the column to add/drop/update
pSchema = &entry.ntbEntry.schema; pSchema = &entry.ntbEntry.schemaRow;
int32_t iCol = 0; int32_t iCol = 0;
for (;;) { for (;;) {
pColumn = NULL; pColumn = NULL;
@ -402,16 +400,16 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
terrno = TSDB_CODE_VND_COL_ALREADY_EXISTS; terrno = TSDB_CODE_VND_COL_ALREADY_EXISTS;
goto _err; goto _err;
} }
pSchema->sver++; pSchema->version++;
pSchema->nCols++; pSchema->nCols++;
pNewSchema = taosMemoryMalloc(sizeof(SSchema) * pSchema->nCols); pNewSchema = taosMemoryMalloc(sizeof(SSchema) * pSchema->nCols);
memcpy(pNewSchema, pSchema->pSchema, sizeof(SSchema) * (pSchema->nCols - 1)); memcpy(pNewSchema, pSchema->pSchema, sizeof(SSchema) * (pSchema->nCols - 1));
pSchema->pSchema = pNewSchema; pSchema->pSchema = pNewSchema;
pSchema->pSchema[entry.ntbEntry.schema.nCols - 1].bytes = pAlterTbReq->bytes; pSchema->pSchema[entry.ntbEntry.schemaRow.nCols - 1].bytes = pAlterTbReq->bytes;
pSchema->pSchema[entry.ntbEntry.schema.nCols - 1].type = pAlterTbReq->type; pSchema->pSchema[entry.ntbEntry.schemaRow.nCols - 1].type = pAlterTbReq->type;
pSchema->pSchema[entry.ntbEntry.schema.nCols - 1].flags = pAlterTbReq->flags; pSchema->pSchema[entry.ntbEntry.schemaRow.nCols - 1].flags = pAlterTbReq->flags;
pSchema->pSchema[entry.ntbEntry.schema.nCols - 1].colId = entry.ntbEntry.ncid++; pSchema->pSchema[entry.ntbEntry.schemaRow.nCols - 1].colId = entry.ntbEntry.ncid++;
strcpy(pSchema->pSchema[entry.ntbEntry.schema.nCols - 1].name, pAlterTbReq->colName); strcpy(pSchema->pSchema[entry.ntbEntry.schemaRow.nCols - 1].name, pAlterTbReq->colName);
break; break;
case TSDB_ALTER_TABLE_DROP_COLUMN: case TSDB_ALTER_TABLE_DROP_COLUMN:
if (pColumn == NULL) { if (pColumn == NULL) {
@ -422,7 +420,7 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION; terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION;
goto _err; goto _err;
} }
pSchema->sver++; pSchema->version++;
tlen = (pSchema->nCols - iCol - 1) * sizeof(SSchema); tlen = (pSchema->nCols - iCol - 1) * sizeof(SSchema);
if (tlen) { if (tlen) {
memmove(pColumn, pColumn + 1, tlen); memmove(pColumn, pColumn + 1, tlen);
@ -438,7 +436,7 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION; terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION;
goto _err; goto _err;
} }
pSchema->sver++; pSchema->version++;
pColumn->bytes = pAlterTbReq->colModBytes; pColumn->bytes = pAlterTbReq->colModBytes;
break; break;
case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME:
@ -446,7 +444,7 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
terrno = TSDB_CODE_VND_TABLE_COL_NOT_EXISTS; terrno = TSDB_CODE_VND_TABLE_COL_NOT_EXISTS;
goto _err; goto _err;
} }
pSchema->sver++; pSchema->version++;
strcpy(pColumn->name, pAlterTbReq->colNewName); strcpy(pColumn->name, pAlterTbReq->colNewName);
break; break;
} }
@ -813,15 +811,15 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
const SSchemaWrapper *pSW; const SSchemaWrapper *pSW;
if (pME->type == TSDB_SUPER_TABLE) { if (pME->type == TSDB_SUPER_TABLE) {
pSW = &pME->stbEntry.schema; pSW = &pME->stbEntry.schemaRow;
} else if (pME->type == TSDB_NORMAL_TABLE) { } else if (pME->type == TSDB_NORMAL_TABLE) {
pSW = &pME->ntbEntry.schema; pSW = &pME->ntbEntry.schemaRow;
} else { } else {
ASSERT(0); ASSERT(0);
} }
skmDbKey.uid = pME->uid; skmDbKey.uid = pME->uid;
skmDbKey.sver = pSW->sver; skmDbKey.sver = pSW->version;
// encode schema // encode schema
int32_t ret = 0; int32_t ret = 0;

6
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -321,13 +321,13 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST; terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
return -1; return -1;
} }
strcat(pRsp->tblFName, mr.me.name); if(pRsp->tblFName) strcat(pRsp->tblFName, mr.me.name);
if (mr.me.type == TSDB_NORMAL_TABLE) { if (mr.me.type == TSDB_NORMAL_TABLE) {
sverNew = mr.me.ntbEntry.schema.sver; sverNew = mr.me.ntbEntry.schemaRow.version;
} else { } else {
metaGetTableEntryByUid(&mr, mr.me.ctbEntry.suid); metaGetTableEntryByUid(&mr, mr.me.ctbEntry.suid);
sverNew = mr.me.stbEntry.schema.sver; sverNew = mr.me.stbEntry.schemaRow.version;
} }
metaReaderClear(&mr); metaReaderClear(&mr);

8
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -64,7 +64,7 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
if (mer1.me.type == TSDB_SUPER_TABLE) { if (mer1.me.type == TSDB_SUPER_TABLE) {
strcpy(metaRsp.stbName, mer1.me.name); strcpy(metaRsp.stbName, mer1.me.name);
schema = mer1.me.stbEntry.schema; schema = mer1.me.stbEntry.schemaRow;
schemaTag = mer1.me.stbEntry.schemaTag; schemaTag = mer1.me.stbEntry.schemaTag;
metaRsp.suid = mer1.me.uid; metaRsp.suid = mer1.me.uid;
} else if (mer1.me.type == TSDB_CHILD_TABLE) { } else if (mer1.me.type == TSDB_CHILD_TABLE) {
@ -73,10 +73,10 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
strcpy(metaRsp.stbName, mer2.me.name); strcpy(metaRsp.stbName, mer2.me.name);
metaRsp.suid = mer2.me.uid; metaRsp.suid = mer2.me.uid;
schema = mer2.me.stbEntry.schema; schema = mer2.me.stbEntry.schemaRow;
schemaTag = mer2.me.stbEntry.schemaTag; schemaTag = mer2.me.stbEntry.schemaTag;
} else if (mer1.me.type == TSDB_NORMAL_TABLE) { } else if (mer1.me.type == TSDB_NORMAL_TABLE) {
schema = mer1.me.ntbEntry.schema; schema = mer1.me.ntbEntry.schemaRow;
} else { } else {
ASSERT(0); ASSERT(0);
} }
@ -84,7 +84,7 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
metaRsp.numOfTags = schemaTag.nCols; metaRsp.numOfTags = schemaTag.nCols;
metaRsp.numOfColumns = schema.nCols; metaRsp.numOfColumns = schema.nCols;
metaRsp.precision = pVnode->config.tsdbCfg.precision; metaRsp.precision = pVnode->config.tsdbCfg.precision;
metaRsp.sversion = schema.sver; metaRsp.sversion = schema.version;
metaRsp.pSchemas = (SSchema *)taosMemoryMalloc(sizeof(SSchema) * (metaRsp.numOfColumns + metaRsp.numOfTags)); metaRsp.pSchemas = (SSchema *)taosMemoryMalloc(sizeof(SSchema) * (metaRsp.numOfColumns + metaRsp.numOfTags));
memcpy(metaRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols); memcpy(metaRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols);

17
source/libs/executor/src/executorimpl.c
View File

@ -4455,15 +4455,15 @@ void extractTableSchemaVersion(SReadHandle* pHandle, uint64_t uid, SExecTaskInfo
pTaskInfo->schemaVer.tablename = strdup(mr.me.name); pTaskInfo->schemaVer.tablename = strdup(mr.me.name);
if (mr.me.type == TSDB_SUPER_TABLE) { if (mr.me.type == TSDB_SUPER_TABLE) {
pTaskInfo->schemaVer.sversion = mr.me.stbEntry.schema.sver; pTaskInfo->schemaVer.sversion = mr.me.stbEntry.schemaRow.version;
pTaskInfo->schemaVer.tversion = mr.me.stbEntry.schemaTag.sver; pTaskInfo->schemaVer.tversion = mr.me.stbEntry.schemaTag.version;
} else if (mr.me.type == TSDB_CHILD_TABLE) { } else if (mr.me.type == TSDB_CHILD_TABLE) {
tb_uid_t suid = mr.me.ctbEntry.suid; tb_uid_t suid = mr.me.ctbEntry.suid;
metaGetTableEntryByUid(&mr, suid); metaGetTableEntryByUid(&mr, suid);
pTaskInfo->schemaVer.sversion = mr.me.stbEntry.schema.sver; pTaskInfo->schemaVer.sversion = mr.me.stbEntry.schemaRow.version;
pTaskInfo->schemaVer.tversion = mr.me.stbEntry.schemaTag.sver; pTaskInfo->schemaVer.tversion = mr.me.stbEntry.schemaTag.version;
} else { } else {
pTaskInfo->schemaVer.sversion = mr.me.ntbEntry.schema.sver; pTaskInfo->schemaVer.sversion = mr.me.ntbEntry.schemaRow.version;
} }
metaReaderClear(&mr); metaReaderClear(&mr);
@ -4668,8 +4668,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc); SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
int32_t tsSlotId = ((SColumnNode*)pSessionNode->window.pTspk)->slotId; int32_t tsSlotId = ((SColumnNode*)pSessionNode->window.pTspk)->slotId;
pOptr = pOptr = createStreamSessionAggOperatorInfo(ops[0], pExprInfo, num, pResBlock, pSessionNode->gap, tsSlotId, &as,
createStreamSessionAggOperatorInfo(ops[0], pExprInfo, num, pResBlock, pSessionNode->gap, tsSlotId, &as, pTaskInfo); pTaskInfo);
} else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) { } else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) {
SPartitionPhysiNode* pPartNode = (SPartitionPhysiNode*)pPhyNode; SPartitionPhysiNode* pPartNode = (SPartitionPhysiNode*)pPhyNode;
@ -5162,8 +5162,7 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SExplainExecInfo
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t initCacheSupporter(SCatchSupporter* pCatchSup, size_t rowSize, const char* pKey, int32_t initCatchSupporter(SCatchSupporter* pCatchSup, size_t rowSize, const char* pKey, const char* pDir) {
const char* pDir) {
pCatchSup->keySize = sizeof(int64_t) + sizeof(int64_t) + sizeof(TSKEY); pCatchSup->keySize = sizeof(int64_t) + sizeof(int64_t) + sizeof(TSKEY);
pCatchSup->pKeyBuf = taosMemoryCalloc(1, pCatchSup->keySize); pCatchSup->pKeyBuf = taosMemoryCalloc(1, pCatchSup->keySize);
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

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

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "function.h"
#include "filter.h" #include "filter.h"
#include "function.h"
#include "functionMgt.h" #include "functionMgt.h"
#include "os.h" #include "os.h"
#include "querynodes.h" #include "querynodes.h"
@ -324,7 +324,6 @@ void addTagPseudoColumnData(STableScanInfo* pTableScanInfo, SSDataBlock* pBlock)
for (int32_t i = 0; i < pBlock->info.rows; ++i) { for (int32_t i = 0; i < pBlock->info.rows; ++i) {
colDataAppend(pColInfoData, i, p, (p == NULL)); colDataAppend(pColInfoData, i, p, (p == NULL));
} }
if (pColInfoData->info.type == TSDB_DATA_TYPE_JSON) { if (pColInfoData->info.type == TSDB_DATA_TYPE_JSON) {
taosMemoryFree((void*)p); taosMemoryFree((void*)p);
} }
@ -409,7 +408,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
STimeWindow* pWin = &pTableScanInfo->cond.twindow; STimeWindow* pWin = &pTableScanInfo->cond.twindow;
qDebug("%s start to repeat ascending order scan data blocks due to query func required, qrange:%" PRId64 qDebug("%s start to repeat ascending order scan data blocks due to query func required, qrange:%" PRId64
"-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey); "-%" PRId64,
GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
// do prepare for the next round table scan operation // do prepare for the next round table scan operation
tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond); tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond);
@ -484,7 +484,8 @@ static void destroyTableScanOperatorInfo(void* param, int32_t numOfOutput) {
} }
} }
SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, tsdbReaderT pDataReader, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo) { SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, tsdbReaderT pDataReader,
SReadHandle* readHandle, SExecTaskInfo* pTaskInfo) {
STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo)); STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
@ -498,7 +499,8 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc; SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
int32_t numOfCols = 0; int32_t numOfCols = 0;
SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, pTaskInfo, COL_MATCH_FROM_COL_ID); SArray* pColList =
extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, pTaskInfo, COL_MATCH_FROM_COL_ID);
int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode); int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
@ -531,7 +533,8 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
pOperator->numOfExprs = numOfCols; pOperator->numOfExprs = numOfCols;
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo, NULL, NULL, getTableScannerExecInfo); pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo,
NULL, NULL, getTableScannerExecInfo);
// for non-blocking operator, the open cost is always 0 // for non-blocking operator, the open cost is always 0
pOperator->cost.openCost = 0; pOperator->cost.openCost = 0;
@ -649,9 +652,7 @@ static void doClearBufferedBlocks(SStreamBlockScanInfo* pInfo) {
taosArrayClear(pInfo->pBlockLists); taosArrayClear(pInfo->pBlockLists);
} }
static bool isSessionWindow(SStreamBlockScanInfo* pInfo) { static bool isSessionWindow(SStreamBlockScanInfo* pInfo) { return pInfo->sessionSup.pStreamAggSup != NULL; }
return pInfo->sessionSup.pStreamAggSup != NULL;
}
static bool prepareDataScan(SStreamBlockScanInfo* pInfo) { static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
SSDataBlock* pSDB = pInfo->pUpdateRes; SSDataBlock* pSDB = pInfo->pUpdateRes;
@ -665,16 +666,16 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
SStreamAggSupporter* pAggSup = pInfo->sessionSup.pStreamAggSup; SStreamAggSupporter* pAggSup = pInfo->sessionSup.pStreamAggSup;
int64_t gap = pInfo->sessionSup.gap; int64_t gap = pInfo->sessionSup.gap;
int32_t winIndex = 0; int32_t winIndex = 0;
SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup->pResultRows, SResultWindowInfo* pCurWin =
tsCols[pInfo->updateResIndex], gap, &winIndex); getSessionTimeWindow(pAggSup->pResultRows, tsCols[pInfo->updateResIndex], gap, &winIndex);
win = pCurWin->win; win = pCurWin->win;
pInfo->updateResIndex += updateSessionWindowInfo(pCurWin, tsCols, pSDB->info.rows, pInfo->updateResIndex +=
pInfo->updateResIndex, gap, NULL); updateSessionWindowInfo(pCurWin, tsCols, pSDB->info.rows, pInfo->updateResIndex, gap, NULL);
} else { } else {
win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[pInfo->updateResIndex], win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[pInfo->updateResIndex], &pInfo->interval,
&pInfo->interval, pInfo->interval.precision, NULL); pInfo->interval.precision, NULL);
pInfo->updateResIndex += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, pInfo->updateResIndex, pInfo->updateResIndex += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, pInfo->updateResIndex, win.ekey,
win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC); binarySearchForKey, NULL, TSDB_ORDER_ASC);
} }
STableScanInfo* pTableScanInfo = pInfo->pOperatorDumy->info; STableScanInfo* pTableScanInfo = pInfo->pOperatorDumy->info;
pTableScanInfo->cond.twindow = win; pTableScanInfo->cond.twindow = win;
@ -737,18 +738,16 @@ void static setSupKeyBuf(SCatchSupporter* pSup, int64_t groupId, int64_t childId
pKey[2] = ts; pKey[2] = ts;
} }
static int32_t catchWidonwInfo(SSDataBlock* pDataBlock, SCatchSupporter* pSup, static int32_t catchWidonwInfo(SSDataBlock* pDataBlock, SCatchSupporter* pSup, int32_t pageId, int32_t tsIndex,
int32_t pageId, int32_t tsIndex, int64_t childId) { int64_t childId) {
SColumnInfoData* pColDataInfo = taosArrayGet(pDataBlock->pDataBlock, tsIndex); SColumnInfoData* pColDataInfo = taosArrayGet(pDataBlock->pDataBlock, tsIndex);
TSKEY* tsCols = (int64_t*)pColDataInfo->pData; TSKEY* tsCols = (int64_t*)pColDataInfo->pData;
for (int32_t i = 0; i < pDataBlock->info.rows; i++) { for (int32_t i = 0; i < pDataBlock->info.rows; i++) {
setSupKeyBuf(pSup, pDataBlock->info.groupId, childId, tsCols[i]); setSupKeyBuf(pSup, pDataBlock->info.groupId, childId, tsCols[i]);
SWindowPosition* p1 = (SWindowPosition*)taosHashGet(pSup->pWindowHashTable, SWindowPosition* p1 = (SWindowPosition*)taosHashGet(pSup->pWindowHashTable, pSup->pKeyBuf, pSup->keySize);
pSup->pKeyBuf, pSup->keySize);
if (p1 == NULL) { if (p1 == NULL) {
SWindowPosition pos = {.pageId = pageId, .rowId = i}; SWindowPosition pos = {.pageId = pageId, .rowId = i};
int32_t code = taosHashPut(pSup->pWindowHashTable, pSup->pKeyBuf, pSup->keySize, &pos, int32_t code = taosHashPut(pSup->pWindowHashTable, pSup->pKeyBuf, pSup->keySize, &pos, sizeof(SWindowPosition));
sizeof(SWindowPosition));
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
} }
@ -760,8 +759,7 @@ static int32_t catchWidonwInfo(SSDataBlock* pDataBlock, SCatchSupporter* pSup,
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t catchDatablock(SSDataBlock* pDataBlock, SCatchSupporter* pSup, static int32_t catchDatablock(SSDataBlock* pDataBlock, SCatchSupporter* pSup, int32_t tsIndex, int64_t childId) {
int32_t tsIndex, int64_t childId) {
int32_t start = 0; int32_t start = 0;
int32_t stop = 0; int32_t stop = 0;
int32_t pageSize = getBufPageSize(pSup->pDataBuf); int32_t pageSize = getBufPageSize(pSup->pDataBuf);
@ -802,10 +800,8 @@ static SSDataBlock* getDataFromCatch(SStreamBlockScanInfo* pInfo) {
int32_t size = taosArrayGetSize(pInfo->childIds); int32_t size = taosArrayGetSize(pInfo->childIds);
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
int64_t id = *(int64_t*)taosArrayGet(pInfo->childIds, i); int64_t id = *(int64_t*)taosArrayGet(pInfo->childIds, i);
setSupKeyBuf(pCSup, pBlock->info.groupId, id, setSupKeyBuf(pCSup, pBlock->info.groupId, id, tsCols[pInfo->updateResIndex]);
tsCols[pInfo->updateResIndex]); SWindowPosition* pos = (SWindowPosition*)taosHashGet(pCSup->pWindowHashTable, pCSup->pKeyBuf, pCSup->keySize);
SWindowPosition* pos = (SWindowPosition*)taosHashGet(pCSup->pWindowHashTable,
pCSup->pKeyBuf, pCSup->keySize);
void* buf = getBufPage(pCSup->pDataBuf, pos->pageId); void* buf = getBufPage(pCSup->pDataBuf, pos->pageId);
SSDataBlock* pDB = createOneDataBlock(pInfo->pRes, false); SSDataBlock* pDB = createOneDataBlock(pInfo->pRes, false);
blockDataFromBuf(pDB, buf); blockDataFromBuf(pDB, buf);
@ -1004,7 +1000,8 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, void* pDataR
pInfo->interval = pSTInfo->interval; pInfo->interval = pSTInfo->interval;
pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1}; pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1};
initCacheSupporter(&pInfo->childAggSup, 1024, "StreamFinalInterval", "/tmp/"); // TODO(liuyao) get row size from phy plan initCatchSupporter(&pInfo->childAggSup, 1024, "StreamFinalInterval",
"/tmp/"); // TODO(liuyao) get row size from phy plan
pOperator->name = "StreamBlockScanOperator"; pOperator->name = "StreamBlockScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
@ -1277,7 +1274,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
// number of columns // number of columns
pColInfoData = taosArrayGet(p->pDataBlock, 3); pColInfoData = taosArrayGet(p->pDataBlock, 3);
colDataAppend(pColInfoData, numOfRows, (char*)&mr.me.stbEntry.schema.nCols, false); colDataAppend(pColInfoData, numOfRows, (char*)&mr.me.stbEntry.schemaRow.nCols, false);
// super table name // super table name
STR_TO_VARSTR(str, mr.me.name); STR_TO_VARSTR(str, mr.me.name);
@ -1301,7 +1298,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
// number of columns // number of columns
pColInfoData = taosArrayGet(p->pDataBlock, 3); pColInfoData = taosArrayGet(p->pDataBlock, 3);
colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.schema.nCols, false); colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.schemaRow.nCols, false);
// super table name // super table name
pColInfoData = taosArrayGet(p->pDataBlock, 4); pColInfoData = taosArrayGet(p->pDataBlock, 4);

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

@ -48,6 +48,7 @@ typedef struct SFullDatabaseName {
static int32_t translateSubquery(STranslateContext* pCxt, SNode* pNode); static int32_t translateSubquery(STranslateContext* pCxt, SNode* pNode);
static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode); static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode);
static EDealRes translateValue(STranslateContext* pCxt, SValueNode* pVal);
static bool afterGroupBy(ESqlClause clause) { return clause > SQL_CLAUSE_GROUP_BY; } static bool afterGroupBy(ESqlClause clause) { return clause > SQL_CLAUSE_GROUP_BY; }
@ -542,17 +543,18 @@ static EDealRes translateColumn(STranslateContext* pCxt, SColumnNode** pCol) {
return res; return res;
} }
static int32_t parseTimeFromValueNode(SValueNode* pVal) { static int32_t parseTimeFromValueNode(STranslateContext* pCxt, SValueNode* pVal) {
if (IS_SIGNED_NUMERIC_TYPE(pVal->node.resType.type)) { if (IS_NUMERIC_TYPE(pVal->node.resType.type) || TSDB_DATA_TYPE_BOOL == pVal->node.resType.type) {
return TSDB_CODE_SUCCESS; if (DEAL_RES_ERROR == translateValue(pCxt, pVal)) {
} else if (IS_UNSIGNED_NUMERIC_TYPE(pVal->node.resType.type)) { return pCxt->errCode;
}
if (IS_UNSIGNED_NUMERIC_TYPE(pVal->node.resType.type)) {
pVal->datum.i = pVal->datum.u; pVal->datum.i = pVal->datum.u;
return TSDB_CODE_SUCCESS;
} else if (IS_FLOAT_TYPE(pVal->node.resType.type)) { } else if (IS_FLOAT_TYPE(pVal->node.resType.type)) {
pVal->datum.i = pVal->datum.d; pVal->datum.i = pVal->datum.d;
return TSDB_CODE_SUCCESS;
} else if (TSDB_DATA_TYPE_BOOL == pVal->node.resType.type) { } else if (TSDB_DATA_TYPE_BOOL == pVal->node.resType.type) {
pVal->datum.i = pVal->datum.b; pVal->datum.i = pVal->datum.b;
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (IS_VAR_DATA_TYPE(pVal->node.resType.type) || TSDB_DATA_TYPE_TIMESTAMP == pVal->node.resType.type) { } else if (IS_VAR_DATA_TYPE(pVal->node.resType.type) || TSDB_DATA_TYPE_TIMESTAMP == pVal->node.resType.type) {
if (TSDB_CODE_SUCCESS == taosParseTime(pVal->literal, &pVal->datum.i, pVal->node.resType.bytes, if (TSDB_CODE_SUCCESS == taosParseTime(pVal->literal, &pVal->datum.i, pVal->node.resType.bytes,
@ -588,62 +590,52 @@ static EDealRes translateValueImpl(STranslateContext* pCxt, SValueNode* pVal, SD
*(bool*)&pVal->typeData = pVal->datum.b; *(bool*)&pVal->typeData = pVal->datum.b;
break; break;
case TSDB_DATA_TYPE_TINYINT: { case TSDB_DATA_TYPE_TINYINT: {
char* endPtr = NULL; pVal->datum.i = taosStr2Int64(pVal->literal, NULL, 10);
pVal->datum.i = taosStr2Int64(pVal->literal, &endPtr, 10);
*(int8_t*)&pVal->typeData = pVal->datum.i; *(int8_t*)&pVal->typeData = pVal->datum.i;
break; break;
} }
case TSDB_DATA_TYPE_SMALLINT: { case TSDB_DATA_TYPE_SMALLINT: {
char* endPtr = NULL; pVal->datum.i = taosStr2Int64(pVal->literal, NULL, 10);
pVal->datum.i = taosStr2Int64(pVal->literal, &endPtr, 10);
*(int16_t*)&pVal->typeData = pVal->datum.i; *(int16_t*)&pVal->typeData = pVal->datum.i;
break; break;
} }
case TSDB_DATA_TYPE_INT: { case TSDB_DATA_TYPE_INT: {
char* endPtr = NULL; pVal->datum.i = taosStr2Int64(pVal->literal, NULL, 10);
pVal->datum.i = taosStr2Int64(pVal->literal, &endPtr, 10);
*(int32_t*)&pVal->typeData = pVal->datum.i; *(int32_t*)&pVal->typeData = pVal->datum.i;
break; break;
} }
case TSDB_DATA_TYPE_BIGINT: { case TSDB_DATA_TYPE_BIGINT: {
char* endPtr = NULL; pVal->datum.i = taosStr2Int64(pVal->literal, NULL, 10);
pVal->datum.i = taosStr2Int64(pVal->literal, &endPtr, 10);
*(int64_t*)&pVal->typeData = pVal->datum.i; *(int64_t*)&pVal->typeData = pVal->datum.i;
break; break;
} }
case TSDB_DATA_TYPE_UTINYINT: { case TSDB_DATA_TYPE_UTINYINT: {
char* endPtr = NULL; pVal->datum.u = taosStr2UInt64(pVal->literal, NULL, 10);
pVal->datum.u = taosStr2UInt64(pVal->literal, &endPtr, 10);
*(uint8_t*)&pVal->typeData = pVal->datum.u; *(uint8_t*)&pVal->typeData = pVal->datum.u;
break; break;
} }
case TSDB_DATA_TYPE_USMALLINT: { case TSDB_DATA_TYPE_USMALLINT: {
char* endPtr = NULL; pVal->datum.u = taosStr2UInt64(pVal->literal, NULL, 10);
pVal->datum.u = taosStr2UInt64(pVal->literal, &endPtr, 10);
*(uint16_t*)&pVal->typeData = pVal->datum.u; *(uint16_t*)&pVal->typeData = pVal->datum.u;
break; break;
} }
case TSDB_DATA_TYPE_UINT: { case TSDB_DATA_TYPE_UINT: {
char* endPtr = NULL; pVal->datum.u = taosStr2UInt64(pVal->literal, NULL, 10);
pVal->datum.u = taosStr2UInt64(pVal->literal, &endPtr, 10);
*(uint32_t*)&pVal->typeData = pVal->datum.u; *(uint32_t*)&pVal->typeData = pVal->datum.u;
break; break;
} }
case TSDB_DATA_TYPE_UBIGINT: { case TSDB_DATA_TYPE_UBIGINT: {
char* endPtr = NULL; pVal->datum.u = taosStr2UInt64(pVal->literal, NULL, 10);
pVal->datum.u = taosStr2UInt64(pVal->literal, &endPtr, 10);
*(uint64_t*)&pVal->typeData = pVal->datum.u; *(uint64_t*)&pVal->typeData = pVal->datum.u;
break; break;
} }
case TSDB_DATA_TYPE_FLOAT: { case TSDB_DATA_TYPE_FLOAT: {
char* endPtr = NULL; pVal->datum.d = taosStr2Double(pVal->literal, NULL);
pVal->datum.d = taosStr2Double(pVal->literal, &endPtr);
*(float*)&pVal->typeData = pVal->datum.d; *(float*)&pVal->typeData = pVal->datum.d;
break; break;
} }
case TSDB_DATA_TYPE_DOUBLE: { case TSDB_DATA_TYPE_DOUBLE: {
char* endPtr = NULL; pVal->datum.d = taosStr2Double(pVal->literal, NULL);
pVal->datum.d = taosStr2Double(pVal->literal, &endPtr);
*(double*)&pVal->typeData = pVal->datum.d; *(double*)&pVal->typeData = pVal->datum.d;
break; break;
} }
@ -659,7 +651,7 @@ static EDealRes translateValueImpl(STranslateContext* pCxt, SValueNode* pVal, SD
break; break;
} }
case TSDB_DATA_TYPE_TIMESTAMP: { case TSDB_DATA_TYPE_TIMESTAMP: {
if (TSDB_CODE_SUCCESS != parseTimeFromValueNode(pVal)) { if (TSDB_CODE_SUCCESS != parseTimeFromValueNode(pCxt, pVal)) {
return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pVal->literal); return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pVal->literal);
} }
*(int64_t*)&pVal->typeData = pVal->datum.i; *(int64_t*)&pVal->typeData = pVal->datum.i;
@ -3951,7 +3943,7 @@ typedef struct SVgroupCreateTableBatch {
static void destroyCreateTbReq(SVCreateTbReq* pReq) { static void destroyCreateTbReq(SVCreateTbReq* pReq) {
taosMemoryFreeClear(pReq->name); taosMemoryFreeClear(pReq->name);
taosMemoryFreeClear(pReq->ntb.schema.pSchema); taosMemoryFreeClear(pReq->ntb.schemaRow.pSchema);
} }
static int32_t buildNormalTableBatchReq(int32_t acctId, const SCreateTableStmt* pStmt, const SVgroupInfo* pVgroupInfo, static int32_t buildNormalTableBatchReq(int32_t acctId, const SCreateTableStmt* pStmt, const SVgroupInfo* pVgroupInfo,
@ -3964,10 +3956,10 @@ static int32_t buildNormalTableBatchReq(int32_t acctId, const SCreateTableStmt*
SVCreateTbReq req = {0}; SVCreateTbReq req = {0};
req.type = TD_NORMAL_TABLE; req.type = TD_NORMAL_TABLE;
req.name = strdup(pStmt->tableName); req.name = strdup(pStmt->tableName);
req.ntb.schema.nCols = LIST_LENGTH(pStmt->pCols); req.ntb.schemaRow.nCols = LIST_LENGTH(pStmt->pCols);
req.ntb.schema.sver = 1; req.ntb.schemaRow.version = 1;
req.ntb.schema.pSchema = taosMemoryCalloc(req.ntb.schema.nCols, sizeof(SSchema)); req.ntb.schemaRow.pSchema = taosMemoryCalloc(req.ntb.schemaRow.nCols, sizeof(SSchema));
if (NULL == req.name || NULL == req.ntb.schema.pSchema) { if (NULL == req.name || NULL == req.ntb.schemaRow.pSchema) {
destroyCreateTbReq(&req); destroyCreateTbReq(&req);
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
@ -3977,7 +3969,7 @@ static int32_t buildNormalTableBatchReq(int32_t acctId, const SCreateTableStmt*
SNode* pCol; SNode* pCol;
col_id_t index = 0; col_id_t index = 0;
FOREACH(pCol, pStmt->pCols) { FOREACH(pCol, pStmt->pCols) {
toSchema((SColumnDefNode*)pCol, index + 1, req.ntb.schema.pSchema + index); toSchema((SColumnDefNode*)pCol, index + 1, req.ntb.schemaRow.pSchema + index);
++index; ++index;
} }
pBatch->info = *pVgroupInfo; pBatch->info = *pVgroupInfo;
@ -4031,7 +4023,7 @@ static void destroyCreateTbReqBatch(SVgroupCreateTableBatch* pTbBatch) {
taosMemoryFreeClear(pTableReq->name); taosMemoryFreeClear(pTableReq->name);
if (pTableReq->type == TSDB_NORMAL_TABLE) { if (pTableReq->type == TSDB_NORMAL_TABLE) {
taosMemoryFreeClear(pTableReq->ntb.schema.pSchema); taosMemoryFreeClear(pTableReq->ntb.schemaRow.pSchema);
} else if (pTableReq->type == TSDB_CHILD_TABLE) { } else if (pTableReq->type == TSDB_CHILD_TABLE) {
taosMemoryFreeClear(pTableReq->ctb.pTag); taosMemoryFreeClear(pTableReq->ctb.pTag);
} }

16
source/libs/parser/src/parser.c
View File

@ -19,7 +19,7 @@
#include "parInt.h" #include "parInt.h"
#include "parToken.h" #include "parToken.h"
bool isInsertSql(const char* pStr, size_t length) { bool qIsInsertSql(const char* pStr, size_t length) {
if (NULL == pStr) { if (NULL == pStr) {
return false; return false;
} }
@ -169,7 +169,7 @@ static void rewriteExprAlias(SNode* pRoot) {
int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery) { int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
if (isInsertSql(pCxt->pSql, pCxt->sqlLen)) { if (qIsInsertSql(pCxt->pSql, pCxt->sqlLen)) {
code = parseInsertSql(pCxt, pQuery); code = parseInsertSql(pCxt, pQuery);
} else { } else {
code = parseSqlIntoAst(pCxt, pQuery); code = parseSqlIntoAst(pCxt, pQuery);
@ -184,6 +184,18 @@ int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** p
return extractResultSchema(pRoot, numOfCols, pSchema); return extractResultSchema(pRoot, numOfCols, pSchema);
} }
int32_t qSetSTableIdForRSma(SNode* pStmt, int64_t uid) {
if (QUERY_NODE_SELECT_STMT == nodeType(pStmt)) {
SNode* pTable = ((SSelectStmt*)pStmt)->pFromTable;
if (QUERY_NODE_REAL_TABLE == nodeType(pTable)) {
((SRealTableNode*)pTable)->pMeta->uid = uid;
((SRealTableNode*)pTable)->pMeta->suid = uid;
return TSDB_CODE_SUCCESS;
}
}
return TSDB_CODE_FAILED;
}
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx) { int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;