This commit is contained in:
yihaoDeng 2024-04-29 00:26:38 +08:00
commit e48fbc4b34
8 changed files with 101 additions and 72 deletions

View File

@ -191,11 +191,11 @@ Left/Right ASOF Join are supported between super tables, normal tables, child ta
| **Operator** | **Meaning for Left ASOF Join** |
| :-------------: | ------------------------ |
| > | Match rows in the right table whose primary key timestamp is less than and the most closed to the left table's primary key timestamp |
| >= | Match rows in the right table whose primary key timestamp is less than or equal to and the most closed to the left table's primary key timestamp |
| > | Match rows in the right table whose primary key timestamp is less than and the most closed to the left table's primary key timestamp |
| >= | Match rows in the right table whose primary key timestamp is less than or equal to and the most closed to the left table's primary key timestamp |
| = | Match rows in the right table whose primary key timestamp is equal to the left table's primary key timestamp |
| < | Match rows in the right table whose the primary key timestamp is greater than and the most closed to the left table's primary key timestamp |
| <= | Match rows in the right table whose primary key timestamp is greater than or equal to and the most closed to the left table's primary key timestamp |
| &lt; | Match rows in the right table whose the primary key timestamp is greater than and the most closed to the left table's primary key timestamp |
| &lt;= | Match rows in the right table whose primary key timestamp is greater than or equal to and the most closed to the left table's primary key timestamp |
For Right ASOF Join, the above operators have the opposite meaning.

View File

@ -185,11 +185,11 @@ SELECT ... FROM table_name1 LEFT|RIGHT ASOF JOIN table_name2 [ON ...] [JLIMIT jl
| **运算符** | **Left ASOF 时含义** |
| :-------------: | ------------------------ |
| > | 匹配右表中主键时间戳小于左表主键时间戳且时间戳最接近的数据行 |
| >= | 匹配右表中主键时间戳小于等于左表主键时间戳且时间戳最接近的数据行 |
| &gt; | 匹配右表中主键时间戳小于左表主键时间戳且时间戳最接近的数据行 |
| &gt;= | 匹配右表中主键时间戳小于等于左表主键时间戳且时间戳最接近的数据行 |
| = | 匹配右表中主键时间戳等于左表主键时间戳的行 |
| < | 匹配右表中主键时间戳大于左表主键时间戳且时间戳最接近的数据行 |
| <= | 匹配右表中主键时间戳大于等于左表主键时间戳且时间戳最接近的数据行 |
| &lt; | 匹配右表中主键时间戳大于左表主键时间戳且时间戳最接近的数据行 |
| &lt;= | 匹配右表中主键时间戳大于等于左表主键时间戳且时间戳最接近的数据行 |
对于 Right ASOF 来说,上述运算符含义正好相反。

View File

@ -22,10 +22,10 @@
#include "mndPrivilege.h"
#include "mndQnode.h"
#include "mndShow.h"
#include "mndSma.h"
#include "mndStb.h"
#include "mndUser.h"
#include "mndView.h"
#include "mndSma.h"
#include "tglobal.h"
#include "tversion.h"
@ -57,6 +57,13 @@ typedef struct {
int64_t lastAccessTimeMs;
} SAppObj;
typedef struct {
int32_t totalDnodes;
int32_t onlineDnodes;
SEpSet epSet;
SArray *pQnodeList;
} SConnPreparedObj;
static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType, uint32_t ip, uint16_t port,
int32_t pid, const char *app, int64_t startTime);
static void mndFreeConn(SConnObj *pConn);
@ -460,7 +467,7 @@ static int32_t mndGetOnlineDnodeNum(SMnode *pMnode, int32_t *num) {
}
static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHbReq *pHbReq,
SClientHbBatchRsp *pBatchRsp) {
SClientHbBatchRsp *pBatchRsp, SConnPreparedObj *pObj) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SClientHbRsp hbRsp = {.connKey = pHbReq->connKey, .status = 0, .info = NULL, .query = NULL};
SRpcConnInfo connInfo = pMsg->info.conn;
@ -503,11 +510,11 @@ static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHb
}
rspBasic->connId = pConn->id;
rspBasic->totalDnodes = mndGetDnodeSize(pMnode);
mndGetOnlineDnodeNum(pMnode, &rspBasic->onlineDnodes);
mndGetMnodeEpSet(pMnode, &rspBasic->epSet);
mndCreateQnodeList(pMnode, &rspBasic->pQnodeList, -1);
rspBasic->connId = pConn->id;
rspBasic->totalDnodes = pObj->totalDnodes;
rspBasic->onlineDnodes = pObj->onlineDnodes;
rspBasic->epSet = pObj->epSet;
rspBasic->pQnodeList = taosArrayDup(pObj->pQnodeList, NULL);
mndReleaseConn(pMnode, pConn, true);
@ -641,6 +648,12 @@ static int32_t mndProcessHeartBeatReq(SRpcMsg *pReq) {
return -1;
}
SConnPreparedObj obj = {0};
obj.totalDnodes = mndGetDnodeSize(pMnode);
mndGetOnlineDnodeNum(pMnode, &obj.onlineDnodes);
mndGetMnodeEpSet(pMnode, &obj.epSet);
mndCreateQnodeList(pMnode, &obj.pQnodeList, -1);
SClientHbBatchRsp batchRsp = {0};
batchRsp.svrTimestamp = taosGetTimestampSec();
batchRsp.rsps = taosArrayInit(0, sizeof(SClientHbRsp));
@ -649,7 +662,7 @@ static int32_t mndProcessHeartBeatReq(SRpcMsg *pReq) {
for (int i = 0; i < sz; i++) {
SClientHbReq *pHbReq = taosArrayGet(batchReq.reqs, i);
if (pHbReq->connKey.connType == CONN_TYPE__QUERY) {
mndProcessQueryHeartBeat(pMnode, pReq, pHbReq, &batchRsp);
mndProcessQueryHeartBeat(pMnode, pReq, pHbReq, &batchRsp, &obj);
} else if (pHbReq->connKey.connType == CONN_TYPE__TMQ) {
SClientHbRsp *pRsp = mndMqHbBuildRsp(pMnode, pHbReq);
if (pRsp != NULL) {
@ -668,6 +681,8 @@ static int32_t mndProcessHeartBeatReq(SRpcMsg *pReq) {
pReq->info.rspLen = tlen;
pReq->info.rsp = buf;
taosArrayDestroy(obj.pQnodeList);
return 0;
}

View File

@ -60,6 +60,7 @@ bool keysHasCol(SNodeList* pKeys);
bool keysHasTbname(SNodeList* pKeys);
SFunctionNode* createGroupKeyAggFunc(SColumnNode* pGroupCol);
int32_t getTimeRangeFromNode(SNode** pPrimaryKeyCond, STimeWindow* pTimeRange, bool* pIsStrict);
int32_t tagScanSetExecutionMode(SScanLogicNode* pScan);
#define CLONE_LIMIT 1
#define CLONE_SLIMIT 1 << 1

View File

@ -392,60 +392,6 @@ static int32_t makeScanLogicNode(SLogicPlanContext* pCxt, SRealTableNode* pRealT
static bool needScanDefaultCol(EScanType scanType) { return SCAN_TYPE_TABLE_COUNT != scanType; }
static EDealRes tagScanNodeHasTbnameFunc(SNode* pNode, void* pContext) {
if (QUERY_NODE_FUNCTION == nodeType(pNode) && FUNCTION_TYPE_TBNAME == ((SFunctionNode*)pNode)->funcType ||
(QUERY_NODE_COLUMN == nodeType(pNode) && COLUMN_TYPE_TBNAME == ((SColumnNode*)pNode)->colType)) {
*(bool*)pContext = true;
return DEAL_RES_END;
}
return DEAL_RES_CONTINUE;
}
static bool tagScanNodeListHasTbname(SNodeList* pCols) {
bool hasTbname = false;
nodesWalkExprs(pCols, tagScanNodeHasTbnameFunc, &hasTbname);
return hasTbname;
}
static bool tagScanNodeHasTbname(SNode* pKeys) {
bool hasTbname = false;
nodesWalkExpr(pKeys, tagScanNodeHasTbnameFunc, &hasTbname);
return hasTbname;
}
static int32_t tagScanSetExecutionMode(SScanLogicNode* pScan) {
pScan->onlyMetaCtbIdx = false;
if (pScan->tableType == TSDB_CHILD_TABLE) {
pScan->onlyMetaCtbIdx = false;
return TSDB_CODE_SUCCESS;
}
if (tagScanNodeListHasTbname(pScan->pScanPseudoCols)) {
pScan->onlyMetaCtbIdx = false;
return TSDB_CODE_SUCCESS;
}
if (pScan->node.pConditions == NULL) {
pScan->onlyMetaCtbIdx = true;
return TSDB_CODE_SUCCESS;
}
SNode* pCond = nodesCloneNode(pScan->node.pConditions);
SNode* pTagCond = NULL;
SNode* pTagIndexCond = NULL;
filterPartitionCond(&pCond, NULL, &pTagIndexCond, &pTagCond, NULL);
if (pTagIndexCond || tagScanNodeHasTbname(pTagCond)) {
pScan->onlyMetaCtbIdx = false;
} else {
pScan->onlyMetaCtbIdx = true;
}
nodesDestroyNode(pCond);
nodesDestroyNode(pTagIndexCond);
nodesDestroyNode(pTagCond);
return TSDB_CODE_SUCCESS;
}
static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SRealTableNode* pRealTable,
SLogicNode** pLogicNode) {
SScanLogicNode* pScan = NULL;

View File

@ -5142,7 +5142,6 @@ int32_t stbJoinOptRewriteToTagScan(SLogicNode* pJoin, SNode* pNode) {
NODES_DESTORY_NODE(pScan->node.pConditions);
pScan->node.requireDataOrder = DATA_ORDER_LEVEL_NONE;
pScan->node.resultDataOrder = DATA_ORDER_LEVEL_NONE;
pScan->onlyMetaCtbIdx = true;
SNodeList* pTags = nodesMakeList();
int32_t code = nodesCollectColumnsFromNode(pJoinNode->pTagEqCond, NULL, COLLECT_COL_TYPE_TAG, &pTags);
@ -5177,6 +5176,8 @@ int32_t stbJoinOptRewriteToTagScan(SLogicNode* pJoin, SNode* pNode) {
code = stbJoinOptAddFuncToScanNode("_vgid", pScan);
}
tagScanSetExecutionMode(pScan);
if (code) {
nodesDestroyList(pTags);
}

View File

@ -615,3 +615,61 @@ int32_t getTimeRangeFromNode(SNode** pPrimaryKeyCond, STimeWindow* pTimeRange, b
}
static EDealRes tagScanNodeHasTbnameFunc(SNode* pNode, void* pContext) {
if (QUERY_NODE_FUNCTION == nodeType(pNode) && FUNCTION_TYPE_TBNAME == ((SFunctionNode*)pNode)->funcType ||
(QUERY_NODE_COLUMN == nodeType(pNode) && COLUMN_TYPE_TBNAME == ((SColumnNode*)pNode)->colType)) {
*(bool*)pContext = true;
return DEAL_RES_END;
}
return DEAL_RES_CONTINUE;
}
static bool tagScanNodeListHasTbname(SNodeList* pCols) {
bool hasTbname = false;
nodesWalkExprs(pCols, tagScanNodeHasTbnameFunc, &hasTbname);
return hasTbname;
}
static bool tagScanNodeHasTbname(SNode* pKeys) {
bool hasTbname = false;
nodesWalkExpr(pKeys, tagScanNodeHasTbnameFunc, &hasTbname);
return hasTbname;
}
int32_t tagScanSetExecutionMode(SScanLogicNode* pScan) {
pScan->onlyMetaCtbIdx = false;
if (pScan->tableType == TSDB_CHILD_TABLE) {
pScan->onlyMetaCtbIdx = false;
return TSDB_CODE_SUCCESS;
}
if (tagScanNodeListHasTbname(pScan->pScanPseudoCols)) {
pScan->onlyMetaCtbIdx = false;
return TSDB_CODE_SUCCESS;
}
if (pScan->node.pConditions == NULL) {
pScan->onlyMetaCtbIdx = true;
return TSDB_CODE_SUCCESS;
}
SNode* pCond = nodesCloneNode(pScan->node.pConditions);
SNode* pTagCond = NULL;
SNode* pTagIndexCond = NULL;
filterPartitionCond(&pCond, NULL, &pTagIndexCond, &pTagCond, NULL);
if (pTagIndexCond || tagScanNodeHasTbname(pTagCond)) {
pScan->onlyMetaCtbIdx = false;
} else {
pScan->onlyMetaCtbIdx = true;
}
nodesDestroyNode(pCond);
nodesDestroyNode(pTagIndexCond);
nodesDestroyNode(pTagCond);
return TSDB_CODE_SUCCESS;
}

View File

@ -194,4 +194,12 @@ if $rows != 144 then
return -1
endi
sql select a.ts, b.ts from tba1 a join sta b on a.ts = b.ts and a.t1 = b.t1;
if $rows != 4 then
return -1
endi
sql select a.ts, b.ts from sta a join sta b on a.ts = b.ts and a.t1 = b.t1;
if $rows != 8 then
return -1
endi