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

[td-11818] fix bug in select * from super_table.

This commit is contained in:
Haojun Liao 2022-01-13 19:01:28 +08:00
parent 23b3981415
commit 5d1238010c
8 changed files with 631 additions and 568 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/planner/planner.h
View File

@ -95,6 +95,7 @@ typedef struct SScanPhyNode {
int8_t tableType;
int32_t order; // scan order: TSDB_ORDER_ASC|TSDB_ORDER_DESC
int32_t count; // repeat count
int32_t reverse; // reverse scan count
} SScanPhyNode;
typedef SScanPhyNode SSystemTableScanPhyNode;

1054
source/client/test/clientTests.cpp
View File

File diff suppressed because it is too large Load Diff

2
source/libs/executor/inc/executorimpl.h
View File

@ -542,7 +542,7 @@ typedef struct SOrderOperatorInfo {
void appendUpstream(SOperatorInfo* p, SOperatorInfo* pUpstream);
SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, STaskRuntimeEnv* pRuntimeEnv, int32_t repeatTime, int32_t reverseTime);
SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, int32_t reverseTime, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, STaskRuntimeEnv* pRuntimeEnv);

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

@ -4844,7 +4844,16 @@ static SSDataBlock* doBlockInfoScan(void* param, bool* newgroup) {
SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, SExecTaskInfo* pTaskInfo) {
assert(repeatTime > 0 && numOfOutput > 0);
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) {
tfree(pInfo);
tfree(pOperator);
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
return NULL;
}
pInfo->pTsdbReadHandle = pTsdbQueryHandle;
pInfo->times = repeatTime;
pInfo->reverseTimes = 0;
@ -4852,7 +4861,6 @@ SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, in
pInfo->current = 0;
pInfo->scanFlag = MAIN_SCAN;
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
pOperator->name = "TableScanOperator";
pOperator->operatorType = OP_TableScan;
pOperator->blockingOptr = false;
@ -4866,6 +4874,39 @@ SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, int32_t order, in
return pOperator;
}
SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, int32_t order, int32_t numOfOutput, int32_t repeatTime, int32_t reverseTime, SExecTaskInfo* pTaskInfo) {
assert(repeatTime > 0);
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) {
tfree(pInfo);
tfree(pOperator);
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
return NULL;
}
pInfo->pTsdbReadHandle = pTsdbQueryHandle;
pInfo->times = repeatTime;
pInfo->reverseTimes = reverseTime;
pInfo->order = order;
pInfo->current = 0;
pInfo->scanFlag = MAIN_SCAN;
pOperator->name = "DataBlocksOptimizedScanOperator";
pOperator->operatorType = OP_DataBlocksOptScan;
pOperator->blockingOptr = false;
pOperator->status = OP_IN_EXECUTING;
pOperator->info = pInfo;
pOperator->numOfOutput = numOfOutput;
pOperator->pRuntimeEnv = NULL;
pOperator->exec = doTableScan;
pOperator->pTaskInfo = pTaskInfo;
return pOperator;
}
SOperatorInfo* createTableSeqScanOperator(void* pTsdbQueryHandle, STaskRuntimeEnv* pRuntimeEnv) {
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
@ -4973,27 +5014,6 @@ void setTableScanFilterOperatorInfo(STableScanInfo* pTableScanInfo, SOperatorInf
}
SOperatorInfo* createDataBlocksOptScanInfo(void* pTsdbQueryHandle, STaskRuntimeEnv* pRuntimeEnv, int32_t repeatTime, int32_t reverseTime) {
assert(repeatTime > 0);
STableScanInfo* pInfo = calloc(1, sizeof(STableScanInfo));
pInfo->pTsdbReadHandle = pTsdbQueryHandle;
pInfo->times = repeatTime;
pInfo->reverseTimes = reverseTime;
pInfo->current = 0;
pInfo->order = pRuntimeEnv->pQueryAttr->order.order;
SOperatorInfo* pOptr = calloc(1, sizeof(SOperatorInfo));
pOptr->name = "DataBlocksOptimizedScanOperator";
// pOptr->operatorType = OP_DataBlocksOptScan;
pOptr->pRuntimeEnv = pRuntimeEnv;
pOptr->blockingOptr = false;
pOptr->info = pInfo;
pOptr->exec = doTableScan;
return pOptr;
}
SArray* getOrderCheckColumns(STaskAttr* pQuery) {
int32_t numOfCols = (pQuery->pGroupbyExpr == NULL)? 0: taosArrayGetSize(pQuery->pGroupbyExpr->columnInfo);
@ -7187,10 +7207,13 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId) {
SOperatorInfo* doCreateOperatorTreeNode(SPhyNode* pPhyNode, SExecTaskInfo* pTaskInfo, void* param) {
if (pPhyNode->pChildren == NULL || taosArrayGetSize(pPhyNode->pChildren) == 0) {
if (pPhyNode->info.type == OP_TableScan) {
SScanPhyNode* pScanPhyNode = (SScanPhyNode*) pPhyNode;
size_t numOfCols = taosArrayGetSize(pPhyNode->pTargets);
SOperatorInfo* pOperatorInfo = createTableScanOperator(param, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pTaskInfo);
pTaskInfo->pRoot = pOperatorInfo;
SScanPhyNode* pScanPhyNode = (SScanPhyNode*)pPhyNode;
size_t numOfCols = taosArrayGetSize(pPhyNode->pTargets);
return createTableScanOperator(param, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pTaskInfo);
} else if (pPhyNode->info.type == OP_DataBlocksOptScan) {
SScanPhyNode* pScanPhyNode = (SScanPhyNode*)pPhyNode;
size_t numOfCols = taosArrayGetSize(pPhyNode->pTargets);
return createDataBlocksOptScanInfo(param, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pScanPhyNode->reverse, pTaskInfo);
} else {
assert(0);
}
@ -7199,21 +7222,20 @@ SOperatorInfo* doCreateOperatorTreeNode(SPhyNode* pPhyNode, SExecTaskInfo* pTask
int32_t doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* readerHandle) {
STsdbQueryCond cond = {.loadExternalRows = false};
cond.twindow.skey = INT64_MIN;
cond.twindow.ekey = INT64_MAX;
uint64_t uid = 0;
SPhyNode* pPhyNode = pPlan->pNode;
if (pPhyNode->info.type == OP_TableScan) {
if (pPhyNode->info.type == OP_TableScan || pPhyNode->info.type == OP_DataBlocksOptScan) {
SScanPhyNode* pScanNode = (SScanPhyNode*) pPhyNode;
uid = pScanNode->uid;
cond.order = pScanNode->order;
cond.numOfCols = taosArrayGetSize(pScanNode->node.pTargets);
STableScanPhyNode* pTableScanNode = (STableScanPhyNode*) pPhyNode;
uid = pTableScanNode->scan.uid;
cond.order = pTableScanNode->scan.order;
cond.numOfCols = taosArrayGetSize(pTableScanNode->scan.node.pTargets);
cond.colList = calloc(cond.numOfCols, sizeof(SColumnInfo));
cond.twindow = pTableScanNode->window;
for(int32_t i = 0; i < cond.numOfCols; ++i) {
SExprInfo* pExprInfo = taosArrayGetP(pScanNode->node.pTargets, i);
SExprInfo* pExprInfo = taosArrayGetP(pTableScanNode->scan.node.pTargets, i);
assert(pExprInfo->pExpr->nodeType == TEXPR_COL_NODE);
SSchema* pSchema = pExprInfo->pExpr->pSchema;
@ -7235,7 +7257,11 @@ int32_t doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* r
*pTaskInfo = createExecTaskInfo((uint64_t)pPlan->id.queryId);
tsdbReadHandleT tsdbReadHandle = tsdbQueryTables(readerHandle, &cond, &group, (*pTaskInfo)->id.queryId, NULL);
doCreateOperatorTreeNode(pPlan->pNode, *pTaskInfo, tsdbReadHandle);
(*pTaskInfo)->pRoot = doCreateOperatorTreeNode(pPlan->pNode, *pTaskInfo, tsdbReadHandle);
if ((*pTaskInfo)->pRoot == NULL) {
return terrno;
}
return TSDB_CODE_SUCCESS;
}

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

@ -917,6 +917,8 @@ int32_t validateLimitNode(SQueryStmtInfo *pQueryInfo, SSqlNode* pSqlNode, SMsgBu
return buildInvalidOperationMsg(pMsgBuf, msg1);
}
}
return TSDB_CODE_SUCCESS;
}
int32_t validateOrderbyNode(SQueryStmtInfo *pQueryInfo, SSqlNode* pSqlNode, SMsgBuf* pMsgBuf) {

4
source/libs/planner/src/physicalPlan.c
View File

@ -262,8 +262,8 @@ static void vgroupMsgToEpSet(const SVgroupMsg* vg, SQueryNodeAddr* execNode) {
}
static uint64_t splitSubplanByTable(SPlanContext* pCxt, SQueryPlanNode* pPlanNode, SQueryTableInfo* pTable) {
SVgroupsInfo* vgroupList = pTable->pMeta->vgroupList;
for (int32_t i = 0; i < pTable->pMeta->vgroupList->numOfVgroups; ++i) {
SVgroupsInfo* pVgroupList = pTable->pMeta->vgroupList;
for (int32_t i = 0; i < pVgroupList->numOfVgroups; ++i) {
STORE_CURRENT_SUBPLAN(pCxt);
SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_SCAN);
subplan->msgType = TDMT_VND_QUERY;

32
source/libs/planner/src/physicalPlanJson.c
View File

@ -530,32 +530,38 @@ static const char* jkScanNodeTableId = "TableId";
static const char* jkScanNodeTableType = "TableType";
static const char* jkScanNodeTableOrder = "Order";
static const char* jkScanNodeTableCount = "Count";
static const char* jkScanNodeTableRevCount = "Reverse";
static bool scanNodeToJson(const void* obj, cJSON* json) {
const SScanPhyNode* scan = (const SScanPhyNode*)obj;
bool res = cJSON_AddNumberToObject(json, jkScanNodeTableId, scan->uid);
const SScanPhyNode* pNode = (const SScanPhyNode*)obj;
bool res = cJSON_AddNumberToObject(json, jkScanNodeTableId, pNode->uid);
if (res) {
res = cJSON_AddNumberToObject(json, jkScanNodeTableType, scan->tableType);
res = cJSON_AddNumberToObject(json, jkScanNodeTableType, pNode->tableType);
}
if (res) {
res = cJSON_AddNumberToObject(json, jkScanNodeTableOrder, scan->order);
res = cJSON_AddNumberToObject(json, jkScanNodeTableOrder, pNode->order);
}
if (res) {
res = cJSON_AddNumberToObject(json, jkScanNodeTableCount, scan->count);
res = cJSON_AddNumberToObject(json, jkScanNodeTableCount, pNode->count);
}
if (res) {
res = cJSON_AddNumberToObject(json, jkScanNodeTableRevCount, pNode->reverse);
}
return res;
}
static bool scanNodeFromJson(const cJSON* json, void* obj) {
SScanPhyNode* scan = (SScanPhyNode*)obj;
scan->uid = getNumber(json, jkScanNodeTableId);
scan->tableType = getNumber(json, jkScanNodeTableType);
scan->count = getNumber(json, jkScanNodeTableCount);
scan->order = getNumber(json, jkScanNodeTableOrder);
SScanPhyNode* pNode = (SScanPhyNode*)obj;
pNode->uid = getNumber(json, jkScanNodeTableId);
pNode->tableType = getNumber(json, jkScanNodeTableType);
pNode->count = getNumber(json, jkScanNodeTableCount);
pNode->order = getNumber(json, jkScanNodeTableOrder);
pNode->reverse = getNumber(json, jkScanNodeTableRevCount);
return true;
}
@ -910,7 +916,7 @@ static SSubplan* subplanFromJson(const cJSON* json) {
}
bool res = fromObject(json, jkSubplanId, subplanIdFromJson, &subplan->id, true);
if (res) {
size_t size = MAX(sizeof(SPhyNode), sizeof(SScanPhyNode));
size_t size = MAX(sizeof(SPhyNode), sizeof(STableScanPhyNode));
res = fromObjectWithAlloc(json, jkSubplanNode, phyNodeFromJson, (void**)&subplan->pNode, size, false);
}
if (res) {
@ -940,8 +946,8 @@ int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
}
*str = cJSON_Print(json);
// printf("====Physical plan:====\n")
// printf("%s\n", *str);
printf("====Physical plan:====\n");
printf("%s\n", *str);
*len = strlen(*str) + 1;
return TSDB_CODE_SUCCESS;
}

6
source/libs/planner/src/planner.c
View File

@ -65,9 +65,9 @@ int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag,
}
if (pLogicPlan->info.type != QNODE_MODIFY) {
// char* str = NULL;
// queryPlanToString(pLogicPlan, &str);
// printf("%s\n", str);
char* str = NULL;
queryPlanToString(pLogicPlan, &str);
printf("%s\n", str);
}
code = optimizeQueryPlan(pLogicPlan);