Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
2c33a51c
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
未验证
提交
2c33a51c
编写于
2月 05, 2022
作者:
D
dapan1121
提交者:
GitHub
2月 05, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #10111 from taosdata/feature/qnode
Feature/qnode
上级
a342af36
dd0a08c9
变更
29
展开全部
隐藏空白更改
内联
并排
Showing
29 changed file
with
575 addition
and
438 deletion
+575
-438
include/common/tmsg.h
include/common/tmsg.h
+7
-7
include/libs/catalog/catalog.h
include/libs/catalog/catalog.h
+2
-2
include/libs/qcom/query.h
include/libs/qcom/query.h
+6
-6
source/client/src/clientHb.c
source/client/src/clientHb.c
+6
-11
source/client/src/clientMsgHandler.c
source/client/src/clientMsgHandler.c
+2
-5
source/dnode/mgmt/impl/test/sut/src/sut.cpp
source/dnode/mgmt/impl/test/sut/src/sut.cpp
+2
-2
source/dnode/mnode/impl/src/mndBnode.c
source/dnode/mnode/impl/src/mndBnode.c
+1
-1
source/dnode/mnode/impl/src/mndCluster.c
source/dnode/mnode/impl/src/mndCluster.c
+2
-2
source/dnode/mnode/impl/src/mndDb.c
source/dnode/mnode/impl/src/mndDb.c
+1
-1
source/dnode/mnode/impl/src/mndDnode.c
source/dnode/mnode/impl/src/mndDnode.c
+2
-2
source/dnode/mnode/impl/src/mndFunc.c
source/dnode/mnode/impl/src/mndFunc.c
+1
-1
source/dnode/mnode/impl/src/mndMnode.c
source/dnode/mnode/impl/src/mndMnode.c
+1
-1
source/dnode/mnode/impl/src/mndProfile.c
source/dnode/mnode/impl/src/mndProfile.c
+2
-2
source/dnode/mnode/impl/src/mndQnode.c
source/dnode/mnode/impl/src/mndQnode.c
+1
-1
source/dnode/mnode/impl/src/mndSnode.c
source/dnode/mnode/impl/src/mndSnode.c
+1
-1
source/dnode/mnode/impl/src/mndStb.c
source/dnode/mnode/impl/src/mndStb.c
+14
-9
source/dnode/mnode/impl/src/mndSubscribe.c
source/dnode/mnode/impl/src/mndSubscribe.c
+2
-2
source/dnode/mnode/impl/src/mndTopic.c
source/dnode/mnode/impl/src/mndTopic.c
+2
-2
source/dnode/mnode/impl/src/mndUser.c
source/dnode/mnode/impl/src/mndUser.c
+1
-1
source/dnode/mnode/impl/src/mndVgroup.c
source/dnode/mnode/impl/src/mndVgroup.c
+2
-2
source/dnode/mnode/impl/test/show/show.cpp
source/dnode/mnode/impl/test/show/show.cpp
+1
-1
source/dnode/mnode/impl/test/stb/stb.cpp
source/dnode/mnode/impl/test/stb/stb.cpp
+5
-3
source/dnode/vnode/src/vnd/vnodeQuery.c
source/dnode/vnode/src/vnd/vnodeQuery.c
+5
-5
source/libs/catalog/inc/catalogInt.h
source/libs/catalog/inc/catalogInt.h
+25
-28
source/libs/catalog/src/catalog.c
source/libs/catalog/src/catalog.c
+413
-290
source/libs/catalog/test/catalogTests.cpp
source/libs/catalog/test/catalogTests.cpp
+38
-22
source/libs/qcom/src/querymsg.c
source/libs/qcom/src/querymsg.c
+27
-25
source/libs/scheduler/inc/schedulerInt.h
source/libs/scheduler/inc/schedulerInt.h
+1
-1
source/libs/scheduler/src/scheduler.c
source/libs/scheduler/src/scheduler.c
+2
-2
未找到文件。
include/common/tmsg.h
浏览文件 @
2c33a51c
...
...
@@ -154,8 +154,8 @@ typedef enum _mgmt_table {
typedef
struct
{
int32_t
vgId
;
char
*
dbName
;
char
*
t
ableFull
Name
;
char
*
db
F
Name
;
char
*
t
b
Name
;
}
SBuildTableMetaInput
;
typedef
struct
{
...
...
@@ -696,8 +696,8 @@ typedef struct {
typedef
struct
{
SMsgHead
header
;
char
dbF
n
ame
[
TSDB_DB_FNAME_LEN
];
char
t
ableFname
[
TSDB_TABLE_F
NAME_LEN
];
char
dbF
N
ame
[
TSDB_DB_FNAME_LEN
];
char
t
bName
[
TSDB_TABLE_
NAME_LEN
];
}
STableInfoReq
;
typedef
struct
{
...
...
@@ -722,9 +722,9 @@ typedef struct {
}
SVgroupsInfo
;
typedef
struct
{
char
tb
Fname
[
TSDB_TABLE_FNAME_LEN
];
// table full name
char
stb
Fname
[
TSDB_TABLE_F
NAME_LEN
];
char
dbF
n
ame
[
TSDB_DB_FNAME_LEN
];
char
tb
Name
[
TSDB_TABLE_NAME_LEN
];
char
stb
Name
[
TSDB_TABLE_
NAME_LEN
];
char
dbF
N
ame
[
TSDB_DB_FNAME_LEN
];
int32_t
numOfTags
;
int32_t
numOfColumns
;
int8_t
precision
;
...
...
include/libs/catalog/catalog.h
浏览文件 @
2c33a51c
...
...
@@ -49,7 +49,7 @@ typedef struct SCatalogCfg {
uint32_t
maxTblCacheNum
;
uint32_t
maxDBCacheNum
;
uint32_t
dbRentSec
;
uint32_t
st
able
RentSec
;
uint32_t
st
b
RentSec
;
}
SCatalogCfg
;
typedef
struct
SSTableMetaVersion
{
...
...
@@ -99,7 +99,7 @@ int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pTransporter, const
int32_t
catalogUpdateDBVgroup
(
struct
SCatalog
*
pCatalog
,
const
char
*
dbName
,
SDBVgroupInfo
*
dbInfo
);
int32_t
catalogRemoveDB
Vgroup
(
struct
SCatalog
*
pCatalog
,
SDbVgVersion
*
dbInfo
);
int32_t
catalogRemoveDB
(
struct
SCatalog
*
pCatalog
,
const
char
*
dbName
,
uint64_t
dbId
);
/**
* Get a table's meta data.
...
...
include/libs/qcom/query.h
浏览文件 @
2c33a51c
...
...
@@ -81,16 +81,15 @@ typedef struct STableMeta {
}
STableMeta
;
typedef
struct
SDBVgroupInfo
{
SRWLatch
lock
;
uint64_t
dbId
;
int32_t
vgVersion
;
int8_t
hashMethod
;
SHashObj
*
vg
Info
;
//key:vgId, value:SVgroupInfo
SHashObj
*
vg
Hash
;
//key:vgId, value:SVgroupInfo
}
SDBVgroupInfo
;
typedef
struct
SUseDbOutput
{
char
db
[
TSDB_DB_FNAME_LEN
];
SDBVgroupInfo
dbVgroup
;
char
db
[
TSDB_DB_FNAME_LEN
];
SDBVgroupInfo
*
dbVgroup
;
}
SUseDbOutput
;
enum
{
...
...
@@ -103,8 +102,9 @@ enum {
typedef
struct
STableMetaOutput
{
int32_t
metaType
;
char
ctbFname
[
TSDB_TABLE_FNAME_LEN
];
char
tbFname
[
TSDB_TABLE_FNAME_LEN
];
char
dbFName
[
TSDB_DB_FNAME_LEN
];
char
ctbName
[
TSDB_TABLE_NAME_LEN
];
char
tbName
[
TSDB_TABLE_NAME_LEN
];
SCTableMeta
ctbMeta
;
STableMeta
*
tbMeta
;
}
STableMetaOutput
;
...
...
source/client/src/clientHb.c
浏览文件 @
2c33a51c
...
...
@@ -41,19 +41,14 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
tscDebug
(
"hb db rsp, db:%s, vgVersion:%d, uid:%"
PRIx64
,
rsp
->
db
,
rsp
->
vgVersion
,
rsp
->
uid
);
if
(
rsp
->
vgVersion
<
0
)
{
SDbVgVersion
dbInfo
;
strcpy
(
dbInfo
.
dbName
,
rsp
->
db
);
dbInfo
.
dbId
=
rsp
->
uid
;
dbInfo
.
vgVersion
=
rsp
->
vgVersion
;
code
=
catalogRemoveDBVgroup
(
pCatalog
,
&
dbInfo
);
code
=
catalogRemoveDB
(
pCatalog
,
rsp
->
db
,
rsp
->
uid
);
}
else
{
SDBVgroupInfo
vgInfo
=
{
0
};
vgInfo
.
dbId
=
rsp
->
uid
;
vgInfo
.
vgVersion
=
rsp
->
vgVersion
;
vgInfo
.
hashMethod
=
rsp
->
hashMethod
;
vgInfo
.
vg
Info
=
taosHashInit
(
rsp
->
vgNum
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
true
,
HASH_ENTRY_LOCK
);
if
(
NULL
==
vgInfo
.
vg
Info
)
{
vgInfo
.
vg
Hash
=
taosHashInit
(
rsp
->
vgNum
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
true
,
HASH_ENTRY_LOCK
);
if
(
NULL
==
vgInfo
.
vg
Hash
)
{
tscError
(
"hash init[%d] failed"
,
rsp
->
vgNum
);
return
TSDB_CODE_TSC_OUT_OF_MEMORY
;
}
...
...
@@ -67,16 +62,16 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
rsp
->
vgroupInfo
[
i
].
epset
.
eps
[
n
].
port
=
ntohs
(
rsp
->
vgroupInfo
[
i
].
epset
.
eps
[
n
].
port
);
}
if
(
0
!=
taosHashPut
(
vgInfo
.
vg
Info
,
&
rsp
->
vgroupInfo
[
i
].
vgId
,
sizeof
(
rsp
->
vgroupInfo
[
i
].
vgId
),
&
rsp
->
vgroupInfo
[
i
],
sizeof
(
rsp
->
vgroupInfo
[
i
])))
{
if
(
0
!=
taosHashPut
(
vgInfo
.
vg
Hash
,
&
rsp
->
vgroupInfo
[
i
].
vgId
,
sizeof
(
rsp
->
vgroupInfo
[
i
].
vgId
),
&
rsp
->
vgroupInfo
[
i
],
sizeof
(
rsp
->
vgroupInfo
[
i
])))
{
tscError
(
"hash push failed, errno:%d"
,
errno
);
taosHashCleanup
(
vgInfo
.
vg
Info
);
taosHashCleanup
(
vgInfo
.
vg
Hash
);
return
TSDB_CODE_TSC_OUT_OF_MEMORY
;
}
}
code
=
catalogUpdateDBVgroup
(
pCatalog
,
rsp
->
db
,
&
vgInfo
);
if
(
code
)
{
taosHashCleanup
(
vgInfo
.
vg
Info
);
taosHashCleanup
(
vgInfo
.
vg
Hash
);
}
}
...
...
source/client/src/clientMsgHandler.c
浏览文件 @
2c33a51c
...
...
@@ -302,15 +302,12 @@ int32_t processDropDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SDropDbRsp
*
rsp
=
(
SDropDbRsp
*
)
pMsg
->
pData
;
SDbVgVersion
dbVer
=
{
0
};
struct
SCatalog
*
pCatalog
=
NULL
;
strncpy
(
dbVer
.
dbName
,
rsp
->
db
,
sizeof
(
dbVer
.
dbName
));
dbVer
.
dbId
=
be64toh
(
rsp
->
uid
);
rsp
->
uid
=
be64toh
(
rsp
->
uid
);
catalogGetHandle
(
pRequest
->
pTscObj
->
pAppInfo
->
clusterId
,
&
pCatalog
);
catalogRemoveDB
Vgroup
(
pCatalog
,
&
dbVer
);
catalogRemoveDB
(
pCatalog
,
rsp
->
db
,
rsp
->
uid
);
tsem_post
(
&
pRequest
->
body
.
rspSem
);
return
code
;
...
...
source/dnode/mgmt/impl/test/sut/src/sut.cpp
浏览文件 @
2c33a51c
...
...
@@ -125,7 +125,7 @@ const char* Testbase::GetMetaName(int32_t index) {
int32_t
Testbase
::
GetMetaNum
()
{
return
pMeta
->
numOfColumns
;
}
const
char
*
Testbase
::
GetMetaTbName
()
{
return
pMeta
->
tb
Fn
ame
;
}
const
char
*
Testbase
::
GetMetaTbName
()
{
return
pMeta
->
tb
N
ame
;
}
void
Testbase
::
SendShowRetrieveReq
()
{
int32_t
contLen
=
sizeof
(
SRetrieveTableReq
);
...
...
@@ -144,7 +144,7 @@ void Testbase::SendShowRetrieveReq() {
pos
=
0
;
}
const
char
*
Testbase
::
GetShowName
()
{
return
pMeta
->
tb
Fn
ame
;
}
const
char
*
Testbase
::
GetShowName
()
{
return
pMeta
->
tb
N
ame
;
}
int8_t
Testbase
::
GetShowInt8
()
{
int8_t
data
=
*
((
int8_t
*
)(
pData
+
pos
));
...
...
source/dnode/mnode/impl/src/mndBnode.c
浏览文件 @
2c33a51c
...
...
@@ -426,7 +426,7 @@ static int32_t mndGetBnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_BNODE
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndCluster.c
浏览文件 @
2c33a51c
...
...
@@ -186,7 +186,7 @@ static int32_t mndGetClusterMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaRsp
cols
++
;
pMeta
->
numOfColumns
=
htonl
(
cols
);
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
pShow
->
numOfColumns
=
cols
;
pShow
->
offset
[
0
]
=
0
;
...
...
@@ -196,7 +196,7 @@ static int32_t mndGetClusterMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaRsp
pShow
->
numOfRows
=
1
;
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @
2c33a51c
...
...
@@ -1113,7 +1113,7 @@ static int32_t mndGetDbMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMe
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_DB
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndDnode.c
浏览文件 @
2c33a51c
...
...
@@ -608,7 +608,7 @@ static int32_t mndGetConfigMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp
pShow
->
numOfRows
=
TSDB_CONFIG_NUMBER
;
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
@@ -715,7 +715,7 @@ static int32_t mndGetDnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_DNODE
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndFunc.c
浏览文件 @
2c33a51c
...
...
@@ -482,7 +482,7 @@ static int32_t mndGetFuncMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *p
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_FUNC
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @
2c33a51c
...
...
@@ -624,7 +624,7 @@ static int32_t mndGetMnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_MNODE
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
mndUpdateMnodeRole
(
pMnode
);
return
0
;
...
...
source/dnode/mnode/impl/src/mndProfile.c
浏览文件 @
2c33a51c
...
...
@@ -623,7 +623,7 @@ static int32_t mndGetConnsMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow
->
numOfRows
=
taosHashGetSize
(
pMgmt
->
cache
->
pHashTable
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
@@ -792,7 +792,7 @@ static int32_t mndGetQueryMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow
->
numOfRows
=
1000000
;
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndQnode.c
浏览文件 @
2c33a51c
...
...
@@ -426,7 +426,7 @@ static int32_t mndGetQnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_QNODE
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndSnode.c
浏览文件 @
2c33a51c
...
...
@@ -428,7 +428,7 @@ static int32_t mndGetSnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_SNODE
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @
2c33a51c
...
...
@@ -723,20 +723,23 @@ static int32_t mndProcessStbMetaReq(SMnodeMsg *pReq) {
SMnode
*
pMnode
=
pReq
->
pMnode
;
STableInfoReq
*
pInfo
=
pReq
->
rpcMsg
.
pCont
;
mDebug
(
"stb:%s, start to retrieve meta"
,
pInfo
->
tableFname
);
char
tbFName
[
TSDB_TABLE_FNAME_LEN
]
=
{
0
};
snprintf
(
tbFName
,
sizeof
(
tbFName
),
"%s.%s"
,
pInfo
->
dbFName
,
pInfo
->
tbName
);
SDbObj
*
pDb
=
mndAcquireDbByStb
(
pMnode
,
pInfo
->
tableFname
);
mDebug
(
"stb:%s, start to retrieve meta"
,
tbFName
);
SDbObj
*
pDb
=
mndAcquireDbByStb
(
pMnode
,
tbFName
);
if
(
pDb
==
NULL
)
{
terrno
=
TSDB_CODE_MND_DB_NOT_SELECTED
;
mError
(
"stb:%s, failed to retrieve meta since %s"
,
pInfo
->
tableFn
ame
,
terrstr
());
mError
(
"stb:%s, failed to retrieve meta since %s"
,
tbFN
ame
,
terrstr
());
return
-
1
;
}
SStbObj
*
pStb
=
mndAcquireStb
(
pMnode
,
pInfo
->
tableFn
ame
);
SStbObj
*
pStb
=
mndAcquireStb
(
pMnode
,
tbFN
ame
);
if
(
pStb
==
NULL
)
{
mndReleaseDb
(
pMnode
,
pDb
);
terrno
=
TSDB_CODE_MND_INVALID_STB
;
mError
(
"stb:%s, failed to get meta since %s"
,
pInfo
->
tableFn
ame
,
terrstr
());
mError
(
"stb:%s, failed to get meta since %s"
,
tbFN
ame
,
terrstr
());
return
-
1
;
}
...
...
@@ -750,11 +753,13 @@ static int32_t mndProcessStbMetaReq(SMnodeMsg *pReq) {
mndReleaseDb
(
pMnode
,
pDb
);
mndReleaseStb
(
pMnode
,
pStb
);
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
mError
(
"stb:%s, failed to get meta since %s"
,
pInfo
->
tableFn
ame
,
terrstr
());
mError
(
"stb:%s, failed to get meta since %s"
,
tbFN
ame
,
terrstr
());
return
-
1
;
}
memcpy
(
pMeta
->
tbFname
,
pStb
->
name
,
TSDB_TABLE_FNAME_LEN
);
strcpy
(
pMeta
->
dbFName
,
pStb
->
db
);
strcpy
(
pMeta
->
tbName
,
pInfo
->
tbName
);
strcpy
(
pMeta
->
stbName
,
pInfo
->
tbName
);
pMeta
->
numOfTags
=
htonl
(
pStb
->
numOfTags
);
pMeta
->
numOfColumns
=
htonl
(
pStb
->
numOfColumns
);
pMeta
->
precision
=
pDb
->
cfg
.
precision
;
...
...
@@ -779,7 +784,7 @@ static int32_t mndProcessStbMetaReq(SMnodeMsg *pReq) {
pReq
->
pCont
=
pMeta
;
pReq
->
contLen
=
contLen
;
mDebug
(
"stb:%s, meta is retrieved, cols:%d tags:%d"
,
pInfo
->
tableFn
ame
,
pStb
->
numOfColumns
,
pStb
->
numOfTags
);
mDebug
(
"stb:%s, meta is retrieved, cols:%d tags:%d"
,
tbFN
ame
,
pStb
->
numOfColumns
,
pStb
->
numOfTags
);
return
0
;
}
...
...
@@ -855,7 +860,7 @@ static int32_t mndGetStbMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pM
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_STB
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndSubscribe.c
浏览文件 @
2c33a51c
...
...
@@ -745,7 +745,7 @@ static int32_t mndProcessConsumerMetaMsg(SMnodeMsg *pMsg) {
SMnode
*
pMnode
=
pMsg
->
pMnode
;
STableInfoReq
*
pInfo
=
pMsg
->
rpcMsg
.
pCont
;
mDebug
(
"subscribe:%s, start to retrieve meta"
,
pInfo
->
t
ableFn
ame
);
mDebug
(
"subscribe:%s, start to retrieve meta"
,
pInfo
->
t
bN
ame
);
#if 0
SDbObj *pDb = mndAcquireDbByConsumer(pMnode, pInfo->tableFname);
...
...
@@ -876,7 +876,7 @@ static int32_t mndGetConsumerMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaRs
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_CONSUMER
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndTopic.c
浏览文件 @
2c33a51c
...
...
@@ -338,7 +338,7 @@ static int32_t mndProcessTopicMetaMsg(SMnodeMsg *pMsg) {
SMnode
*
pMnode
=
pMsg
->
pMnode
;
STableInfoReq
*
pInfo
=
pMsg
->
rpcMsg
.
pCont
;
mDebug
(
"topic:%s, start to retrieve meta"
,
pInfo
->
t
ableFn
ame
);
mDebug
(
"topic:%s, start to retrieve meta"
,
pInfo
->
t
bN
ame
);
#if 0
SDbObj *pDb = mndAcquireDbByTopic(pMnode, pInfo->tableFname);
...
...
@@ -469,7 +469,7 @@ static int32_t mndGetTopicMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaRsp *
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_TOPIC
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndUser.c
浏览文件 @
2c33a51c
...
...
@@ -474,7 +474,7 @@ static int32_t mndGetUserMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *p
pShow
->
numOfRows
=
sdbGetSize
(
pSdb
,
SDB_USER
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @
2c33a51c
...
...
@@ -525,7 +525,7 @@ static int32_t mndGetVgroupMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp
}
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
@@ -638,7 +638,7 @@ static int32_t mndGetVnodeMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow
->
replica
=
dnodeId
;
pShow
->
numOfRows
=
mndGetVnodesNum
(
pMnode
,
dnodeId
);
pShow
->
rowSize
=
pShow
->
offset
[
cols
-
1
]
+
pShow
->
bytes
[
cols
-
1
];
strcpy
(
pMeta
->
tb
Fn
ame
,
mndShowStr
(
pShow
->
type
));
strcpy
(
pMeta
->
tb
N
ame
,
mndShowStr
(
pShow
->
type
));
return
0
;
}
...
...
source/dnode/mnode/impl/test/show/show.cpp
浏览文件 @
2c33a51c
...
...
@@ -64,7 +64,7 @@ TEST_F(MndTestShow, 03_ShowMsg_Conn) {
test
.
SendShowMetaReq
(
TSDB_MGMT_TABLE_CONNS
,
""
);
STableMetaRsp
*
pMeta
=
test
.
GetShowMeta
();
EXPECT_STREQ
(
pMeta
->
tb
Fn
ame
,
"show connections"
);
EXPECT_STREQ
(
pMeta
->
tb
N
ame
,
"show connections"
);
EXPECT_EQ
(
pMeta
->
numOfTags
,
0
);
EXPECT_EQ
(
pMeta
->
numOfColumns
,
7
);
EXPECT_EQ
(
pMeta
->
precision
,
0
);
...
...
source/dnode/mnode/impl/test/stb/stb.cpp
浏览文件 @
2c33a51c
...
...
@@ -126,7 +126,8 @@ TEST_F(MndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
int32_t
contLen
=
sizeof
(
STableInfoReq
);
STableInfoReq
*
pReq
=
(
STableInfoReq
*
)
rpcMallocCont
(
contLen
);
strcpy
(
pReq
->
tableFname
,
"1.d1.stb"
);
strcpy
(
pReq
->
dbFName
,
"1.d1"
);
strcpy
(
pReq
->
tbName
,
"stb"
);
SRpcMsg
*
pMsg
=
test
.
SendReq
(
TDMT_MND_STB_META
,
pReq
,
contLen
);
ASSERT_NE
(
pMsg
,
nullptr
);
...
...
@@ -146,8 +147,9 @@ TEST_F(MndTestStb, 01_Create_Show_Meta_Drop_Restart_Stb) {
pSchema
->
bytes
=
htonl
(
pSchema
->
bytes
);
}
EXPECT_STREQ
(
pRsp
->
tbFname
,
"1.d1.stb"
);
EXPECT_STREQ
(
pRsp
->
stbFname
,
""
);
EXPECT_STREQ
(
pRsp
->
dbFName
,
"1.d1"
);
EXPECT_STREQ
(
pRsp
->
tbName
,
"stb"
);
EXPECT_STREQ
(
pRsp
->
stbName
,
"stb"
);
EXPECT_EQ
(
pRsp
->
numOfColumns
,
2
);
EXPECT_EQ
(
pRsp
->
numOfTags
,
3
);
EXPECT_EQ
(
pRsp
->
precision
,
TSDB_TIME_PRECISION_MILLI
);
...
...
source/dnode/vnode/src/vnd/vnodeQuery.c
浏览文件 @
2c33a51c
...
...
@@ -84,7 +84,7 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
int
msgLen
=
0
;
int32_t
code
=
TSDB_CODE_VND_APP_ERROR
;
pTbCfg
=
metaGetTbInfoByName
(
pVnode
->
pMeta
,
pReq
->
t
ableFn
ame
,
&
uid
);
pTbCfg
=
metaGetTbInfoByName
(
pVnode
->
pMeta
,
pReq
->
t
bN
ame
,
&
uid
);
if
(
pTbCfg
==
NULL
)
{
code
=
TSDB_CODE_VND_TB_NOT_EXIST
;
goto
_exit
;
...
...
@@ -119,13 +119,13 @@ static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
goto
_exit
;
}
memcpy
(
pTbMetaMsg
->
dbF
name
,
pReq
->
dbFname
,
sizeof
(
pTbMetaMsg
->
dbFn
ame
));
strcpy
(
pTbMetaMsg
->
tb
Fname
,
pTbCfg
->
n
ame
);
memcpy
(
pTbMetaMsg
->
dbF
Name
,
pReq
->
dbFName
,
sizeof
(
pTbMetaMsg
->
dbFN
ame
));
strcpy
(
pTbMetaMsg
->
tb
Name
,
pReq
->
tbN
ame
);
if
(
pTbCfg
->
type
==
META_CHILD_TABLE
)
{
strcpy
(
pTbMetaMsg
->
stb
Fn
ame
,
pStbCfg
->
name
);
strcpy
(
pTbMetaMsg
->
stb
N
ame
,
pStbCfg
->
name
);
pTbMetaMsg
->
suid
=
htobe64
(
pTbCfg
->
ctbCfg
.
suid
);
}
else
if
(
pTbCfg
->
type
==
META_SUPER_TABLE
)
{
strcpy
(
pTbMetaMsg
->
stb
Fn
ame
,
pTbCfg
->
name
);
strcpy
(
pTbMetaMsg
->
stb
N
ame
,
pTbCfg
->
name
);
pTbMetaMsg
->
suid
=
htobe64
(
uid
);
}
pTbMetaMsg
->
numOfTags
=
htonl
(
nTagCols
);
...
...
source/libs/catalog/inc/catalogInt.h
浏览文件 @
2c33a51c
...
...
@@ -47,55 +47,52 @@ enum {
CTG_RENT_STABLE
,
};
typedef
struct
SC
TG
Debug
{
typedef
struct
SC
tg
Debug
{
int32_t
lockDebug
;
}
SC
TG
Debug
;
}
SC
tg
Debug
;
typedef
struct
SVgroupListCache
{
int32_t
vgroupVersion
;
SHashObj
*
cache
;
// key:vgId, value:SVgroupInfo
}
SVgroupListCache
;
typedef
struct
SCtgTbMetaCache
{
SRWLatch
stbLock
;
SHashObj
*
cache
;
//key:tbname, value:STableMeta
SHashObj
*
stbCache
;
//key:suid, value:STableMeta*
}
SCtgTbMetaCache
;
typedef
struct
SDBVgroupCache
{
SHashObj
*
cache
;
//key:dbname, value:SDBVgroupInfo
}
SDBVgroupCache
;
typedef
struct
SCtgDBCache
{
SRWLatch
vgLock
;
int8_t
deleted
;
SDBVgroupInfo
*
vgInfo
;
SCtgTbMetaCache
tbCache
;
}
SCtgDBCache
;
typedef
struct
STableMetaCache
{
SRWLatch
stableLock
;
SHashObj
*
cache
;
//key:fulltablename, value:STableMeta
SHashObj
*
stableCache
;
//key:suid, value:STableMeta*
}
STableMetaCache
;
typedef
struct
SRentSlotInfo
{
typedef
struct
SCtgRentSlot
{
SRWLatch
lock
;
bool
needSort
;
SArray
*
meta
;
// element is SDbVgVersion or SSTableMetaVersion
}
S
RentSlotInfo
;
}
S
CtgRentSlot
;
typedef
struct
S
Meta
RentMgmt
{
typedef
struct
S
Ctg
RentMgmt
{
int8_t
type
;
uint16_t
slotNum
;
uint16_t
slotRIdx
;
int64_t
lastReadMsec
;
S
RentSlotInfo
*
slots
;
}
S
Meta
RentMgmt
;
S
CtgRentSlot
*
slots
;
}
S
Ctg
RentMgmt
;
typedef
struct
SCatalog
{
uint64_t
clusterId
;
SDBVgroupCache
dbCache
;
STableMetaCache
tableCache
;
SMetaRentMgmt
dbRent
;
SMetaRentMgmt
stableRent
;
uint64_t
clusterId
;
SHashObj
*
dbCache
;
//key:dbname, value:SCtgDBCache
SCtgRentMgmt
dbRent
;
SCtgRentMgmt
stbRent
;
}
SCatalog
;
typedef
struct
SCtgApiStat
{
}
SCtgApiStat
;
typedef
struct
SCtgR
esourc
eStat
{
typedef
struct
SCtgR
untim
eStat
{
}
SCtgR
esourc
eStat
;
}
SCtgR
untim
eStat
;
typedef
struct
SCtgCacheStat
{
...
...
@@ -103,7 +100,7 @@ typedef struct SCtgCacheStat {
typedef
struct
SCatalogStat
{
SCtgApiStat
api
;
SCtgR
esourceStat
resourc
e
;
SCtgR
untimeStat
runtim
e
;
SCtgCacheStat
cache
;
}
SCatalogStat
;
...
...
source/libs/catalog/src/catalog.c
浏览文件 @
2c33a51c
此差异已折叠。
点击以展开。
source/libs/catalog/test/catalogTests.cpp
浏览文件 @
2c33a51c
...
...
@@ -128,15 +128,14 @@ void ctgTestBuildCTableMetaOutput(STableMetaOutput *output) {
strcpy
(
sn
.
dbname
,
"db1"
);
strcpy
(
sn
.
tname
,
ctgTestSTablename
);
char
tbFullName
[
TSDB_TABLE_FNAME_LEN
]
;
tName
ExtractFullName
(
&
cn
,
tbFullName
);
char
db
[
TSDB_DB_FNAME_LEN
]
=
{
0
}
;
tName
GetFullDbName
(
&
cn
,
db
);
strcpy
(
output
->
dbFName
,
db
);
SET_META_TYPE_BOTH_TABLE
(
output
->
metaType
);
strcpy
(
output
->
ctbFname
,
tbFullName
);
tNameExtractFullName
(
&
cn
,
tbFullName
);
strcpy
(
output
->
tbFname
,
tbFullName
);
strcpy
(
output
->
ctbName
,
cn
.
tname
);
strcpy
(
output
->
tbName
,
sn
.
tname
);
output
->
ctbMeta
.
vgId
=
9
;
output
->
ctbMeta
.
tableType
=
TSDB_CHILD_TABLE
;
...
...
@@ -175,10 +174,11 @@ void ctgTestBuildCTableMetaOutput(STableMetaOutput *output) {
strcpy
(
s
->
name
,
"tag1s"
);
}
void
ctgTestBuildDBVgroup
(
SDBVgroupInfo
*
dbVgroup
)
{
void
ctgTestBuildDBVgroup
(
SDBVgroupInfo
*
*
p
dbVgroup
)
{
static
int32_t
vgVersion
=
ctgTestVgVersion
+
1
;
int32_t
vgNum
=
0
;
SVgroupInfo
vgInfo
=
{
0
};
SDBVgroupInfo
*
dbVgroup
=
(
SDBVgroupInfo
*
)
calloc
(
1
,
sizeof
(
SDBVgroupInfo
));
dbVgroup
->
vgVersion
=
vgVersion
++
;
...
...
@@ -186,7 +186,7 @@ void ctgTestBuildDBVgroup(SDBVgroupInfo *dbVgroup) {
dbVgroup
->
hashMethod
=
0
;
dbVgroup
->
dbId
=
ctgTestDbId
;
dbVgroup
->
vg
Info
=
taosHashInit
(
ctgTestVgNum
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
true
,
HASH_ENTRY_LOCK
);
dbVgroup
->
vg
Hash
=
taosHashInit
(
ctgTestVgNum
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
true
,
HASH_ENTRY_LOCK
);
vgNum
=
ctgTestGetVgNumFromVgVersion
(
dbVgroup
->
vgVersion
);
uint32_t
hashUnit
=
UINT32_MAX
/
vgNum
;
...
...
@@ -203,8 +203,10 @@ void ctgTestBuildDBVgroup(SDBVgroupInfo *dbVgroup) {
addr
->
port
=
htons
(
n
+
22
);
}
taosHashPut
(
dbVgroup
->
vg
Info
,
&
vgInfo
.
vgId
,
sizeof
(
vgInfo
.
vgId
),
&
vgInfo
,
sizeof
(
vgInfo
));
taosHashPut
(
dbVgroup
->
vg
Hash
,
&
vgInfo
.
vgId
,
sizeof
(
vgInfo
.
vgId
),
&
vgInfo
,
sizeof
(
vgInfo
));
}
*
pdbVgroup
=
dbVgroup
;
}
void
ctgTestPrepareDbVgroups
(
void
*
shandle
,
SEpSet
*
pEpSet
,
SRpcMsg
*
pMsg
,
SRpcMsg
*
pRsp
)
{
...
...
@@ -250,7 +252,8 @@ void ctgTestPrepareTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcM
pRsp
->
contLen
=
sizeof
(
STableMetaRsp
)
+
(
ctgTestColNum
+
ctgTestTagNum
)
*
sizeof
(
SSchema
);
pRsp
->
pCont
=
calloc
(
1
,
pRsp
->
contLen
);
rspMsg
=
(
STableMetaRsp
*
)
pRsp
->
pCont
;
sprintf
(
rspMsg
->
tbFname
,
"%s.%s"
,
ctgTestDbname
,
ctgTestTablename
);
strcpy
(
rspMsg
->
dbFName
,
ctgTestDbname
);
strcpy
(
rspMsg
->
tbName
,
ctgTestTablename
);
rspMsg
->
numOfTags
=
0
;
rspMsg
->
numOfColumns
=
htonl
(
ctgTestColNum
);
rspMsg
->
precision
=
1
;
...
...
@@ -285,8 +288,9 @@ void ctgTestPrepareCTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpc
pRsp
->
contLen
=
sizeof
(
STableMetaRsp
)
+
(
ctgTestColNum
+
ctgTestTagNum
)
*
sizeof
(
SSchema
);
pRsp
->
pCont
=
calloc
(
1
,
pRsp
->
contLen
);
rspMsg
=
(
STableMetaRsp
*
)
pRsp
->
pCont
;
sprintf
(
rspMsg
->
tbFname
,
"%s.%s"
,
ctgTestDbname
,
ctgTestCTablename
);
sprintf
(
rspMsg
->
stbFname
,
"%s.%s"
,
ctgTestDbname
,
ctgTestSTablename
);
strcpy
(
rspMsg
->
dbFName
,
ctgTestDbname
);
strcpy
(
rspMsg
->
tbName
,
ctgTestCTablename
);
strcpy
(
rspMsg
->
stbName
,
ctgTestSTablename
);
rspMsg
->
numOfTags
=
htonl
(
ctgTestTagNum
);
rspMsg
->
numOfColumns
=
htonl
(
ctgTestColNum
);
rspMsg
->
precision
=
1
;
...
...
@@ -327,8 +331,9 @@ void ctgTestPrepareSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpc
pRsp
->
contLen
=
sizeof
(
STableMetaRsp
)
+
(
ctgTestColNum
+
ctgTestTagNum
)
*
sizeof
(
SSchema
);
pRsp
->
pCont
=
calloc
(
1
,
pRsp
->
contLen
);
rspMsg
=
(
STableMetaRsp
*
)
pRsp
->
pCont
;
sprintf
(
rspMsg
->
tbFname
,
"%s.%s"
,
ctgTestDbname
,
ctgTestSTablename
);
sprintf
(
rspMsg
->
stbFname
,
"%s.%s"
,
ctgTestDbname
,
ctgTestSTablename
);
strcpy
(
rspMsg
->
dbFName
,
ctgTestDbname
);
strcpy
(
rspMsg
->
tbName
,
ctgTestSTablename
);
strcpy
(
rspMsg
->
stbName
,
ctgTestSTablename
);
rspMsg
->
numOfTags
=
htonl
(
ctgTestTagNum
);
rspMsg
->
numOfColumns
=
htonl
(
ctgTestColNum
);
rspMsg
->
precision
=
1
;
...
...
@@ -370,8 +375,9 @@ void ctgTestPrepareMultiSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg,
pRsp
->
contLen
=
sizeof
(
STableMetaRsp
)
+
(
ctgTestColNum
+
ctgTestTagNum
)
*
sizeof
(
SSchema
);
pRsp
->
pCont
=
calloc
(
1
,
pRsp
->
contLen
);
rspMsg
=
(
STableMetaRsp
*
)
pRsp
->
pCont
;
sprintf
(
rspMsg
->
tbFname
,
"%s.%s_%d"
,
ctgTestDbname
,
ctgTestSTablename
,
idx
);
sprintf
(
rspMsg
->
stbFname
,
"%s.%s_%d"
,
ctgTestDbname
,
ctgTestSTablename
,
idx
);
strcpy
(
rspMsg
->
dbFName
,
ctgTestDbname
);
sprintf
(
rspMsg
->
tbName
,
"%s_%d"
,
ctgTestSTablename
,
idx
);
sprintf
(
rspMsg
->
stbName
,
"%s_%d"
,
ctgTestSTablename
,
idx
);
rspMsg
->
numOfTags
=
htonl
(
ctgTestTagNum
);
rspMsg
->
numOfColumns
=
htonl
(
ctgTestColNum
);
rspMsg
->
precision
=
1
;
...
...
@@ -589,12 +595,12 @@ void *ctgTestGetDbVgroupThread(void *param) {
void
*
ctgTestSetDbVgroupThread
(
void
*
param
)
{
struct
SCatalog
*
pCtg
=
(
struct
SCatalog
*
)
param
;
int32_t
code
=
0
;
SDBVgroupInfo
dbVgroup
=
{
0
}
;
SDBVgroupInfo
*
dbVgroup
=
NULL
;
int32_t
n
=
0
;
while
(
!
ctgTestStop
)
{
ctgTestBuildDBVgroup
(
&
dbVgroup
);
code
=
catalogUpdateDBVgroup
(
pCtg
,
ctgTestDbname
,
&
dbVgroup
);
code
=
catalogUpdateDBVgroup
(
pCtg
,
ctgTestDbname
,
dbVgroup
);
if
(
code
)
{
assert
(
0
);
}
...
...
@@ -669,6 +675,7 @@ void *ctgTestSetCtableMetaThread(void *param) {
return
NULL
;
}
TEST
(
tableMeta
,
normalTable
)
{
struct
SCatalog
*
pCtg
=
NULL
;
void
*
mockPointer
=
(
void
*
)
0x1
;
...
...
@@ -1062,9 +1069,11 @@ TEST(dbVgroup, getSetDbVgroupCase) {
void
*
mockPointer
=
(
void
*
)
0x1
;
SVgroupInfo
vgInfo
=
{
0
};
SVgroupInfo
*
pvgInfo
=
NULL
;
SDBVgroupInfo
dbVgroup
=
{
0
}
;
SDBVgroupInfo
*
dbVgroup
=
NULL
;
SArray
*
vgList
=
NULL
;
ctgTestInitLogFile
();
ctgTestSetPrepareDbVgroupsAndNormalMeta
();
initQueryModuleMsgHandle
();
...
...
@@ -1099,7 +1108,7 @@ TEST(dbVgroup, getSetDbVgroupCase) {
taosArrayDestroy
(
vgList
);
ctgTestBuildDBVgroup
(
&
dbVgroup
);
code
=
catalogUpdateDBVgroup
(
pCtg
,
ctgTestDbname
,
&
dbVgroup
);
code
=
catalogUpdateDBVgroup
(
pCtg
,
ctgTestDbname
,
dbVgroup
);
ASSERT_EQ
(
code
,
0
);
code
=
catalogGetTableHashVgroup
(
pCtg
,
mockPointer
,
(
const
SEpSet
*
)
mockPointer
,
&
n
,
&
vgInfo
);
...
...
@@ -1169,6 +1178,7 @@ TEST(multiThread, getSetDbVgroupCase) {
catalogDestroy
();
}
TEST
(
multiThread
,
ctableMeta
)
{
struct
SCatalog
*
pCtg
=
NULL
;
void
*
mockPointer
=
(
void
*
)
0x1
;
...
...
@@ -1178,6 +1188,8 @@ TEST(multiThread, ctableMeta) {
SArray
*
vgList
=
NULL
;
ctgTestStop
=
false
;
ctgTestInitLogFile
();
ctgTestSetPrepareDbVgroupsAndChildMeta
();
initQueryModuleMsgHandle
();
...
...
@@ -1212,11 +1224,13 @@ TEST(multiThread, ctableMeta) {
}
ctgTestStop
=
true
;
sleep
(
1
);
sleep
(
2
);
catalogDestroy
();
}
TEST
(
rentTest
,
allRent
)
{
struct
SCatalog
*
pCtg
=
NULL
;
void
*
mockPointer
=
(
void
*
)
0x1
;
...
...
@@ -1229,6 +1243,8 @@ TEST(rentTest, allRent) {
SSTableMetaVersion
*
stable
=
NULL
;
uint32_t
num
=
0
;
ctgTestInitLogFile
();
ctgTestSetPrepareDbVgroupsAndMultiSuperMeta
();
initQueryModuleMsgHandle
();
...
...
@@ -1291,4 +1307,4 @@ int main(int argc, char **argv) {
return
RUN_ALL_TESTS
();
}
#pragma GCC diagnostic pop
\ No newline at end of file
#pragma GCC diagnostic pop
source/libs/qcom/src/querymsg.c
浏览文件 @
2c33a51c
...
...
@@ -45,11 +45,11 @@ int32_t queryBuildTableMetaReqMsg(void* input, char **msg, int32_t msgSize, int3
bMsg
->
header
.
vgId
=
htonl
(
bInput
->
vgId
);
if
(
bInput
->
dbName
)
{
tstrncpy
(
bMsg
->
dbF
name
,
bInput
->
dbName
,
tListLen
(
bMsg
->
dbFn
ame
));
if
(
bInput
->
db
F
Name
)
{
tstrncpy
(
bMsg
->
dbF
Name
,
bInput
->
dbFName
,
tListLen
(
bMsg
->
dbFN
ame
));
}
tstrncpy
(
bMsg
->
t
ableFname
,
bInput
->
tableFullName
,
tListLen
(
bMsg
->
tableFn
ame
));
tstrncpy
(
bMsg
->
t
bName
,
bInput
->
tbName
,
tListLen
(
bMsg
->
tbN
ame
));
*
msgLen
=
(
int32_t
)
sizeof
(
*
bMsg
);
return
TSDB_CODE_SUCCESS
;
...
...
@@ -113,12 +113,19 @@ int32_t queryProcessUseDBRsp(void* output, char *msg, int32_t msgSize) {
return
TSDB_CODE_TSC_VALUE_OUT_OF_RANGE
;
}
pOut
->
dbVgroup
.
vgVersion
=
pRsp
->
vgVersion
;
pOut
->
dbVgroup
.
hashMethod
=
pRsp
->
hashMethod
;
pOut
->
dbVgroup
.
dbId
=
pRsp
->
uid
;
pOut
->
dbVgroup
.
vgInfo
=
taosHashInit
(
pRsp
->
vgNum
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
true
,
HASH_ENTRY_LOCK
);
if
(
NULL
==
pOut
->
dbVgroup
.
vgInfo
)
{
qError
(
"hash init[%d] failed"
,
pRsp
->
vgNum
);
pOut
->
dbVgroup
=
calloc
(
1
,
sizeof
(
SDBVgroupInfo
));
if
(
NULL
==
pOut
->
dbVgroup
)
{
qError
(
"calloc %d failed"
,
(
int32_t
)
sizeof
(
SDBVgroupInfo
));
return
TSDB_CODE_TSC_OUT_OF_MEMORY
;
}
pOut
->
dbVgroup
->
vgVersion
=
pRsp
->
vgVersion
;
pOut
->
dbVgroup
->
hashMethod
=
pRsp
->
hashMethod
;
pOut
->
dbVgroup
->
dbId
=
pRsp
->
uid
;
pOut
->
dbVgroup
->
vgHash
=
taosHashInit
(
pRsp
->
vgNum
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
true
,
HASH_ENTRY_LOCK
);
if
(
NULL
==
pOut
->
dbVgroup
->
vgHash
)
{
qError
(
"taosHashInit %d failed"
,
pRsp
->
vgNum
);
tfree
(
pOut
->
dbVgroup
);
return
TSDB_CODE_TSC_OUT_OF_MEMORY
;
}
...
...
@@ -131,8 +138,8 @@ int32_t queryProcessUseDBRsp(void* output, char *msg, int32_t msgSize) {
pRsp
->
vgroupInfo
[
i
].
epset
.
eps
[
n
].
port
=
ntohs
(
pRsp
->
vgroupInfo
[
i
].
epset
.
eps
[
n
].
port
);
}
if
(
0
!=
taosHashPut
(
pOut
->
dbVgroup
.
vgInfo
,
&
pRsp
->
vgroupInfo
[
i
].
vgId
,
sizeof
(
pRsp
->
vgroupInfo
[
i
].
vgId
),
&
pRsp
->
vgroupInfo
[
i
],
sizeof
(
pRsp
->
vgroupInfo
[
i
])))
{
qError
(
"
hash push
failed"
);
if
(
0
!=
taosHashPut
(
pOut
->
dbVgroup
->
vgHash
,
&
pRsp
->
vgroupInfo
[
i
].
vgId
,
sizeof
(
pRsp
->
vgroupInfo
[
i
].
vgId
),
&
pRsp
->
vgroupInfo
[
i
],
sizeof
(
pRsp
->
vgroupInfo
[
i
])))
{
qError
(
"
taosHashPut
failed"
);
goto
_return
;
}
}
...
...
@@ -142,8 +149,10 @@ int32_t queryProcessUseDBRsp(void* output, char *msg, int32_t msgSize) {
return
code
;
_return:
if
(
pOut
)
{
tfree
(
pOut
->
dbVgroup
.
vgInfo
);
taosHashCleanup
(
pOut
->
dbVgroup
->
vgHash
);
tfree
(
pOut
->
dbVgroup
);
}
return
code
;
...
...
@@ -248,16 +257,13 @@ int32_t queryProcessTableMetaRsp(void* output, char *msg, int32_t msgSize) {
return
TSDB_CODE_TSC_INVALID_VALUE
;
}
strcpy
(
pOut
->
dbFName
,
pMetaMsg
->
dbFName
);
if
(
pMetaMsg
->
tableType
==
TSDB_CHILD_TABLE
)
{
SET_META_TYPE_BOTH_TABLE
(
pOut
->
metaType
);
if
(
pMetaMsg
->
dbFname
[
0
])
{
snprintf
(
pOut
->
ctbFname
,
sizeof
(
pOut
->
ctbFname
),
"%s.%s"
,
pMetaMsg
->
dbFname
,
pMetaMsg
->
tbFname
);
snprintf
(
pOut
->
tbFname
,
sizeof
(
pOut
->
tbFname
),
"%s.%s"
,
pMetaMsg
->
dbFname
,
pMetaMsg
->
stbFname
);
}
else
{
memcpy
(
pOut
->
ctbFname
,
pMetaMsg
->
tbFname
,
sizeof
(
pOut
->
ctbFname
));
memcpy
(
pOut
->
tbFname
,
pMetaMsg
->
stbFname
,
sizeof
(
pOut
->
tbFname
));
}
strcpy
(
pOut
->
ctbName
,
pMetaMsg
->
tbName
);
strcpy
(
pOut
->
tbName
,
pMetaMsg
->
stbName
);
pOut
->
ctbMeta
.
vgId
=
pMetaMsg
->
vgId
;
pOut
->
ctbMeta
.
tableType
=
pMetaMsg
->
tableType
;
...
...
@@ -268,11 +274,7 @@ int32_t queryProcessTableMetaRsp(void* output, char *msg, int32_t msgSize) {
}
else
{
SET_META_TYPE_TABLE
(
pOut
->
metaType
);
if
(
pMetaMsg
->
dbFname
[
0
])
{
snprintf
(
pOut
->
tbFname
,
sizeof
(
pOut
->
tbFname
),
"%s.%s"
,
pMetaMsg
->
dbFname
,
pMetaMsg
->
tbFname
);
}
else
{
memcpy
(
pOut
->
tbFname
,
pMetaMsg
->
tbFname
,
sizeof
(
pOut
->
tbFname
));
}
strcpy
(
pOut
->
tbName
,
pMetaMsg
->
tbName
);
code
=
queryCreateTableMetaFromMsg
(
pMetaMsg
,
(
pMetaMsg
->
tableType
==
TSDB_SUPER_TABLE
),
&
pOut
->
tbMeta
);
}
...
...
@@ -291,4 +293,4 @@ void initQueryModuleMsgHandle() {
queryProcessMsgRsp
[
TMSG_INDEX
(
TDMT_MND_USE_DB
)]
=
queryProcessUseDBRsp
;
}
#pragma GCC diagnostic pop
\ No newline at end of file
#pragma GCC diagnostic pop
source/libs/scheduler/inc/schedulerInt.h
浏览文件 @
2c33a51c
...
...
@@ -132,7 +132,7 @@ typedef struct SSchJob {
SQueryProfileSummary
summary
;
}
SSchJob
;
#define SCH_TASK_READY_TO_LUNCH(
task) (atomic_load_32(&(task)->childReady
) >= taosArrayGetSize((task)->children))
#define SCH_TASK_READY_TO_LUNCH(
readyNum, task) ((readyNum
) >= taosArrayGetSize((task)->children))
#define SCH_IS_DATA_SRC_TASK(task) ((task)->plan->type == QUERY_TYPE_SCAN)
#define SCH_TASK_NEED_WAIT_ALL(task) ((task)->plan->type == QUERY_TYPE_MODIFY)
...
...
source/libs/scheduler/src/scheduler.c
浏览文件 @
2c33a51c
...
...
@@ -773,14 +773,14 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
SSchTask
*
par
=
*
(
SSchTask
**
)
taosArrayGet
(
pTask
->
parents
,
i
);
pErrTask
=
par
;
atomic_add_fetch_32
(
&
par
->
childReady
,
1
);
int32_t
readyNum
=
atomic_add_fetch_32
(
&
par
->
childReady
,
1
);
SCH_LOCK
(
SCH_WRITE
,
&
par
->
lock
);
SDownstreamSource
source
=
{.
taskId
=
pTask
->
taskId
,
.
schedId
=
schMgmt
.
sId
,
.
addr
=
pTask
->
succeedAddr
};
qSetSubplanExecutionNode
(
par
->
plan
,
pTask
->
plan
->
id
.
templateId
,
&
source
);
SCH_UNLOCK
(
SCH_WRITE
,
&
par
->
lock
);
if
(
SCH_TASK_READY_TO_LUNCH
(
par
))
{
if
(
SCH_TASK_READY_TO_LUNCH
(
readyNum
,
par
))
{
SCH_ERR_RET
(
schLaunchTask
(
pJob
,
par
));
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录