Merge pull request #15237 from taosdata/feature/3.0_debug_wxy
fix: plan problem caused by project eliminate optimize
This commit is contained in:
commit
dceec792bc
|
@ -197,6 +197,7 @@ bool fmIsSystemInfoFunc(int32_t funcId);
|
|||
bool fmIsImplicitTsFunc(int32_t funcId);
|
||||
bool fmIsClientPseudoColumnFunc(int32_t funcId);
|
||||
bool fmIsMultiRowsFunc(int32_t funcId);
|
||||
bool fmIsKeepOrderFunc(int32_t funcId);
|
||||
|
||||
int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);
|
||||
|
||||
|
|
|
@ -29,9 +29,17 @@ extern "C" {
|
|||
typedef enum EDataOrderLevel {
|
||||
DATA_ORDER_LEVEL_NONE = 1,
|
||||
DATA_ORDER_LEVEL_IN_BLOCK,
|
||||
DATA_ORDER_LEVEL_IN_GROUP
|
||||
DATA_ORDER_LEVEL_IN_GROUP,
|
||||
DATA_ORDER_LEVEL_GLOBAL
|
||||
} EDataOrderLevel;
|
||||
|
||||
typedef enum EGroupAction {
|
||||
GROUP_ACTION_NONE = 1,
|
||||
GROUP_ACTION_SET,
|
||||
GROUP_ACTION_KEEP,
|
||||
GROUP_ACTION_CLEAR
|
||||
} EGroupAction;
|
||||
|
||||
typedef struct SLogicNode {
|
||||
ENodeType type;
|
||||
SNodeList* pTargets; // SColumnNode
|
||||
|
@ -44,6 +52,7 @@ typedef struct SLogicNode {
|
|||
SNode* pSlimit;
|
||||
EDataOrderLevel requireDataOrder; // requirements for input data
|
||||
EDataOrderLevel resultDataOrder; // properties of the output data
|
||||
EGroupAction groupAction;
|
||||
} SLogicNode;
|
||||
|
||||
typedef enum EScanType {
|
||||
|
|
|
@ -2868,7 +2868,7 @@ int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scan
|
|||
*order = TSDB_ORDER_ASC;
|
||||
*scanFlag = MAIN_SCAN;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
} else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
|
||||
} else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN) {
|
||||
STableScanInfo* pTableScanInfo = pOperator->info;
|
||||
*order = pTableScanInfo->cond.order;
|
||||
*scanFlag = pTableScanInfo->scanFlag;
|
||||
|
|
|
@ -47,6 +47,7 @@ extern "C" {
|
|||
#define FUNC_MGT_SYSTEM_INFO_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(18)
|
||||
#define FUNC_MGT_CLIENT_PC_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(19)
|
||||
#define FUNC_MGT_MULTI_ROWS_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(20)
|
||||
#define FUNC_MGT_KEEP_ORDER_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(21)
|
||||
|
||||
#define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
|
||||
|
||||
|
|
|
@ -2097,7 +2097,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
|
|||
{
|
||||
.name = "top",
|
||||
.type = FUNCTION_TYPE_TOP,
|
||||
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
|
||||
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_KEEP_ORDER_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
|
||||
.translateFunc = translateTopBot,
|
||||
.getEnvFunc = getTopBotFuncEnv,
|
||||
.initFunc = topBotFunctionSetup,
|
||||
|
@ -2112,7 +2112,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
|
|||
{
|
||||
.name = "bottom",
|
||||
.type = FUNCTION_TYPE_BOTTOM,
|
||||
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
|
||||
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_KEEP_ORDER_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
|
||||
.translateFunc = translateTopBot,
|
||||
.getEnvFunc = getTopBotFuncEnv,
|
||||
.initFunc = topBotFunctionSetup,
|
||||
|
@ -2480,7 +2480,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
|
|||
{
|
||||
.name = "sample",
|
||||
.type = FUNCTION_TYPE_SAMPLE,
|
||||
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
|
||||
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_MULTI_ROWS_FUNC | FUNC_MGT_KEEP_ORDER_FUNC | FUNC_MGT_FORBID_STREAM_FUNC,
|
||||
.translateFunc = translateSample,
|
||||
.getEnvFunc = getSampleFuncEnv,
|
||||
.initFunc = sampleFunctionSetup,
|
||||
|
@ -2906,7 +2906,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
|
|||
{
|
||||
.name = "_select_value",
|
||||
.type = FUNCTION_TYPE_SELECT_VALUE,
|
||||
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC,
|
||||
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_KEEP_ORDER_FUNC,
|
||||
.translateFunc = translateSelectValue,
|
||||
.getEnvFunc = getSelectivityFuncEnv, // todo remove this function later.
|
||||
.initFunc = functionSetup,
|
||||
|
|
|
@ -183,6 +183,8 @@ bool fmIsClientPseudoColumnFunc(int32_t funcId) { return isSpecificClassifyFunc(
|
|||
|
||||
bool fmIsMultiRowsFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_MULTI_ROWS_FUNC); }
|
||||
|
||||
bool fmIsKeepOrderFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_KEEP_ORDER_FUNC); }
|
||||
|
||||
bool fmIsInterpFunc(int32_t funcId) {
|
||||
if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
|
||||
return false;
|
||||
|
|
|
@ -332,6 +332,9 @@ static int32_t logicNodeCopy(const SLogicNode* pSrc, SLogicNode* pDst) {
|
|||
COPY_SCALAR_FIELD(precision);
|
||||
CLONE_NODE_FIELD(pLimit);
|
||||
CLONE_NODE_FIELD(pSlimit);
|
||||
COPY_SCALAR_FIELD(requireDataOrder);
|
||||
COPY_SCALAR_FIELD(resultDataOrder);
|
||||
COPY_SCALAR_FIELD(groupAction);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
|
|
@ -504,6 +504,9 @@ static const char* jkLogicPlanConditions = "Conditions";
|
|||
static const char* jkLogicPlanChildren = "Children";
|
||||
static const char* jkLogicPlanLimit = "Limit";
|
||||
static const char* jkLogicPlanSlimit = "SLimit";
|
||||
static const char* jkLogicPlanRequireDataOrder = "RequireDataOrder";
|
||||
static const char* jkLogicPlanResultDataOrder = "ResultDataOrder";
|
||||
static const char* jkLogicPlanGroupAction = "GroupAction";
|
||||
|
||||
static int32_t logicPlanNodeToJson(const void* pObj, SJson* pJson) {
|
||||
const SLogicNode* pNode = (const SLogicNode*)pObj;
|
||||
|
@ -521,6 +524,15 @@ static int32_t logicPlanNodeToJson(const void* pObj, SJson* pJson) {
|
|||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = tjsonAddObject(pJson, jkLogicPlanSlimit, nodeToJson, pNode->pSlimit);
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = tjsonAddIntegerToObject(pJson, jkLogicPlanRequireDataOrder, pNode->requireDataOrder);
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = tjsonAddIntegerToObject(pJson, jkLogicPlanResultDataOrder, pNode->resultDataOrder);
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = tjsonAddIntegerToObject(pJson, jkLogicPlanGroupAction, pNode->groupAction);
|
||||
}
|
||||
|
||||
return code;
|
||||
}
|
||||
|
@ -541,6 +553,15 @@ static int32_t jsonToLogicPlanNode(const SJson* pJson, void* pObj) {
|
|||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = jsonToNodeObject(pJson, jkLogicPlanSlimit, &pNode->pSlimit);
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
tjsonGetNumberValue(pJson, jkLogicPlanRequireDataOrder, pNode->requireDataOrder, code);
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
tjsonGetNumberValue(pJson, jkLogicPlanResultDataOrder, pNode->resultDataOrder, code);
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
tjsonGetNumberValue(pJson, jkLogicPlanGroupAction, pNode->groupAction, code);
|
||||
}
|
||||
|
||||
return code;
|
||||
}
|
||||
|
|
|
@ -434,8 +434,12 @@ static FORCE_INLINE int32_t checkAndTrimValue(SToken* pToken, char* tmpTokenBuf,
|
|||
}
|
||||
|
||||
static bool isNullStr(SToken* pToken) {
|
||||
return (pToken->type == TK_NULL) || ((pToken->type == TK_NK_STRING) && (pToken->n != 0) &&
|
||||
(strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0));
|
||||
return ((pToken->type == TK_NK_STRING) && (pToken->n != 0) &&
|
||||
(strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0));
|
||||
}
|
||||
|
||||
static bool isNullValue(int8_t dataType, SToken* pToken) {
|
||||
return TK_NULL == pToken->type || (!IS_STR_DATA_TYPE(dataType) && isNullStr(pToken));
|
||||
}
|
||||
|
||||
static FORCE_INLINE int32_t toDouble(SToken* pToken, double* value, char** endPtr) {
|
||||
|
@ -461,7 +465,7 @@ static int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int
|
|||
return code;
|
||||
}
|
||||
|
||||
if (isNullStr(pToken)) {
|
||||
if (isNullValue(pSchema->type, pToken)) {
|
||||
if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type && PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
|
||||
return buildSyntaxErrMsg(pMsgBuf, "primary timestamp should not be null", pToken->z);
|
||||
}
|
||||
|
@ -735,11 +739,12 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, SParsedDataColInfo*
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static void buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTag, int64_t suid, const char* sname, SArray* tagName) {
|
||||
static void buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTag, int64_t suid, const char* sname,
|
||||
SArray* tagName) {
|
||||
pTbReq->type = TD_CHILD_TABLE;
|
||||
pTbReq->name = strdup(tname);
|
||||
pTbReq->ctb.suid = suid;
|
||||
if(sname) pTbReq->ctb.name = strdup(sname);
|
||||
if (sname) pTbReq->ctb.name = strdup(sname);
|
||||
pTbReq->ctb.pTag = (uint8_t*)pTag;
|
||||
pTbReq->ctb.tagName = taosArrayDup(tagName);
|
||||
pTbReq->commentLen = -1;
|
||||
|
@ -753,7 +758,7 @@ static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema, int16
|
|||
uint64_t uv;
|
||||
char* endptr = NULL;
|
||||
|
||||
if (isNullStr(pToken)) {
|
||||
if (isNullValue(pSchema->type, pToken)) {
|
||||
if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type && PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
|
||||
return buildSyntaxErrMsg(pMsgBuf, "primary timestamp should not be null", pToken->z);
|
||||
}
|
||||
|
@ -761,7 +766,7 @@ static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema, int16
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
// strcpy(val->colName, pSchema->name);
|
||||
// strcpy(val->colName, pSchema->name);
|
||||
val->cid = pSchema->colId;
|
||||
val->type = pSchema->type;
|
||||
|
||||
|
@ -971,7 +976,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
|
|||
goto end;
|
||||
}
|
||||
|
||||
if (!isNullStr(&sToken)) {
|
||||
if (!isNullValue(pTagSchema->type, &sToken)) {
|
||||
taosArrayPush(tagName, pTagSchema->name);
|
||||
}
|
||||
if (pTagSchema->type == TSDB_DATA_TYPE_JSON) {
|
||||
|
@ -980,7 +985,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
|
|||
taosMemoryFree(tmpTokenBuf);
|
||||
goto end;
|
||||
}
|
||||
if (isNullStr(&sToken)) {
|
||||
if (isNullValue(pTagSchema->type, &sToken)) {
|
||||
code = tTagNew(pTagVals, 1, true, &pTag);
|
||||
} else {
|
||||
code = parseJsontoTagData(sToken.z, pTagVals, &pTag, &pCxt->msg);
|
||||
|
@ -1321,7 +1326,11 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, TdFilePtr fp, STableDataB
|
|||
|
||||
static int32_t parseDataFromFile(SInsertParseContext* pCxt, SToken filePath, STableDataBlocks* dataBuf) {
|
||||
char filePathStr[TSDB_FILENAME_LEN] = {0};
|
||||
strncpy(filePathStr, filePath.z, filePath.n);
|
||||
if (TK_NK_STRING == filePath.type) {
|
||||
trimString(filePath.z, filePath.n, filePathStr, sizeof(filePathStr));
|
||||
} else {
|
||||
strncpy(filePathStr, filePath.z, filePath.n);
|
||||
}
|
||||
TdFilePtr fp = taosOpenFile(filePathStr, TD_FILE_READ | TD_FILE_STREAM);
|
||||
if (NULL == fp) {
|
||||
return TAOS_SYSTEM_ERROR(errno);
|
||||
|
@ -1804,8 +1813,8 @@ int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const char* sTableName, char* tName, TAOS_MULTI_BIND* bind,
|
||||
char* msgBuf, int32_t msgBufLen) {
|
||||
int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const char* sTableName, char* tName,
|
||||
TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen) {
|
||||
STableDataBlocks* pDataBlock = (STableDataBlocks*)pBlock;
|
||||
SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};
|
||||
SParsedDataColInfo* tags = (SParsedDataColInfo*)boundTags;
|
||||
|
@ -1856,7 +1865,7 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const ch
|
|||
}
|
||||
} else {
|
||||
STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
|
||||
// strcpy(val.colName, pTagSchema->name);
|
||||
// strcpy(val.colName, pTagSchema->name);
|
||||
if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) {
|
||||
val.pData = (uint8_t*)bind[c].buffer;
|
||||
val.nData = colLen;
|
||||
|
@ -2247,7 +2256,8 @@ static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SS
|
|||
* @param msg
|
||||
* @return int32_t
|
||||
*/
|
||||
static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* pSchema, STag** ppTag, SArray** tagName, SMsgBuf* msg) {
|
||||
static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* pSchema, STag** ppTag, SArray** tagName,
|
||||
SMsgBuf* msg) {
|
||||
SArray* pTagArray = taosArrayInit(tags->numOfBound, sizeof(STagVal));
|
||||
if (!pTagArray) {
|
||||
return TSDB_CODE_TSC_OUT_OF_MEMORY;
|
||||
|
@ -2264,7 +2274,7 @@ static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* p
|
|||
|
||||
taosArrayPush(*tagName, pTagSchema->name);
|
||||
STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
|
||||
// strcpy(val.colName, pTagSchema->name);
|
||||
// strcpy(val.colName, pTagSchema->name);
|
||||
if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) {
|
||||
val.pData = (uint8_t*)kv->value;
|
||||
val.nData = kv->length;
|
||||
|
@ -2320,7 +2330,7 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols
|
|||
buildInvalidOperationMsg(&pBuf, "bound tags error");
|
||||
return ret;
|
||||
}
|
||||
STag* pTag = NULL;
|
||||
STag* pTag = NULL;
|
||||
SArray* tagName = NULL;
|
||||
ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tags, pTagsSchema, &pTag, &tagName, &pBuf);
|
||||
if (ret != TSDB_CODE_SUCCESS) {
|
||||
|
@ -2404,9 +2414,9 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols
|
|||
} else {
|
||||
int32_t colLen = kv->length;
|
||||
if (pColSchema->type == TSDB_DATA_TYPE_TIMESTAMP) {
|
||||
// uError("SML:data before:%ld, precision:%d", kv->i, pTableMeta->tableInfo.precision);
|
||||
// uError("SML:data before:%ld, precision:%d", kv->i, pTableMeta->tableInfo.precision);
|
||||
kv->i = convertTimePrecision(kv->i, TSDB_TIME_PRECISION_NANO, pTableMeta->tableInfo.precision);
|
||||
// uError("SML:data after:%ld, precision:%d", kv->i, pTableMeta->tableInfo.precision);
|
||||
// uError("SML:data after:%ld, precision:%d", kv->i, pTableMeta->tableInfo.precision);
|
||||
}
|
||||
|
||||
if (IS_VAR_DATA_TYPE(kv->type)) {
|
||||
|
|
|
@ -1089,7 +1089,7 @@ static int32_t translateScanPseudoColumnFunc(STranslateContext* pCxt, SFunctionN
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
if (0 == LIST_LENGTH(pFunc->pParameterList)) {
|
||||
if (!isSelectStmt(pCxt->pCurrStmt) ||
|
||||
if (!isSelectStmt(pCxt->pCurrStmt) || NULL == ((SSelectStmt*)pCxt->pCurrStmt)->pFromTable ||
|
||||
QUERY_NODE_REAL_TABLE != nodeType(((SSelectStmt*)pCxt->pCurrStmt)->pFromTable)) {
|
||||
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_TBNAME);
|
||||
}
|
||||
|
|
|
@ -36,7 +36,7 @@ bool qIsInsertValuesSql(const char* pStr, size_t length) {
|
|||
pStr += index;
|
||||
index = 0;
|
||||
t = tStrGetToken((char*)pStr, &index, false);
|
||||
if (TK_USING == t.type || TK_VALUES == t.type) {
|
||||
if (TK_USING == t.type || TK_VALUES == t.type || TK_FILE == t.type) {
|
||||
return true;
|
||||
} else if (TK_SELECT == t.type) {
|
||||
return false;
|
||||
|
|
|
@ -444,4 +444,11 @@ TEST_F(ParserSelectTest, withoutFrom) {
|
|||
run("SELECT USER()");
|
||||
}
|
||||
|
||||
TEST_F(ParserSelectTest, withoutFromSemanticCheck) {
|
||||
useDb("root", "test");
|
||||
|
||||
run("SELECT c1", TSDB_CODE_PAR_INVALID_COLUMN);
|
||||
run("SELECT TBNAME", TSDB_CODE_PAR_INVALID_TBNAME);
|
||||
}
|
||||
|
||||
} // namespace ParserTest
|
||||
|
|
|
@ -35,6 +35,7 @@ int32_t generateUsageErrMsg(char* pBuf, int32_t len, int32_t errCode, ...);
|
|||
int32_t createColumnByRewriteExprs(SNodeList* pExprs, SNodeList** pList);
|
||||
int32_t createColumnByRewriteExpr(SNode* pExpr, SNodeList** pList);
|
||||
int32_t replaceLogicNode(SLogicSubplan* pSubplan, SLogicNode* pOld, SLogicNode* pNew);
|
||||
int32_t adjustLogicNodeDataRequirement(SLogicNode* pNode, EDataOrderLevel requirement);
|
||||
|
||||
int32_t createLogicPlan(SPlanContext* pCxt, SLogicSubplan** pLogicSubplan);
|
||||
int32_t optimizeLogicPlan(SPlanContext* pCxt, SLogicSubplan* pLogicSubplan);
|
||||
|
|
|
@ -250,6 +250,9 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
|
|||
SScanLogicNode* pScan = NULL;
|
||||
int32_t code = makeScanLogicNode(pCxt, pRealTable, pSelect->hasRepeatScanFuncs, (SLogicNode**)&pScan);
|
||||
|
||||
pScan->node.groupAction = GROUP_ACTION_NONE;
|
||||
pScan->node.resultDataOrder = DATA_ORDER_LEVEL_IN_BLOCK;
|
||||
|
||||
// set columns to scan
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = nodesCollectColumns(pSelect, SQL_CLAUSE_FROM, pRealTable->table.tableAlias, COLLECT_COL_TYPE_COL,
|
||||
|
@ -336,6 +339,9 @@ static int32_t createJoinLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
|
|||
|
||||
pJoin->joinType = pJoinTable->joinType;
|
||||
pJoin->isSingleTableJoin = pJoinTable->table.singleTable;
|
||||
pJoin->node.groupAction = GROUP_ACTION_CLEAR;
|
||||
pJoin->node.requireDataOrder = DATA_ORDER_LEVEL_GLOBAL;
|
||||
pJoin->node.requireDataOrder = DATA_ORDER_LEVEL_GLOBAL;
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
|
||||
|
@ -472,6 +478,9 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
|
|||
}
|
||||
|
||||
pAgg->hasLastRow = pSelect->hasLastRowFunc;
|
||||
pAgg->node.groupAction = GROUP_ACTION_SET;
|
||||
pAgg->node.requireDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
pAgg->node.resultDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
|
||||
|
@ -540,6 +549,10 @@ static int32_t createIndefRowsFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt
|
|||
pIdfRowsFunc->isTailFunc = pSelect->hasTailFunc;
|
||||
pIdfRowsFunc->isUniqueFunc = pSelect->hasUniqueFunc;
|
||||
pIdfRowsFunc->isTimeLineFunc = pSelect->hasTimeLineFunc;
|
||||
pIdfRowsFunc->node.groupAction = GROUP_ACTION_KEEP;
|
||||
pIdfRowsFunc->node.requireDataOrder =
|
||||
pIdfRowsFunc->isTimeLineFunc ? DATA_ORDER_LEVEL_IN_GROUP : DATA_ORDER_LEVEL_NONE;
|
||||
pIdfRowsFunc->node.resultDataOrder = pIdfRowsFunc->node.requireDataOrder;
|
||||
|
||||
// indefinite rows functions and _select_values functions
|
||||
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsVectorFunc, &pIdfRowsFunc->pFuncs);
|
||||
|
@ -571,6 +584,10 @@ static int32_t createInterpFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt* p
|
|||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
pInterpFunc->node.groupAction = GROUP_ACTION_KEEP;
|
||||
pInterpFunc->node.requireDataOrder = DATA_ORDER_LEVEL_IN_GROUP;
|
||||
pInterpFunc->node.resultDataOrder = pInterpFunc->node.requireDataOrder;
|
||||
|
||||
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsInterpFunc, &pInterpFunc->pFuncs);
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = rewriteExprsForSelect(pInterpFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT);
|
||||
|
@ -642,10 +659,12 @@ static int32_t createWindowLogicNodeByState(SLogicPlanContext* pCxt, SStateWindo
|
|||
}
|
||||
|
||||
pWindow->winType = WINDOW_TYPE_STATE;
|
||||
pWindow->node.groupAction = GROUP_ACTION_KEEP;
|
||||
pWindow->node.requireDataOrder = pCxt->pPlanCxt->streamQuery ? DATA_ORDER_LEVEL_IN_BLOCK : DATA_ORDER_LEVEL_IN_GROUP;
|
||||
pWindow->node.resultDataOrder = DATA_ORDER_LEVEL_IN_GROUP;
|
||||
pWindow->pStateExpr = nodesCloneNode(pState->pExpr);
|
||||
|
||||
pWindow->pTspk = nodesCloneNode(pState->pCol);
|
||||
if (NULL == pWindow->pTspk) {
|
||||
if (NULL == pWindow->pStateExpr || NULL == pWindow->pTspk) {
|
||||
nodesDestroyNode((SNode*)pWindow);
|
||||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
|
@ -663,6 +682,9 @@ static int32_t createWindowLogicNodeBySession(SLogicPlanContext* pCxt, SSessionW
|
|||
pWindow->winType = WINDOW_TYPE_SESSION;
|
||||
pWindow->sessionGap = ((SValueNode*)pSession->pGap)->datum.i;
|
||||
pWindow->windowAlgo = pCxt->pPlanCxt->streamQuery ? SESSION_ALGO_STREAM_SINGLE : SESSION_ALGO_MERGE;
|
||||
pWindow->node.groupAction = GROUP_ACTION_KEEP;
|
||||
pWindow->node.requireDataOrder = pCxt->pPlanCxt->streamQuery ? DATA_ORDER_LEVEL_IN_BLOCK : DATA_ORDER_LEVEL_IN_GROUP;
|
||||
pWindow->node.resultDataOrder = DATA_ORDER_LEVEL_IN_GROUP;
|
||||
|
||||
pWindow->pTspk = nodesCloneNode((SNode*)pSession->pCol);
|
||||
if (NULL == pWindow->pTspk) {
|
||||
|
@ -689,6 +711,9 @@ static int32_t createWindowLogicNodeByInterval(SLogicPlanContext* pCxt, SInterva
|
|||
pWindow->slidingUnit =
|
||||
(NULL != pInterval->pSliding ? ((SValueNode*)pInterval->pSliding)->unit : pWindow->intervalUnit);
|
||||
pWindow->windowAlgo = pCxt->pPlanCxt->streamQuery ? INTERVAL_ALGO_STREAM_SINGLE : INTERVAL_ALGO_HASH;
|
||||
pWindow->node.groupAction = GROUP_ACTION_KEEP;
|
||||
pWindow->node.requireDataOrder = DATA_ORDER_LEVEL_IN_BLOCK;
|
||||
pWindow->node.resultDataOrder = DATA_ORDER_LEVEL_IN_GROUP;
|
||||
|
||||
pWindow->pTspk = nodesCloneNode(pInterval->pCol);
|
||||
if (NULL == pWindow->pTspk) {
|
||||
|
@ -734,6 +759,10 @@ static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
|
|||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
pFill->node.groupAction = GROUP_ACTION_KEEP;
|
||||
pFill->node.requireDataOrder = DATA_ORDER_LEVEL_IN_GROUP;
|
||||
pFill->node.resultDataOrder = DATA_ORDER_LEVEL_IN_GROUP;
|
||||
|
||||
int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_WINDOW, NULL, COLLECT_COL_TYPE_ALL, &pFill->node.pTargets);
|
||||
if (TSDB_CODE_SUCCESS == code && NULL == pFill->node.pTargets) {
|
||||
code = nodesListMakeStrictAppend(&pFill->node.pTargets,
|
||||
|
@ -768,6 +797,9 @@ static int32_t createSortLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
|
|||
}
|
||||
|
||||
pSort->groupSort = pSelect->groupSort;
|
||||
pSort->node.groupAction = pSort->groupSort ? GROUP_ACTION_KEEP : GROUP_ACTION_CLEAR;
|
||||
pSort->node.requireDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
pSort->node.resultDataOrder = pSort->groupSort ? DATA_ORDER_LEVEL_IN_GROUP : DATA_ORDER_LEVEL_GLOBAL;
|
||||
|
||||
int32_t code = nodesCollectColumns(pSelect, SQL_CLAUSE_ORDER_BY, NULL, COLLECT_COL_TYPE_ALL, &pSort->node.pTargets);
|
||||
if (TSDB_CODE_SUCCESS == code && NULL == pSort->node.pTargets) {
|
||||
|
@ -818,6 +850,9 @@ static int32_t createProjectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSel
|
|||
|
||||
TSWAP(pProject->node.pLimit, pSelect->pLimit);
|
||||
TSWAP(pProject->node.pSlimit, pSelect->pSlimit);
|
||||
pProject->node.groupAction = GROUP_ACTION_CLEAR;
|
||||
pProject->node.requireDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
pProject->node.resultDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
|
||||
|
@ -850,6 +885,10 @@ static int32_t createPartitionLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pS
|
|||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
pPartition->node.groupAction = GROUP_ACTION_SET;
|
||||
pPartition->node.requireDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
pPartition->node.resultDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
|
||||
int32_t code =
|
||||
nodesCollectColumns(pSelect, SQL_CLAUSE_PARTITION_BY, NULL, COLLECT_COL_TYPE_ALL, &pPartition->node.pTargets);
|
||||
if (TSDB_CODE_SUCCESS == code && NULL == pPartition->node.pTargets) {
|
||||
|
@ -882,6 +921,10 @@ static int32_t createDistinctLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSe
|
|||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
pAgg->node.groupAction = GROUP_ACTION_SET;
|
||||
pAgg->node.requireDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
pAgg->node.resultDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
// set grouyp keys, agg funcs and having conditions
|
||||
SNodeList* pGroupKeys = NULL;
|
||||
|
@ -1369,6 +1412,7 @@ int32_t createLogicPlan(SPlanContext* pCxt, SLogicSubplan** pLogicSubplan) {
|
|||
if (TSDB_CODE_SUCCESS == code) {
|
||||
setLogicNodeParent(pSubplan->pNode);
|
||||
setLogicSubplanType(cxt.hasScan, pSubplan);
|
||||
code = adjustLogicNodeDataRequirement(pSubplan->pNode, DATA_ORDER_LEVEL_NONE);
|
||||
}
|
||||
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
|
|
|
@ -1579,6 +1579,34 @@ static bool eliminateProjOptMayBeOptimized(SLogicNode* pNode) {
|
|||
return eliminateProjOptCheckProjColumnNames(pProjectNode);
|
||||
}
|
||||
|
||||
typedef struct CheckNewChildTargetsCxt {
|
||||
SNodeList* pNewChildTargets;
|
||||
bool canUse;
|
||||
} CheckNewChildTargetsCxt;
|
||||
|
||||
static EDealRes eliminateProjOptCanUseNewChildTargetsImpl(SNode* pNode, void* pContext) {
|
||||
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
|
||||
CheckNewChildTargetsCxt* pCxt = pContext;
|
||||
SNode* pTarget = NULL;
|
||||
FOREACH(pTarget, pCxt->pNewChildTargets) {
|
||||
if (!nodesEqualNode(pTarget, pNode)) {
|
||||
pCxt->canUse = false;
|
||||
return DEAL_RES_END;
|
||||
}
|
||||
}
|
||||
}
|
||||
return DEAL_RES_CONTINUE;
|
||||
}
|
||||
|
||||
static bool eliminateProjOptCanUseNewChildTargets(SLogicNode* pChild, SNodeList* pNewChildTargets) {
|
||||
if (NULL == pChild->pConditions) {
|
||||
return true;
|
||||
}
|
||||
CheckNewChildTargetsCxt cxt = {.pNewChildTargets = pNewChildTargets, .canUse = true};
|
||||
nodesWalkExpr(pChild->pConditions, eliminateProjOptCanUseNewChildTargetsImpl, &cxt);
|
||||
return cxt.canUse;
|
||||
}
|
||||
|
||||
static int32_t eliminateProjOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubplan,
|
||||
SProjectLogicNode* pProjectNode) {
|
||||
SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pProjectNode->node.pChildren, 0);
|
||||
|
@ -1594,8 +1622,13 @@ static int32_t eliminateProjOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan*
|
|||
}
|
||||
}
|
||||
}
|
||||
nodesDestroyList(pChild->pTargets);
|
||||
pChild->pTargets = pNewChildTargets;
|
||||
if (eliminateProjOptCanUseNewChildTargets(pChild, pNewChildTargets)) {
|
||||
nodesDestroyList(pChild->pTargets);
|
||||
pChild->pTargets = pNewChildTargets;
|
||||
} else {
|
||||
nodesDestroyList(pNewChildTargets);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t code = replaceLogicNode(pLogicSubplan, (SLogicNode*)pProjectNode, pChild);
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
|
@ -1873,6 +1906,8 @@ static int32_t rewriteUniqueOptCreateAgg(SIndefRowsFuncLogicNode* pIndef, SLogic
|
|||
TSWAP(pAgg->node.pChildren, pIndef->node.pChildren);
|
||||
optResetParent((SLogicNode*)pAgg);
|
||||
pAgg->node.precision = pIndef->node.precision;
|
||||
pAgg->node.requireDataOrder = DATA_ORDER_LEVEL_IN_BLOCK; // first function requirement
|
||||
pAgg->node.resultDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
bool hasSelectPrimaryKey = false;
|
||||
|
@ -1945,6 +1980,8 @@ static int32_t rewriteUniqueOptCreateProject(SIndefRowsFuncLogicNode* pIndef, SL
|
|||
|
||||
TSWAP(pProject->node.pTargets, pIndef->node.pTargets);
|
||||
pProject->node.precision = pIndef->node.precision;
|
||||
pProject->node.requireDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
pProject->node.resultDataOrder = DATA_ORDER_LEVEL_NONE;
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
SNode* pNode = NULL;
|
||||
|
@ -1973,12 +2010,17 @@ static int32_t rewriteUniqueOptimizeImpl(SOptimizeContext* pCxt, SLogicSubplan*
|
|||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = nodesListMakeAppend(&pProject->pChildren, (SNode*)pAgg);
|
||||
pAgg->pParent = pProject;
|
||||
pAgg = NULL;
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
pAgg->pParent = pProject;
|
||||
pAgg = NULL;
|
||||
code = replaceLogicNode(pLogicSubplan, (SLogicNode*)pIndef, pProject);
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
code = adjustLogicNodeDataRequirement(
|
||||
pProject, NULL == pProject->pParent ? DATA_ORDER_LEVEL_NONE : pProject->pParent->requireDataOrder);
|
||||
pProject = NULL;
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
nodesDestroyNode((SNode*)pIndef);
|
||||
} else {
|
||||
|
@ -2145,8 +2187,8 @@ static bool tagScanMayBeOptimized(SLogicNode* pNode) {
|
|||
}
|
||||
|
||||
SAggLogicNode* pAgg = (SAggLogicNode*)(pNode->pParent);
|
||||
if (NULL == pAgg->pGroupKeys || NULL != pAgg->pAggFuncs ||
|
||||
planOptNodeListHasCol(pAgg->pGroupKeys) || !planOptNodeListHasTbname(pAgg->pGroupKeys)) {
|
||||
if (NULL == pAgg->pGroupKeys || NULL != pAgg->pAggFuncs || planOptNodeListHasCol(pAgg->pGroupKeys) ||
|
||||
!planOptNodeListHasTbname(pAgg->pGroupKeys)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
|
@ -2171,11 +2213,12 @@ static int32_t tagScanOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubp
|
|||
pScanNode->scanType = SCAN_TYPE_TAG;
|
||||
SNode* pTarget = NULL;
|
||||
FOREACH(pTarget, pScanNode->node.pTargets) {
|
||||
if (PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)(pTarget))->colId) {
|
||||
ERASE_NODE(pScanNode->node.pTargets);
|
||||
break;
|
||||
}
|
||||
if (PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)(pTarget))->colId) {
|
||||
ERASE_NODE(pScanNode->node.pTargets);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
NODES_DESTORY_LIST(pScanNode->pScanCols);
|
||||
|
||||
SLogicNode* pAgg = pScanNode->node.pParent;
|
||||
|
@ -2185,8 +2228,8 @@ static int32_t tagScanOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSubp
|
|||
SNode* pAggTarget = NULL;
|
||||
FOREACH(pAggTarget, pAgg->pTargets) {
|
||||
SNode* pScanTarget = NULL;
|
||||
FOREACH(pScanTarget, pScanNode->node.pTargets) {
|
||||
if (0 == strcmp( ((SColumnNode*)pAggTarget)->colName, ((SColumnNode*)pAggTarget)->colName )) {
|
||||
FOREACH(pScanTarget, pScanNode->node.pTargets) {
|
||||
if (0 == strcmp(((SColumnNode*)pAggTarget)->colName, ((SColumnNode*)pAggTarget)->colName)) {
|
||||
nodesListAppend(pScanTargets, nodesCloneNode(pScanTarget));
|
||||
break;
|
||||
}
|
||||
|
|
|
@ -974,6 +974,17 @@ static int32_t createInterpFuncPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pCh
|
|||
return code;
|
||||
}
|
||||
|
||||
static bool projectCanMergeDataBlock(SProjectLogicNode* pProject) {
|
||||
if (DATA_ORDER_LEVEL_NONE == pProject->node.resultDataOrder) {
|
||||
return true;
|
||||
}
|
||||
if (1 != LIST_LENGTH(pProject->node.pChildren)) {
|
||||
return false;
|
||||
}
|
||||
SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pProject->node.pChildren, 0);
|
||||
return DATA_ORDER_LEVEL_GLOBAL == pChild->resultDataOrder ? true : false;
|
||||
}
|
||||
|
||||
static int32_t createProjectPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChildren,
|
||||
SProjectLogicNode* pProjectLogicNode, SPhysiNode** pPhyNode) {
|
||||
SProjectPhysiNode* pProject =
|
||||
|
@ -982,6 +993,8 @@ static int32_t createProjectPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChild
|
|||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
pProject->mergeDataBlock = projectCanMergeDataBlock(pProjectLogicNode);
|
||||
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
if (0 == LIST_LENGTH(pChildren)) {
|
||||
pProject->pProjections = nodesCloneList(pProjectLogicNode->pProjections);
|
||||
|
|
|
@ -657,6 +657,9 @@ static int32_t stbSplSplitWindowForPartTable(SSplitContext* pCxt, SStableSplitIn
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
if (NULL != pInfo->pSplitNode->pParent && QUERY_NODE_LOGIC_PLAN_FILL == nodeType(pInfo->pSplitNode->pParent)) {
|
||||
pInfo->pSplitNode = pInfo->pSplitNode->pParent;
|
||||
}
|
||||
SExchangeLogicNode* pExchange = NULL;
|
||||
int32_t code = splCreateExchangeNode(pCxt, pInfo->pSplitNode, &pExchange);
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
|
|
|
@ -13,6 +13,7 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include "functionMgt.h"
|
||||
#include "planInt.h"
|
||||
|
||||
static char* getUsageErrFormat(int32_t errCode) {
|
||||
|
@ -121,3 +122,185 @@ int32_t replaceLogicNode(SLogicSubplan* pSubplan, SLogicNode* pOld, SLogicNode*
|
|||
}
|
||||
return TSDB_CODE_PLAN_INTERNAL_ERROR;
|
||||
}
|
||||
|
||||
static int32_t adjustScanDataRequirement(SScanLogicNode* pScan, EDataOrderLevel requirement) {
|
||||
if (SCAN_TYPE_TABLE != pScan->scanType || SCAN_TYPE_TABLE_MERGE != pScan->scanType) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
// The lowest sort level of scan output data is DATA_ORDER_LEVEL_IN_BLOCK
|
||||
if (requirement < DATA_ORDER_LEVEL_IN_BLOCK) {
|
||||
requirement = DATA_ORDER_LEVEL_IN_BLOCK;
|
||||
}
|
||||
if (DATA_ORDER_LEVEL_IN_BLOCK == requirement) {
|
||||
pScan->scanType = SCAN_TYPE_TABLE;
|
||||
} else if (TSDB_SUPER_TABLE == pScan->tableType) {
|
||||
pScan->scanType = SCAN_TYPE_TABLE_MERGE;
|
||||
}
|
||||
pScan->node.resultDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustJoinDataRequirement(SJoinLogicNode* pJoin, EDataOrderLevel requirement) {
|
||||
// The lowest sort level of join input and output data is DATA_ORDER_LEVEL_GLOBAL
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static bool isKeepOrderAggFunc(SNodeList* pFuncs) {
|
||||
SNode* pFunc = NULL;
|
||||
FOREACH(pFunc, pFuncs) {
|
||||
if (!fmIsKeepOrderFunc(((SFunctionNode*)pFunc)->funcId)) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
static int32_t adjustAggDataRequirement(SAggLogicNode* pAgg, EDataOrderLevel requirement) {
|
||||
// The sort level of agg with group by output data can only be DATA_ORDER_LEVEL_NONE
|
||||
if (requirement > DATA_ORDER_LEVEL_NONE && (NULL != pAgg->pGroupKeys || !isKeepOrderAggFunc(pAgg->pAggFuncs))) {
|
||||
return TSDB_CODE_PLAN_INTERNAL_ERROR;
|
||||
}
|
||||
pAgg->node.resultDataOrder = requirement;
|
||||
pAgg->node.requireDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustProjectDataRequirement(SProjectLogicNode* pProject, EDataOrderLevel requirement) {
|
||||
pProject->node.resultDataOrder = requirement;
|
||||
pProject->node.requireDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustIntervalDataRequirement(SWindowLogicNode* pWindow, EDataOrderLevel requirement) {
|
||||
// The lowest sort level of interval output data is DATA_ORDER_LEVEL_IN_GROUP
|
||||
if (requirement < DATA_ORDER_LEVEL_IN_GROUP) {
|
||||
requirement = DATA_ORDER_LEVEL_IN_GROUP;
|
||||
}
|
||||
// The sort level of interval input data is always DATA_ORDER_LEVEL_IN_BLOCK
|
||||
pWindow->node.resultDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustSessionDataRequirement(SWindowLogicNode* pWindow, EDataOrderLevel requirement) {
|
||||
if (requirement <= pWindow->node.resultDataOrder) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
pWindow->node.resultDataOrder = requirement;
|
||||
pWindow->node.requireDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustStateDataRequirement(SWindowLogicNode* pWindow, EDataOrderLevel requirement) {
|
||||
if (requirement <= pWindow->node.resultDataOrder) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
pWindow->node.resultDataOrder = requirement;
|
||||
pWindow->node.requireDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustWindowDataRequirement(SWindowLogicNode* pWindow, EDataOrderLevel requirement) {
|
||||
switch (pWindow->winType) {
|
||||
case WINDOW_TYPE_INTERVAL:
|
||||
return adjustIntervalDataRequirement(pWindow, requirement);
|
||||
case WINDOW_TYPE_SESSION:
|
||||
return adjustSessionDataRequirement(pWindow, requirement);
|
||||
case WINDOW_TYPE_STATE:
|
||||
return adjustStateDataRequirement(pWindow, requirement);
|
||||
default:
|
||||
break;
|
||||
}
|
||||
return TSDB_CODE_PLAN_INTERNAL_ERROR;
|
||||
}
|
||||
|
||||
static int32_t adjustFillDataRequirement(SFillLogicNode* pFill, EDataOrderLevel requirement) {
|
||||
if (requirement <= pFill->node.requireDataOrder) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
pFill->node.resultDataOrder = requirement;
|
||||
pFill->node.requireDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustSortDataRequirement(SSortLogicNode* pSort, EDataOrderLevel requirement) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustPartitionDataRequirement(SPartitionLogicNode* pPart, EDataOrderLevel requirement) {
|
||||
if (DATA_ORDER_LEVEL_GLOBAL == requirement) {
|
||||
return TSDB_CODE_PLAN_INTERNAL_ERROR;
|
||||
}
|
||||
pPart->node.resultDataOrder = requirement;
|
||||
pPart->node.requireDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustIndefRowsDataRequirement(SIndefRowsFuncLogicNode* pIndef, EDataOrderLevel requirement) {
|
||||
if (requirement <= pIndef->node.resultDataOrder) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
pIndef->node.resultDataOrder = requirement;
|
||||
pIndef->node.requireDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t adjustInterpDataRequirement(SInterpFuncLogicNode* pInterp, EDataOrderLevel requirement) {
|
||||
if (requirement <= pInterp->node.requireDataOrder) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
pInterp->node.resultDataOrder = requirement;
|
||||
pInterp->node.requireDataOrder = requirement;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t adjustLogicNodeDataRequirement(SLogicNode* pNode, EDataOrderLevel requirement) {
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
switch (nodeType(pNode)) {
|
||||
case QUERY_NODE_LOGIC_PLAN_SCAN:
|
||||
code = adjustScanDataRequirement((SScanLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_JOIN:
|
||||
code = adjustJoinDataRequirement((SJoinLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_AGG:
|
||||
code = adjustAggDataRequirement((SAggLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_PROJECT:
|
||||
code = adjustProjectDataRequirement((SProjectLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_VNODE_MODIFY:
|
||||
case QUERY_NODE_LOGIC_PLAN_EXCHANGE:
|
||||
case QUERY_NODE_LOGIC_PLAN_MERGE:
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_WINDOW:
|
||||
code = adjustWindowDataRequirement((SWindowLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_FILL:
|
||||
code = adjustFillDataRequirement((SFillLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_SORT:
|
||||
code = adjustSortDataRequirement((SSortLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_PARTITION:
|
||||
code = adjustPartitionDataRequirement((SPartitionLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_INDEF_ROWS_FUNC:
|
||||
code = adjustIndefRowsDataRequirement((SIndefRowsFuncLogicNode*)pNode, requirement);
|
||||
break;
|
||||
case QUERY_NODE_LOGIC_PLAN_INTERP_FUNC:
|
||||
code = adjustInterpDataRequirement((SInterpFuncLogicNode*)pNode, requirement);
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
if (TSDB_CODE_SUCCESS == code) {
|
||||
SNode* pChild = NULL;
|
||||
FOREACH(pChild, pNode->pChildren) {
|
||||
code = adjustLogicNodeDataRequirement((SLogicNode*)pChild, pNode->requireDataOrder);
|
||||
if (TSDB_CODE_SUCCESS != code) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
return code;
|
||||
}
|
||||
|
|
|
@ -38,9 +38,15 @@ TEST_F(PlanIntervalTest, fill) {
|
|||
run("SELECT COUNT(*) FROM t1 WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59' "
|
||||
"INTERVAL(10s) FILL(LINEAR)");
|
||||
|
||||
run("SELECT COUNT(*) FROM st1 WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59' "
|
||||
"INTERVAL(10s) FILL(LINEAR)");
|
||||
|
||||
run("SELECT COUNT(*), SUM(c1) FROM t1 "
|
||||
"WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59' "
|
||||
"INTERVAL(10s) FILL(VALUE, 10, 20)");
|
||||
|
||||
run("SELECT COUNT(*) FROM st1 WHERE ts > TIMESTAMP '2022-04-01 00:00:00' and ts < TIMESTAMP '2022-04-30 23:59:59' "
|
||||
"PARTITION BY TBNAME interval(10s) fill(prev)");
|
||||
}
|
||||
|
||||
TEST_F(PlanIntervalTest, selectFunc) {
|
||||
|
|
|
@ -48,10 +48,28 @@ TEST_F(PlanSubqeuryTest, doubleGroupBy) {
|
|||
"WHERE a > 100 GROUP BY b");
|
||||
}
|
||||
|
||||
TEST_F(PlanSubqeuryTest, withSetOperator) {
|
||||
TEST_F(PlanSubqeuryTest, innerSetOperator) {
|
||||
useDb("root", "test");
|
||||
|
||||
run("SELECT c1 FROM (SELECT c1 FROM t1 UNION ALL SELECT c1 FROM t1)");
|
||||
|
||||
run("SELECT c1 FROM (SELECT c1 FROM t1 UNION SELECT c1 FROM t1)");
|
||||
}
|
||||
|
||||
TEST_F(PlanSubqeuryTest, innerFill) {
|
||||
useDb("root", "test");
|
||||
|
||||
run("SELECT cnt FROM (SELECT _WSTART ts, COUNT(*) cnt FROM t1 "
|
||||
"WHERE ts > '2022-04-01 00:00:00' and ts < '2022-04-30 23:59:59' INTERVAL(10s) FILL(LINEAR)) "
|
||||
"WHERE ts > '2022-04-06 00:00:00'");
|
||||
}
|
||||
|
||||
TEST_F(PlanSubqeuryTest, outerInterval) {
|
||||
useDb("root", "test");
|
||||
|
||||
run("SELECT COUNT(*) FROM (SELECT * FROM st1) INTERVAL(5s)");
|
||||
|
||||
run("SELECT COUNT(*) + SUM(c1) FROM (SELECT * FROM st1) INTERVAL(5s)");
|
||||
|
||||
run("SELECT COUNT(*) FROM (SELECT ts, TOP(c1, 10) FROM st1s1) INTERVAL(5s)");
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue