Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
e11147dd
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
e11147dd
编写于
4月 27, 2022
作者:
C
Cary Xu
提交者:
GitHub
4月 27, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #11964 from taosdata/feature/TD-14481-3.0
feat: rollup sma refactor
上级
6e591748
05a9af5c
变更
9
显示空白变更内容
内联
并排
Showing
9 changed file
with
101 addition
and
59 deletion
+101
-59
include/common/tmsg.h
include/common/tmsg.h
+2
-2
include/common/trow.h
include/common/trow.h
+8
-0
source/common/src/tmsg.c
source/common/src/tmsg.c
+43
-6
source/common/src/trow.c
source/common/src/trow.c
+1
-1
source/dnode/vnode/src/inc/tsdbSma.h
source/dnode/vnode/src/inc/tsdbSma.h
+6
-14
source/dnode/vnode/src/inc/vnodeInt.h
source/dnode/vnode/src/inc/vnodeInt.h
+10
-1
source/dnode/vnode/src/tsdb/tsdbRead.c
source/dnode/vnode/src/tsdb/tsdbRead.c
+2
-4
source/dnode/vnode/src/tsdb/tsdbSma.c
source/dnode/vnode/src/tsdb/tsdbSma.c
+20
-27
source/dnode/vnode/src/vnd/vnodeSvr.c
source/dnode/vnode/src/vnd/vnodeSvr.c
+9
-4
未找到文件。
include/common/tmsg.h
浏览文件 @
e11147dd
...
...
@@ -1521,8 +1521,8 @@ typedef struct {
char
*
qmsg2
;
// pAst2:qmsg2:SRetention2 => trigger aggr task2
}
SRSmaParam
;
int
tEncodeSRSmaParam
(
SCoder
*
pCoder
,
const
SRSmaParam
*
pRSmaParam
);
int
tDecodeSRSmaParam
(
SCoder
*
pCoder
,
SRSmaParam
*
pRSmaParam
);
int
32_t
tEncodeSRSmaParam
(
SCoder
*
pCoder
,
const
SRSmaParam
*
pRSmaParam
);
int
32_t
tDecodeSRSmaParam
(
SCoder
*
pCoder
,
SRSmaParam
*
pRSmaParam
);
typedef
struct
SVCreateStbReq
{
const
char
*
name
;
...
...
include/common/trow.h
浏览文件 @
e11147dd
...
...
@@ -214,6 +214,14 @@ STSRow *tdRowDup(STSRow *row);
static
FORCE_INLINE
SKvRowIdx
*
tdKvRowColIdxAt
(
STSRow
*
pRow
,
col_id_t
idx
)
{
return
(
SKvRowIdx
*
)
TD_ROW_COL_IDX
(
pRow
)
+
idx
;
}
static
FORCE_INLINE
int16_t
tdKvRowColIdAt
(
STSRow
*
pRow
,
col_id_t
idx
)
{
ASSERT
(
idx
>=
0
);
if
(
idx
==
0
)
{
return
PRIMARYKEY_TIMESTAMP_COL_ID
;
}
return
((
SKvRowIdx
*
)
TD_ROW_COL_IDX
(
pRow
)
+
idx
-
1
)
->
colId
;
}
static
FORCE_INLINE
void
*
tdKVRowColVal
(
STSRow
*
pRow
,
SKvRowIdx
*
pIdx
)
{
return
POINTER_SHIFT
(
pRow
,
pIdx
->
offset
);
}
#define TD_ROW_OFFSET(p) ((p)->toffset); // During ParseInsert when without STSchema, how to get the offset for STpRow?
...
...
source/common/src/tmsg.c
浏览文件 @
e11147dd
...
...
@@ -3621,6 +3621,43 @@ void tFreeSCMCreateStreamReq(SCMCreateStreamReq *pReq) {
taosMemoryFreeClear
(
pReq
->
ast
);
}
int32_t
tEncodeSRSmaParam
(
SCoder
*
pCoder
,
const
SRSmaParam
*
pRSmaParam
)
{
if
(
tEncodeFloat
(
pCoder
,
pRSmaParam
->
xFilesFactor
)
<
0
)
return
-
1
;
if
(
tEncodeI32v
(
pCoder
,
pRSmaParam
->
delay
)
<
0
)
return
-
1
;
if
(
tEncodeI32v
(
pCoder
,
pRSmaParam
->
qmsg1Len
)
<
0
)
return
-
1
;
if
(
tEncodeI32v
(
pCoder
,
pRSmaParam
->
qmsg2Len
)
<
0
)
return
-
1
;
if
(
pRSmaParam
->
qmsg1Len
>
0
)
{
if
(
tEncodeBinary
(
pCoder
,
pRSmaParam
->
qmsg1
,
(
uint64_t
)
pRSmaParam
->
qmsg1Len
)
<
0
)
// qmsg1Len contains len of '\0'
return
-
1
;
}
if
(
pRSmaParam
->
qmsg2Len
>
0
)
{
if
(
tEncodeBinary
(
pCoder
,
pRSmaParam
->
qmsg2
,
(
uint64_t
)
pRSmaParam
->
qmsg2Len
)
<
0
)
// qmsg2Len contains len of '\0'
return
-
1
;
}
return
0
;
}
int32_t
tDecodeSRSmaParam
(
SCoder
*
pCoder
,
SRSmaParam
*
pRSmaParam
)
{
if
(
tDecodeFloat
(
pCoder
,
&
pRSmaParam
->
xFilesFactor
)
<
0
)
return
-
1
;
if
(
tDecodeI32v
(
pCoder
,
&
pRSmaParam
->
delay
)
<
0
)
return
-
1
;
if
(
tDecodeI32v
(
pCoder
,
&
pRSmaParam
->
qmsg1Len
)
<
0
)
return
-
1
;
if
(
tDecodeI32v
(
pCoder
,
&
pRSmaParam
->
qmsg2Len
)
<
0
)
return
-
1
;
if
(
pRSmaParam
->
qmsg1Len
>
0
)
{
uint64_t
len
;
if
(
tDecodeBinaryAlloc
(
pCoder
,
(
void
**
)
&
pRSmaParam
->
qmsg1
,
&
len
)
<
0
)
return
-
1
;
// qmsg1Len contains len of '\0'
}
else
{
pRSmaParam
->
qmsg1
=
NULL
;
}
if
(
pRSmaParam
->
qmsg2Len
>
0
)
{
uint64_t
len
;
if
(
tDecodeBinaryAlloc
(
pCoder
,
(
void
**
)
&
pRSmaParam
->
qmsg2
,
&
len
)
<
0
)
return
-
1
;
// qmsg2Len contains len of '\0'
}
else
{
pRSmaParam
->
qmsg2
=
NULL
;
}
return
0
;
}
int
tEncodeSVCreateStbReq
(
SCoder
*
pCoder
,
const
SVCreateStbReq
*
pReq
)
{
if
(
tStartEncode
(
pCoder
)
<
0
)
return
-
1
;
...
...
@@ -3629,9 +3666,9 @@ int tEncodeSVCreateStbReq(SCoder *pCoder, const SVCreateStbReq *pReq) {
if
(
tEncodeI8
(
pCoder
,
pReq
->
rollup
)
<
0
)
return
-
1
;
if
(
tEncodeSSchemaWrapper
(
pCoder
,
&
pReq
->
schema
)
<
0
)
return
-
1
;
if
(
tEncodeSSchemaWrapper
(
pCoder
,
&
pReq
->
schemaTag
)
<
0
)
return
-
1
;
//
if (pReq->rollup) {
// if (tEncodeSRSmaParam(pCoder,
pReq->pRSmaParam) < 0) return -1;
//
}
if
(
pReq
->
rollup
)
{
if
(
tEncodeSRSmaParam
(
pCoder
,
&
pReq
->
pRSmaParam
)
<
0
)
return
-
1
;
}
tEndEncode
(
pCoder
);
return
0
;
...
...
@@ -3645,9 +3682,9 @@ int tDecodeSVCreateStbReq(SCoder *pCoder, SVCreateStbReq *pReq) {
if
(
tDecodeI8
(
pCoder
,
&
pReq
->
rollup
)
<
0
)
return
-
1
;
if
(
tDecodeSSchemaWrapper
(
pCoder
,
&
pReq
->
schema
)
<
0
)
return
-
1
;
if
(
tDecodeSSchemaWrapper
(
pCoder
,
&
pReq
->
schemaTag
)
<
0
)
return
-
1
;
//
if (pReq->rollup) {
// if (tDecodeSRSmaParam(pCoder,
pReq->pRSmaParam) < 0) return -1;
//
}
if
(
pReq
->
rollup
)
{
if
(
tDecodeSRSmaParam
(
pCoder
,
&
pReq
->
pRSmaParam
)
<
0
)
return
-
1
;
}
tEndDecode
(
pCoder
);
return
0
;
...
...
source/common/src/trow.c
浏览文件 @
e11147dd
...
...
@@ -220,7 +220,7 @@ static uint8_t tdGetMergedBitmapByte(uint8_t byte) {
}
/**
* @brief Merge bitmap from 2 bits to 1 bit
s
, and the memory buffer should be guaranteed by the invoker.
* @brief Merge bitmap from 2 bits to 1 bit, and the memory buffer should be guaranteed by the invoker.
*
* @param srcBitmap
* @param nBits
...
...
source/dnode/vnode/src/inc/tsdbSma.h
浏览文件 @
e11147dd
...
...
@@ -22,13 +22,13 @@
extern
"C"
{
#endif
typedef
int32_t
(
*
__tb_ddl_fn_t
)(
void
*
ahandle
,
void
**
result
,
void
*
p1
,
void
*
p2
);
//
typedef int32_t (*__tb_ddl_fn_t)(void *ahandle, void **result, void *p1, void *p2);
struct
STbDdlH
{
void
*
ahandle
;
void
*
result
;
__tb_ddl_fn_t
fp
;
};
//
struct STbDdlH {
//
void *ahandle;
//
void *result;
//
__tb_ddl_fn_t fp;
//
};
static
FORCE_INLINE
int32_t
tsdbUidStoreInit
(
STbUidStore
**
pStore
)
{
ASSERT
(
*
pStore
==
NULL
);
...
...
@@ -40,14 +40,6 @@ static FORCE_INLINE int32_t tsdbUidStoreInit(STbUidStore **pStore) {
return
TSDB_CODE_SUCCESS
;
}
int32_t
tsdbUidStorePut
(
STbUidStore
*
pStore
,
tb_uid_t
suid
,
tb_uid_t
*
uid
);
void
tsdbUidStoreDestory
(
STbUidStore
*
pStore
);
void
*
tsdbUidStoreFree
(
STbUidStore
*
pStore
);
int32_t
tsdbRegisterRSma
(
STsdb
*
pTsdb
,
SMeta
*
pMeta
,
SVCreateTbReq
*
pReq
);
int32_t
tsdbFetchTbUidList
(
void
*
pTsdb
,
void
**
result
,
void
*
suid
,
void
*
uid
);
int32_t
tsdbUpdateTbUidList
(
STsdb
*
pTsdb
,
STbUidStore
*
pUidStore
);
int32_t
tsdbTriggerRSma
(
STsdb
*
pTsdb
,
SMeta
*
pMeta
,
void
*
pMsg
,
int32_t
inputType
);
#ifdef __cplusplus
}
...
...
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @
e11147dd
...
...
@@ -107,6 +107,15 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen);
int32_t
tqProcessStreamTrigger
(
STQ
*
pTq
,
void
*
data
,
int32_t
dataLen
,
int32_t
workerId
);
int32_t
tqProcessPollReq
(
STQ
*
pTq
,
SRpcMsg
*
pMsg
,
int32_t
workerId
);
// sma
int32_t
tsdbRegisterRSma
(
STsdb
*
pTsdb
,
SMeta
*
pMeta
,
SVCreateStbReq
*
pReq
);
int32_t
tsdbFetchTbUidList
(
STsdb
*
pTsdb
,
STbUidStore
**
ppStore
,
tb_uid_t
suid
,
tb_uid_t
uid
);
int32_t
tsdbUpdateTbUidList
(
STsdb
*
pTsdb
,
STbUidStore
*
pUidStore
);
void
tsdbUidStoreDestory
(
STbUidStore
*
pStore
);
void
*
tsdbUidStoreFree
(
STbUidStore
*
pStore
);
int32_t
tsdbTriggerRSma
(
STsdb
*
pTsdb
,
SMeta
*
pMeta
,
void
*
pMsg
,
int32_t
inputType
);
typedef
struct
{
int8_t
streamType
;
// sma or other
int8_t
dstType
;
...
...
@@ -162,7 +171,7 @@ struct STbUidStore {
#define TD_VID(PVNODE) (PVNODE)->config.vgId
typedef
struct
STbDdlH
STbDdlH
;
//
typedef struct STbDdlH STbDdlH;
// sma
void
smaHandleRes
(
void
*
pVnode
,
int64_t
smaId
,
const
SArray
*
data
);
...
...
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @
e11147dd
...
...
@@ -1519,8 +1519,7 @@ static void mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capacit
}
else
if
(
isRow1DataRow
)
{
colIdOfRow1
=
pSchema1
->
columns
[
j
].
colId
;
}
else
{
SKvRowIdx
*
pColIdx
=
tdKvRowColIdxAt
(
row1
,
j
);
colIdOfRow1
=
pColIdx
->
colId
;
colIdOfRow1
=
tdKvRowColIdAt
(
row1
,
j
);
}
int32_t
colIdOfRow2
;
...
...
@@ -1529,8 +1528,7 @@ static void mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capacit
}
else
if
(
isRow2DataRow
)
{
colIdOfRow2
=
pSchema2
->
columns
[
k
].
colId
;
}
else
{
SKvRowIdx
*
pColIdx
=
tdKvRowColIdxAt
(
row2
,
k
);
colIdOfRow2
=
pColIdx
->
colId
;
colIdOfRow2
=
tdKvRowColIdAt
(
row2
,
j
);
}
if
(
colIdOfRow1
==
colIdOfRow2
)
{
...
...
source/dnode/vnode/src/tsdb/tsdbSma.c
浏览文件 @
e11147dd
...
...
@@ -173,6 +173,7 @@ static void tsdbGetSmaDir(int32_t vgId, ETsdbSmaType smaType, char dirName[])
static
int32_t
tsdbInsertTSmaDataImpl
(
STsdb
*
pTsdb
,
int64_t
indexUid
,
const
char
*
msg
);
static
int32_t
tsdbInsertRSmaDataImpl
(
STsdb
*
pTsdb
,
const
char
*
msg
);
static
FORCE_INLINE
int32_t
tsdbUidStorePut
(
STbUidStore
*
pStore
,
tb_uid_t
suid
,
tb_uid_t
*
uid
);
static
FORCE_INLINE
int32_t
tsdbUpdateTbUidListImpl
(
STsdb
*
pTsdb
,
tb_uid_t
*
suid
,
SArray
*
tbUids
);
// mgmt interface
static
int32_t
tsdbDropTSmaDataImpl
(
STsdb
*
pTsdb
,
int64_t
indexUid
);
...
...
@@ -1692,18 +1693,16 @@ int32_t tsdbDropTSma(STsdb *pTsdb, char *pMsg) {
* @param pReq
* @return int32_t
*/
int32_t
tsdbRegisterRSma
(
STsdb
*
pTsdb
,
SMeta
*
pMeta
,
SVCreateTbReq
*
pReq
)
{
#if 0
SRSmaParam *param = pReq->stbCfg.pRSmaParam;
if (!param) {
tsdbDebug("vgId:%d return directly since no rollup for stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name,
pReq->stbCfg.suid);
int32_t
tsdbRegisterRSma
(
STsdb
*
pTsdb
,
SMeta
*
pMeta
,
SVCreateStbReq
*
pReq
)
{
if
(
!
pReq
->
rollup
)
{
tsdbDebug
(
"vgId:%d return directly since no rollup for stable %s %"
PRIi64
,
REPO_ID
(
pTsdb
),
pReq
->
name
,
pReq
->
suid
);
return
TSDB_CODE_SUCCESS
;
}
SRSmaParam
*
param
=
&
pReq
->
pRSmaParam
;
if
((
param
->
qmsg1Len
==
0
)
&&
(
param
->
qmsg2Len
==
0
))
{
tsdbWarn("vgId:%d no qmsg1/qmsg2 for rollup stable %s %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->s
tbCfg.s
uid);
tsdbWarn
(
"vgId:%d no qmsg1/qmsg2 for rollup stable %s %"
PRIi64
,
REPO_ID
(
pTsdb
),
pReq
->
name
,
pReq
->
suid
);
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -1716,9 +1715,9 @@ int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq) {
SSmaStat
*
pStat
=
SMA_ENV_STAT
(
pEnv
);
SRSmaInfo
*
pRSmaInfo
=
NULL
;
pRSmaInfo = taosHashGet(SMA_STAT_INFO_HASH(pStat), &pReq->s
tbCfg.s
uid, sizeof(tb_uid_t));
pRSmaInfo
=
taosHashGet
(
SMA_STAT_INFO_HASH
(
pStat
),
&
pReq
->
suid
,
sizeof
(
tb_uid_t
));
if
(
pRSmaInfo
)
{
tsdbWarn("vgId:%d rsma info already exists for stb: %s, %" PRIi64, REPO_ID(pTsdb), pReq->name, pReq->s
tbCfg.s
uid);
tsdbWarn
(
"vgId:%d rsma info already exists for stb: %s, %"
PRIi64
,
REPO_ID
(
pTsdb
),
pReq
->
name
,
pReq
->
suid
);
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -1758,14 +1757,13 @@ int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq) {
}
}
if (taosHashPut(SMA_STAT_INFO_HASH(pStat), &pReq->s
tbCfg.s
uid, sizeof(tb_uid_t), &pRSmaInfo, sizeof(pRSmaInfo)) !=
if
(
taosHashPut
(
SMA_STAT_INFO_HASH
(
pStat
),
&
pReq
->
suid
,
sizeof
(
tb_uid_t
),
&
pRSmaInfo
,
sizeof
(
pRSmaInfo
))
!=
TSDB_CODE_SUCCESS
)
{
return
TSDB_CODE_FAILED
;
}
else
{
tsdbDebug("vgId:%d register rsma info succeed for suid:%" PRIi64, REPO_ID(pTsdb), pReq->stbCfg.
suid);
tsdbDebug
(
"vgId:%d register rsma info succeed for suid:%"
PRIi64
,
REPO_ID
(
pTsdb
),
pReq
->
suid
);
}
#endif
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -1777,7 +1775,7 @@ int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq) {
* @param uid
* @return int32_t
*/
int32_t
tsdbUidStorePut
(
STbUidStore
*
pStore
,
tb_uid_t
suid
,
tb_uid_t
*
uid
)
{
static
int32_t
tsdbUidStorePut
(
STbUidStore
*
pStore
,
tb_uid_t
suid
,
tb_uid_t
*
uid
)
{
// prefer to store suid/uids in array
if
((
suid
==
pStore
->
suid
)
||
(
pStore
->
suid
==
0
))
{
if
(
pStore
->
suid
==
0
)
{
...
...
@@ -1833,6 +1831,7 @@ void tsdbUidStoreDestory(STbUidStore *pStore) {
if
(
pStore
)
{
if
(
pStore
->
uidHash
)
{
if
(
pStore
->
tbUids
)
{
// When pStore->tbUids not NULL, the pStore->uidHash has k/v; otherwise pStore->uidHash only has keys.
void
*
pIter
=
taosHashIterate
(
pStore
->
uidHash
,
NULL
);
while
(
pIter
)
{
SArray
*
arr
=
*
(
SArray
**
)
pIter
;
...
...
@@ -1847,8 +1846,10 @@ void tsdbUidStoreDestory(STbUidStore *pStore) {
}
void
*
tsdbUidStoreFree
(
STbUidStore
*
pStore
)
{
if
(
pStore
)
{
tsdbUidStoreDestory
(
pStore
);
taosMemoryFree
(
pStore
);
}
return
NULL
;
}
...
...
@@ -1861,7 +1862,7 @@ void *tsdbUidStoreFree(STbUidStore *pStore) {
* @param uid
* @return int32_t
*/
int32_t
tsdbFetchTbUidList
(
void
*
pTsdb
,
void
**
ppStore
,
void
*
suid
,
void
*
uid
)
{
int32_t
tsdbFetchTbUidList
(
STsdb
*
pTsdb
,
STbUidStore
**
ppStore
,
tb_uid_t
suid
,
tb_uid_t
uid
)
{
SSmaEnv
*
pEnv
=
REPO_RSMA_ENV
((
STsdb
*
)
pTsdb
);
// only applicable to rollup SMA ctables
...
...
@@ -1877,7 +1878,7 @@ int32_t tsdbFetchTbUidList(void *pTsdb, void **ppStore, void *suid, void *uid) {
}
// info cached when create rsma stable and return directly for non-rsma ctables
if
(
!
taosHashGet
(
infoHash
,
suid
,
sizeof
(
tb_uid_t
)))
{
if
(
!
taosHashGet
(
infoHash
,
&
suid
,
sizeof
(
tb_uid_t
)))
{
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -1887,7 +1888,7 @@ int32_t tsdbFetchTbUidList(void *pTsdb, void **ppStore, void *suid, void *uid) {
}
}
if
(
tsdbUidStorePut
(
*
ppStore
,
*
(
tb_uid_t
*
)
suid
,
(
tb_uid_t
*
)
uid
)
!=
0
)
{
if
(
tsdbUidStorePut
(
*
ppStore
,
suid
,
&
uid
)
!=
0
)
{
*
ppStore
=
tsdbUidStoreFree
(
*
ppStore
);
return
TSDB_CODE_FAILED
;
}
...
...
@@ -1935,12 +1936,10 @@ static FORCE_INLINE int32_t tsdbUpdateTbUidListImpl(STsdb *pTsdb, tb_uid_t *suid
int32_t
tsdbUpdateTbUidList
(
STsdb
*
pTsdb
,
STbUidStore
*
pStore
)
{
if
(
!
pStore
||
(
taosArrayGetSize
(
pStore
->
tbUids
)
==
0
))
{
tsdbDebug
(
"vgId:%d no need to update tbUids since empty uidStore"
,
REPO_ID
(
pTsdb
));
tsdbUidStoreFree
(
pStore
);
return
TSDB_CODE_SUCCESS
;
}
if
(
tsdbUpdateTbUidListImpl
(
pTsdb
,
&
pStore
->
suid
,
pStore
->
tbUids
)
!=
TSDB_CODE_SUCCESS
)
{
tsdbUidStoreFree
(
pStore
);
return
TSDB_CODE_FAILED
;
}
...
...
@@ -1951,15 +1950,11 @@ int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pStore) {
if
(
tsdbUpdateTbUidListImpl
(
pTsdb
,
pTbSuid
,
pTbUids
)
!=
TSDB_CODE_SUCCESS
)
{
taosHashCancelIterate
(
pStore
->
uidHash
,
pIter
);
tsdbUidStoreFree
(
pStore
);
return
TSDB_CODE_FAILED
;
}
pIter
=
taosHashIterate
(
pStore
->
uidHash
,
pIter
);
}
tsdbUidStoreFree
(
pStore
);
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -1971,8 +1966,6 @@ static int32_t tsdbFetchSubmitReqSuids(SSubmitReq *pMsg, STbUidStore *pStore) {
STSRow
*
row
=
NULL
;
terrno
=
TSDB_CODE_SUCCESS
;
// pMsg->length = htonl(pMsg->length);
// pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
if
(
tInitSubmitMsgIterEx
(
pMsg
,
&
msgIter
)
<
0
)
return
-
1
;
while
(
true
)
{
...
...
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @
e11147dd
...
...
@@ -300,13 +300,13 @@ static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq,
goto
_err
;
}
// tsdbRegisterRSma(pVnode->pTsdb, pVnode->pMeta, &vCreateTbReq);
if
(
metaCreateSTable
(
pVnode
->
pMeta
,
version
,
&
req
)
<
0
)
{
pRsp
->
code
=
terrno
;
goto
_err
;
}
tsdbRegisterRSma
(
pVnode
->
pTsdb
,
pVnode
->
pMeta
,
&
req
);
tCoderClear
(
&
coder
);
return
0
;
...
...
@@ -323,6 +323,7 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
SVCreateTbBatchRsp
rsp
=
{
0
};
SVCreateTbRsp
cRsp
=
{
0
};
char
tbName
[
TSDB_TABLE_FNAME_LEN
];
STbUidStore
*
pStore
=
NULL
;
pRsp
->
msgType
=
TDMT_VND_CREATE_TABLE_RSP
;
pRsp
->
code
=
TSDB_CODE_SUCCESS
;
...
...
@@ -361,6 +362,7 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
cRsp
.
code
=
terrno
;
}
else
{
cRsp
.
code
=
TSDB_CODE_SUCCESS
;
tsdbFetchTbUidList
(
pVnode
->
pTsdb
,
&
pStore
,
pCreateReq
->
ctb
.
suid
,
pCreateReq
->
uid
);
}
taosArrayPush
(
rsp
.
pArray
,
&
cRsp
);
...
...
@@ -368,6 +370,9 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
tCoderClear
(
&
coder
);
tsdbUpdateTbUidList
(
pVnode
->
pTsdb
,
pStore
);
tsdbUidStoreFree
(
pStore
);
// prepare rsp
int32_t
ret
=
0
;
tEncodeSize
(
tEncodeSVCreateTbBatchRsp
,
&
rsp
,
pRsp
->
contLen
,
ret
);
...
...
@@ -426,7 +431,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
SSubmitRsp
rsp
=
{
0
};
pRsp
->
code
=
0
;
tsdbTriggerRSma
(
pVnode
->
pTsdb
,
pVnode
->
pMeta
,
pReq
,
STREAM_DATA_TYPE_SUBMIT_BLOCK
);
// handle the request
if
(
tsdbInsertData
(
pVnode
->
pTsdb
,
version
,
pSubmitReq
,
&
rsp
)
<
0
)
{
pRsp
->
code
=
terrno
;
...
...
@@ -435,7 +440,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
// pRsp->msgType = TDMT_VND_SUBMIT_RSP;
// vnodeProcessSubmitReq(pVnode, ptr, pRsp);
// tsdbTriggerRSma(pVnode->pTsdb, pVnode->pMeta, p
tr
, STREAM_DATA_TYPE_SUBMIT_BLOCK);
// tsdbTriggerRSma(pVnode->pTsdb, pVnode->pMeta, p
Req
, STREAM_DATA_TYPE_SUBMIT_BLOCK);
// encode the response (TODO)
pRsp
->
pCont
=
rpcMallocCont
(
sizeof
(
SSubmitRsp
));
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录