Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
640b2f4d
TDengine
项目概览
taosdata
/
TDengine
接近 1 年 前同步成功
通知
1181
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,发现更多精彩内容 >>
提交
640b2f4d
编写于
3月 30, 2022
作者:
C
Cary Xu
提交者:
cpwu
3月 30, 2022
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
insert tSma by SSDataBlock
上级
1cce35c0
变更
6
展开全部
隐藏空白更改
内联
并排
Showing
6 changed file
with
309 addition
and
208 deletion
+309
-208
include/common/tmsg.h
include/common/tmsg.h
+0
-27
source/dnode/vnode/inc/tsdb.h
source/dnode/vnode/inc/tsdb.h
+3
-7
source/dnode/vnode/src/inc/tsdbSma.h
source/dnode/vnode/src/inc/tsdbSma.h
+2
-3
source/dnode/vnode/src/tsdb/tsdbSma.c
source/dnode/vnode/src/tsdb/tsdbSma.c
+190
-143
source/dnode/vnode/src/vnd/vnodeWrite.c
source/dnode/vnode/src/vnd/vnodeWrite.c
+3
-1
source/dnode/vnode/test/tsdbSmaTest.cpp
source/dnode/vnode/test/tsdbSmaTest.cpp
+111
-27
未找到文件。
include/common/tmsg.h
浏览文件 @
640b2f4d
...
...
@@ -2016,7 +2016,6 @@ typedef struct {
int8_t
slidingUnit
;
// MACRO: TIME_UNIT_XXX
int8_t
timezoneInt
;
// sma data expired if timezone changes.
char
indexName
[
TSDB_INDEX_NAME_LEN
];
char
timezone
[
TD_TIMEZONE_LEN
];
int32_t
exprLen
;
int32_t
tagsFilterLen
;
int64_t
indexUid
;
...
...
@@ -2054,32 +2053,6 @@ void* tDeserializeSVCreateTSmaReq(void* buf, SVCreateTSmaReq* pReq);
int32_t
tSerializeSVDropTSmaReq
(
void
**
buf
,
SVDropTSmaReq
*
pReq
);
void
*
tDeserializeSVDropTSmaReq
(
void
*
buf
,
SVDropTSmaReq
*
pReq
);
typedef
struct
{
col_id_t
colId
;
uint16_t
blockSize
;
// sma data block size
char
data
[];
}
STSmaColData
;
typedef
struct
{
tb_uid_t
tableUid
;
// super/child/normal table uid
int32_t
dataLen
;
// not including head
char
data
[];
}
STSmaTbData
;
typedef
struct
{
int64_t
indexUid
;
TSKEY
skey
;
// startKey of one interval/sliding window
int64_t
interval
;
int32_t
dataLen
;
// not including head
int8_t
intervalUnit
;
char
data
[];
}
STSmaDataWrapper
;
// sma data for a interval/sliding window
// interval/sliding => window
// => window->table->colId
// => 当一个window下所有的表均计算完成时,流计算告知tsdb清除window的过期标记
// RSma: Rollup SMA
typedef
struct
{
int64_t
interval
;
...
...
source/dnode/vnode/inc/tsdb.h
浏览文件 @
640b2f4d
...
...
@@ -100,10 +100,11 @@ int32_t tsdbUpdateSmaWindow(STsdb *pTsdb, const char *msg);
* @brief Insert tSma(Time-range-wise SMA) data from stream computing engine
*
* @param pTsdb
* @param indexUid
* @param msg
* @return int32_t
*/
int32_t
tsdbInsertTSmaData
(
STsdb
*
pTsdb
,
char
*
msg
);
int32_t
tsdbInsertTSmaData
(
STsdb
*
pTsdb
,
int64_t
indexUid
,
const
char
*
msg
);
/**
* @brief Drop tSma data and local cache.
...
...
@@ -130,16 +131,11 @@ int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg);
* @param pTsdb
* @param pData
* @param indexUid
* @param interval
* @param intervalUnit
* @param tableUid
* @param colId
* @param querySKey
* @param nMaxResult
* @return int32_t
*/
int32_t
tsdbGetTSmaData
(
STsdb
*
pTsdb
,
STSmaDataWrapper
*
pData
,
int64_t
indexUid
,
int64_t
interval
,
int8_t
intervalUnit
,
tb_uid_t
tableUid
,
col_id_t
colId
,
TSKEY
querySKey
,
int32_t
nMaxResult
);
int32_t
tsdbGetTSmaData
(
STsdb
*
pTsdb
,
char
*
pData
,
int64_t
indexUid
,
TSKEY
querySKey
,
int32_t
nMaxResult
);
// STsdbCfg
int
tsdbOptionsInit
(
STsdbCfg
*
);
...
...
source/dnode/vnode/src/inc/tsdbSma.h
浏览文件 @
640b2f4d
...
...
@@ -44,11 +44,10 @@ int32_t tsdbRemoveTSmaData(STsdb *pTsdb, STSma *param, STimeWindow *pWin);
#endif
// internal func
static
FORCE_INLINE
int32_t
tsdbEncodeTSmaKey
(
tb_uid_t
tableUid
,
col_id_t
col
Id
,
TSKEY
tsKey
,
void
**
pData
)
{
static
FORCE_INLINE
int32_t
tsdbEncodeTSmaKey
(
int64_t
group
Id
,
TSKEY
tsKey
,
void
**
pData
)
{
int32_t
len
=
0
;
len
+=
taosEncodeFixedI64
(
pData
,
tableUid
);
len
+=
taosEncodeFixedU16
(
pData
,
colId
);
len
+=
taosEncodeFixedI64
(
pData
,
tsKey
);
len
+=
taosEncodeFixedI64
(
pData
,
groupId
);
return
len
;
}
...
...
source/dnode/vnode/src/tsdb/tsdbSma.c
浏览文件 @
640b2f4d
此差异已折叠。
点击以展开。
source/dnode/vnode/src/vnd/vnodeWrite.c
浏览文件 @
640b2f4d
...
...
@@ -17,7 +17,9 @@
void
smaHandleRes
(
void
*
pVnode
,
int64_t
smaId
,
const
SArray
*
data
)
{
// TODO
blockDebugShowData
(
data
);
tsdbInsertTSmaData
(((
SVnode
*
)
pVnode
)
->
pTsdb
,
smaId
,
(
const
char
*
)
data
);
}
void
vnodeProcessWMsgs
(
SVnode
*
pVnode
,
SArray
*
pMsgs
)
{
...
...
@@ -202,7 +204,7 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
vCreateSmaReq
.
tSma
.
indexUid
);
// record current timezone of server side
tstrncpy
(
vCreateSmaReq
.
tSma
.
timezone
,
tsTimezoneStr
,
TD_TIMEZONE_LEN
)
;
vCreateSmaReq
.
tSma
.
timezoneInt
=
tsTimezone
;
if
(
metaCreateTSma
(
pVnode
->
pMeta
,
&
vCreateSmaReq
)
<
0
)
{
// TODO: handle error
...
...
source/dnode/vnode/test/tsdbSmaTest.cpp
浏览文件 @
640b2f4d
...
...
@@ -15,6 +15,7 @@
#include <gtest/gtest.h>
#include <tsdbDef.h>
#include <taoserror.h>
#include <tglobal.h>
#include <iostream>
...
...
@@ -280,7 +281,7 @@ TEST(testCase, tSma_metaDB_Put_Get_Del_Test) {
}
#endif
#if
0
#if
1
TEST
(
testCase
,
tSma_Data_Insert_Query_Test
)
{
// step 1: prepare meta
const
char
*
smaIndexName1
=
"sma_index_test_1"
;
...
...
@@ -299,9 +300,9 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
// encode
STSma
tSma
=
{
0
};
tSma
.
version
=
0
;
tSma.intervalUnit = TIME_UNIT_
DAY
;
tSma
.
intervalUnit
=
TIME_UNIT_
MINUTE
;
tSma
.
interval
=
1
;
tSma.slidingUnit = TIME_UNIT_
HOUR
;
tSma
.
slidingUnit
=
TIME_UNIT_
MINUTE
;
tSma
.
sliding
=
1
;
// sliding = interval when it's convert window
tSma
.
indexUid
=
indexUid1
;
tstrncpy
(
tSma
.
indexName
,
smaIndexName1
,
TSDB_INDEX_NAME_LEN
);
...
...
@@ -330,8 +331,8 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
ASSERT_EQ
(
metaSaveSmaToDB
(
pMeta
,
pSmaCfg
),
0
);
// step 2: insert data
STsdb
*pTsdb = (STsdb *)taosMemoryCalloc(1, sizeof(STsdb));
STsdbCfg
*pCfg = &pTsdb->config;
STsdb
*
pTsdb
=
(
STsdb
*
)
taosMemoryCalloc
(
1
,
sizeof
(
STsdb
));
STsdbCfg
*
pCfg
=
&
pTsdb
->
config
;
pTsdb
->
pMeta
=
pMeta
;
pTsdb
->
vgId
=
2
;
...
...
@@ -405,15 +406,94 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
ASSERT_EQ
(
tsdbUpdateSmaWindow
(
pTsdb
,
(
const
char
*
)
pMsg
),
0
);
// init
int32_t allocCnt = 0;
int32_t allocStep = 16384;
int32_t buffer = 1024;
void *buf = NULL;
ASSERT_EQ(tsdbMakeRoom(&buf, allocStep), 0);
int32_t bufSize = taosTSizeof(buf);
int32_t numOfTables = 10;
col_id_t numOfCols = 4096;
ASSERT_GT(numOfCols, 0);
const
int32_t
tSmaGroupSize
=
4
;
const
int32_t
tSmaNumOfTags
=
2
;
const
int64_t
tSmaGroupId
=
12345670
;
const
col_id_t
tSmaNumOfCols
=
9
;
// binary/nchar/varbinary/varchar are only used for tags for group by conditions.
const
int32_t
tSmaNumOfRows
=
2
;
SArray
*
pDataBlocks
=
taosArrayInit
(
tSmaGroupSize
,
sizeof
(
SSDataBlock
*
));
ASSERT_NE
(
pDataBlocks
,
nullptr
);
int32_t
tSmaTypeArray
[
tSmaNumOfCols
]
=
{
TSDB_DATA_TYPE_TIMESTAMP
,
TSDB_DATA_TYPE_BOOL
,
TSDB_DATA_TYPE_INT
,
TSDB_DATA_TYPE_UBIGINT
,
TSDB_DATA_TYPE_SMALLINT
,
TSDB_DATA_TYPE_FLOAT
,
TSDB_DATA_TYPE_DOUBLE
,
TSDB_DATA_TYPE_VARCHAR
,
TSDB_DATA_TYPE_NCHAR
};
// last 2 columns for group by tags
// int32_t tSmaTypeArray[tSmaNumOfCols] = {TSDB_DATA_TYPE_TIMESTAMP, TSDB_DATA_TYPE_BOOL};
const
char
*
tSmaGroupbyTags
[
tSmaGroupSize
*
tSmaNumOfTags
]
=
{
"BeiJing"
,
"HaiDian"
,
"BeiJing"
,
"ChaoYang"
,
"ShangHai"
,
"PuDong"
,
"ShangHai"
,
"MinHang"
};
TSKEY
tSmaSKeyMs
=
(
int64_t
)
1648535332
*
1000
;
int64_t
tSmaIntervalMs
=
tSma
.
interval
*
60
*
1000
;
int64_t
tSmaInitVal
=
0
;
for
(
int32_t
g
=
0
;
g
<
tSmaGroupSize
;
++
g
)
{
SSDataBlock
*
pDataBlock
=
(
SSDataBlock
*
)
taosMemoryCalloc
(
1
,
sizeof
(
SSDataBlock
));
ASSERT_NE
(
pDataBlock
,
nullptr
);
pDataBlock
->
pBlockAgg
=
NULL
;
pDataBlock
->
info
.
numOfCols
=
tSmaNumOfCols
;
pDataBlock
->
info
.
rows
=
tSmaNumOfRows
;
pDataBlock
->
info
.
groupId
=
tSmaGroupId
+
g
;
pDataBlock
->
pDataBlock
=
taosArrayInit
(
tSmaNumOfCols
,
sizeof
(
SColumnInfoData
*
));
ASSERT_NE
(
pDataBlock
->
pDataBlock
,
nullptr
);
for
(
int32_t
c
=
0
;
c
<
tSmaNumOfCols
;
++
c
)
{
SColumnInfoData
*
pColInfoData
=
(
SColumnInfoData
*
)
taosMemoryCalloc
(
1
,
sizeof
(
SColumnInfoData
));
ASSERT_NE
(
pColInfoData
,
nullptr
);
pColInfoData
->
info
.
type
=
tSmaTypeArray
[
c
];
if
(
IS_VAR_DATA_TYPE
(
pColInfoData
->
info
.
type
))
{
pColInfoData
->
info
.
bytes
=
100
;
// update accordingly
}
else
{
pColInfoData
->
info
.
bytes
=
TYPE_BYTES
[
pColInfoData
->
info
.
type
];
}
pColInfoData
->
pData
=
(
char
*
)
taosMemoryCalloc
(
1
,
tSmaNumOfRows
*
pColInfoData
->
info
.
bytes
);
for
(
int32_t
r
=
0
;
r
<
tSmaNumOfRows
;
++
r
)
{
void
*
pCellData
=
pColInfoData
->
pData
+
r
*
pColInfoData
->
info
.
bytes
;
switch
(
pColInfoData
->
info
.
type
)
{
case
TSDB_DATA_TYPE_TIMESTAMP
:
*
(
TSKEY
*
)
pCellData
=
tSmaSKeyMs
+
tSmaIntervalMs
*
r
;
break
;
case
TSDB_DATA_TYPE_BOOL
:
*
(
bool
*
)
pCellData
=
(
bool
)
tSmaInitVal
++
;
break
;
case
TSDB_DATA_TYPE_INT
:
*
(
int
*
)
pCellData
=
(
int
)
tSmaInitVal
++
;
break
;
case
TSDB_DATA_TYPE_UBIGINT
:
*
(
uint64_t
*
)
pCellData
=
(
uint64_t
)
tSmaInitVal
++
;
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
*
(
int16_t
*
)
pCellData
=
(
int16_t
)
tSmaInitVal
++
;
break
;
case
TSDB_DATA_TYPE_FLOAT
:
*
(
float
*
)
pCellData
=
(
float
)
tSmaInitVal
++
;
break
;
case
TSDB_DATA_TYPE_DOUBLE
:
*
(
double
*
)
pCellData
=
(
double
)
tSmaInitVal
++
;
break
;
case
TSDB_DATA_TYPE_VARCHAR
:
// city
varDataSetLen
(
pCellData
,
strlen
(
tSmaGroupbyTags
[
g
*
2
]));
memcpy
(
varDataVal
(
pCellData
),
tSmaGroupbyTags
[
g
*
2
],
varDataLen
(
pCellData
));
break
;
case
TSDB_DATA_TYPE_NCHAR
:
// district
varDataSetLen
(
pCellData
,
strlen
(
tSmaGroupbyTags
[
g
*
2
+
1
]));
memcpy
(
varDataVal
(
pCellData
),
tSmaGroupbyTags
[
g
*
2
+
1
],
varDataLen
(
pCellData
));
break
;
default:
ASSERT_EQ
(
0
,
1
);
// add definition
break
;
}
}
// push SColumnInfoData
taosArrayPush
(
pDataBlock
->
pDataBlock
,
&
pColInfoData
);
}
// push SSDataBlock
taosArrayPush
(
pDataBlocks
,
&
pDataBlock
);
}
// execute
ASSERT_EQ
(
tsdbInsertTSmaData
(
pTsdb
,
tSma
.
indexUid
,
(
const
char
*
)
pDataBlocks
),
TSDB_CODE_SUCCESS
);
#if 0
STSmaDataWrapper *pSmaData = NULL;
...
...
@@ -464,26 +544,30 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
ASSERT_EQ(tsdbInsertTSmaData(pTsdb, (char *)pSmaData), TSDB_CODE_SUCCESS);
#endif
SSDataBlock *pSmaData = (SSDataBlock *)taosMemoryCalloc(1, sizeof(SSDataBlock));
// step 3: query
uint32_t
checkDataCnt
=
0
;
for (int32_t t = 0; t < numOfTables; ++t) {
for (col_id_t c = 0; c < numOfCols; ++c) {
ASSERT_EQ(tsdbGetTSmaData(pTsdb, NULL, indexUid1, interval1, intervalUnit1, tbUid + t,
c + PRIMARYKEY_TIMESTAMP_COL_ID, skey1, 1),
TSDB_CODE_SUCCESS);
++checkDataCnt;
}
}
ASSERT_EQ
(
tsdbGetTSmaData
(
pTsdb
,
NULL
,
indexUid1
,
skey1
,
1
),
TSDB_CODE_SUCCESS
);
++
checkDataCnt
;
printf
(
"%s:%d The sma data check count for insert and query is %"
PRIu32
"
\n
"
,
__FILE__
,
__LINE__
,
checkDataCnt
);
// release data
taosMemoryFreeClear
(
pMsg
);
taosTZfree(buf);
for
(
int32_t
i
=
0
;
i
<
taosArrayGetSize
(
pDataBlocks
);
++
i
)
{
SSDataBlock
*
pDataBlock
=
(
SSDataBlock
*
)
taosArrayGet
(
pDataBlocks
,
i
);
int32_t
numOfOutput
=
taosArrayGetSize
(
pDataBlock
->
pDataBlock
);
for
(
int32_t
j
=
0
;
j
<
numOfOutput
;
++
j
)
{
SColumnInfoData
*
pColInfoData
=
(
SColumnInfoData
*
)
taosArrayGet
(
pDataBlock
->
pDataBlock
,
j
);
colDataDestroy
(
pColInfoData
);
}
taosArrayDestroy
(
pDataBlock
->
pDataBlock
);
taosMemoryFreeClear
(
pDataBlock
->
pBlockAgg
);
taosMemoryFreeClear
(
pDataBlock
);
}
taosArrayDestroy
(
pDataBlocks
);
// release meta
tdDestroyTSma
(
&
tSma
);
tfsClose
(
pTsdb
->
pTfs
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录