Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
哥们要飞
TDengine
提交
d97941aa
T
TDengine
项目概览
哥们要飞
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
1
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
d97941aa
编写于
2月 10, 2020
作者:
S
slguan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
rename from meter to table
上级
fd424455
变更
14
隐藏空白更改
内联
并排
Showing
14 changed file
with
189 addition
and
162 deletion
+189
-162
src/client/inc/tscUtil.h
src/client/inc/tscUtil.h
+2
-2
src/client/src/tscLocal.c
src/client/src/tscLocal.c
+2
-2
src/client/src/tscServer.c
src/client/src/tscServer.c
+2
-2
src/inc/mnode.h
src/inc/mnode.h
+4
-7
src/inc/taosmsg.h
src/inc/taosmsg.h
+8
-5
src/mnode/CMakeLists.txt
src/mnode/CMakeLists.txt
+9
-11
src/mnode/inc/mgmtTable.h
src/mnode/inc/mgmtTable.h
+38
-0
src/mnode/inc/mgmtUtil.h
src/mnode/inc/mgmtUtil.h
+12
-7
src/mnode/src/mgmtDnodeInt.c
src/mnode/src/mgmtDnodeInt.c
+1
-1
src/mnode/src/mgmtMeter.c
src/mnode/src/mgmtMeter.c
+62
-62
src/mnode/src/mgmtShell.c
src/mnode/src/mgmtShell.c
+13
-13
src/mnode/src/mgmtSupertableQuery.c
src/mnode/src/mgmtSupertableQuery.c
+16
-16
src/mnode/src/mgmtUtil.c
src/mnode/src/mgmtUtil.c
+19
-33
src/mnode/src/mgmtVgroup.c
src/mnode/src/mgmtVgroup.c
+1
-1
未找到文件。
src/client/inc/tscUtil.h
浏览文件 @
d97941aa
...
...
@@ -30,10 +30,10 @@ extern "C" {
#include "tsdb.h"
#define UTIL_METER_IS_SUPERTABLE(metaInfo) \
(((metaInfo)->pMeterMeta != NULL) && ((metaInfo)->pMeterMeta->meterType == TSDB_
METER_METRIC
))
(((metaInfo)->pMeterMeta != NULL) && ((metaInfo)->pMeterMeta->meterType == TSDB_
TABLE_TYPE_SUPER_TABLE
))
#define UTIL_METER_IS_NOMRAL_METER(metaInfo) (!(UTIL_METER_IS_SUPERTABLE(metaInfo)))
#define UTIL_METER_IS_CREATE_FROM_METRIC(metaInfo) \
(((metaInfo)->pMeterMeta != NULL) && ((metaInfo)->pMeterMeta->meterType == TSDB_
METER_M
TABLE))
(((metaInfo)->pMeterMeta != NULL) && ((metaInfo)->pMeterMeta->meterType == TSDB_
TABLE_TYPE_CREATE_FROM_S
TABLE))
#define TSDB_COL_IS_TAG(f) (((f)&TSDB_COL_TAG) != 0)
...
...
src/client/src/tscLocal.c
浏览文件 @
d97941aa
...
...
@@ -79,8 +79,8 @@ static int32_t getToStringLength(const char *pData, int32_t length, int32_t type
static
int32_t
tscMaxLengthOfTagsFields
(
SSqlObj
*
pSql
)
{
SMeterMeta
*
pMeta
=
tscGetMeterMetaInfo
(
&
pSql
->
cmd
,
0
,
0
)
->
pMeterMeta
;
if
(
pMeta
->
meterType
==
TSDB_
METER_METRIC
||
pMeta
->
meterType
==
TSDB_METER_O
TABLE
||
pMeta
->
meterType
==
TSDB_
METER_S
TABLE
)
{
if
(
pMeta
->
meterType
==
TSDB_
TABLE_TYPE_SUPER_TABLE
||
pMeta
->
meterType
==
TSDB_TABLE_TYPE_NORMAL_
TABLE
||
pMeta
->
meterType
==
TSDB_
TABLE_TYPE_STREAM_
TABLE
)
{
return
0
;
}
...
...
src/client/src/tscServer.c
浏览文件 @
d97941aa
...
...
@@ -2991,7 +2991,7 @@ int tscProcessMeterMetaRsp(SSqlObj *pSql) {
int32_t
tagLen
=
0
;
SSchema
*
pTagsSchema
=
tsGetTagSchema
(
pMeta
);
if
(
pMeta
->
meterType
==
TSDB_
METER_M
TABLE
)
{
if
(
pMeta
->
meterType
==
TSDB_
TABLE_TYPE_CREATE_FROM_S
TABLE
)
{
for
(
int32_t
i
=
0
;
i
<
pMeta
->
numOfTags
;
++
i
)
{
tagLen
+=
pTagsSchema
[
i
].
bytes
;
}
...
...
@@ -3106,7 +3106,7 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
int32_t
tagLen
=
0
;
SSchema
*
pTagsSchema
=
tsGetTagSchema
(
pMeta
);
if
(
pMeta
->
meterType
==
TSDB_
METER_M
TABLE
)
{
if
(
pMeta
->
meterType
==
TSDB_
TABLE_TYPE_CREATE_FROM_S
TABLE
)
{
for
(
int32_t
j
=
0
;
j
<
pMeta
->
numOfTags
;
++
j
)
{
tagLen
+=
pTagsSchema
[
j
].
bytes
;
}
...
...
src/inc/mnode.h
浏览文件 @
d97941aa
...
...
@@ -324,20 +324,17 @@ void mgmtCleanUpVgroups();
// meter API
int
mgmtInitMeters
();
STabObj
*
mgmtGet
Meter
(
char
*
meterId
);
STabObj
*
mgmtGet
Meter
Info
(
char
*
src
,
char
*
tags
[]);
STabObj
*
mgmtGet
Table
(
char
*
meterId
);
STabObj
*
mgmtGet
Table
Info
(
char
*
src
,
char
*
tags
[]);
int
mgmtRetrieveMetricMeta
(
SConnObj
*
pConn
,
char
**
pStart
,
SMetricMetaMsg
*
pInfo
);
int
mgmtCreateMeter
(
SDbObj
*
pDb
,
SCreateTableMsg
*
pCreate
);
int
mgmtDropMeter
(
SDbObj
*
pDb
,
char
*
meterId
,
int
ignore
);
int
mgmtAlterMeter
(
SDbObj
*
pDb
,
SAlterTableMsg
*
pAlter
);
int
mgmtGet
Meter
Meta
(
SMeterMeta
*
pMeta
,
SShowObj
*
pShow
,
SConnObj
*
pConn
);
int
mgmtGet
Table
Meta
(
SMeterMeta
*
pMeta
,
SShowObj
*
pShow
,
SConnObj
*
pConn
);
int
mgmtRetrieveMeters
(
SShowObj
*
pShow
,
char
*
data
,
int
rows
,
SConnObj
*
pConn
);
void
mgmtCleanUpMeters
();
SSchema
*
mgmtGet
Meter
Schema
(
STabObj
*
pMeter
);
// get schema for a meter
SSchema
*
mgmtGet
Table
Schema
(
STabObj
*
pMeter
);
// get schema for a meter
bool
mgmtMeterCreateFromMetric
(
STabObj
*
pMeterObj
);
bool
mgmtIsMetric
(
STabObj
*
pMeterObj
);
bool
mgmtIsNormalMeter
(
STabObj
*
pMeterObj
);
// dnode API
int
mgmtInitDnodes
();
...
...
src/inc/taosmsg.h
浏览文件 @
d97941aa
...
...
@@ -181,11 +181,14 @@ enum _mgmt_table {
#define TSDB_KILL_MSG_LEN 30
#define TSDB_METER_METRIC 0 // metric
#define TSDB_METER_MTABLE 1 // table created from metric
#define TSDB_METER_OTABLE 2 // ordinary table
#define TSDB_METER_STABLE 3 // table created from stream computing
#define TSDB_MAX_METER_TYPES 4
enum
{
TSDB_TABLE_TYPE_SUPER_TABLE
=
0
,
// super table
TSDB_TABLE_TYPE_CREATE_FROM_STABLE
=
1
,
// table created from super table
TSDB_TABLE_TYPE_NORMAL_TABLE
=
2
,
// ordinary table
TSDB_TABLE_TYPE_STREAM_TABLE
=
3
,
// table created from stream computing
TSDB_TABLE_TYPE_MAX
=
4
}
ETableType
;
#define TSDB_VN_READ_ACCCESS ((char)0x1)
#define TSDB_VN_WRITE_ACCCESS ((char)0x2)
...
...
src/mnode/CMakeLists.txt
浏览文件 @
d97941aa
...
...
@@ -2,22 +2,20 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT
(
TDengine
)
IF
((
TD_LINUX_64
)
OR
(
TD_LINUX_32 AND TD_ARM
))
INCLUDE_DIRECTORIES
(
${
TD_COMMUNITY_DIR
}
/src/dnode/inc
)
INCLUDE_DIRECTORIES
(
${
TD_COMMUNITY_DIR
}
/src/mnode/detail/inc
)
INCLUDE_DIRECTORIES
(
${
TD_COMMUNITY_DIR
}
/src/vnode/detail/inc
)
INCLUDE_DIRECTORIES
(
${
TD_COMMUNITY_DIR
}
/src/client/inc
)
INCLUDE_DIRECTORIES
(
${
TD_COMMUNITY_DIR
}
/src/inc
)
INCLUDE_DIRECTORIES
(
${
TD_OS_DIR
}
/inc
)
INCLUDE_DIRECTORIES
(
${
TD_COMMUNITY_DIR
}
/src/inc
)
INCLUDE_DIRECTORIES
(
${
TD_COMMUNITY_DIR
}
/src/util/inc
)
INCLUDE_DIRECTORIES
(
${
TD_COMMUNITY_DIR
}
/src/dnode/inc
)
INCLUDE_DIRECTORIES
(
inc
)
AUX_SOURCE_DIRECTORY
(
src SRC
)
ADD_LIBRARY
(
mnode
${
SRC
}
)
TARGET_LINK_LIBRARIES
(
mnode trpc tutil sdb pthread
)
#
TARGET_LINK_LIBRARIES(mnode trpc tutil sdb pthread)
IF
(
TD_CLUSTER
)
TARGET_LINK_LIBRARIES
(
mnode mcluster
)
ELSEIF
(
TD_LITE
)
TARGET_LINK_LIBRARIES
(
mnode mlite
)
ENDIF
()
#IF (TD_CLUSTER)
# TARGET_LINK_LIBRARIES(mnode mcluster)
#ENDIF ()
ENDIF
()
src/mnode/inc/mgmtTable.h
0 → 100644
浏览文件 @
d97941aa
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include "tast.h"
#ifndef TBASE_MNODE_UTIL_H
#define TBASE_MNODE_UTIL_H
bool
mgmtTableCreateFromSuperTable
(
STabObj
*
pTableObj
);
bool
mgmtIsSuperTable
(
STabObj
*
pTableObj
);
bool
mgmtIsNormalTable
(
STabObj
*
pTableObj
);
typedef
struct
SSyntaxTreeFilterSupporter
{
SSchema
*
pTagSchema
;
int32_t
numOfTags
;
int32_t
optr
;
}
SSyntaxTreeFilterSupporter
;
char
*
mgmtTableGetTag
(
STabObj
*
pMeter
,
int32_t
col
,
SSchema
*
pTagColSchema
);
int32_t
mgmtGetTagsLength
(
STabObj
*
pMetric
,
int32_t
col
);
bool
mgmtCheckIsMonitorDB
(
char
*
db
,
char
*
monitordb
);
int32_t
mgmtCheckDBParams
(
SCreateDbMsg
*
pCreate
);
#endif
src/mnode/inc/mgmtUtil.h
浏览文件 @
d97941aa
...
...
@@ -17,8 +17,12 @@
#include "tast.h"
#ifndef TBASE_MGMTUTIL_H
#define TBASE_MGMTUTIL_H
#ifndef TBASE_MNODE_UTIL_H
#define TBASE_MNODE_UTIL_H
bool
mgmtTableCreateFromSuperTable
(
STabObj
*
pTableObj
);
bool
mgmtIsSuperTable
(
STabObj
*
pTableObj
);
bool
mgmtIsNormalTable
(
STabObj
*
pTableObj
);
typedef
struct
SSyntaxTreeFilterSupporter
{
SSchema
*
pTagSchema
;
...
...
@@ -26,17 +30,18 @@ typedef struct SSyntaxTreeFilterSupporter {
int32_t
optr
;
}
SSyntaxTreeFilterSupporter
;
char
*
mgmtMeterGetTag
(
STabObj
*
pMeter
,
int32_t
col
,
SSchema
*
pTagColSchema
);
int32_t
mgmtFindTagCol
(
STabObj
*
pMetric
,
const
char
*
tagName
);
char
*
mgmtTableGetTag
(
STabObj
*
pMeter
,
int32_t
col
,
SSchema
*
pTagColSchema
);
int32_t
mgmtGetTagsLength
(
STabObj
*
pMetric
,
int32_t
col
);
bool
mgmtCheckIsMonitorDB
(
char
*
db
,
char
*
monitordb
);
bool
mgmtCheckIsMonitorDB
(
char
*
db
,
char
*
monitordb
);
int32_t
mgmtCheckDBParams
(
SCreateDbMsg
*
pCreate
);
int32_t
mgmtFindTagCol
(
STabObj
*
pMetric
,
const
char
*
tagName
);
int32_t
mgmtRetrieveMetersFromMetric
(
SMetricMetaMsg
*
pInfo
,
int32_t
tableIndex
,
tQueryResultset
*
pRes
);
int32_t
mgmtDoJoin
(
SMetricMetaMsg
*
pMetricMetaMsg
,
tQueryResultset
*
pRes
);
void
mgmtReorganizeMetersInMetricMeta
(
SMetricMetaMsg
*
pInfo
,
int32_t
index
,
tQueryResultset
*
pRes
);
bool
tSkipListNodeFilterCallback
(
const
void
*
pNode
,
void
*
param
);
#endif //TBASE_MGMTUTIL_H
#endif
src/mnode/src/mgmtDnodeInt.c
浏览文件 @
d97941aa
...
...
@@ -206,7 +206,7 @@ char *mgmtBuildCreateMeterIe(STabObj *pMeter, char *pMsg, int vnode) {
*/
pCreateMeter
->
sversion
=
htonl
(
pMeter
->
sversion
);
pCreateMeter
->
numOfColumns
=
htons
(
pMeter
->
numOfColumns
);
SSchema
*
pSchema
=
mgmtGet
Meter
Schema
(
pMeter
);
SSchema
*
pSchema
=
mgmtGet
Table
Schema
(
pMeter
);
for
(
int
i
=
0
;
i
<
pMeter
->
numOfColumns
;
++
i
)
{
pCreateMeter
->
schema
[
i
].
type
=
pSchema
[
i
].
type
;
...
...
src/mnode/src/mgmtMeter.c
浏览文件 @
d97941aa
...
...
@@ -109,7 +109,7 @@ void mgmtMeterActionInit() {
}
static
int32_t
mgmtGetReqTagsLength
(
STabObj
*
pMetric
,
int16_t
*
cols
,
int32_t
numOfCols
)
{
assert
(
mgmtIs
Metric
(
pMetric
)
&&
numOfCols
>=
0
&&
numOfCols
<=
TSDB_MAX_TAGS
+
1
);
assert
(
mgmtIs
SuperTable
(
pMetric
)
&&
numOfCols
>=
0
&&
numOfCols
<=
TSDB_MAX_TAGS
+
1
);
int32_t
len
=
0
;
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
...
...
@@ -142,7 +142,7 @@ void *mgmtMeterActionReset(void *row, char *str, int size, int *ssize) {
pMeter
->
schema
=
(
char
*
)
realloc
(
pMeter
->
schema
,
pMeter
->
schemaSize
);
memcpy
(
pMeter
->
schema
,
str
+
tsize
,
pMeter
->
schemaSize
);
if
(
mgmt
MeterCreateFromMetric
(
pMeter
))
{
if
(
mgmt
TableCreateFromSuperTable
(
pMeter
))
{
pMeter
->
pTagData
=
pMeter
->
schema
;
}
...
...
@@ -164,7 +164,7 @@ void *mgmtMeterActionInsert(void *row, char *str, int size, int *ssize) {
pMeter
=
(
STabObj
*
)
row
;
if
(
mgmtIsNormal
Meter
(
pMeter
))
{
if
(
mgmtIsNormal
Table
(
pMeter
))
{
pVgroup
=
mgmtGetVgroup
(
pMeter
->
gid
.
vgId
);
if
(
pVgroup
==
NULL
)
{
mError
(
"id:%s not in vgroup:%d"
,
pMeter
->
meterId
,
pMeter
->
gid
.
vgId
);
...
...
@@ -185,17 +185,17 @@ void *mgmtMeterActionInsert(void *row, char *str, int size, int *ssize) {
}
}
if
(
mgmt
MeterCreateFromMetric
(
pMeter
))
{
if
(
mgmt
TableCreateFromSuperTable
(
pMeter
))
{
pMeter
->
pTagData
=
(
char
*
)
pMeter
->
schema
;
pMetric
=
mgmtGet
Meter
(
pMeter
->
pTagData
);
pMetric
=
mgmtGet
Table
(
pMeter
->
pTagData
);
assert
(
pMetric
!=
NULL
);
}
if
(
pMeter
->
meterType
==
TSDB_
METER_S
TABLE
)
{
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_STREAM_
TABLE
)
{
pMeter
->
pSql
=
(
char
*
)
pMeter
->
schema
+
sizeof
(
SSchema
)
*
pMeter
->
numOfColumns
;
}
if
(
mgmtIsNormal
Meter
(
pMeter
))
{
if
(
mgmtIsNormal
Table
(
pMeter
))
{
if
(
pMetric
)
mgmtAddMeterIntoMetric
(
pMetric
,
pMeter
);
if
(
!
sdbMaster
)
{
...
...
@@ -233,7 +233,7 @@ void *mgmtMeterActionDelete(void *row, char *str, int size, int *ssize) {
pMeter
=
(
STabObj
*
)
row
;
if
(
mgmtIsNormal
Meter
(
pMeter
))
{
if
(
mgmtIsNormal
Table
(
pMeter
))
{
if
(
pMeter
->
gid
.
vgId
==
0
)
{
return
NULL
;
}
...
...
@@ -251,13 +251,13 @@ void *mgmtMeterActionDelete(void *row, char *str, int size, int *ssize) {
}
}
if
(
mgmt
MeterCreateFromMetric
(
pMeter
))
{
if
(
mgmt
TableCreateFromSuperTable
(
pMeter
))
{
pMeter
->
pTagData
=
(
char
*
)
pMeter
->
schema
;
pMetric
=
mgmtGet
Meter
(
pMeter
->
pTagData
);
pMetric
=
mgmtGet
Table
(
pMeter
->
pTagData
);
assert
(
pMetric
!=
NULL
);
}
if
(
mgmtIsNormal
Meter
(
pMeter
))
{
if
(
mgmtIsNormal
Table
(
pMeter
))
{
if
(
pMetric
)
mgmtRemoveMeterFromMetric
(
pMetric
,
pMeter
);
pVgroup
->
meterList
[
pMeter
->
gid
.
sid
]
=
NULL
;
...
...
@@ -285,7 +285,7 @@ void *mgmtMeterActionUpdate(void *row, char *str, int size, int *ssize) {
STabObj
*
pNew
=
(
STabObj
*
)
str
;
if
(
pNew
->
isDirty
)
{
pMetric
=
mgmtGet
Meter
(
pMeter
->
pTagData
);
pMetric
=
mgmtGet
Table
(
pMeter
->
pTagData
);
removeMeterFromMetricIndex
(
pMetric
,
pMeter
);
}
mgmtMeterActionReset
(
pMeter
,
str
,
size
,
NULL
);
...
...
@@ -353,7 +353,7 @@ void *mgmtMeterActionBatchUpdate(void *row, char *str, int size, int *ssize) {
STabObj
*
pMeter
=
(
STabObj
*
)
row
;
SMeterBatchUpdateMsg
*
msg
=
(
SMeterBatchUpdateMsg
*
)
str
;
if
(
mgmtIs
Metric
(
pMeter
))
{
if
(
mgmtIs
SuperTable
(
pMeter
))
{
if
(
msg
->
type
==
SDB_TYPE_INSERT
)
{
// Insert schema
uint32_t
total_cols
=
pMeter
->
numOfColumns
+
pMeter
->
numOfTags
;
pMeter
->
schema
=
realloc
(
pMeter
->
schema
,
(
total_cols
+
msg
->
cols
)
*
sizeof
(
SSchema
));
...
...
@@ -378,7 +378,7 @@ void *mgmtMeterActionBatchUpdate(void *row, char *str, int size, int *ssize) {
return
pMeter
->
pHead
;
}
else
if
(
mgmt
MeterCreateFromMetric
(
pMeter
))
{
}
else
if
(
mgmt
TableCreateFromSuperTable
(
pMeter
))
{
if
(
msg
->
type
==
SDB_TYPE_INSERT
)
{
SSchema
*
schemas
=
(
SSchema
*
)
msg
->
data
;
int
total_size
=
0
;
...
...
@@ -452,7 +452,7 @@ int mgmtInitMeters() {
while
(
1
)
{
pNode
=
sdbFetchRow
(
meterSdb
,
pNode
,
(
void
**
)
&
pMeter
);
if
(
pMeter
==
NULL
)
break
;
if
(
mgmtIs
Metric
(
pMeter
))
pMeter
->
numOfMeters
=
0
;
if
(
mgmtIs
SuperTable
(
pMeter
))
pMeter
->
numOfMeters
=
0
;
}
pNode
=
NULL
;
...
...
@@ -470,7 +470,7 @@ int mgmtInitMeters() {
continue
;
}
if
(
mgmtIsNormal
Meter
(
pMeter
))
{
if
(
mgmtIsNormal
Table
(
pMeter
))
{
pVgroup
=
mgmtGetVgroup
(
pMeter
->
gid
.
vgId
);
if
(
pVgroup
==
NULL
)
{
...
...
@@ -501,13 +501,13 @@ int mgmtInitMeters() {
pVgroup
->
meterList
[
pMeter
->
gid
.
sid
]
=
pMeter
;
taosIdPoolMarkStatus
(
pVgroup
->
idPool
,
pMeter
->
gid
.
sid
,
1
);
if
(
pMeter
->
meterType
==
TSDB_
METER_S
TABLE
)
{
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_STREAM_
TABLE
)
{
pMeter
->
pSql
=
(
char
*
)
pMeter
->
schema
+
sizeof
(
SSchema
)
*
pMeter
->
numOfColumns
;
}
if
(
mgmt
MeterCreateFromMetric
(
pMeter
))
{
if
(
mgmt
TableCreateFromSuperTable
(
pMeter
))
{
pMeter
->
pTagData
=
(
char
*
)
pMeter
->
schema
;
// + sizeof(SSchema)*pMeter->numOfColumns;
pMetric
=
mgmtGet
Meter
(
pMeter
->
pTagData
);
pMetric
=
mgmtGet
Table
(
pMeter
->
pTagData
);
if
(
pMetric
)
mgmtAddMeterIntoMetric
(
pMetric
,
pMeter
);
}
...
...
@@ -524,7 +524,7 @@ int mgmtInitMeters() {
return
0
;
}
STabObj
*
mgmtGet
Meter
(
char
*
meterId
)
{
return
(
STabObj
*
)
sdbGetRow
(
meterSdb
,
meterId
);
}
STabObj
*
mgmtGet
Table
(
char
*
meterId
)
{
return
(
STabObj
*
)
sdbGetRow
(
meterSdb
,
meterId
);
}
int
mgmtCreateMeter
(
SDbObj
*
pDb
,
SCreateTableMsg
*
pCreate
)
{
STabObj
*
pMeter
=
NULL
;
...
...
@@ -548,7 +548,7 @@ int mgmtCreateMeter(SDbObj *pDb, SCreateTableMsg *pCreate) {
}
// does table exist?
pMeter
=
mgmtGet
Meter
(
pCreate
->
meterId
);
pMeter
=
mgmtGet
Table
(
pCreate
->
meterId
);
if
(
pMeter
)
{
if
(
pCreate
->
igExists
)
{
return
TSDB_CODE_SUCCESS
;
...
...
@@ -563,9 +563,9 @@ int mgmtCreateMeter(SDbObj *pDb, SCreateTableMsg *pCreate) {
memset
(
pMeter
,
0
,
sizeof
(
STabObj
));
if
(
pCreate
->
numOfColumns
==
0
&&
pCreate
->
numOfTags
==
0
)
{
// MTABLE
pMeter
->
meterType
=
TSDB_
METER_M
TABLE
;
pMeter
->
meterType
=
TSDB_
TABLE_TYPE_CREATE_FROM_S
TABLE
;
char
*
pTagData
=
(
char
*
)
pCreate
->
schema
;
// it is a tag key
pMetric
=
mgmtGet
Meter
(
pTagData
);
pMetric
=
mgmtGet
Table
(
pTagData
);
if
(
pMetric
==
NULL
)
{
mError
(
"table:%s, corresponding super table does not exist"
,
pCreate
->
meterId
);
return
TSDB_CODE_INVALID_TABLE
;
...
...
@@ -615,16 +615,16 @@ int mgmtCreateMeter(SDbObj *pDb, SCreateTableMsg *pCreate) {
}
if
(
pCreate
->
sqlLen
>
0
)
{
pMeter
->
meterType
=
TSDB_
METER_S
TABLE
;
pMeter
->
meterType
=
TSDB_
TABLE_TYPE_STREAM_
TABLE
;
pMeter
->
pSql
=
pMeter
->
schema
+
numOfCols
*
sizeof
(
SSchema
);
memcpy
(
pMeter
->
pSql
,
(
char
*
)(
pCreate
->
schema
)
+
numOfCols
*
sizeof
(
SSchema
),
pCreate
->
sqlLen
);
pMeter
->
pSql
[
pCreate
->
sqlLen
-
1
]
=
0
;
mTrace
(
"table:%s, stream sql len:%d sql:%s"
,
pCreate
->
meterId
,
pCreate
->
sqlLen
,
pMeter
->
pSql
);
}
else
{
if
(
pCreate
->
numOfTags
>
0
)
{
pMeter
->
meterType
=
TSDB_
METER_METRIC
;
pMeter
->
meterType
=
TSDB_
TABLE_TYPE_SUPER_TABLE
;
}
else
{
pMeter
->
meterType
=
TSDB_
METER_O
TABLE
;
pMeter
->
meterType
=
TSDB_
TABLE_TYPE_NORMAL_
TABLE
;
}
}
}
...
...
@@ -723,7 +723,7 @@ int mgmtDropMeter(SDbObj *pDb, char *meterId, int ignore) {
STabObj
*
pMeter
;
SAcctObj
*
pAcct
;
pMeter
=
mgmtGet
Meter
(
meterId
);
pMeter
=
mgmtGet
Table
(
meterId
);
if
(
pMeter
==
NULL
)
{
if
(
ignore
)
{
return
TSDB_CODE_SUCCESS
;
...
...
@@ -739,7 +739,7 @@ int mgmtDropMeter(SDbObj *pDb, char *meterId, int ignore) {
return
TSDB_CODE_MONITOR_DB_FORBEIDDEN
;
}
if
(
mgmtIsNormal
Meter
(
pMeter
))
{
if
(
mgmtIsNormal
Table
(
pMeter
))
{
return
dropMeterImp
(
pDb
,
pMeter
,
pAcct
);
}
else
{
// remove a metric
...
...
@@ -762,7 +762,7 @@ int mgmtDropMeter(SDbObj *pDb, char *meterId, int ignore) {
int
mgmtAlterMeter
(
SDbObj
*
pDb
,
SAlterTableMsg
*
pAlter
)
{
STabObj
*
pMeter
;
pMeter
=
mgmtGet
Meter
(
pAlter
->
meterId
);
pMeter
=
mgmtGet
Table
(
pAlter
->
meterId
);
if
(
pMeter
==
NULL
)
{
return
TSDB_CODE_INVALID_TABLE
;
}
...
...
@@ -771,7 +771,7 @@ int mgmtAlterMeter(SDbObj *pDb, SAlterTableMsg *pAlter) {
if
(
mgmtCheckIsMonitorDB
(
pDb
->
name
,
tsMonitorDbName
))
return
TSDB_CODE_MONITOR_DB_FORBEIDDEN
;
if
(
pAlter
->
type
==
TSDB_ALTER_TABLE_UPDATE_TAG_VAL
)
{
if
(
!
mgmtIsNormal
Meter
(
pMeter
)
||
!
mgmtMeterCreateFromMetric
(
pMeter
))
{
if
(
!
mgmtIsNormal
Table
(
pMeter
)
||
!
mgmtTableCreateFromSuperTable
(
pMeter
))
{
return
TSDB_CODE_OPS_NOT_SUPPORT
;
}
}
...
...
@@ -927,7 +927,7 @@ int mgmtRemoveMeterFromMetric(STabObj *pMetric, STabObj *pMeter) {
void
mgmtCleanUpMeters
()
{
sdbCloseTable
(
meterSdb
);
}
int
mgmtGet
Meter
Meta
(
SMeterMeta
*
pMeta
,
SShowObj
*
pShow
,
SConnObj
*
pConn
)
{
int
mgmtGet
Table
Meta
(
SMeterMeta
*
pMeta
,
SShowObj
*
pShow
,
SConnObj
*
pConn
)
{
int
cols
=
0
;
SDbObj
*
pDb
=
NULL
;
...
...
@@ -974,16 +974,16 @@ int mgmtGetMeterMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn) {
return
0
;
}
SSchema
*
mgmtGet
Meter
Schema
(
STabObj
*
pMeter
)
{
SSchema
*
mgmtGet
Table
Schema
(
STabObj
*
pMeter
)
{
if
(
pMeter
==
NULL
)
{
return
NULL
;
}
if
(
!
mgmt
MeterCreateFromMetric
(
pMeter
))
{
if
(
!
mgmt
TableCreateFromSuperTable
(
pMeter
))
{
return
(
SSchema
*
)
pMeter
->
schema
;
}
STabObj
*
pMetric
=
mgmtGet
Meter
(
pMeter
->
pTagData
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pMeter
->
pTagData
);
assert
(
pMetric
!=
NULL
);
return
(
SSchema
*
)
pMetric
->
schema
;
...
...
@@ -1001,7 +1001,7 @@ static int32_t mgmtSerializeTagValue(char* pMsg, STabObj* pMeter, int16_t* tagsI
offset
+=
TSDB_METER_NAME_LEN
;
}
else
{
SSchema
s
=
{
0
};
char
*
tag
=
mgmt
Meter
GetTag
(
pMeter
,
tagsId
[
j
],
&
s
);
char
*
tag
=
mgmt
Table
GetTag
(
pMeter
,
tagsId
[
j
],
&
s
);
memcpy
(
pMsg
+
offset
,
tag
,
(
size_t
)
s
.
bytes
);
offset
+=
s
.
bytes
;
...
...
@@ -1201,9 +1201,9 @@ int mgmtRetrieveMetricMeta(SConnObj *pConn, char **pStart, SMetricMetaMsg *pMetr
for
(
int32_t
i
=
0
;
i
<
pMetricMetaMsg
->
numOfMeters
;
++
i
)
{
SMetricMetaElemMsg
*
pElem
=
doConvertMetricMetaMsg
(
pMetricMetaMsg
,
i
);
STabObj
*
pMetric
=
mgmtGet
Meter
(
pElem
->
meterId
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pElem
->
meterId
);
if
(
!
mgmtIs
Metric
(
pMetric
))
{
if
(
!
mgmtIs
SuperTable
(
pMetric
))
{
ret
=
TSDB_CODE_NOT_SUPER_TABLE
;
break
;
}
...
...
@@ -1218,7 +1218,7 @@ int mgmtRetrieveMetricMeta(SConnObj *pConn, char **pStart, SMetricMetaMsg *pMetr
for (int32_t i = 0; i < pMetricMetaMsg->numOfMeters; ++i) {
SMetricMetaElemMsg *pElem = (SMetricMetaElemMsg*) ((char *) pMetricMetaMsg + pMetricMetaMsg->metaElem[i]);
STabObj *pMetric = mgmtGet
Meter
(pElem->meterId);
STabObj *pMetric = mgmtGet
Table
(pElem->meterId);
if (pMetric->pSkipList->nSize > num) {
index = i;
...
...
@@ -1294,7 +1294,7 @@ int mgmtRetrieveMeters(SShowObj *pShow, char *data, int rows, SConnObj *pConn) {
pShow
->
pNode
=
sdbFetchRow
(
meterSdb
,
pShow
->
pNode
,
(
void
**
)
&
pMeter
);
if
(
pMeter
==
NULL
)
break
;
if
(
mgmtIs
Metric
(
pMeter
))
continue
;
if
(
mgmtIs
SuperTable
(
pMeter
))
continue
;
// not belong to current db
if
(
strncmp
(
pMeter
->
meterId
,
prefix
,
prefixLen
))
continue
;
...
...
@@ -1455,7 +1455,7 @@ int mgmtRetrieveMetrics(SShowObj *pShow, char *data, int rows, SConnObj *pConn)
}
int32_t
mgmtFindTagCol
(
STabObj
*
pMetric
,
const
char
*
tagName
)
{
if
(
!
mgmtIs
Metric
(
pMetric
))
return
-
1
;
if
(
!
mgmtIs
SuperTable
(
pMetric
))
return
-
1
;
SSchema
*
schema
=
NULL
;
...
...
@@ -1474,7 +1474,7 @@ int32_t mgmtMeterModifyTagNameByCol(STabObj *pMetric, uint32_t col, const char *
uint32_t
len
=
strlen
(
nname
);
if
(
pMetric
==
NULL
||
(
!
mgmtIs
Metric
(
pMetric
))
||
col
>=
pMetric
->
numOfTags
||
len
>=
TSDB_COL_NAME_LEN
||
if
(
pMetric
==
NULL
||
(
!
mgmtIs
SuperTable
(
pMetric
))
||
col
>=
pMetric
->
numOfTags
||
len
>=
TSDB_COL_NAME_LEN
||
mgmtFindTagCol
(
pMetric
,
nname
)
>=
0
)
return
TSDB_CODE_APP_ERROR
;
...
...
@@ -1503,7 +1503,7 @@ int32_t mgmtMeterModifyTagNameByCol(STabObj *pMetric, uint32_t col, const char *
}
int32_t
mgmtMeterModifyTagNameByName
(
STabObj
*
pMetric
,
const
char
*
oname
,
const
char
*
nname
)
{
if
(
pMetric
==
NULL
||
(
!
mgmtIs
Metric
(
pMetric
)))
return
TSDB_CODE_APP_ERROR
;
if
(
pMetric
==
NULL
||
(
!
mgmtIs
SuperTable
(
pMetric
)))
return
TSDB_CODE_APP_ERROR
;
int
index
=
mgmtFindTagCol
(
pMetric
,
oname
);
if
(
index
<
0
)
{
...
...
@@ -1517,9 +1517,9 @@ int32_t mgmtMeterModifyTagNameByName(STabObj *pMetric, const char *oname, const
int32_t
mgmtMeterModifyTagValueByCol
(
STabObj
*
pMeter
,
int
col
,
const
char
*
nContent
)
{
int
rowSize
=
0
;
if
(
pMeter
==
NULL
||
nContent
==
NULL
||
(
!
mgmt
MeterCreateFromMetric
(
pMeter
)))
return
TSDB_CODE_APP_ERROR
;
if
(
pMeter
==
NULL
||
nContent
==
NULL
||
(
!
mgmt
TableCreateFromSuperTable
(
pMeter
)))
return
TSDB_CODE_APP_ERROR
;
STabObj
*
pMetric
=
mgmtGet
Meter
(
pMeter
->
pTagData
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pMeter
->
pTagData
);
assert
(
pMetric
!=
NULL
);
if
(
col
<
0
||
col
>
pMetric
->
numOfTags
)
return
TSDB_CODE_APP_ERROR
;
...
...
@@ -1561,10 +1561,10 @@ int32_t mgmtMeterModifyTagValueByCol(STabObj *pMeter, int col, const char *nCont
}
int32_t
mgmtMeterModifyTagValueByName
(
STabObj
*
pMeter
,
char
*
tagName
,
char
*
nContent
)
{
if
(
pMeter
==
NULL
||
tagName
==
NULL
||
nContent
==
NULL
||
(
!
mgmt
MeterCreateFromMetric
(
pMeter
)))
if
(
pMeter
==
NULL
||
tagName
==
NULL
||
nContent
==
NULL
||
(
!
mgmt
TableCreateFromSuperTable
(
pMeter
)))
return
TSDB_CODE_INVALID_MSG_TYPE
;
STabObj
*
pMetric
=
mgmtGet
Meter
(
pMeter
->
pTagData
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pMeter
->
pTagData
);
if
(
pMetric
==
NULL
)
return
TSDB_CODE_APP_ERROR
;
int
col
=
mgmtFindTagCol
(
pMetric
,
tagName
);
...
...
@@ -1574,7 +1574,7 @@ int32_t mgmtMeterModifyTagValueByName(STabObj *pMeter, char *tagName, char *nCon
}
int32_t
mgmtMeterAddTags
(
STabObj
*
pMetric
,
SSchema
schema
[],
int
ntags
)
{
if
(
pMetric
==
NULL
||
(
!
mgmtIs
Metric
(
pMetric
)))
return
TSDB_CODE_INVALID_TABLE
;
if
(
pMetric
==
NULL
||
(
!
mgmtIs
SuperTable
(
pMetric
)))
return
TSDB_CODE_INVALID_TABLE
;
if
(
pMetric
->
numOfTags
+
ntags
>
TSDB_MAX_TAGS
)
return
TSDB_CODE_APP_ERROR
;
...
...
@@ -1615,7 +1615,7 @@ int32_t mgmtMeterAddTags(STabObj *pMetric, SSchema schema[], int ntags) {
}
int32_t
mgmtMeterDropTagByCol
(
STabObj
*
pMetric
,
int
col
)
{
if
(
pMetric
==
NULL
||
(
!
mgmtIs
Metric
(
pMetric
))
||
col
<=
0
||
col
>=
pMetric
->
numOfTags
)
return
TSDB_CODE_APP_ERROR
;
if
(
pMetric
==
NULL
||
(
!
mgmtIs
SuperTable
(
pMetric
))
||
col
<=
0
||
col
>=
pMetric
->
numOfTags
)
return
TSDB_CODE_APP_ERROR
;
// Pack message to do batch update
uint32_t
size
=
sizeof
(
SMeterBatchUpdateMsg
)
+
sizeof
(
SchemaUnit
);
...
...
@@ -1643,7 +1643,7 @@ int32_t mgmtMeterDropTagByCol(STabObj *pMetric, int col) {
}
int32_t
mgmtMeterDropTagByName
(
STabObj
*
pMetric
,
char
*
name
)
{
if
(
pMetric
==
NULL
||
(
!
mgmtIs
Metric
(
pMetric
)))
{
if
(
pMetric
==
NULL
||
(
!
mgmtIs
SuperTable
(
pMetric
)))
{
mTrace
(
"Failed to drop tag name: %s from table: %s"
,
name
,
pMetric
->
meterId
);
return
TSDB_CODE_INVALID_TABLE
;
}
...
...
@@ -1657,7 +1657,7 @@ int32_t mgmtFindColumnIndex(STabObj *pMeter, const char *colName) {
STabObj
*
pMetric
=
NULL
;
SSchema
*
schema
=
NULL
;
if
(
pMeter
->
meterType
==
TSDB_
METER_OTABLE
||
pMeter
->
meterType
==
TSDB_METER_METRIC
)
{
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_NORMAL_TABLE
||
pMeter
->
meterType
==
TSDB_TABLE_TYPE_SUPER_TABLE
)
{
schema
=
(
SSchema
*
)
pMeter
->
schema
;
for
(
int32_t
i
=
0
;
i
<
pMeter
->
numOfColumns
;
i
++
)
{
if
(
strcasecmp
(
schema
[
i
].
name
,
colName
)
==
0
)
{
...
...
@@ -1665,8 +1665,8 @@ int32_t mgmtFindColumnIndex(STabObj *pMeter, const char *colName) {
}
}
}
else
if
(
pMeter
->
meterType
==
TSDB_
METER_M
TABLE
)
{
pMetric
=
mgmtGet
Meter
(
pMeter
->
pTagData
);
}
else
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_CREATE_FROM_S
TABLE
)
{
pMetric
=
mgmtGet
Table
(
pMeter
->
pTagData
);
if
(
pMetric
==
NULL
)
{
mError
(
"MTable not belongs to any metric, meter: %s"
,
pMeter
->
meterId
);
return
-
1
;
...
...
@@ -1686,7 +1686,7 @@ int32_t mgmtMeterAddColumn(STabObj *pMeter, SSchema schema[], int ncols) {
SAcctObj
*
pAcct
=
NULL
;
SDbObj
*
pDb
=
NULL
;
if
(
pMeter
==
NULL
||
pMeter
->
meterType
==
TSDB_
METER_MTABLE
||
pMeter
->
meterType
==
TSDB_METER_S
TABLE
||
ncols
<=
0
)
if
(
pMeter
==
NULL
||
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_CREATE_FROM_STABLE
||
pMeter
->
meterType
==
TSDB_TABLE_TYPE_STREAM_
TABLE
||
ncols
<=
0
)
return
TSDB_CODE_APP_ERROR
;
// ASSUMPTION: no two tags are the same
...
...
@@ -1707,9 +1707,9 @@ int32_t mgmtMeterAddColumn(STabObj *pMeter, SSchema schema[], int ncols) {
pMeter
->
schema
=
realloc
(
pMeter
->
schema
,
pMeter
->
schemaSize
+
sizeof
(
SSchema
)
*
ncols
);
if
(
pMeter
->
meterType
==
TSDB_
METER_O
TABLE
)
{
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_NORMAL_
TABLE
)
{
memcpy
(
pMeter
->
schema
+
pMeter
->
schemaSize
,
schema
,
sizeof
(
SSchema
)
*
ncols
);
}
else
if
(
pMeter
->
meterType
==
TSDB_
METER_METRIC
)
{
}
else
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_SUPER_TABLE
)
{
memmove
(
pMeter
->
schema
+
sizeof
(
SSchema
)
*
(
pMeter
->
numOfColumns
+
ncols
),
pMeter
->
schema
+
sizeof
(
SSchema
)
*
pMeter
->
numOfColumns
,
sizeof
(
SSchema
)
*
pMeter
->
numOfTags
);
memcpy
(
pMeter
->
schema
+
sizeof
(
SSchema
)
*
pMeter
->
numOfColumns
,
schema
,
sizeof
(
SSchema
)
*
ncols
);
...
...
@@ -1721,13 +1721,13 @@ int32_t mgmtMeterAddColumn(STabObj *pMeter, SSchema schema[], int ncols) {
pMeter
->
schemaSize
+=
sizeof
(
SSchema
)
*
ncols
;
pMeter
->
numOfColumns
+=
ncols
;
pMeter
->
sversion
++
;
if
(
mgmtIsNormal
Meter
(
pMeter
))
if
(
mgmtIsNormal
Table
(
pMeter
))
pAcct
->
acctInfo
.
numOfTimeSeries
+=
ncols
;
else
pAcct
->
acctInfo
.
numOfTimeSeries
+=
(
ncols
*
pMeter
->
numOfMeters
);
sdbUpdateRow
(
meterSdb
,
pMeter
,
0
,
1
);
if
(
pMeter
->
meterType
==
TSDB_
METER_METRIC
)
{
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_SUPER_TABLE
)
{
for
(
STabObj
*
pObj
=
pMeter
->
pHead
;
pObj
!=
NULL
;
pObj
=
pObj
->
next
)
{
pObj
->
numOfColumns
++
;
pObj
->
nextColId
=
pMeter
->
nextColId
;
...
...
@@ -1743,7 +1743,7 @@ int32_t mgmtMeterDropColumnByName(STabObj *pMeter, const char *name) {
SAcctObj
*
pAcct
=
NULL
;
SDbObj
*
pDb
=
NULL
;
if
(
pMeter
==
NULL
||
pMeter
->
meterType
==
TSDB_
METER_MTABLE
||
pMeter
->
meterType
==
TSDB_METER_S
TABLE
)
if
(
pMeter
==
NULL
||
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_CREATE_FROM_STABLE
||
pMeter
->
meterType
==
TSDB_TABLE_TYPE_STREAM_
TABLE
)
return
TSDB_CODE_APP_ERROR
;
int32_t
index
=
mgmtFindColumnIndex
(
pMeter
,
name
);
...
...
@@ -1761,16 +1761,16 @@ int32_t mgmtMeterDropColumnByName(STabObj *pMeter, const char *name) {
return
TSDB_CODE_APP_ERROR
;
}
if
(
pMeter
->
meterType
==
TSDB_
METER_O
TABLE
)
{
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_NORMAL_
TABLE
)
{
memmove
(
pMeter
->
schema
+
sizeof
(
SSchema
)
*
index
,
pMeter
->
schema
+
sizeof
(
SSchema
)
*
(
index
+
1
),
sizeof
(
SSchema
)
*
(
pMeter
->
numOfColumns
-
index
-
1
));
}
else
if
(
pMeter
->
meterType
==
TSDB_
METER_METRIC
)
{
}
else
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_SUPER_TABLE
)
{
memmove
(
pMeter
->
schema
+
sizeof
(
SSchema
)
*
index
,
pMeter
->
schema
+
sizeof
(
SSchema
)
*
(
index
+
1
),
sizeof
(
SSchema
)
*
(
pMeter
->
numOfColumns
+
pMeter
->
numOfTags
-
index
-
1
));
}
pMeter
->
schemaSize
-=
sizeof
(
SSchema
);
pMeter
->
numOfColumns
--
;
if
(
mgmtIsNormal
Meter
(
pMeter
))
if
(
mgmtIsNormal
Table
(
pMeter
))
pAcct
->
acctInfo
.
numOfTimeSeries
--
;
else
pAcct
->
acctInfo
.
numOfTimeSeries
-=
(
pMeter
->
numOfMeters
);
...
...
@@ -1779,7 +1779,7 @@ int32_t mgmtMeterDropColumnByName(STabObj *pMeter, const char *name) {
pMeter
->
sversion
++
;
sdbUpdateRow
(
meterSdb
,
pMeter
,
0
,
1
);
if
(
pMeter
->
meterType
==
TSDB_
METER_METRIC
)
{
if
(
pMeter
->
meterType
==
TSDB_
TABLE_TYPE_SUPER_TABLE
)
{
for
(
STabObj
*
pObj
=
pMeter
->
pHead
;
pObj
!=
NULL
;
pObj
=
pObj
->
next
)
{
pObj
->
numOfColumns
--
;
pObj
->
sversion
=
pMeter
->
sversion
;
...
...
src/mnode/src/mgmtShell.c
浏览文件 @
d97941aa
...
...
@@ -157,7 +157,7 @@ bool mgmtCheckMeterMetaMsgType(char *pMsg) {
SMeterInfoMsg
*
pInfo
=
(
SMeterInfoMsg
*
)
pMsg
;
int16_t
autoCreate
=
htons
(
pInfo
->
createFlag
);
STabObj
*
pMeterObj
=
mgmtGet
Meter
(
pInfo
->
meterId
);
STabObj
*
pMeterObj
=
mgmtGet
Table
(
pInfo
->
meterId
);
// If table does not exists and autoCreate flag is set, we add the handler into another task queue, namely tranQueue
bool
addIntoTranQueue
=
(
pMeterObj
==
NULL
&&
autoCreate
==
1
);
...
...
@@ -199,7 +199,7 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
goto
_exit_code
;
}
pMeterObj
=
mgmtGet
Meter
(
pInfo
->
meterId
);
pMeterObj
=
mgmtGet
Table
(
pInfo
->
meterId
);
// on demand create table from super table if meter does not exists
if
(
pMeterObj
==
NULL
&&
pInfo
->
createFlag
==
1
)
{
...
...
@@ -243,7 +243,7 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
goto
_exit_code
;
}
pMeterObj
=
mgmtGet
Meter
(
pInfo
->
meterId
);
pMeterObj
=
mgmtGet
Table
(
pInfo
->
meterId
);
}
if
((
pStart
=
mgmtAllocMsg
(
pConn
,
size
,
&
pMsg
,
&
pRsp
))
==
NULL
)
{
...
...
@@ -279,10 +279,10 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
pMsg
+=
sizeof
(
SMeterMeta
);
pSchema
=
(
SSchema
*
)
pMsg
;
// schema locates at the end of SMeterMeta struct
if
(
mgmt
MeterCreateFromMetric
(
pMeterObj
))
{
if
(
mgmt
TableCreateFromSuperTable
(
pMeterObj
))
{
assert
(
pMeterObj
->
numOfTags
==
0
);
STabObj
*
pMetric
=
mgmtGet
Meter
(
pMeterObj
->
pTagData
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pMeterObj
->
pTagData
);
uint32_t
numOfTotalCols
=
(
uint32_t
)
pMetric
->
numOfTags
+
pMetric
->
numOfColumns
;
pMeta
->
numOfTags
=
pMetric
->
numOfTags
;
// update the numOfTags info
...
...
@@ -302,7 +302,7 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
pMsg
+=
numOfTotalCols
*
sizeof
(
SSchema
);
}
if
(
mgmtIsNormal
Meter
(
pMeterObj
))
{
if
(
mgmtIsNormal
Table
(
pMeterObj
))
{
pVgroup
=
mgmtGetVgroup
(
pMeterObj
->
gid
.
vgId
);
if
(
pVgroup
==
NULL
)
{
pRsp
->
code
=
TSDB_CODE_INVALID_TABLE
;
...
...
@@ -397,7 +397,7 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
}
// get meter schema, and fill into resp payload
pMeterObj
=
mgmtGet
Meter
(
tblName
);
pMeterObj
=
mgmtGet
Table
(
tblName
);
pDbObj
=
mgmtGetDbByMeterId
(
tblName
);
if
(
pMeterObj
==
NULL
||
(
pDbObj
==
NULL
))
{
...
...
@@ -419,10 +419,10 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
pCurMeter
+=
sizeof
(
SMultiMeterMeta
);
pSchema
=
(
SSchema
*
)
pCurMeter
;
// schema locates at the end of SMeterMeta struct
if
(
mgmt
MeterCreateFromMetric
(
pMeterObj
))
{
if
(
mgmt
TableCreateFromSuperTable
(
pMeterObj
))
{
assert
(
pMeterObj
->
numOfTags
==
0
);
STabObj
*
pMetric
=
mgmtGet
Meter
(
pMeterObj
->
pTagData
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pMeterObj
->
pTagData
);
uint32_t
numOfTotalCols
=
(
uint32_t
)
pMetric
->
numOfTags
+
pMetric
->
numOfColumns
;
pMeta
->
meta
.
numOfTags
=
pMetric
->
numOfTags
;
// update the numOfTags info
...
...
@@ -442,7 +442,7 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
pCurMeter
+=
numOfTotalCols
*
sizeof
(
SSchema
);
}
if
(
mgmtIsNormal
Meter
(
pMeterObj
))
{
if
(
mgmtIsNormal
Table
(
pMeterObj
))
{
pVgroup
=
mgmtGetVgroup
(
pMeterObj
->
gid
.
vgId
);
if
(
pVgroup
==
NULL
)
{
pRsp
->
code
=
TSDB_CODE_INVALID_TABLE
;
...
...
@@ -509,7 +509,7 @@ int mgmtProcessMetricMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
}
SMetricMetaElemMsg
*
pElem
=
(
SMetricMetaElemMsg
*
)(((
char
*
)
pMetricMetaMsg
)
+
pMetricMetaMsg
->
metaElem
[
0
]);
pMetric
=
mgmtGet
Meter
(
pElem
->
meterId
);
pMetric
=
mgmtGet
Table
(
pElem
->
meterId
);
SDbObj
*
pDb
=
NULL
;
if
(
pConn
->
pDb
!=
NULL
)
pDb
=
mgmtGetDb
(
pConn
->
pDb
->
name
);
...
...
@@ -880,7 +880,7 @@ int mgmtProcessUseDbMsg(char *pMsg, int msgLen, SConnObj *pConn) {
}
int
(
*
mgmtGetMetaFp
[])(
SMeterMeta
*
pMeta
,
SShowObj
*
pShow
,
SConnObj
*
pConn
)
=
{
mgmtGetAcctMeta
,
mgmtGetUserMeta
,
mgmtGetDbMeta
,
mgmtGet
Meter
Meta
,
mgmtGetDnodeMeta
,
mgmtGetAcctMeta
,
mgmtGetUserMeta
,
mgmtGetDbMeta
,
mgmtGet
Table
Meta
,
mgmtGetDnodeMeta
,
mgmtGetMnodeMeta
,
mgmtGetVgroupMeta
,
mgmtGetMetricMeta
,
mgmtGetModuleMeta
,
mgmtGetQueryMeta
,
mgmtGetStreamMeta
,
mgmtGetConfigMeta
,
mgmtGetConnsMeta
,
mgmtGetScoresMeta
,
grantGetGrantsMeta
,
mgmtGetVnodeMeta
,
...
...
@@ -1076,7 +1076,7 @@ int mgmtProcessCreateTableMsg(char *pMsg, int msgLen, SConnObj *pConn) {
}
else
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
if
(
code
==
TSDB_CODE_TABLE_ALREADY_EXIST
)
{
// table already created when the second attempt to create table
STabObj
*
pMeter
=
mgmtGet
Meter
(
pCreate
->
meterId
);
STabObj
*
pMeter
=
mgmtGet
Table
(
pCreate
->
meterId
);
assert
(
pMeter
!=
NULL
);
mWarn
(
"table:%s, table already created, failed to create table, ts:%"
PRId64
", code:%d"
,
pCreate
->
meterId
,
...
...
src/mnode/src/mgmtSupertableQuery.c
浏览文件 @
d97941aa
...
...
@@ -84,8 +84,8 @@ static int32_t tabObjResultComparator(const void* p1, const void* p2, void* para
schema
.
type
=
TSDB_DATA_TYPE_BINARY
;
schema
.
bytes
=
TSDB_METER_ID_LEN
;
}
else
{
f1
=
mgmt
Meter
GetTag
(
pNode1
,
colIdx
,
NULL
);
f2
=
mgmt
Meter
GetTag
(
pNode2
,
colIdx
,
&
schema
);
f1
=
mgmt
Table
GetTag
(
pNode1
,
colIdx
,
NULL
);
f2
=
mgmt
Table
GetTag
(
pNode2
,
colIdx
,
&
schema
);
assert
(
schema
.
type
==
pOrderDesc
->
pTagSchema
->
pSchema
[
colIdx
].
type
);
}
...
...
@@ -134,7 +134,7 @@ void mgmtReorganizeMetersInMetricMeta(SMetricMetaMsg* pMetricMetaMsg, int32_t ta
SMetricMetaElemMsg
*
pElem
=
(
SMetricMetaElemMsg
*
)((
char
*
)
pMetricMetaMsg
+
pMetricMetaMsg
->
metaElem
[
tableIndex
]);
STabObj
*
pMetric
=
mgmtGet
Meter
(
pElem
->
meterId
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pElem
->
meterId
);
SSchema
*
pTagSchema
=
(
SSchema
*
)(
pMetric
->
schema
+
pMetric
->
numOfColumns
*
sizeof
(
SSchema
));
/*
...
...
@@ -181,7 +181,7 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p
pRes
->
num
=
0
;
for
(
pToken
=
strsep
(
&
str
,
sep
);
pToken
!=
NULL
;
pToken
=
strsep
(
&
str
,
sep
))
{
STabObj
*
pMeterObj
=
mgmtGet
Meter
(
pToken
);
STabObj
*
pMeterObj
=
mgmtGet
Table
(
pToken
);
if
(
pMeterObj
==
NULL
)
{
mWarn
(
"metric:%s error in metric query expression, invalid meter id:%s"
,
pMetric
->
meterId
,
pToken
);
continue
;
...
...
@@ -193,7 +193,7 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p
}
/* not a table created from metric, ignore */
if
(
pMeterObj
->
meterType
!=
TSDB_
METER_M
TABLE
)
{
if
(
pMeterObj
->
meterType
!=
TSDB_
TABLE_TYPE_CREATE_FROM_S
TABLE
)
{
continue
;
}
...
...
@@ -201,7 +201,7 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p
* queried meter not belongs to this metric, ignore, metric does not have
* uid, so compare according to meterid
*/
STabObj
*
parentMetric
=
mgmtGet
Meter
(
pMeterObj
->
pTagData
);
STabObj
*
parentMetric
=
mgmtGet
Table
(
pMeterObj
->
pTagData
);
if
(
strncasecmp
(
parentMetric
->
meterId
,
pMetric
->
meterId
,
TSDB_METER_ID_LEN
)
!=
0
||
(
parentMetric
->
uid
!=
pMetric
->
uid
))
{
continue
;
...
...
@@ -256,7 +256,7 @@ UNUSED_FUNC static bool mgmtJoinFilterCallback(tSkipListNode* pNode, void* param
SJoinSupporter
*
pSupporter
=
(
SJoinSupporter
*
)
param
;
SSchema
s
=
{
0
};
char
*
v
=
mgmt
Meter
GetTag
((
STabObj
*
)
pNode
->
pData
,
pSupporter
->
colIndex
,
&
s
);
char
*
v
=
mgmt
Table
GetTag
((
STabObj
*
)
pNode
->
pData
,
pSupporter
->
colIndex
,
&
s
);
for
(
int32_t
i
=
0
;
i
<
pSupporter
->
size
;
++
i
)
{
int32_t
ret
=
doCompare
(
v
,
pSupporter
->
val
[
i
],
pSupporter
->
type
,
s
.
bytes
);
...
...
@@ -288,7 +288,7 @@ static void orderResult(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes, i
tOrderDescriptor
*
descriptor
=
(
tOrderDescriptor
*
)
calloc
(
1
,
sizeof
(
tOrderDescriptor
)
+
sizeof
(
int32_t
)
*
1
);
// only one column for join
STabObj
*
pMetric
=
mgmtGet
Meter
(
pElem
->
meterId
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pElem
->
meterId
);
SSchema
*
pTagSchema
=
(
SSchema
*
)(
pMetric
->
schema
+
pMetric
->
numOfColumns
*
sizeof
(
SSchema
));
descriptor
->
pTagSchema
=
tCreateTagSchema
(
pTagSchema
,
pMetric
->
numOfTags
);
...
...
@@ -311,8 +311,8 @@ static int32_t mgmtCheckForDuplicateTagValue(tQueryResultset* pRes, int32_t inde
STabObj
*
pObj1
=
pRes
[
index
].
pRes
[
k
-
1
];
STabObj
*
pObj2
=
pRes
[
index
].
pRes
[
k
];
char
*
val1
=
mgmt
Meter
GetTag
(
pObj1
,
tagCol
,
&
s
);
char
*
val2
=
mgmt
Meter
GetTag
(
pObj2
,
tagCol
,
NULL
);
char
*
val1
=
mgmt
Table
GetTag
(
pObj1
,
tagCol
,
&
s
);
char
*
val2
=
mgmt
Table
GetTag
(
pObj2
,
tagCol
,
NULL
);
if
(
doCompare
(
val1
,
val2
,
s
.
type
,
s
.
bytes
)
==
0
)
{
return
TSDB_CODE_DUPLICATE_TAGS
;
...
...
@@ -354,8 +354,8 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) {
strcpy
(
right
,
cond
+
TSDB_METER_ID_LEN
+
sizeof
(
int16_t
));
int16_t
rightTagColIndex
=
*
(
int16_t
*
)(
cond
+
TSDB_METER_ID_LEN
*
2
+
sizeof
(
int16_t
));
STabObj
*
pLeftMetric
=
mgmtGet
Meter
(
left
);
STabObj
*
pRightMetric
=
mgmtGet
Meter
(
right
);
STabObj
*
pLeftMetric
=
mgmtGet
Table
(
left
);
STabObj
*
pRightMetric
=
mgmtGet
Table
(
right
);
// decide the pRes belongs to
int32_t
leftIndex
=
0
;
...
...
@@ -363,7 +363,7 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) {
for
(
int32_t
i
=
0
;
i
<
pMetricMetaMsg
->
numOfMeters
;
++
i
)
{
STabObj
*
pObj
=
(
STabObj
*
)
pRes
[
i
].
pRes
[
0
];
STabObj
*
pMetric1
=
mgmtGet
Meter
(
pObj
->
pTagData
);
STabObj
*
pMetric1
=
mgmtGet
Table
(
pObj
->
pTagData
);
if
(
pMetric1
==
pLeftMetric
)
{
leftIndex
=
i
;
}
else
if
(
pMetric1
==
pRightMetric
)
{
...
...
@@ -391,8 +391,8 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) {
STabObj
*
pLeftObj
=
pRes
[
leftIndex
].
pRes
[
i
];
STabObj
*
pRightObj
=
pRes
[
rightIndex
].
pRes
[
j
];
char
*
v1
=
mgmt
Meter
GetTag
(
pLeftObj
,
leftTagColIndex
,
&
s
);
char
*
v2
=
mgmt
Meter
GetTag
(
pRightObj
,
rightTagColIndex
,
NULL
);
char
*
v1
=
mgmt
Table
GetTag
(
pLeftObj
,
leftTagColIndex
,
&
s
);
char
*
v2
=
mgmt
Table
GetTag
(
pRightObj
,
rightTagColIndex
,
NULL
);
int32_t
ret
=
doCompare
(
v1
,
v2
,
s
.
type
,
s
.
bytes
);
if
(
ret
==
0
)
{
// qualified
...
...
@@ -729,7 +729,7 @@ static int32_t mgmtFilterMeterByIndex(STabObj* pMetric, tQueryResultset* pRes, c
int
mgmtRetrieveMetersFromMetric
(
SMetricMetaMsg
*
pMsg
,
int32_t
tableIndex
,
tQueryResultset
*
pRes
)
{
SMetricMetaElemMsg
*
pElem
=
(
SMetricMetaElemMsg
*
)((
char
*
)
pMsg
+
pMsg
->
metaElem
[
tableIndex
]);
STabObj
*
pMetric
=
mgmtGet
Meter
(
pElem
->
meterId
);
STabObj
*
pMetric
=
mgmtGet
Table
(
pElem
->
meterId
);
char
*
pCond
=
NULL
;
char
*
tmpTableNameCond
=
NULL
;
...
...
src/mnode/src/mgmtUtil.c
浏览文件 @
d97941aa
...
...
@@ -21,32 +21,18 @@
#include "textbuffer.h"
#include "tschemautil.h"
#include "tsqlfunction.h"
#include "vnodeTagMgmt.h"
extern
int
cksumsize
;
uint64_t
mgmtGetCheckSum
(
FILE
*
fp
,
int
offset
)
{
uint64_t
checksum
=
0
;
uint64_t
data
;
int
bytes
;
while
(
1
)
{
data
=
0
;
bytes
=
fread
(
&
data
,
sizeof
(
data
),
1
,
fp
);
if
(
bytes
!=
sizeof
(
data
))
break
;
checksum
+=
data
;
}
return
checksum
;
bool
mgmtTableCreateFromSuperTable
(
STabObj
*
pTableObj
)
{
return
pTableObj
->
meterType
==
TSDB_TABLE_TYPE_CREATE_FROM_STABLE
;
}
bool
mgmt
MeterCreateFromMetric
(
STabObj
*
pMeterObj
)
{
return
pMeterObj
->
meterType
==
TSDB_METER_MTABLE
;
}
bool
mgmtIsMetric
(
STabObj
*
pMeterObj
)
{
return
pMeterObj
->
meterType
==
TSDB_METER_METRIC
;
}
bool
mgmt
IsSuperTable
(
STabObj
*
pTableObj
)
{
return
pTableObj
->
meterType
==
TSDB_TABLE_TYPE_SUPER_TABLE
;
}
bool
mgmtIsNormalMeter
(
STabObj
*
pMeterObj
)
{
return
!
mgmtIsMetric
(
pMeterObj
);
}
bool
mgmtIsNormalTable
(
STabObj
*
pTableObj
)
{
return
!
mgmtIsSuperTable
(
pTableObj
);
}
/**
* TODO: the tag offset value should be kept in memory to avoid dynamically calculating the value
...
...
@@ -56,30 +42,30 @@ bool mgmtIsNormalMeter(STabObj* pMeterObj) { return !mgmtIsMetric(pMeterObj); }
* @param pTagColSchema
* @return
*/
char
*
mgmt
MeterGetTag
(
STabObj
*
pMeter
,
int32_t
col
,
SSchema
*
pTagColSchema
)
{
if
(
!
mgmt
MeterCreateFromMetric
(
pMeter
))
{
char
*
mgmt
TableGetTag
(
STabObj
*
pTable
,
int32_t
col
,
SSchema
*
pTagColSchema
)
{
if
(
!
mgmt
TableCreateFromSuperTable
(
pTable
))
{
return
NULL
;
}
STabObj
*
p
Metric
=
mgmtGetMeter
(
pMeter
->
pTagData
);
int32_t
offset
=
mgmtGetTagsLength
(
p
Metric
,
col
)
+
TSDB_METER_ID_LEN
;
STabObj
*
p
SuperTable
=
mgmtGetTable
(
pTable
->
pTagData
);
int32_t
offset
=
mgmtGetTagsLength
(
p
SuperTable
,
col
)
+
TSDB_METER_ID_LEN
;
assert
(
offset
>
0
);
if
(
pTagColSchema
!=
NULL
)
{
*
pTagColSchema
=
((
SSchema
*
)
p
Metric
->
schema
)[
pMetric
->
numOfColumns
+
col
];
*
pTagColSchema
=
((
SSchema
*
)
p
SuperTable
->
schema
)[
pSuperTable
->
numOfColumns
+
col
];
}
return
(
p
Meter
->
pTagData
+
offset
);
return
(
p
Table
->
pTagData
+
offset
);
}
int32_t
mgmtGetTagsLength
(
STabObj
*
p
Metric
,
int32_t
col
)
{
// length before column col
assert
(
mgmtIs
Metric
(
pMetric
)
&&
col
>=
0
);
int32_t
mgmtGetTagsLength
(
STabObj
*
p
SuperTable
,
int32_t
col
)
{
// length before column col
assert
(
mgmtIs
SuperTable
(
pSuperTable
)
&&
col
>=
0
);
int32_t
len
=
0
;
int32_t
tagColumnIndexOffset
=
p
Metric
->
numOfColumns
;
int32_t
tagColumnIndexOffset
=
p
SuperTable
->
numOfColumns
;
for
(
int32_t
i
=
0
;
i
<
p
Metric
->
numOfTags
&&
i
<
col
;
++
i
)
{
len
+=
((
SSchema
*
)
p
Metric
->
schema
)[
tagColumnIndexOffset
+
i
].
bytes
;
for
(
int32_t
i
=
0
;
i
<
p
SuperTable
->
numOfTags
&&
i
<
col
;
++
i
)
{
len
+=
((
SSchema
*
)
p
SuperTable
->
schema
)[
tagColumnIndexOffset
+
i
].
bytes
;
}
return
len
;
...
...
src/mnode/src/mgmtVgroup.c
浏览文件 @
d97941aa
...
...
@@ -241,7 +241,7 @@ int mgmtGetVgroupMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn) {
SVgObj
*
pVgroup
=
NULL
;
STabObj
*
pMeter
=
NULL
;
if
(
pShow
->
payloadLen
>
0
)
{
pMeter
=
mgmtGet
Meter
(
pShow
->
payload
);
pMeter
=
mgmtGet
Table
(
pShow
->
payload
);
if
(
NULL
==
pMeter
)
{
return
TSDB_CODE_INVALID_METER_ID
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录