Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
dgr6878
TDengine
提交
fdad406e
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,发现更多精彩内容 >>
提交
fdad406e
编写于
5月 25, 2020
作者:
S
Shengliang Guan
提交者:
Shuduo Sang
5月 25, 2020
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #2024 from taosdata/feature/query
Feature/query
上级
5ab5dcea
0c4b90f3
变更
13
显示空白变更内容
内联
并排
Showing
13 changed file
with
173 addition
and
129 deletion
+173
-129
src/client/inc/tscLog.h
src/client/inc/tscLog.h
+1
-1
src/client/inc/tsclient.h
src/client/inc/tsclient.h
+1
-5
src/common/inc/tdataformat.h
src/common/inc/tdataformat.h
+4
-4
src/common/src/tdataformat.c
src/common/src/tdataformat.c
+26
-26
src/common/src/ttypes.c
src/common/src/ttypes.c
+42
-2
src/query/src/qExecutor.c
src/query/src/qExecutor.c
+10
-10
src/tsdb/inc/tsdbMain.h
src/tsdb/inc/tsdbMain.h
+3
-3
src/tsdb/src/tsdbCache.c
src/tsdb/src/tsdbCache.c
+2
-2
src/tsdb/src/tsdbFile.c
src/tsdb/src/tsdbFile.c
+2
-2
src/tsdb/src/tsdbMain.c
src/tsdb/src/tsdbMain.c
+6
-6
src/tsdb/src/tsdbRWHelper.c
src/tsdb/src/tsdbRWHelper.c
+47
-47
src/tsdb/src/tsdbRead.c
src/tsdb/src/tsdbRead.c
+29
-20
tests/pytest/import_merge/importToCommit.py
tests/pytest/import_merge/importToCommit.py
+0
-1
未找到文件。
src/client/inc/tscLog.h
浏览文件 @
fdad406e
src/client/inc/tsclient.h
浏览文件 @
fdad406e
...
...
@@ -53,11 +53,7 @@ typedef struct STableComInfo {
}
STableComInfo
;
typedef
struct
STableMeta
{
// super table if it is created according to super table, otherwise, tableInfo is used
union
{
struct
STableMeta
*
pSTable
;
STableComInfo
tableInfo
;
};
uint8_t
tableType
;
int16_t
sversion
;
SCMVgroupInfo
vgroupInfo
;
...
...
src/common/inc/tdataformat.h
浏览文件 @
fdad406e
...
...
@@ -154,8 +154,8 @@ typedef struct SDataCol {
static
FORCE_INLINE
void
dataColReset
(
SDataCol
*
pDataCol
)
{
pDataCol
->
len
=
0
;
}
void
dataColInit
(
SDataCol
*
pDataCol
,
STColumn
*
pCol
,
void
**
pBuf
,
int
maxPoints
);
void
dataColAppendVal
(
SDataCol
*
pCol
,
void
*
value
,
int
numOf
Point
s
,
int
maxPoints
);
void
dataColPopPoints
(
SDataCol
*
pCol
,
int
pointsToPop
,
int
numOf
Point
s
);
void
dataColAppendVal
(
SDataCol
*
pCol
,
void
*
value
,
int
numOf
Row
s
,
int
maxPoints
);
void
dataColPopPoints
(
SDataCol
*
pCol
,
int
pointsToPop
,
int
numOf
Row
s
);
void
dataColSetOffset
(
SDataCol
*
pCol
,
int
nEle
);
bool
isNEleNull
(
SDataCol
*
pCol
,
int
nEle
);
...
...
@@ -195,7 +195,7 @@ typedef struct {
int
maxPoints
;
// max number of points
int
bufSize
;
int
numOf
Point
s
;
int
numOf
Row
s
;
int
numOfCols
;
// Total number of cols
int
sversion
;
// TODO: set sversion
void
*
buf
;
...
...
@@ -205,7 +205,7 @@ typedef struct {
#define keyCol(pCols) (&((pCols)->cols[0])) // Key column
#define dataColsKeyAt(pCols, idx) ((TSKEY *)(keyCol(pCols)->pData))[(idx)]
#define dataColsKeyFirst(pCols) dataColsKeyAt(pCols, 0)
#define dataColsKeyLast(pCols) ((pCols->numOf
Points == 0) ? 0 : dataColsKeyAt(pCols, (pCols)->numOfPoint
s - 1))
#define dataColsKeyLast(pCols) ((pCols->numOf
Rows == 0) ? 0 : dataColsKeyAt(pCols, (pCols)->numOfRow
s - 1))
SDataCols
*
tdNewDataCols
(
int
maxRowSize
,
int
maxCols
,
int
maxRows
);
void
tdResetDataCols
(
SDataCols
*
pCols
);
...
...
src/common/src/tdataformat.c
浏览文件 @
fdad406e
...
...
@@ -187,29 +187,29 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, void **pBuf, int maxPoints)
}
void
dataColAppendVal
(
SDataCol
*
pCol
,
void
*
value
,
int
numOf
Point
s
,
int
maxPoints
)
{
void
dataColAppendVal
(
SDataCol
*
pCol
,
void
*
value
,
int
numOf
Row
s
,
int
maxPoints
)
{
ASSERT
(
pCol
!=
NULL
&&
value
!=
NULL
);
switch
(
pCol
->
type
)
{
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_NCHAR
:
// set offset
pCol
->
dataOff
[
numOf
Point
s
]
=
pCol
->
len
;
pCol
->
dataOff
[
numOf
Row
s
]
=
pCol
->
len
;
// Copy data
memcpy
(
POINTER_SHIFT
(
pCol
->
pData
,
pCol
->
len
),
value
,
varDataTLen
(
value
));
// Update the length
pCol
->
len
+=
varDataTLen
(
value
);
break
;
default:
ASSERT
(
pCol
->
len
==
TYPE_BYTES
[
pCol
->
type
]
*
numOf
Point
s
);
ASSERT
(
pCol
->
len
==
TYPE_BYTES
[
pCol
->
type
]
*
numOf
Row
s
);
memcpy
(
POINTER_SHIFT
(
pCol
->
pData
,
pCol
->
len
),
value
,
pCol
->
bytes
);
pCol
->
len
+=
pCol
->
bytes
;
break
;
}
}
void
dataColPopPoints
(
SDataCol
*
pCol
,
int
pointsToPop
,
int
numOf
Point
s
)
{
int
pointsLeft
=
numOf
Point
s
-
pointsToPop
;
void
dataColPopPoints
(
SDataCol
*
pCol
,
int
pointsToPop
,
int
numOf
Row
s
)
{
int
pointsLeft
=
numOf
Row
s
-
pointsToPop
;
ASSERT
(
pointsLeft
>
0
);
...
...
@@ -221,7 +221,7 @@ void dataColPopPoints(SDataCol *pCol, int pointsToPop, int numOfPoints) {
memmove
(
pCol
->
pData
,
POINTER_SHIFT
(
pCol
->
pData
,
toffset
),
pCol
->
len
);
dataColSetOffset
(
pCol
,
pointsLeft
);
}
else
{
ASSERT
(
pCol
->
len
==
TYPE_BYTES
[
pCol
->
type
]
*
numOf
Point
s
);
ASSERT
(
pCol
->
len
==
TYPE_BYTES
[
pCol
->
type
]
*
numOf
Row
s
);
pCol
->
len
=
TYPE_BYTES
[
pCol
->
type
]
*
pointsLeft
;
memmove
(
pCol
->
pData
,
POINTER_SHIFT
(
pCol
->
pData
,
TYPE_BYTES
[
pCol
->
type
]
*
pointsToPop
),
pCol
->
len
);
}
...
...
@@ -322,7 +322,7 @@ SDataCols *tdDupDataCols(SDataCols *pDataCols, bool keepData) {
pRet
->
numOfCols
=
pDataCols
->
numOfCols
;
pRet
->
sversion
=
pDataCols
->
sversion
;
if
(
keepData
)
pRet
->
numOf
Points
=
pDataCols
->
numOfPoint
s
;
if
(
keepData
)
pRet
->
numOf
Rows
=
pDataCols
->
numOfRow
s
;
for
(
int
i
=
0
;
i
<
pDataCols
->
numOfCols
;
i
++
)
{
pRet
->
cols
[
i
].
type
=
pDataCols
->
cols
[
i
].
type
;
...
...
@@ -352,7 +352,7 @@ SDataCols *tdDupDataCols(SDataCols *pDataCols, bool keepData) {
}
void
tdResetDataCols
(
SDataCols
*
pCols
)
{
pCols
->
numOf
Point
s
=
0
;
pCols
->
numOf
Row
s
=
0
;
for
(
int
i
=
0
;
i
<
pCols
->
maxCols
;
i
++
)
{
dataColReset
(
pCols
->
cols
+
i
);
}
...
...
@@ -365,14 +365,14 @@ void tdAppendDataRowToDataCol(SDataRow row, SDataCols *pCols) {
SDataCol
*
pCol
=
pCols
->
cols
+
i
;
void
*
value
=
tdGetRowDataOfCol
(
row
,
pCol
->
type
,
pCol
->
offset
);
dataColAppendVal
(
pCol
,
value
,
pCols
->
numOf
Point
s
,
pCols
->
maxPoints
);
dataColAppendVal
(
pCol
,
value
,
pCols
->
numOf
Row
s
,
pCols
->
maxPoints
);
}
pCols
->
numOf
Point
s
++
;
pCols
->
numOf
Row
s
++
;
}
// Pop pointsToPop points from the SDataCols
void
tdPopDataColsPoints
(
SDataCols
*
pCols
,
int
pointsToPop
)
{
int
pointsLeft
=
pCols
->
numOf
Point
s
-
pointsToPop
;
int
pointsLeft
=
pCols
->
numOf
Row
s
-
pointsToPop
;
if
(
pointsLeft
<=
0
)
{
tdResetDataCols
(
pCols
);
return
;
...
...
@@ -380,14 +380,14 @@ void tdPopDataColsPoints(SDataCols *pCols, int pointsToPop) {
for
(
int
iCol
=
0
;
iCol
<
pCols
->
numOfCols
;
iCol
++
)
{
SDataCol
*
pCol
=
pCols
->
cols
+
iCol
;
dataColPopPoints
(
pCol
,
pointsToPop
,
pCols
->
numOf
Point
s
);
dataColPopPoints
(
pCol
,
pointsToPop
,
pCols
->
numOf
Row
s
);
}
pCols
->
numOf
Point
s
=
pointsLeft
;
pCols
->
numOf
Row
s
=
pointsLeft
;
}
int
tdMergeDataCols
(
SDataCols
*
target
,
SDataCols
*
source
,
int
rowsToMerge
)
{
ASSERT
(
rowsToMerge
>
0
&&
rowsToMerge
<=
source
->
numOf
Point
s
);
ASSERT
(
target
->
numOf
Point
s
+
rowsToMerge
<=
target
->
maxPoints
);
ASSERT
(
rowsToMerge
>
0
&&
rowsToMerge
<=
source
->
numOf
Row
s
);
ASSERT
(
target
->
numOf
Row
s
+
rowsToMerge
<=
target
->
maxPoints
);
ASSERT
(
target
->
numOfCols
==
source
->
numOfCols
);
SDataCols
*
pTarget
=
NULL
;
...
...
@@ -395,10 +395,10 @@ int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge) {
if
(
dataColsKeyLast
(
target
)
<
dataColsKeyFirst
(
source
))
{
// No overlap
for
(
int
i
=
0
;
i
<
rowsToMerge
;
i
++
)
{
for
(
int
j
=
0
;
j
<
source
->
numOfCols
;
j
++
)
{
dataColAppendVal
(
target
->
cols
+
j
,
tdGetColDataOfRow
(
source
->
cols
+
j
,
i
),
target
->
numOf
Point
s
,
dataColAppendVal
(
target
->
cols
+
j
,
tdGetColDataOfRow
(
source
->
cols
+
j
,
i
),
target
->
numOf
Row
s
,
target
->
maxPoints
);
}
target
->
numOf
Point
s
++
;
target
->
numOf
Row
s
++
;
}
}
else
{
pTarget
=
tdDupDataCols
(
target
,
true
);
...
...
@@ -406,7 +406,7 @@ int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge) {
int
iter1
=
0
;
int
iter2
=
0
;
tdMergeTwoDataCols
(
target
,
pTarget
,
&
iter1
,
source
,
&
iter2
,
pTarget
->
numOf
Point
s
+
rowsToMerge
);
tdMergeTwoDataCols
(
target
,
pTarget
,
&
iter1
,
source
,
&
iter2
,
pTarget
->
numOf
Row
s
+
rowsToMerge
);
}
tdFreeDataCols
(
pTarget
);
...
...
@@ -421,30 +421,30 @@ void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, SDataCol
// TODO: add resolve duplicate key here
tdResetDataCols
(
target
);
while
(
target
->
numOf
Point
s
<
tRows
)
{
if
(
*
iter1
>=
src1
->
numOf
Points
&&
*
iter2
>=
src2
->
numOfPoint
s
)
break
;
while
(
target
->
numOf
Row
s
<
tRows
)
{
if
(
*
iter1
>=
src1
->
numOf
Rows
&&
*
iter2
>=
src2
->
numOfRow
s
)
break
;
TSKEY
key1
=
(
*
iter1
>=
src1
->
numOf
Point
s
)
?
INT64_MAX
:
((
TSKEY
*
)(
src1
->
cols
[
0
].
pData
))[
*
iter1
];
TSKEY
key2
=
(
*
iter2
>=
src2
->
numOf
Point
s
)
?
INT64_MAX
:
((
TSKEY
*
)(
src2
->
cols
[
0
].
pData
))[
*
iter2
];
TSKEY
key1
=
(
*
iter1
>=
src1
->
numOf
Row
s
)
?
INT64_MAX
:
((
TSKEY
*
)(
src1
->
cols
[
0
].
pData
))[
*
iter1
];
TSKEY
key2
=
(
*
iter2
>=
src2
->
numOf
Row
s
)
?
INT64_MAX
:
((
TSKEY
*
)(
src2
->
cols
[
0
].
pData
))[
*
iter2
];
if
(
key1
<=
key2
)
{
for
(
int
i
=
0
;
i
<
src1
->
numOfCols
;
i
++
)
{
ASSERT
(
target
->
cols
[
i
].
type
==
src1
->
cols
[
i
].
type
);
dataColAppendVal
(
&
(
target
->
cols
[
i
]),
tdGetColDataOfRow
(
src1
->
cols
+
i
,
*
iter1
),
target
->
numOf
Point
s
,
dataColAppendVal
(
&
(
target
->
cols
[
i
]),
tdGetColDataOfRow
(
src1
->
cols
+
i
,
*
iter1
),
target
->
numOf
Row
s
,
target
->
maxPoints
);
}
target
->
numOf
Point
s
++
;
target
->
numOf
Row
s
++
;
(
*
iter1
)
++
;
if
(
key1
==
key2
)
(
*
iter2
)
++
;
}
else
{
for
(
int
i
=
0
;
i
<
src2
->
numOfCols
;
i
++
)
{
ASSERT
(
target
->
cols
[
i
].
type
==
src2
->
cols
[
i
].
type
);
dataColAppendVal
(
&
(
target
->
cols
[
i
]),
tdGetColDataOfRow
(
src2
->
cols
+
i
,
*
iter2
),
target
->
numOf
Point
s
,
dataColAppendVal
(
&
(
target
->
cols
[
i
]),
tdGetColDataOfRow
(
src2
->
cols
+
i
,
*
iter2
),
target
->
numOf
Row
s
,
target
->
maxPoints
);
}
target
->
numOf
Point
s
++
;
target
->
numOf
Row
s
++
;
(
*
iter2
)
++
;
}
}
...
...
src/common/src/ttypes.c
浏览文件 @
fdad406e
...
...
@@ -273,6 +273,46 @@ static void getStatics_d(const TSKEY *primaryKey, const void *pData, int32_t num
#endif
}
static
void
getStatics_bin
(
const
TSKEY
*
primaryKey
,
const
void
*
pData
,
int32_t
numOfRow
,
int64_t
*
min
,
int64_t
*
max
,
int64_t
*
sum
,
int16_t
*
minIndex
,
int16_t
*
maxIndex
,
int16_t
*
numOfNull
)
{
const
char
*
data
=
pData
;
ASSERT
(
numOfRow
<=
INT16_MAX
);
for
(
int32_t
i
=
0
;
i
<
numOfRow
;
++
i
)
{
if
(
isNull
((
const
char
*
)
varDataVal
(
data
),
TSDB_DATA_TYPE_BINARY
))
{
(
*
numOfNull
)
+=
1
;
}
data
+=
varDataLen
(
data
);
}
*
sum
=
0
;
*
max
=
0
;
*
min
=
0
;
*
minIndex
=
0
;
*
maxIndex
=
0
;
}
static
void
getStatics_nchr
(
const
TSKEY
*
primaryKey
,
const
void
*
pData
,
int32_t
numOfRow
,
int64_t
*
min
,
int64_t
*
max
,
int64_t
*
sum
,
int16_t
*
minIndex
,
int16_t
*
maxIndex
,
int16_t
*
numOfNull
)
{
const
char
*
data
=
pData
;
ASSERT
(
numOfRow
<=
INT16_MAX
);
for
(
int32_t
i
=
0
;
i
<
numOfRow
;
++
i
)
{
if
(
isNull
((
const
char
*
)
varDataVal
(
data
),
TSDB_DATA_TYPE_NCHAR
))
{
(
*
numOfNull
)
+=
1
;
}
data
+=
varDataLen
(
data
);
}
*
sum
=
0
;
*
max
=
0
;
*
min
=
0
;
*
minIndex
=
0
;
*
maxIndex
=
0
;
}
tDataTypeDescriptor
tDataTypeDesc
[
11
]
=
{
{
TSDB_DATA_TYPE_NULL
,
6
,
1
,
"NOTYPE"
,
NULL
,
NULL
,
NULL
},
{
TSDB_DATA_TYPE_BOOL
,
4
,
CHAR_BYTES
,
"BOOL"
,
tsCompressBool
,
tsDecompressBool
,
getStatics_i8
},
...
...
@@ -282,9 +322,9 @@ tDataTypeDescriptor tDataTypeDesc[11] = {
{
TSDB_DATA_TYPE_BIGINT
,
6
,
LONG_BYTES
,
"BIGINT"
,
tsCompressBigint
,
tsDecompressBigint
,
getStatics_i64
},
{
TSDB_DATA_TYPE_FLOAT
,
5
,
FLOAT_BYTES
,
"FLOAT"
,
tsCompressFloat
,
tsDecompressFloat
,
getStatics_f
},
{
TSDB_DATA_TYPE_DOUBLE
,
6
,
DOUBLE_BYTES
,
"DOUBLE"
,
tsCompressDouble
,
tsDecompressDouble
,
getStatics_d
},
{
TSDB_DATA_TYPE_BINARY
,
6
,
0
,
"BINARY"
,
tsCompressString
,
tsDecompressString
,
NULL
},
{
TSDB_DATA_TYPE_BINARY
,
6
,
0
,
"BINARY"
,
tsCompressString
,
tsDecompressString
,
getStatics_bin
},
{
TSDB_DATA_TYPE_TIMESTAMP
,
9
,
LONG_BYTES
,
"TIMESTAMP"
,
tsCompressTimestamp
,
tsDecompressTimestamp
,
getStatics_i64
},
{
TSDB_DATA_TYPE_NCHAR
,
5
,
8
,
"NCHAR"
,
tsCompressString
,
tsDecompressString
,
NULL
},
{
TSDB_DATA_TYPE_NCHAR
,
5
,
8
,
"NCHAR"
,
tsCompressString
,
tsDecompressString
,
getStatics_nchr
},
};
char
tTokenTypeSwitcher
[
13
]
=
{
...
...
src/query/src/qExecutor.c
浏览文件 @
fdad406e
...
...
@@ -200,7 +200,7 @@ bool getNeighborPoints(SQInfo *pQInfo, void *pMeterObj, SPointInterpoSupporter *
return false;
} else { // prev has been located
if (pQuery->fileId >= 0) {
pQuery->pos = pQuery->pBlock[pQuery->slot].numOf
Point
s - 1;
pQuery->pos = pQuery->pBlock[pQuery->slot].numOf
Row
s - 1;
getOneRowFromDataBlock(pRuntimeEnv, pPointInterpSupporter->pPrevPoint, pQuery->pos);
qTrace("QInfo:%p get prev data point, fileId:%d, slot:%d, pos:%d, pQuery->pos:%d", GET_QINFO_ADDR(pQuery),
...
...
@@ -210,11 +210,11 @@ bool getNeighborPoints(SQInfo *pQInfo, void *pMeterObj, SPointInterpoSupporter *
assert(vnodeIsDatablockLoaded(pRuntimeEnv, pMeterObj, -1, true) == DISK_BLOCK_NO_NEED_TO_LOAD);
pBlock = &pRuntimeEnv->cacheBlock;
pQuery->pos = pBlock->numOf
Point
s - 1;
pQuery->pos = pBlock->numOf
Row
s - 1;
getOneRowFromDataBlock(pRuntimeEnv, pPointInterpSupporter->pPrevPoint, pQuery->pos);
qTrace("QInfo:%p get prev data point, fileId:%d, slot:%d, pos:%d, pQuery->pos:%d", GET_QINFO_ADDR(pQuery),
pQuery->fileId, pQuery->slot, pBlock->numOf
Point
s - 1, pQuery->pos);
pQuery->fileId, pQuery->slot, pBlock->numOf
Row
s - 1, pQuery->pos);
}
}
}
...
...
@@ -603,9 +603,9 @@ static SWindowStatus *getTimeWindowResStatus(SWindowResInfo *pWindowResInfo, int
return
&
pWindowResInfo
->
pResult
[
slot
].
status
;
}
static
int32_t
getForwardStepsInBlock
(
int32_t
numOf
Point
s
,
__block_search_fn_t
searchFn
,
TSKEY
ekey
,
int16_t
pos
,
static
int32_t
getForwardStepsInBlock
(
int32_t
numOf
Row
s
,
__block_search_fn_t
searchFn
,
TSKEY
ekey
,
int16_t
pos
,
int16_t
order
,
int64_t
*
pData
)
{
int32_t
endPos
=
searchFn
((
char
*
)
pData
,
numOf
Point
s
,
ekey
,
order
);
int32_t
endPos
=
searchFn
((
char
*
)
pData
,
numOf
Row
s
,
ekey
,
order
);
int32_t
forwardStep
=
0
;
if
(
endPos
>=
0
)
{
...
...
@@ -2329,7 +2329,7 @@ SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle,
int32_t
binarySearchForKey
(
char
*
pValue
,
int
num
,
TSKEY
key
,
int
order
)
{
int32_t
midPos
=
-
1
;
int32_t
numOf
Point
s
;
int32_t
numOf
Row
s
;
if
(
num
<=
0
)
{
return
-
1
;
...
...
@@ -2348,8 +2348,8 @@ int32_t binarySearchForKey(char *pValue, int num, TSKEY key, int order) {
if
(
key
==
keyList
[
firstPos
])
return
firstPos
;
if
(
key
<
keyList
[
firstPos
])
return
firstPos
-
1
;
numOf
Point
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Point
s
>>
1
)
+
firstPos
;
numOf
Row
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Row
s
>>
1
)
+
firstPos
;
if
(
key
<
keyList
[
midPos
])
{
lastPos
=
midPos
-
1
;
...
...
@@ -2374,8 +2374,8 @@ int32_t binarySearchForKey(char *pValue, int num, TSKEY key, int order) {
return
lastPos
;
}
numOf
Point
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Point
s
>>
1
)
+
firstPos
;
numOf
Row
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Row
s
>>
1
)
+
firstPos
;
if
(
key
<
keyList
[
midPos
])
{
lastPos
=
midPos
-
1
;
...
...
src/tsdb/inc/tsdbMain.h
浏览文件 @
fdad406e
...
...
@@ -74,7 +74,7 @@ void tsdbCloseMetaFile(SMetaFile *mfh);
typedef
struct
{
TSKEY
keyFirst
;
TSKEY
keyLast
;
int32_t
numOf
Point
s
;
int32_t
numOf
Row
s
;
void
*
pData
;
}
SMemTable
;
...
...
@@ -173,7 +173,7 @@ typedef struct {
typedef
struct
{
TSKEY
keyFirst
;
TSKEY
keyLast
;
int64_t
numOf
Point
s
;
int64_t
numOf
Row
s
;
SList
*
list
;
}
SCacheMem
;
...
...
@@ -294,7 +294,7 @@ typedef struct {
int64_t
last
:
1
;
// If the block in data file or last file
int64_t
offset
:
63
;
// Offset of data block or sub-block index depending on numOfSubBlocks
int32_t
algorithm
:
8
;
// Compression algorithm
int32_t
numOf
Points
:
24
;
// Number of total points
int32_t
numOf
Rows
:
24
;
// Number of total points
int32_t
sversion
;
// Schema version
int32_t
len
;
// Data block length or nothing
int16_t
numOfSubBlocks
;
// Number of sub-blocks;
...
...
src/tsdb/src/tsdbCache.c
浏览文件 @
fdad406e
...
...
@@ -82,7 +82,7 @@ void *tsdbAllocFromCache(STsdbCache *pCache, int bytes, TSKEY key) {
memset
(
ptr
,
0
,
bytes
);
if
(
key
<
pCache
->
mem
->
keyFirst
)
pCache
->
mem
->
keyFirst
=
key
;
if
(
key
>
pCache
->
mem
->
keyLast
)
pCache
->
mem
->
keyLast
=
key
;
pCache
->
mem
->
numOf
Point
s
++
;
pCache
->
mem
->
numOf
Row
s
++
;
return
ptr
;
}
...
...
@@ -127,7 +127,7 @@ static int tsdbAllocBlockFromPool(STsdbCache *pCache) {
if
(
pCache
->
mem
==
NULL
)
return
-
1
;
pCache
->
mem
->
keyFirst
=
INT64_MAX
;
pCache
->
mem
->
keyLast
=
0
;
pCache
->
mem
->
numOf
Point
s
=
0
;
pCache
->
mem
->
numOf
Row
s
=
0
;
pCache
->
mem
->
list
=
tdListNew
(
sizeof
(
STsdbCacheBlock
*
));
}
...
...
src/tsdb/src/tsdbFile.c
浏览文件 @
fdad406e
...
...
@@ -233,10 +233,10 @@ SFileGroup *tsdbGetFileGroupNext(SFileGroupIter *pIter) {
// SCompBlock *pBlock = pStartBlock;
// for (int i = 0; i < numOfBlocks; i++) {
// if (tsdbLoadCompCols(pFile, pBlock, (void *)pCompData) < 0) return -1;
// pCols->numOf
Point
s += (pCompData->cols[0].len / 8);
// pCols->numOf
Row
s += (pCompData->cols[0].len / 8);
// for (int iCol = 0; iCol < pBlock->numOfCols; iCol++) {
// SCompCol *pCompCol = &(pCompData->cols[iCol]);
// // pCols->numOf
Points += pBlock->numOfPoint
s;
// // pCols->numOf
Rows += pBlock->numOfRow
s;
// int k = 0;
// for (; k < pCols->numOfCols; k++) {
// if (pCompCol->colId == pCols->cols[k].colId) break;
...
...
src/tsdb/src/tsdbMain.c
浏览文件 @
fdad406e
...
...
@@ -830,7 +830,7 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
tSkipListNewNodeInfo
(
pTable
->
mem
->
pData
,
&
level
,
&
headSize
);
TSKEY
key
=
dataRowKey
(
row
);
// printf("insert:%lld, size:%d\n", key, pTable->mem->numOf
Point
s);
// printf("insert:%lld, size:%d\n", key, pTable->mem->numOf
Row
s);
// Copy row into the memory
SSkipListNode
*
pNode
=
tsdbAllocFromCache
(
pRepo
->
tsdbCache
,
headSize
+
dataRowLen
(
row
),
key
);
...
...
@@ -854,7 +854,7 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
if
(
key
<
pTable
->
mem
->
keyFirst
)
pTable
->
mem
->
keyFirst
=
key
;
if
(
key
>
pTable
->
lastKey
)
pTable
->
lastKey
=
key
;
pTable
->
mem
->
numOf
Point
s
=
tSkipListGetSize
(
pTable
->
mem
->
pData
);
pTable
->
mem
->
numOf
Row
s
=
tSkipListGetSize
(
pTable
->
mem
->
pData
);
tsdbTrace
(
"vgId:%d, tid:%d, uid:%"
PRId64
", table:%s a row is inserted to table! key:%"
PRId64
,
pRepo
->
config
.
tsdbId
,
pTable
->
tableId
.
tid
,
pTable
->
tableId
.
uid
,
varDataVal
(
pTable
->
name
),
dataRowKey
(
row
));
...
...
@@ -1063,7 +1063,7 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SSkipListIterator **iters
while
(
true
)
{
int
rowsRead
=
tsdbReadRowsFromCache
(
pIter
,
maxKey
,
maxRowsToRead
,
pDataCols
);
assert
(
rowsRead
>=
0
);
if
(
pDataCols
->
numOf
Point
s
==
0
)
break
;
if
(
pDataCols
->
numOf
Row
s
==
0
)
break
;
nLoop
++
;
ASSERT
(
dataColsKeyFirst
(
pDataCols
)
>=
minKey
&&
dataColsKeyFirst
(
pDataCols
)
<=
maxKey
);
...
...
@@ -1072,13 +1072,13 @@ static int tsdbCommitToFile(STsdbRepo *pRepo, int fid, SSkipListIterator **iters
int
rowsWritten
=
tsdbWriteDataBlock
(
pHelper
,
pDataCols
);
ASSERT
(
rowsWritten
!=
0
);
if
(
rowsWritten
<
0
)
goto
_err
;
ASSERT
(
rowsWritten
<=
pDataCols
->
numOf
Point
s
);
ASSERT
(
rowsWritten
<=
pDataCols
->
numOf
Row
s
);
tdPopDataColsPoints
(
pDataCols
,
rowsWritten
);
maxRowsToRead
=
pCfg
->
maxRowsPerFileBlock
*
4
/
5
-
pDataCols
->
numOf
Point
s
;
maxRowsToRead
=
pCfg
->
maxRowsPerFileBlock
*
4
/
5
-
pDataCols
->
numOf
Row
s
;
}
ASSERT
(
pDataCols
->
numOf
Point
s
==
0
);
ASSERT
(
pDataCols
->
numOf
Row
s
==
0
);
// Move the last block to the new .l file if neccessary
if
(
tsdbMoveLastBlockIfNeccessary
(
pHelper
)
<
0
)
{
...
...
src/tsdb/src/tsdbRWHelper.c
浏览文件 @
fdad406e
...
...
@@ -307,7 +307,7 @@ void tsdbSetHelperTable(SRWHelper *pHelper, STable *pTable, STsdbRepo *pRepo) {
*/
int
tsdbWriteDataBlock
(
SRWHelper
*
pHelper
,
SDataCols
*
pDataCols
)
{
ASSERT
(
TSDB_HELPER_TYPE
(
pHelper
)
==
TSDB_WRITE_HELPER
);
ASSERT
(
pDataCols
->
numOf
Point
s
>
0
);
ASSERT
(
pDataCols
->
numOf
Row
s
>
0
);
SCompBlock
compBlock
;
int
rowsToWrite
=
0
;
...
...
@@ -322,7 +322,7 @@ int tsdbWriteDataBlock(SRWHelper *pHelper, SDataCols *pDataCols) {
if
(
pIdx
->
offset
==
0
||
(
!
pIdx
->
hasLast
&&
keyFirst
>
pIdx
->
maxKey
))
{
// Just append as a super block
ASSERT
(
pHelper
->
hasOldLastBlock
==
false
);
rowsToWrite
=
pDataCols
->
numOf
Point
s
;
rowsToWrite
=
pDataCols
->
numOf
Row
s
;
SFile
*
pWFile
=
NULL
;
bool
isLast
=
false
;
...
...
@@ -380,10 +380,10 @@ int tsdbMoveLastBlockIfNeccessary(SRWHelper *pHelper) {
if
(
pCompBlock
->
numOfSubBlocks
>
1
)
{
if
(
tsdbLoadBlockData
(
pHelper
,
blockAtIdx
(
pHelper
,
pIdx
->
numOfBlocks
-
1
),
NULL
)
<
0
)
return
-
1
;
ASSERT
(
pHelper
->
pDataCols
[
0
]
->
numOf
Point
s
>
0
&&
pHelper
->
pDataCols
[
0
]
->
numOf
Point
s
<
pHelper
->
config
.
minRowsPerFileBlock
);
ASSERT
(
pHelper
->
pDataCols
[
0
]
->
numOf
Row
s
>
0
&&
pHelper
->
pDataCols
[
0
]
->
numOf
Row
s
<
pHelper
->
config
.
minRowsPerFileBlock
);
if
(
tsdbWriteBlockToFile
(
pHelper
,
&
(
pHelper
->
files
.
nLastF
),
pHelper
->
pDataCols
[
0
],
pHelper
->
pDataCols
[
0
]
->
numOf
Point
s
,
&
compBlock
,
true
,
true
)
<
0
)
pHelper
->
pDataCols
[
0
]
->
numOf
Row
s
,
&
compBlock
,
true
,
true
)
<
0
)
return
-
1
;
if
(
tsdbUpdateSuperBlock
(
pHelper
,
&
compBlock
,
pIdx
->
numOfBlocks
-
1
)
<
0
)
return
-
1
;
...
...
@@ -625,13 +625,13 @@ int tsdbLoadBlockDataCols(SRWHelper *pHelper, SDataCols *pDataCols, int blkIdx,
for
(
int
i
=
1
;
i
<
numOfSubBlocks
;
i
++
)
{
pStartBlock
++
;
if
(
tsdbLoadSingleBlockDataCols
(
pHelper
,
pStartBlock
,
colIds
,
numOfColIds
,
pHelper
->
pDataCols
[
1
])
<
0
)
return
-
1
;
tdMergeDataCols
(
pDataCols
,
pHelper
->
pDataCols
[
1
],
pHelper
->
pDataCols
[
1
]
->
numOf
Point
s
);
tdMergeDataCols
(
pDataCols
,
pHelper
->
pDataCols
[
1
],
pHelper
->
pDataCols
[
1
]
->
numOf
Row
s
);
}
return
0
;
}
static
int
tsdbCheckAndDecodeColumnData
(
SDataCol
*
pDataCol
,
char
*
content
,
int32_t
len
,
int8_t
comp
,
int
numOf
Point
s
,
static
int
tsdbCheckAndDecodeColumnData
(
SDataCol
*
pDataCol
,
char
*
content
,
int32_t
len
,
int8_t
comp
,
int
numOf
Row
s
,
int
maxPoints
,
char
*
buffer
,
int
bufferSize
)
{
// Verify by checksum
if
(
!
taosCheckChecksumWhole
((
uint8_t
*
)
content
,
len
))
return
-
1
;
...
...
@@ -640,16 +640,16 @@ static int tsdbCheckAndDecodeColumnData(SDataCol *pDataCol, char *content, int32
if
(
comp
)
{
// // Need to decompress
pDataCol
->
len
=
(
*
(
tDataTypeDesc
[
pDataCol
->
type
].
decompFunc
))(
content
,
len
-
sizeof
(
TSCKSUM
),
numOf
Point
s
,
pDataCol
->
pData
,
pDataCol
->
spaceSize
,
comp
,
buffer
,
bufferSize
);
content
,
len
-
sizeof
(
TSCKSUM
),
numOf
Row
s
,
pDataCol
->
pData
,
pDataCol
->
spaceSize
,
comp
,
buffer
,
bufferSize
);
if
(
pDataCol
->
type
==
TSDB_DATA_TYPE_BINARY
||
pDataCol
->
type
==
TSDB_DATA_TYPE_NCHAR
)
{
dataColSetOffset
(
pDataCol
,
numOf
Point
s
);
dataColSetOffset
(
pDataCol
,
numOf
Row
s
);
}
}
else
{
// No need to decompress, just memcpy it
pDataCol
->
len
=
len
-
sizeof
(
TSCKSUM
);
memcpy
(
pDataCol
->
pData
,
content
,
pDataCol
->
len
);
if
(
pDataCol
->
type
==
TSDB_DATA_TYPE_BINARY
||
pDataCol
->
type
==
TSDB_DATA_TYPE_NCHAR
)
{
dataColSetOffset
(
pDataCol
,
numOf
Point
s
);
dataColSetOffset
(
pDataCol
,
numOf
Row
s
);
}
}
return
0
;
...
...
@@ -673,7 +673,7 @@ static int tsdbLoadBlockDataImpl(SRWHelper *pHelper, SCompBlock *pCompBlock, SDa
int32_t
tsize
=
sizeof
(
SCompData
)
+
sizeof
(
SCompCol
)
*
pCompBlock
->
numOfCols
+
sizeof
(
TSCKSUM
);
if
(
!
taosCheckChecksumWhole
((
uint8_t
*
)
pCompData
,
tsize
))
goto
_err
;
pDataCols
->
numOf
Points
=
pCompBlock
->
numOfPoint
s
;
pDataCols
->
numOf
Rows
=
pCompBlock
->
numOfRow
s
;
// Recover the data
int
ccol
=
0
;
...
...
@@ -682,7 +682,7 @@ static int tsdbLoadBlockDataImpl(SRWHelper *pHelper, SCompBlock *pCompBlock, SDa
SDataCol
*
pDataCol
=
&
(
pDataCols
->
cols
[
dcol
]);
if
(
ccol
>=
pCompData
->
numOfCols
)
{
// Set current column as NULL and forward
dataColSetNEleNull
(
pDataCol
,
pCompBlock
->
numOf
Point
s
,
pDataCols
->
maxPoints
);
dataColSetNEleNull
(
pDataCol
,
pCompBlock
->
numOf
Row
s
,
pDataCols
->
maxPoints
);
dcol
++
;
continue
;
}
...
...
@@ -691,15 +691,15 @@ static int tsdbLoadBlockDataImpl(SRWHelper *pHelper, SCompBlock *pCompBlock, SDa
if
(
pCompCol
->
colId
==
pDataCol
->
colId
)
{
if
(
pCompBlock
->
algorithm
==
TWO_STAGE_COMP
)
{
int
zsize
=
pDataCol
->
bytes
*
pCompBlock
->
numOf
Point
s
+
COMP_OVERFLOW_BYTES
;
int
zsize
=
pDataCol
->
bytes
*
pCompBlock
->
numOf
Row
s
+
COMP_OVERFLOW_BYTES
;
if
(
pCompCol
->
type
==
TSDB_DATA_TYPE_BINARY
||
pCompCol
->
type
==
TSDB_DATA_TYPE_NCHAR
)
{
zsize
+=
(
sizeof
(
VarDataLenT
)
*
pCompBlock
->
numOf
Point
s
);
zsize
+=
(
sizeof
(
VarDataLenT
)
*
pCompBlock
->
numOf
Row
s
);
}
pHelper
->
compBuffer
=
trealloc
(
pHelper
->
compBuffer
,
zsize
);
if
(
pHelper
->
compBuffer
==
NULL
)
goto
_err
;
}
if
(
tsdbCheckAndDecodeColumnData
(
pDataCol
,
(
char
*
)
pCompData
+
tsize
+
pCompCol
->
offset
,
pCompCol
->
len
,
pCompBlock
->
algorithm
,
pCompBlock
->
numOf
Point
s
,
pDataCols
->
maxPoints
,
pCompBlock
->
algorithm
,
pCompBlock
->
numOf
Row
s
,
pDataCols
->
maxPoints
,
pHelper
->
compBuffer
,
tsizeof
(
pHelper
->
compBuffer
))
<
0
)
goto
_err
;
dcol
++
;
...
...
@@ -708,7 +708,7 @@ static int tsdbLoadBlockDataImpl(SRWHelper *pHelper, SCompBlock *pCompBlock, SDa
ccol
++
;
}
else
{
// Set current column as NULL and forward
dataColSetNEleNull
(
pDataCol
,
pCompBlock
->
numOf
Point
s
,
pDataCols
->
maxPoints
);
dataColSetNEleNull
(
pDataCol
,
pCompBlock
->
numOf
Row
s
,
pDataCols
->
maxPoints
);
dcol
++
;
}
}
...
...
@@ -732,7 +732,7 @@ int tsdbLoadBlockData(SRWHelper *pHelper, SCompBlock *pCompBlock, SDataCols *tar
tdResetDataCols
(
pHelper
->
pDataCols
[
1
]);
pCompBlock
++
;
if
(
tsdbLoadBlockDataImpl
(
pHelper
,
pCompBlock
,
pHelper
->
pDataCols
[
1
])
<
0
)
goto
_err
;
if
(
tdMergeDataCols
(
pHelper
->
pDataCols
[
0
],
pHelper
->
pDataCols
[
1
],
pHelper
->
pDataCols
[
1
]
->
numOf
Point
s
)
<
0
)
goto
_err
;
if
(
tdMergeDataCols
(
pHelper
->
pDataCols
[
0
],
pHelper
->
pDataCols
[
1
],
pHelper
->
pDataCols
[
1
]
->
numOf
Row
s
)
<
0
)
goto
_err
;
}
// if (target) TODO
...
...
@@ -753,7 +753,7 @@ static bool tsdbShouldCreateNewLast(SRWHelper *pHelper) {
static
int
tsdbWriteBlockToFile
(
SRWHelper
*
pHelper
,
SFile
*
pFile
,
SDataCols
*
pDataCols
,
int
rowsToWrite
,
SCompBlock
*
pCompBlock
,
bool
isLast
,
bool
isSuperBlock
)
{
ASSERT
(
rowsToWrite
>
0
&&
rowsToWrite
<=
pDataCols
->
numOf
Point
s
&&
ASSERT
(
rowsToWrite
>
0
&&
rowsToWrite
<=
pDataCols
->
numOf
Row
s
&&
rowsToWrite
<=
pHelper
->
config
.
maxRowsPerFileBlock
);
SCompData
*
pCompData
=
(
SCompData
*
)(
pHelper
->
pBuffer
);
...
...
@@ -840,7 +840,7 @@ static int tsdbWriteBlockToFile(SRWHelper *pHelper, SFile *pFile, SDataCols *pDa
pCompBlock
->
last
=
isLast
;
pCompBlock
->
offset
=
offset
;
pCompBlock
->
algorithm
=
pHelper
->
config
.
compress
;
pCompBlock
->
numOf
Point
s
=
rowsToWrite
;
pCompBlock
->
numOf
Row
s
=
rowsToWrite
;
pCompBlock
->
sversion
=
pHelper
->
tableInfo
.
sversion
;
pCompBlock
->
len
=
(
int32_t
)
lsize
;
pCompBlock
->
numOfSubBlocks
=
isSuperBlock
?
1
:
0
;
...
...
@@ -877,7 +877,7 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
int
rowsWritten
=
0
;
SCompBlock
compBlock
=
{
0
};
ASSERT
(
pDataCols
->
numOf
Point
s
>
0
);
ASSERT
(
pDataCols
->
numOf
Row
s
>
0
);
TSKEY
keyFirst
=
dataColsKeyFirst
(
pDataCols
);
SCompIdx
*
pIdx
=
pHelper
->
pCompIdx
+
pHelper
->
tableInfo
.
tid
;
...
...
@@ -889,32 +889,32 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
// ASSERT(compareKeyBlock((void *)&keyFirst, (void *)pCompBlock) == 0);
if
(
keyFirst
>
blockAtIdx
(
pHelper
,
blkIdx
)
->
keyLast
)
{
// Merge with the last block by append
ASSERT
(
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Point
s
<
pHelper
->
config
.
minRowsPerFileBlock
&&
blkIdx
==
pIdx
->
numOfBlocks
-
1
);
ASSERT
(
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Row
s
<
pHelper
->
config
.
minRowsPerFileBlock
&&
blkIdx
==
pIdx
->
numOfBlocks
-
1
);
int
defaultRowsToWrite
=
pHelper
->
config
.
maxRowsPerFileBlock
*
4
/
5
;
// TODO: make a interface
rowsWritten
=
MIN
((
defaultRowsToWrite
-
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Points
),
pDataCols
->
numOfPoint
s
);
rowsWritten
=
MIN
((
defaultRowsToWrite
-
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Rows
),
pDataCols
->
numOfRow
s
);
if
((
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOfSubBlocks
<
TSDB_MAX_SUBBLOCKS
)
&&
(
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Point
s
+
rowsWritten
<
pHelper
->
config
.
minRowsPerFileBlock
)
&&
(
pHelper
->
files
.
nLastF
.
fd
)
>
0
)
{
(
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Row
s
+
rowsWritten
<
pHelper
->
config
.
minRowsPerFileBlock
)
&&
(
pHelper
->
files
.
nLastF
.
fd
)
>
0
)
{
if
(
tsdbWriteBlockToFile
(
pHelper
,
&
(
pHelper
->
files
.
lastF
),
pDataCols
,
rowsWritten
,
&
compBlock
,
true
,
false
)
<
0
)
goto
_err
;
if
(
tsdbAddSubBlock
(
pHelper
,
&
compBlock
,
blkIdx
,
rowsWritten
)
<
0
)
goto
_err
;
}
else
{
// Load
if
(
tsdbLoadBlockData
(
pHelper
,
blockAtIdx
(
pHelper
,
blkIdx
),
NULL
)
<
0
)
goto
_err
;
ASSERT
(
pHelper
->
pDataCols
[
0
]
->
numOf
Points
==
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOfPoint
s
);
ASSERT
(
pHelper
->
pDataCols
[
0
]
->
numOf
Rows
==
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOfRow
s
);
// Merge
if
(
tdMergeDataCols
(
pHelper
->
pDataCols
[
0
],
pDataCols
,
rowsWritten
)
<
0
)
goto
_err
;
// Write
SFile
*
pWFile
=
NULL
;
bool
isLast
=
false
;
if
(
pHelper
->
pDataCols
[
0
]
->
numOf
Point
s
>=
pHelper
->
config
.
minRowsPerFileBlock
)
{
if
(
pHelper
->
pDataCols
[
0
]
->
numOf
Row
s
>=
pHelper
->
config
.
minRowsPerFileBlock
)
{
pWFile
=
&
(
pHelper
->
files
.
dataF
);
}
else
{
isLast
=
true
;
pWFile
=
(
pHelper
->
files
.
nLastF
.
fd
>
0
)
?
&
(
pHelper
->
files
.
nLastF
)
:
&
(
pHelper
->
files
.
lastF
);
}
if
(
tsdbWriteBlockToFile
(
pHelper
,
pWFile
,
pHelper
->
pDataCols
[
0
],
pHelper
->
pDataCols
[
0
]
->
numOf
Point
s
,
&
compBlock
,
isLast
,
true
)
<
0
)
pHelper
->
pDataCols
[
0
]
->
numOf
Row
s
,
&
compBlock
,
isLast
,
true
)
<
0
)
goto
_err
;
if
(
tsdbUpdateSuperBlock
(
pHelper
,
&
compBlock
,
blkIdx
)
<
0
)
goto
_err
;
}
...
...
@@ -931,7 +931,7 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
// rows1: number of rows must merge in this block
int
rows1
=
tsdbGetRowsInRange
(
pDataCols
,
blockAtIdx
(
pHelper
,
blkIdx
)
->
keyFirst
,
blockAtIdx
(
pHelper
,
blkIdx
)
->
keyLast
);
// rows2: max nuber of rows the block can have more
int
rows2
=
pHelper
->
config
.
maxRowsPerFileBlock
-
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Point
s
;
int
rows2
=
pHelper
->
config
.
maxRowsPerFileBlock
-
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Row
s
;
// rows3: number of rows between this block and the next block
int
rows3
=
tsdbGetRowsInRange
(
pDataCols
,
blockAtIdx
(
pHelper
,
blkIdx
)
->
keyFirst
,
keyLimit
);
...
...
@@ -939,7 +939,7 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
if
((
rows2
>=
rows1
)
&&
((
blockAtIdx
(
pHelper
,
blkIdx
)
->
last
)
||
((
rows1
+
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Point
s
<
pHelper
->
config
.
minRowsPerFileBlock
)
&&
(
pHelper
->
files
.
nLastF
.
fd
<
0
))))
{
((
rows1
+
blockAtIdx
(
pHelper
,
blkIdx
)
->
numOf
Row
s
<
pHelper
->
config
.
minRowsPerFileBlock
)
&&
(
pHelper
->
files
.
nLastF
.
fd
<
0
))))
{
rowsWritten
=
rows1
;
bool
isLast
=
false
;
SFile
*
pFile
=
NULL
;
...
...
@@ -965,11 +965,11 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
int
round
=
0
;
// tdResetDataCols(pHelper->pDataCols[1]);
while
(
true
)
{
if
(
iter1
>=
pHelper
->
pDataCols
[
0
]
->
numOf
Point
s
&&
iter2
>=
rows3
)
break
;
if
(
iter1
>=
pHelper
->
pDataCols
[
0
]
->
numOf
Row
s
&&
iter2
>=
rows3
)
break
;
tdMergeTwoDataCols
(
pHelper
->
pDataCols
[
1
],
pHelper
->
pDataCols
[
0
],
&
iter1
,
pDataCols
,
&
iter2
,
pHelper
->
config
.
maxRowsPerFileBlock
*
4
/
5
);
ASSERT
(
pHelper
->
pDataCols
[
1
]
->
numOf
Point
s
>
0
);
ASSERT
(
pHelper
->
pDataCols
[
1
]
->
numOf
Row
s
>
0
);
if
(
tsdbWriteBlockToFile
(
pHelper
,
&
(
pHelper
->
files
.
dataF
),
pHelper
->
pDataCols
[
1
],
pHelper
->
pDataCols
[
1
]
->
numOf
Point
s
,
&
compBlock
,
false
,
true
)
<
0
)
pHelper
->
pDataCols
[
1
]
->
numOf
Row
s
,
&
compBlock
,
false
,
true
)
<
0
)
goto
_err
;
if
(
round
==
0
)
{
tsdbUpdateSuperBlock
(
pHelper
,
&
compBlock
,
blkIdx
);
...
...
@@ -980,17 +980,17 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
blkIdx
++
;
// TODO: the blkIdx here is not correct
// if (iter1 >= pHelper->pDataCols[0]->numOf
Point
s && iter2 >= rows3) {
// if (pHelper->pDataCols[1]->numOf
Point
s > 0) {
// if (iter1 >= pHelper->pDataCols[0]->numOf
Row
s && iter2 >= rows3) {
// if (pHelper->pDataCols[1]->numOf
Row
s > 0) {
// if (tsdbWriteBlockToFile(pHelper, &pHelper->files.dataF, pHelper->pDataCols[1],
// pHelper->pDataCols[1]->numOf
Point
s, &compBlock, false, true) < 0)
// pHelper->pDataCols[1]->numOf
Row
s, &compBlock, false, true) < 0)
// goto _err;
// // TODO: the blkIdx here is not correct
// tsdbAddSubBlock(pHelper, &compBlock, blkIdx, pHelper->pDataCols[1]->numOf
Point
s);
// tsdbAddSubBlock(pHelper, &compBlock, blkIdx, pHelper->pDataCols[1]->numOf
Row
s);
// }
// }
// TSKEY key1 = iter1 >= pHelper->pDataCols[0]->numOf
Point
s
// TSKEY key1 = iter1 >= pHelper->pDataCols[0]->numOf
Row
s
// ? INT64_MAX
// : ((int64_t *)(pHelper->pDataCols[0]->cols[0].pData))[iter1];
// TSKEY key2 = iter2 >= rowsWritten ? INT64_MAX : ((int64_t *)(pDataCols->cols[0].pData))[iter2];
...
...
@@ -998,11 +998,11 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
// if (key1 < key2) {
// for (int i = 0; i < pDataCols->numOfCols; i++) {
// SDataCol *pDataCol = pHelper->pDataCols[1]->cols + i;
// memcpy(((char *)pDataCol->pData + TYPE_BYTES[pDataCol->type] * pHelper->pDataCols[1]->numOf
Point
s),
// memcpy(((char *)pDataCol->pData + TYPE_BYTES[pDataCol->type] * pHelper->pDataCols[1]->numOf
Row
s),
// ((char *)pHelper->pDataCols[0]->cols[i].pData + TYPE_BYTES[pDataCol->type] * iter1),
// TYPE_BYTES[pDataCol->type]);
// }
// pHelper->pDataCols[1]->numOf
Point
s++;
// pHelper->pDataCols[1]->numOf
Row
s++;
// iter1++;
// } else if (key1 == key2) {
// // TODO: think about duplicate key cases
...
...
@@ -1010,17 +1010,17 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
// } else {
// for (int i = 0; i < pDataCols->numOfCols; i++) {
// SDataCol *pDataCol = pHelper->pDataCols[1]->cols + i;
// memcpy(((char *)pDataCol->pData + TYPE_BYTES[pDataCol->type] * pHelper->pDataCols[1]->numOf
Point
s),
// memcpy(((char *)pDataCol->pData + TYPE_BYTES[pDataCol->type] * pHelper->pDataCols[1]->numOf
Row
s),
// ((char *)pDataCols->cols[i].pData +
// TYPE_BYTES[pDataCol->type] * iter2),
// TYPE_BYTES[pDataCol->type]);
// }
// pHelper->pDataCols[1]->numOf
Point
s++;
// pHelper->pDataCols[1]->numOf
Row
s++;
// iter2++;
// }
// if (pHelper->pDataCols[0]->numOf
Point
s >= pHelper->config.maxRowsPerFileBlock * 4 / 5) {
// if (tsdbWriteBlockToFile(pHelper, &pHelper->files.dataF, pHelper->pDataCols[1], pHelper->pDataCols[1]->numOf
Point
s, &compBlock, false, true) < 0) goto _err;
// if (pHelper->pDataCols[0]->numOf
Row
s >= pHelper->config.maxRowsPerFileBlock * 4 / 5) {
// if (tsdbWriteBlockToFile(pHelper, &pHelper->files.dataF, pHelper->pDataCols[1], pHelper->pDataCols[1]->numOf
Row
s, &compBlock, false, true) < 0) goto _err;
// // TODO: blkIdx here is not correct, fix it
// tsdbInsertSuperBlock(pHelper, &compBlock, blkIdx);
...
...
@@ -1133,7 +1133,7 @@ static int tsdbAddSubBlock(SRWHelper *pHelper, SCompBlock *pCompBlock, int blkId
pSCompBlock
->
numOfSubBlocks
++
;
ASSERT
(
pSCompBlock
->
numOfSubBlocks
<=
TSDB_MAX_SUBBLOCKS
);
pSCompBlock
->
len
+=
sizeof
(
SCompBlock
);
pSCompBlock
->
numOf
Point
s
+=
rowsAdded
;
pSCompBlock
->
numOf
Row
s
+=
rowsAdded
;
pSCompBlock
->
keyFirst
=
MIN
(
pSCompBlock
->
keyFirst
,
pCompBlock
->
keyFirst
);
pSCompBlock
->
keyLast
=
MAX
(
pSCompBlock
->
keyLast
,
pCompBlock
->
keyLast
);
pIdx
->
len
+=
sizeof
(
SCompBlock
);
...
...
@@ -1164,7 +1164,7 @@ static int tsdbAddSubBlock(SRWHelper *pHelper, SCompBlock *pCompBlock, int blkId
((
SCompBlock
*
)
ptr
)[
1
]
=
*
pCompBlock
;
pSCompBlock
->
numOfSubBlocks
=
2
;
pSCompBlock
->
numOf
Point
s
+=
rowsAdded
;
pSCompBlock
->
numOf
Row
s
+=
rowsAdded
;
pSCompBlock
->
offset
=
((
char
*
)
ptr
)
-
((
char
*
)
pHelper
->
pCompInfo
);
pSCompBlock
->
len
=
sizeof
(
SCompBlock
)
*
2
;
pSCompBlock
->
keyFirst
=
MIN
(((
SCompBlock
*
)
ptr
)[
0
].
keyFirst
,
((
SCompBlock
*
)
ptr
)[
1
].
keyFirst
);
...
...
@@ -1219,7 +1219,7 @@ static int tsdbUpdateSuperBlock(SRWHelper *pHelper, SCompBlock *pCompBlock, int
// Get the number of rows in range [minKey, maxKey]
static
int
tsdbGetRowsInRange
(
SDataCols
*
pDataCols
,
TSKEY
minKey
,
TSKEY
maxKey
)
{
if
(
pDataCols
->
numOf
Point
s
==
0
)
return
0
;
if
(
pDataCols
->
numOf
Row
s
==
0
)
return
0
;
ASSERT
(
minKey
<=
maxKey
);
TSKEY
keyFirst
=
dataColsKeyFirst
(
pDataCols
);
...
...
@@ -1228,11 +1228,11 @@ static int tsdbGetRowsInRange(SDataCols *pDataCols, TSKEY minKey, TSKEY maxKey)
if
(
minKey
>
keyLast
||
maxKey
<
keyFirst
)
return
0
;
void
*
ptr1
=
taosbsearch
((
void
*
)
&
minKey
,
(
void
*
)
pDataCols
->
cols
[
0
].
pData
,
pDataCols
->
numOf
Point
s
,
sizeof
(
TSKEY
),
void
*
ptr1
=
taosbsearch
((
void
*
)
&
minKey
,
(
void
*
)
pDataCols
->
cols
[
0
].
pData
,
pDataCols
->
numOf
Row
s
,
sizeof
(
TSKEY
),
compTSKEY
,
TD_GE
);
ASSERT
(
ptr1
!=
NULL
);
void
*
ptr2
=
taosbsearch
((
void
*
)
&
maxKey
,
(
void
*
)
pDataCols
->
cols
[
0
].
pData
,
pDataCols
->
numOf
Point
s
,
sizeof
(
TSKEY
),
void
*
ptr2
=
taosbsearch
((
void
*
)
&
maxKey
,
(
void
*
)
pDataCols
->
cols
[
0
].
pData
,
pDataCols
->
numOf
Row
s
,
sizeof
(
TSKEY
),
compTSKEY
,
TD_LE
);
ASSERT
(
ptr2
!=
NULL
);
...
...
src/tsdb/src/tsdbRead.c
浏览文件 @
fdad406e
...
...
@@ -436,7 +436,7 @@ static SDataBlockInfo getTrueDataBlockInfo(STableCheckInfo* pCheckInfo, SCompBlo
SDataBlockInfo
info
=
{
.
window
=
{.
skey
=
pBlock
->
keyFirst
,
.
ekey
=
pBlock
->
keyLast
},
.
numOfCols
=
pBlock
->
numOfCols
,
.
rows
=
pBlock
->
numOf
Point
s
,
.
rows
=
pBlock
->
numOf
Row
s
,
.
tid
=
pCheckInfo
->
tableId
.
tid
,
.
uid
=
pCheckInfo
->
tableId
.
uid
,
};
...
...
@@ -608,11 +608,11 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
}
SDataCols
*
pTSCol
=
pQueryHandle
->
rhelper
.
pDataCols
[
0
];
assert
(
pTSCol
->
cols
->
type
==
TSDB_DATA_TYPE_TIMESTAMP
&&
pTSCol
->
numOf
Points
==
pBlock
->
numOfPoint
s
);
assert
(
pTSCol
->
cols
->
type
==
TSDB_DATA_TYPE_TIMESTAMP
&&
pTSCol
->
numOf
Rows
==
pBlock
->
numOfRow
s
);
if
(
pCheckInfo
->
lastKey
>
pBlock
->
keyFirst
)
{
cur
->
pos
=
binarySearchForKey
(
pTSCol
->
cols
[
0
].
pData
,
pBlock
->
numOf
Point
s
,
pCheckInfo
->
lastKey
,
pQueryHandle
->
order
);
binarySearchForKey
(
pTSCol
->
cols
[
0
].
pData
,
pBlock
->
numOf
Row
s
,
pCheckInfo
->
lastKey
,
pQueryHandle
->
order
);
}
else
{
cur
->
pos
=
0
;
}
...
...
@@ -630,9 +630,9 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
SDataCols
*
pDataCols
=
pCheckInfo
->
pDataCols
;
if
(
pCheckInfo
->
lastKey
<
pBlock
->
keyLast
)
{
cur
->
pos
=
binarySearchForKey
(
pDataCols
->
cols
[
0
].
pData
,
pBlock
->
numOf
Point
s
,
pCheckInfo
->
lastKey
,
pQueryHandle
->
order
);
binarySearchForKey
(
pDataCols
->
cols
[
0
].
pData
,
pBlock
->
numOf
Row
s
,
pCheckInfo
->
lastKey
,
pQueryHandle
->
order
);
}
else
{
cur
->
pos
=
pBlock
->
numOf
Point
s
-
1
;
cur
->
pos
=
pBlock
->
numOf
Row
s
-
1
;
}
doMergeTwoLevelData
(
pQueryHandle
,
pCheckInfo
,
pBlock
,
sa
);
...
...
@@ -647,7 +647,7 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
static
int
vnodeBinarySearchKey
(
char
*
pValue
,
int
num
,
TSKEY
key
,
int
order
)
{
int
firstPos
,
lastPos
,
midPos
=
-
1
;
int
numOf
Point
s
;
int
numOf
Row
s
;
TSKEY
*
keyList
;
assert
(
order
==
TSDB_ORDER_ASC
||
order
==
TSDB_ORDER_DESC
);
...
...
@@ -665,8 +665,8 @@ static int vnodeBinarySearchKey(char* pValue, int num, TSKEY key, int order) {
if
(
key
==
keyList
[
firstPos
])
return
firstPos
;
if
(
key
<
keyList
[
firstPos
])
return
firstPos
-
1
;
numOf
Point
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Point
s
>>
1
)
+
firstPos
;
numOf
Row
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Row
s
>>
1
)
+
firstPos
;
if
(
key
<
keyList
[
midPos
])
{
lastPos
=
midPos
-
1
;
...
...
@@ -691,8 +691,8 @@ static int vnodeBinarySearchKey(char* pValue, int num, TSKEY key, int order) {
return
lastPos
;
}
numOf
Point
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Point
s
>>
1
)
+
firstPos
;
numOf
Row
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Row
s
>>
1
)
+
firstPos
;
if
(
key
<
keyList
[
midPos
])
{
lastPos
=
midPos
-
1
;
...
...
@@ -810,7 +810,7 @@ static void doMergeTwoLevelData(STsdbQueryHandle* pQueryHandle, STableCheckInfo*
cur
->
mixBlock
=
(
cur
->
pos
!=
blockInfo
.
rows
-
1
);
}
else
{
int32_t
order
=
(
pQueryHandle
->
order
==
TSDB_ORDER_ASC
)
?
TSDB_ORDER_DESC
:
TSDB_ORDER_ASC
;
endPos
=
vnodeBinarySearchKey
(
pCols
->
cols
[
0
].
pData
,
pCols
->
numOf
Point
s
,
pQueryHandle
->
window
.
ekey
,
order
);
endPos
=
vnodeBinarySearchKey
(
pCols
->
cols
[
0
].
pData
,
pCols
->
numOf
Row
s
,
pQueryHandle
->
window
.
ekey
,
order
);
cur
->
mixBlock
=
true
;
}
...
...
@@ -904,7 +904,7 @@ static void doMergeTwoLevelData(STsdbQueryHandle* pQueryHandle, STableCheckInfo*
}
int32_t
order
=
ASCENDING_TRAVERSE
(
pQueryHandle
->
order
)
?
TSDB_ORDER_DESC
:
TSDB_ORDER_ASC
;
int32_t
end
=
vnodeBinarySearchKey
(
pCols
->
cols
[
0
].
pData
,
pCols
->
numOf
Point
s
,
key
,
order
);
int32_t
end
=
vnodeBinarySearchKey
(
pCols
->
cols
[
0
].
pData
,
pCols
->
numOf
Row
s
,
key
,
order
);
if
(
tsArray
[
end
]
==
key
)
{
// the value of key in cache equals to the end timestamp value, ignore it
tSkipListIterNext
(
pCheckInfo
->
iter
);
}
...
...
@@ -1002,7 +1002,7 @@ static void doMergeTwoLevelData(STsdbQueryHandle* pQueryHandle, STableCheckInfo*
int32_t
binarySearchForKey
(
char
*
pValue
,
int
num
,
TSKEY
key
,
int
order
)
{
int
firstPos
,
lastPos
,
midPos
=
-
1
;
int
numOf
Point
s
;
int
numOf
Row
s
;
TSKEY
*
keyList
;
if
(
num
<=
0
)
return
-
1
;
...
...
@@ -1018,8 +1018,8 @@ int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
if
(
key
==
keyList
[
firstPos
])
return
firstPos
;
if
(
key
<
keyList
[
firstPos
])
return
firstPos
-
1
;
numOf
Point
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Point
s
>>
1
)
+
firstPos
;
numOf
Row
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Row
s
>>
1
)
+
firstPos
;
if
(
key
<
keyList
[
midPos
])
{
lastPos
=
midPos
-
1
;
...
...
@@ -1044,8 +1044,8 @@ int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
return
lastPos
;
}
numOf
Point
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Point
s
>>
1
)
+
firstPos
;
numOf
Row
s
=
lastPos
-
firstPos
+
1
;
midPos
=
(
numOf
Row
s
>>
1
)
+
firstPos
;
if
(
key
<
keyList
[
midPos
])
{
lastPos
=
midPos
-
1
;
...
...
@@ -1066,7 +1066,6 @@ static void cleanBlockOrderSupporter(SBlockOrderSupporter* pSupporter, int32_t n
for
(
int32_t
i
=
0
;
i
<
numOfTables
;
++
i
)
{
STableBlockInfo
*
pBlockInfo
=
pSupporter
->
pDataBlockInfo
[
i
];
// tfree(pBlockInfo->statis);
tfree
(
pBlockInfo
);
}
...
...
@@ -1539,9 +1538,19 @@ int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT* pQueryHandle, SDataSta
STableBlockInfo
*
pBlockInfo
=
&
pHandle
->
pDataBlockInfo
[
cur
->
slot
];
tsdbLoadCompData
(
&
pHandle
->
rhelper
,
pBlockInfo
->
compBlock
,
NULL
);
tsdbGetDataStatis
(
&
pHandle
->
rhelper
,
pHandle
->
statis
,
QH_GET_NUM_OF_COLS
(
pHandle
));
size_t
numOfCols
=
QH_GET_NUM_OF_COLS
(
pHandle
);
memset
(
pHandle
->
statis
,
0
,
sizeof
(
SDataStatis
)
*
numOfCols
);
tsdbGetDataStatis
(
&
pHandle
->
rhelper
,
pHandle
->
statis
,
numOfCols
);
*
pBlockStatis
=
pHandle
->
statis
;
//update the number of NULL data rows
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
if
(
pHandle
->
statis
[
i
].
numOfNull
==
-
1
)
{
pHandle
->
statis
[
i
].
numOfNull
=
pBlockInfo
->
compBlock
->
numOfRows
;
}
}
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -1575,7 +1584,7 @@ SArray* tsdbRetrieveDataBlock(TsdbQueryHandleT* pQueryHandle, SArray* pIdList) {
doLoadFileDataBlock
(
pHandle
,
pBlock
,
pCheckInfo
);
// todo refactor
int32_t
numOfRows
=
copyDataFromFileBlock
(
pHandle
,
pHandle
->
outputCapacity
,
0
,
0
,
pBlock
->
numOf
Point
s
-
1
);
int32_t
numOfRows
=
copyDataFromFileBlock
(
pHandle
,
pHandle
->
outputCapacity
,
0
,
0
,
pBlock
->
numOf
Row
s
-
1
);
// if the buffer is not full in case of descending order query, move the data in the front of the buffer
if
(
!
ASCENDING_TRAVERSE
(
pHandle
->
order
)
&&
numOfRows
<
pHandle
->
outputCapacity
)
{
...
...
tests/pytest/import_merge/importToCommit.py
浏览文件 @
fdad406e
...
...
@@ -46,7 +46,6 @@ class TDTestCase:
tdLog
.
info
(
'insert data until the first commit'
)
dnodesDir
=
tdDnodes
.
getDnodesRootDir
()
dataDir
=
dnodesDir
+
'/dnode1/data/vnode'
tdLog
.
info
(
'CBD: dataDir=%s'
%
dataDir
)
startTime
=
self
.
startTime
rid0
=
1
while
(
True
):
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录