Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
dgr6878
TDengine
提交
0be65a23
T
TDengine
项目概览
dgr6878
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
8
Star
0
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,发现更多精彩内容 >>
提交
0be65a23
编写于
5月 28, 2020
作者:
H
Hongze Cheng
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'develop' into feature/2.0tsdb
上级
1da4fa6a
f71c5aea
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
230 addition
and
21 deletion
+230
-21
src/common/inc/tdataformat.h
src/common/inc/tdataformat.h
+53
-0
src/common/src/tdataformat.c
src/common/src/tdataformat.c
+146
-0
src/tsdb/src/tsdbMeta.c
src/tsdb/src/tsdbMeta.c
+16
-10
src/tsdb/src/tsdbRead.c
src/tsdb/src/tsdbRead.c
+11
-9
src/vnode/src/vnodeWrite.c
src/vnode/src/vnodeWrite.c
+4
-2
未找到文件。
src/common/inc/tdataformat.h
浏览文件 @
0be65a23
...
...
@@ -217,6 +217,59 @@ void tdPopDataColsPoints(SDataCols *pCols, int pointsToPop); //!!!!
int
tdMergeDataCols
(
SDataCols
*
target
,
SDataCols
*
src
,
int
rowsToMerge
);
void
tdMergeTwoDataCols
(
SDataCols
*
target
,
SDataCols
*
src1
,
int
*
iter1
,
SDataCols
*
src2
,
int
*
iter2
,
int
tRows
);
// ----------------- Tag row structure
/* A tag row, the format is like below:
+----------+----------------------------------------------------------------+
| STagRow | STagCol | STagCol | STagCol | STagCol | ...| STagCol | STagCol |
+----------+----------------------------------------------------------------+
pData
+----------+----------------------------------------------------------------+
| value 1 | value 2 | value 3 | value 4 | ....|value n |
+----------+----------------------------------------------------------------+
*/
#define TD_TAG_ROW_HEAD_SIZE sizeof(int16_t)
#define tagRowNum(r) (*(int16_t *)(r))
#define tagRowArray(r) POINTER_SHIFT(r, TD_TAG_ROW_HEAD_SIZE)
//#define dataRowKey(r) (*(TSKEY *)(dataRowTuple(r)))
//#define dataRowSetLen(r, l) (dataRowLen(r) = (l))
//#define dataRowCpy(dst, r) memcpy((dst), (r), dataRowLen(r))
//#define dataRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_DATA_ROW_HEAD_SIZE)
typedef
struct
{
int16_t
colId
;
// column ID
int16_t
colType
;
uint16_t
offset
;
//to store value for numeric col or offset for binary/Nchar
}
STagCol
;
typedef
struct
{
int32_t
len
;
void
*
pData
;
// Space to store the tag value
uint16_t
dataLen
;
int16_t
ncols
;
// Total columns allocated
STagCol
tagCols
[];
}
STagRow
;
#define tagColSize(r) (sizeof(STagCol) + r.colLen)
int
tdSetTagCol
(
SDataRow
row
,
void
*
value
,
int16_t
len
,
int8_t
type
,
int16_t
colId
);
//insert tag value and update all the information
int
tdDeleteTagCol
(
SDataRow
row
,
int16_t
colId
);
// delete tag value and update all the information
void
*
tdQueryTagByID
(
SDataRow
row
,
int16_t
colId
,
int16_t
*
type
);
//if find tag, 0, else return -1;
int
tdAppendTagColVal
(
SDataRow
row
,
void
*
value
,
int8_t
type
,
int32_t
bytes
,
int16_t
colId
);
SDataRow
tdTagRowDup
(
SDataRow
row
);
void
tdFreeTagRow
(
SDataRow
row
);
SDataRow
tdTagRowDecode
(
SDataRow
row
);
int
tdTagRowCpy
(
SDataRow
dst
,
SDataRow
src
);
void
*
tdNewTagRowFromSchema
(
STSchema
*
pSchema
,
int16_t
numofTags
);
STSchema
*
tdGetSchemaFromData
(
SDataRow
*
row
);
#ifdef __cplusplus
}
#endif
...
...
src/common/src/tdataformat.c
浏览文件 @
0be65a23
...
...
@@ -14,6 +14,7 @@
*/
#include "tdataformat.h"
#include "wchar.h"
#include "talgo.h"
/**
* Create a SSchema object with nCols columns
...
...
@@ -151,6 +152,151 @@ SDataRow tdNewDataRowFromSchema(STSchema *pSchema) {
return
row
;
}
int
tdSetTagCol
(
SDataRow
row
,
void
*
value
,
int16_t
len
,
int8_t
type
,
int16_t
colId
){
//insert/update tag value and update all the information
ASSERT
(((
STagRow
*
)
row
)
->
pData
!=
NULL
);
//STagCol * stCol = tdQueryTagColByID()
return
0
;
};
int
tdDeleteTagCol
(
SDataRow
row
,
int16_t
colId
){
// delete tag value and update all the information
//todo
return
0
;
};
static
int
compTagId
(
const
void
*
key1
,
const
void
*
key2
)
{
if
(((
STagCol
*
)
key1
)
->
colId
>
((
STagCol
*
)
key2
)
->
colId
)
{
return
1
;
}
else
if
(((
STagCol
*
)
key1
)
->
colId
==
((
STagCol
*
)
key2
)
->
colId
)
{
return
0
;
}
else
{
return
-
1
;
}
}
/**
* Find tag structure by colId, if find, return tag structure, else return NULL;
*/
STagCol
*
tdQueryTagColByID
(
SDataRow
row
,
int16_t
colId
,
int
flags
)
{
//if find tag, 0, else return -1;
ASSERT
(((
STagRow
*
)
row
)
->
pData
!=
NULL
);
STagCol
*
pBase
=
((
STagRow
*
)
row
)
->
tagCols
;
int16_t
nCols
=
((
STagRow
*
)
row
)
->
ncols
;
STagCol
key
=
{
colId
,
0
,
0
};
STagCol
*
stCol
=
taosbsearch
(
&
key
,
pBase
,
nCols
,
sizeof
(
STagCol
),
compTagId
,
flags
);
return
stCol
;
};
/**
* Find tag value by colId, if find, return tag value, else return NULL;
*/
void
*
tdQueryTagByID
(
SDataRow
row
,
int16_t
colId
,
int16_t
*
type
)
{
ASSERT
(((
STagRow
*
)
row
)
->
pData
!=
NULL
);
STagCol
*
pBase
=
((
STagRow
*
)
row
)
->
tagCols
;
int16_t
nCols
=
((
STagRow
*
)
row
)
->
ncols
;
STagCol
key
=
{
colId
,
0
,
0
};
STagCol
*
stCol
=
taosbsearch
(
&
key
,
pBase
,
nCols
,
sizeof
(
STagCol
),
compTagId
,
TD_EQ
);
if
(
NULL
==
stCol
)
{
return
NULL
;
}
void
*
pData
=
((
STagRow
*
)
row
)
->
pData
;
*
type
=
stCol
->
colType
;
return
pData
+
stCol
->
offset
;
};
int
tdAppendTagColVal
(
SDataRow
row
,
void
*
value
,
int8_t
type
,
int32_t
bytes
,
int16_t
colId
){
ASSERT
(
value
!=
NULL
);
//ASSERT(bytes-2 == varDataTLen(value));
ASSERT
(
row
!=
NULL
);
STagRow
*
pTagrow
=
row
;
pTagrow
->
tagCols
[
pTagrow
->
ncols
].
colId
=
colId
;
pTagrow
->
tagCols
[
pTagrow
->
ncols
].
colType
=
type
;
pTagrow
->
tagCols
[
pTagrow
->
ncols
].
offset
=
pTagrow
->
dataLen
;
switch
(
type
)
{
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_NCHAR
:
memcpy
((
char
*
)
pTagrow
->
pData
+
pTagrow
->
dataLen
,
value
,
varDataTLen
(
value
));
pTagrow
->
dataLen
+=
varDataTLen
(
value
);
break
;
default:
memcpy
((
char
*
)
pTagrow
->
pData
+
pTagrow
->
dataLen
,
value
,
TYPE_BYTES
[
type
]);
pTagrow
->
dataLen
+=
TYPE_BYTES
[
type
];
break
;
}
pTagrow
->
ncols
++
;
return
0
;
};
void
*
tdNewTagRowFromSchema
(
STSchema
*
pSchema
,
int16_t
numofTags
)
{
int32_t
size
=
sizeof
(
STagRow
)
+
numofTags
*
sizeof
(
STagCol
);
STagRow
*
row
=
malloc
(
size
);
if
(
row
==
NULL
)
return
NULL
;
int32_t
datasize
=
pSchema
->
tlen
;
row
->
pData
=
malloc
(
datasize
);
if
(
NULL
==
row
->
pData
)
{
free
(
row
);
return
NULL
;
}
row
->
len
=
size
;
row
->
dataLen
=
0
;
row
->
ncols
=
0
;
return
row
;
}
/**
* free tag row
*/
void
tdFreeTagRow
(
SDataRow
row
)
{
if
(
row
)
{
free
(((
STagRow
*
)
row
)
->
pData
);
free
(
row
);
}
}
SDataRow
tdTagRowDup
(
SDataRow
row
)
{
STagRow
*
trow
=
malloc
(
dataRowLen
(
row
));
if
(
trow
==
NULL
)
return
NULL
;
dataRowCpy
(
trow
,
row
);
trow
->
pData
=
malloc
(
trow
->
dataLen
);
if
(
NULL
==
trow
->
pData
)
{
free
(
trow
);
return
NULL
;
}
memcpy
(
trow
->
pData
,
((
STagRow
*
)
row
)
->
pData
,
trow
->
dataLen
);
return
trow
;
}
SDataRow
tdTagRowDecode
(
SDataRow
row
)
{
STagRow
*
trow
=
malloc
(
dataRowLen
(
row
));
if
(
trow
==
NULL
)
return
NULL
;
dataRowCpy
(
trow
,
row
);
trow
->
pData
=
malloc
(
trow
->
dataLen
);
if
(
NULL
==
trow
->
pData
)
{
free
(
trow
);
return
NULL
;
}
char
*
pData
=
(
char
*
)
row
+
dataRowLen
(
row
);
memcpy
(
trow
->
pData
,
pData
,
trow
->
dataLen
);
return
trow
;
}
int
tdTagRowCpy
(
SDataRow
dst
,
SDataRow
src
)
{
if
(
src
==
NULL
)
return
-
1
;
dataRowCpy
(
dst
,
src
);
void
*
pData
=
dst
+
dataRowLen
(
src
);
memcpy
(
pData
,
((
STagRow
*
)
src
)
->
pData
,
((
STagRow
*
)
src
)
->
dataLen
);
return
0
;
}
/**
* Free the SDataRow object
*/
...
...
src/tsdb/src/tsdbMeta.c
浏览文件 @
0be65a23
...
...
@@ -47,8 +47,8 @@ void tsdbEncodeTable(STable *pTable, char *buf, int *contLen) {
ptr
=
tdEncodeSchema
(
ptr
,
pTable
->
schema
);
ptr
=
tdEncodeSchema
(
ptr
,
pTable
->
tagSchema
);
}
else
if
(
pTable
->
type
==
TSDB_CHILD_TABLE
)
{
data
RowCpy
(
ptr
,
pTable
->
tagVal
);
ptr
=
POINTER_SHIFT
(
ptr
,
dataRowLen
(
pTable
->
tagVal
));
tdTag
RowCpy
(
ptr
,
pTable
->
tagVal
);
ptr
=
POINTER_SHIFT
(
ptr
,
dataRowLen
(
pTable
->
tagVal
)
+
((
STagRow
*
)
pTable
->
tagVal
)
->
dataLen
);
}
else
{
ptr
=
tdEncodeSchema
(
ptr
,
pTable
->
schema
);
}
...
...
@@ -94,8 +94,8 @@ STable *tsdbDecodeTable(void *cont, int contLen) {
pTable
->
schema
=
tdDecodeSchema
(
&
ptr
);
pTable
->
tagSchema
=
tdDecodeSchema
(
&
ptr
);
}
else
if
(
pTable
->
type
==
TSDB_CHILD_TABLE
)
{
pTable
->
tagVal
=
td
DataRowDup
(
ptr
);
ptr
=
POINTER_SHIFT
(
ptr
,
dataRowLen
(
pTable
->
tagVal
));
pTable
->
tagVal
=
td
TagRowDecode
(
ptr
);
ptr
=
POINTER_SHIFT
(
ptr
,
dataRowLen
(
pTable
->
tagVal
)
+
((
STagRow
*
)
pTable
->
tagVal
)
->
dataLen
);
}
else
{
pTable
->
schema
=
tdDecodeSchema
(
&
ptr
);
}
...
...
@@ -117,8 +117,10 @@ static char* getTagIndexKey(const void* pData) {
SDataRow
row
=
elem
->
pTable
->
tagVal
;
STSchema
*
pSchema
=
tsdbGetTableTagSchema
(
elem
->
pMeta
,
elem
->
pTable
);
STColumn
*
pCol
=
&
pSchema
->
columns
[
DEFAULT_TAG_INDEX_COLUMN
];
return
tdGetRowDataOfCol
(
row
,
pCol
->
type
,
TD_DATA_ROW_HEAD_SIZE
+
pCol
->
offset
);
int16_t
type
=
0
;
void
*
res
=
tdQueryTagByID
(
row
,
pCol
->
colId
,
&
type
);
ASSERT
(
type
==
pCol
->
type
);
return
res
;
}
int
tsdbRestoreTable
(
void
*
pHandle
,
void
*
cont
,
int
contLen
)
{
...
...
@@ -258,8 +260,9 @@ int32_t tsdbGetTableTagVal(TsdbRepoT* repo, STableId* id, int32_t colId, int16_t
}
SDataRow
row
=
(
SDataRow
)
pTable
->
tagVal
;
char
*
d
=
tdGetRowDataOfCol
(
row
,
pCol
->
type
,
TD_DATA_ROW_HEAD_SIZE
+
pCol
->
offset
);
int16_t
tagtype
=
0
;
char
*
d
=
tdQueryTagByID
(
row
,
pCol
->
colId
,
&
tagtype
);
//ASSERT((int8_t)tagtype == pCol->type)
*
val
=
d
;
*
type
=
pCol
->
type
;
*
bytes
=
pCol
->
bytes
;
...
...
@@ -472,7 +475,7 @@ static int tsdbFreeTable(STable *pTable) {
if
(
pTable
==
NULL
)
return
0
;
if
(
pTable
->
type
==
TSDB_CHILD_TABLE
)
{
tdFree
Data
Row
(
pTable
->
tagVal
);
tdFree
Tag
Row
(
pTable
->
tagVal
);
}
else
{
tdFreeSchema
(
pTable
->
schema
);
}
...
...
@@ -620,7 +623,9 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
STSchema
*
pSchema
=
tsdbGetTableTagSchema
(
pMeta
,
pTable
);
STColumn
*
pCol
=
&
pSchema
->
columns
[
DEFAULT_TAG_INDEX_COLUMN
];
char
*
key
=
tdGetRowDataOfCol
(
pTable
->
tagVal
,
pCol
->
type
,
TD_DATA_ROW_HEAD_SIZE
+
pCol
->
offset
);
int16_t
tagtype
=
0
;
char
*
key
=
tdQueryTagByID
(
pTable
->
tagVal
,
pCol
->
colId
,
&
tagtype
);
ASSERT
(
pCol
->
type
==
tagtype
);
SArray
*
res
=
tSkipListGet
(
pSTable
->
pIndex
,
key
);
size_t
size
=
taosArrayGetSize
(
res
);
...
...
@@ -639,6 +644,7 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
return
0
;
}
char
*
getTSTupleKey
(
const
void
*
data
)
{
SDataRow
row
=
(
SDataRow
)
data
;
return
POINTER_SHIFT
(
row
,
TD_DATA_ROW_HEAD_SIZE
);
...
...
src/tsdb/src/tsdbRead.c
浏览文件 @
0be65a23
...
...
@@ -1753,9 +1753,9 @@ int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) {
STColumn
*
pCol
=
schemaColAt
(
pTableGroupSupp
->
pTagSchema
,
colIndex
);
bytes
=
pCol
->
bytes
;
type
=
pCol
->
type
;
f1
=
td
GetRowDataOfCol
(
pTable1
->
tagVal
,
pCol
->
type
,
TD_DATA_ROW_HEAD_SIZE
+
pCol
->
offset
);
f2
=
td
GetRowDataOfCol
(
pTable2
->
tagVal
,
pCol
->
type
,
TD_DATA_ROW_HEAD_SIZE
+
pCol
->
offset
);
int16_t
tgtype1
,
tgtype2
=
0
;
f1
=
td
QueryTagByID
(
pTable1
->
tagVal
,
pCol
->
colId
,
&
tgtype1
);
f2
=
td
QueryTagByID
(
pTable2
->
tagVal
,
pCol
->
colId
,
&
tgtype2
);
}
int32_t
ret
=
doCompare
(
f1
,
f2
,
type
,
bytes
);
...
...
@@ -1843,12 +1843,14 @@ bool indexedNodeFilterFp(const void* pNode, void* param) {
val
=
(
char
*
)
elem
->
pTable
->
name
;
type
=
TSDB_DATA_TYPE_BINARY
;
}
else
{
STSchema
*
pTSchema
=
(
STSchema
*
)
pInfo
->
param
;
// todo table schema is identical to stable schema??
int32_t
offset
=
pTSchema
->
columns
[
pInfo
->
colIndex
].
offset
;
val
=
tdGetRowDataOfCol
(
elem
->
pTable
->
tagVal
,
pInfo
->
sch
.
type
,
TD_DATA_ROW_HEAD_SIZE
+
offset
);
}
// STSchema* pTSchema = (STSchema*) pInfo->param; // todo table schema is identical to stable schema??
int16_t
type
;
// int32_t offset = pTSchema->columns[pInfo->colIndex].offset;
// val = tdGetRowDataOfCol(elem->pTable->tagVal, pInfo->sch.type, TD_DATA_ROW_HEAD_SIZE + offset);
val
=
tdQueryTagByID
(
elem
->
pTable
->
tagVal
,
pInfo
->
sch
.
colId
,
&
type
);
// ASSERT(pInfo->sch.type == type);
}
//todo :the val is possible to be null, so check it out carefully
int32_t
ret
=
0
;
if
(
type
==
TSDB_DATA_TYPE_BINARY
||
type
==
TSDB_DATA_TYPE_NCHAR
)
{
if
(
pInfo
->
optr
==
TSDB_RELATION_IN
)
{
...
...
src/vnode/src/vnodeWrite.c
浏览文件 @
0be65a23
...
...
@@ -140,11 +140,13 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe
char
*
pTagData
=
pTable
->
data
+
totalCols
*
sizeof
(
SSchema
);
int
accumBytes
=
0
;
dataRow
=
tdNewDataRowFromSchema
(
pDestTagSchema
);
//dataRow = tdNewDataRowFromSchema(pDestTagSchema);
dataRow
=
tdNewTagRowFromSchema
(
pDestTagSchema
,
numOfTags
);
for
(
int
i
=
0
;
i
<
numOfTags
;
i
++
)
{
STColumn
*
pTCol
=
schemaColAt
(
pDestTagSchema
,
i
);
tdAppendColVal
(
dataRow
,
pTagData
+
accumBytes
,
pTCol
->
type
,
pTCol
->
bytes
,
pTCol
->
offset
);
// tdAppendColVal(dataRow, pTagData + accumBytes, pTCol->type, pTCol->bytes, pTCol->offset);
tdAppendTagColVal
(
dataRow
,
pTagData
+
accumBytes
,
pTCol
->
type
,
pTCol
->
bytes
,
pTCol
->
colId
);
accumBytes
+=
htons
(
pSchema
[
i
+
numOfColumns
].
bytes
);
}
tsdbTableSetTagValue
(
&
tCfg
,
dataRow
,
false
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录