Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
7314a0dd
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
7314a0dd
编写于
11月 08, 2020
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[TD-1978]
上级
ff8930f9
变更
8
显示空白变更内容
内联
并排
Showing
8 changed file
with
323 addition
and
517 deletion
+323
-517
src/client/src/tscFunctionImpl.c
src/client/src/tscFunctionImpl.c
+12
-150
src/client/src/tscLocalMerge.c
src/client/src/tscLocalMerge.c
+4
-4
src/common/src/ttypes.c
src/common/src/ttypes.c
+2
-32
src/inc/taosdef.h
src/inc/taosdef.h
+43
-45
src/query/inc/qFill.h
src/query/inc/qFill.h
+27
-23
src/query/src/qExecutor.c
src/query/src/qExecutor.c
+5
-5
src/query/src/qFill.c
src/query/src/qFill.c
+230
-257
src/util/src/tcompare.c
src/util/src/tcompare.c
+0
-1
未找到文件。
src/client/src/tscFunctionImpl.c
浏览文件 @
7314a0dd
...
@@ -28,6 +28,7 @@
...
@@ -28,6 +28,7 @@
#include "tscompression.h"
#include "tscompression.h"
#include "tsqlfunction.h"
#include "tsqlfunction.h"
#include "tutil.h"
#include "tutil.h"
#include "ttype.h"
#define GET_INPUT_CHAR(x) (((char *)((x)->aInputElemBuf)) + ((x)->startOffset) * ((x)->inputBytes))
#define GET_INPUT_CHAR(x) (((char *)((x)->aInputElemBuf)) + ((x)->startOffset) * ((x)->inputBytes))
#define GET_INPUT_CHAR_INDEX(x, y) (GET_INPUT_CHAR(x) + (y) * (x)->inputBytes)
#define GET_INPUT_CHAR_INDEX(x, y) (GET_INPUT_CHAR(x) + (y) * (x)->inputBytes)
...
@@ -2479,28 +2480,8 @@ static void percentile_function(SQLFunctionCtx *pCtx) {
...
@@ -2479,28 +2480,8 @@ static void percentile_function(SQLFunctionCtx *pCtx) {
continue
;
continue
;
}
}
// TODO extract functions
double
v
=
0
;
double
v
=
0
;
switch
(
pCtx
->
inputType
)
{
GET_TYPED_DATA
(
v
,
pCtx
->
inputType
,
data
);
case
TSDB_DATA_TYPE_TINYINT
:
v
=
GET_INT8_VAL
(
data
);
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
v
=
GET_INT16_VAL
(
data
);
break
;
case
TSDB_DATA_TYPE_BIGINT
:
v
=
(
double
)(
GET_INT64_VAL
(
data
));
break
;
case
TSDB_DATA_TYPE_FLOAT
:
v
=
GET_FLOAT_VAL
(
data
);
break
;
case
TSDB_DATA_TYPE_DOUBLE
:
v
=
GET_DOUBLE_VAL
(
data
);
break
;
default:
v
=
GET_INT32_VAL
(
data
);
break
;
}
if
(
v
<
GET_DOUBLE_VAL
(
&
pInfo
->
minval
))
{
if
(
v
<
GET_DOUBLE_VAL
(
&
pInfo
->
minval
))
{
SET_DOUBLE_VAL
(
&
pInfo
->
minval
,
v
);
SET_DOUBLE_VAL
(
&
pInfo
->
minval
,
v
);
...
@@ -2541,30 +2522,10 @@ static void percentile_function_f(SQLFunctionCtx *pCtx, int32_t index) {
...
@@ -2541,30 +2522,10 @@ static void percentile_function_f(SQLFunctionCtx *pCtx, int32_t index) {
SResultRowCellInfo
*
pResInfo
=
GET_RES_INFO
(
pCtx
);
SResultRowCellInfo
*
pResInfo
=
GET_RES_INFO
(
pCtx
);
SPercentileInfo
*
pInfo
=
(
SPercentileInfo
*
)
GET_ROWCELL_INTERBUF
(
pResInfo
);
SPercentileInfo
*
pInfo
=
(
SPercentileInfo
*
)
GET_ROWCELL_INTERBUF
(
pResInfo
);
if
(
pInfo
->
stage
==
0
)
{
if
(
pInfo
->
stage
==
0
)
{
// TODO extract functions
double
v
=
0
;
double
v
=
0
;
switch
(
pCtx
->
inputType
)
{
GET_TYPED_DATA
(
v
,
pCtx
->
inputType
,
pData
);
case
TSDB_DATA_TYPE_TINYINT
:
v
=
GET_INT8_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
v
=
GET_INT16_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_BIGINT
:
v
=
(
double
)(
GET_INT64_VAL
(
pData
));
break
;
case
TSDB_DATA_TYPE_FLOAT
:
v
=
GET_FLOAT_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_DOUBLE
:
v
=
GET_DOUBLE_VAL
(
pData
);
break
;
default:
v
=
GET_INT32_VAL
(
pData
);
break
;
}
if
(
v
<
GET_DOUBLE_VAL
(
&
pInfo
->
minval
))
{
if
(
v
<
GET_DOUBLE_VAL
(
&
pInfo
->
minval
))
{
SET_DOUBLE_VAL
(
&
pInfo
->
minval
,
v
);
SET_DOUBLE_VAL
(
&
pInfo
->
minval
,
v
);
...
@@ -2653,29 +2614,9 @@ static void apercentile_function(SQLFunctionCtx *pCtx) {
...
@@ -2653,29 +2614,9 @@ static void apercentile_function(SQLFunctionCtx *pCtx) {
}
}
notNullElems
+=
1
;
notNullElems
+=
1
;
double
v
=
0
;
switch
(
pCtx
->
inputType
)
{
case
TSDB_DATA_TYPE_TINYINT
:
v
=
GET_INT8_VAL
(
data
);
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
v
=
GET_INT16_VAL
(
data
);
break
;
case
TSDB_DATA_TYPE_BIGINT
:
v
=
(
double
)(
GET_INT64_VAL
(
data
));
break
;
case
TSDB_DATA_TYPE_FLOAT
:
v
=
GET_FLOAT_VAL
(
data
);
break
;
case
TSDB_DATA_TYPE_DOUBLE
:
v
=
GET_DOUBLE_VAL
(
data
);
break
;
default:
v
=
GET_INT32_VAL
(
data
);
break
;
}
double
v
=
0
;
GET_TYPED_DATA
(
v
,
pCtx
->
inputType
,
data
);
tHistogramAdd
(
&
pInfo
->
pHisto
,
v
);
tHistogramAdd
(
&
pInfo
->
pHisto
,
v
);
}
}
...
@@ -2700,26 +2641,7 @@ static void apercentile_function_f(SQLFunctionCtx *pCtx, int32_t index) {
...
@@ -2700,26 +2641,7 @@ static void apercentile_function_f(SQLFunctionCtx *pCtx, int32_t index) {
SAPercentileInfo
*
pInfo
=
getAPerctInfo
(
pCtx
);
SAPercentileInfo
*
pInfo
=
getAPerctInfo
(
pCtx
);
double
v
=
0
;
double
v
=
0
;
switch
(
pCtx
->
inputType
)
{
GET_TYPED_DATA
(
v
,
pCtx
->
inputType
,
pData
);
case
TSDB_DATA_TYPE_TINYINT
:
v
=
GET_INT8_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
v
=
GET_INT16_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_BIGINT
:
v
=
(
double
)(
GET_INT64_VAL
(
pData
));
break
;
case
TSDB_DATA_TYPE_FLOAT
:
v
=
GET_FLOAT_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_DOUBLE
:
v
=
GET_DOUBLE_VAL
(
pData
);
break
;
default:
v
=
GET_INT32_VAL
(
pData
);
break
;
}
tHistogramAdd
(
&
pInfo
->
pHisto
,
v
);
tHistogramAdd
(
&
pInfo
->
pHisto
,
v
);
...
@@ -4142,22 +4064,7 @@ static void rate_function(SQLFunctionCtx *pCtx) {
...
@@ -4142,22 +4064,7 @@ static void rate_function(SQLFunctionCtx *pCtx) {
notNullElems
++
;
notNullElems
++
;
int64_t
v
=
0
;
int64_t
v
=
0
;
switch
(
pCtx
->
inputType
)
{
GET_TYPED_DATA
(
v
,
pCtx
->
inputType
,
pData
);
case
TSDB_DATA_TYPE_TINYINT
:
v
=
(
int64_t
)
GET_INT8_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
v
=
(
int64_t
)
GET_INT16_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_INT
:
v
=
(
int64_t
)
GET_INT32_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_BIGINT
:
v
=
(
int64_t
)
GET_INT64_VAL
(
pData
);
break
;
default:
assert
(
0
);
}
if
((
INT64_MIN
==
pRateInfo
->
firstValue
)
||
(
INT64_MIN
==
pRateInfo
->
firstKey
))
{
if
((
INT64_MIN
==
pRateInfo
->
firstValue
)
||
(
INT64_MIN
==
pRateInfo
->
firstKey
))
{
pRateInfo
->
firstValue
=
v
;
pRateInfo
->
firstValue
=
v
;
...
@@ -4207,22 +4114,7 @@ static void rate_function_f(SQLFunctionCtx *pCtx, int32_t index) {
...
@@ -4207,22 +4114,7 @@ static void rate_function_f(SQLFunctionCtx *pCtx, int32_t index) {
TSKEY
*
primaryKey
=
pCtx
->
ptsList
;
TSKEY
*
primaryKey
=
pCtx
->
ptsList
;
int64_t
v
=
0
;
int64_t
v
=
0
;
switch
(
pCtx
->
inputType
)
{
GET_TYPED_DATA
(
v
,
pCtx
->
inputType
,
pData
);
case
TSDB_DATA_TYPE_TINYINT
:
v
=
(
int64_t
)
GET_INT8_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
v
=
(
int64_t
)
GET_INT16_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_INT
:
v
=
(
int64_t
)
GET_INT32_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_BIGINT
:
v
=
(
int64_t
)
GET_INT64_VAL
(
pData
);
break
;
default:
assert
(
0
);
}
if
((
INT64_MIN
==
pRateInfo
->
firstValue
)
||
(
INT64_MIN
==
pRateInfo
->
firstKey
))
{
if
((
INT64_MIN
==
pRateInfo
->
firstValue
)
||
(
INT64_MIN
==
pRateInfo
->
firstKey
))
{
pRateInfo
->
firstValue
=
v
;
pRateInfo
->
firstValue
=
v
;
...
@@ -4349,22 +4241,7 @@ static void irate_function(SQLFunctionCtx *pCtx) {
...
@@ -4349,22 +4241,7 @@ static void irate_function(SQLFunctionCtx *pCtx) {
notNullElems
++
;
notNullElems
++
;
int64_t
v
=
0
;
int64_t
v
=
0
;
switch
(
pCtx
->
inputType
)
{
GET_TYPED_DATA
(
v
,
pCtx
->
inputType
,
pData
);
case
TSDB_DATA_TYPE_TINYINT
:
v
=
(
int64_t
)
GET_INT8_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
v
=
(
int64_t
)
GET_INT16_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_INT
:
v
=
(
int64_t
)
GET_INT32_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_BIGINT
:
v
=
(
int64_t
)
GET_INT64_VAL
(
pData
);
break
;
default:
assert
(
0
);
}
// TODO: calc once if only call this function once ????
// TODO: calc once if only call this function once ????
if
((
INT64_MIN
==
pRateInfo
->
lastKey
)
||
(
INT64_MIN
==
pRateInfo
->
lastValue
))
{
if
((
INT64_MIN
==
pRateInfo
->
lastKey
)
||
(
INT64_MIN
==
pRateInfo
->
lastValue
))
{
...
@@ -4409,22 +4286,7 @@ static void irate_function_f(SQLFunctionCtx *pCtx, int32_t index) {
...
@@ -4409,22 +4286,7 @@ static void irate_function_f(SQLFunctionCtx *pCtx, int32_t index) {
TSKEY
*
primaryKey
=
pCtx
->
ptsList
;
TSKEY
*
primaryKey
=
pCtx
->
ptsList
;
int64_t
v
=
0
;
int64_t
v
=
0
;
switch
(
pCtx
->
inputType
)
{
GET_TYPED_DATA
(
v
,
pCtx
->
inputType
,
pData
);
case
TSDB_DATA_TYPE_TINYINT
:
v
=
(
int64_t
)
GET_INT8_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_SMALLINT
:
v
=
(
int64_t
)
GET_INT16_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_INT
:
v
=
(
int64_t
)
GET_INT32_VAL
(
pData
);
break
;
case
TSDB_DATA_TYPE_BIGINT
:
v
=
(
int64_t
)
GET_INT64_VAL
(
pData
);
break
;
default:
assert
(
0
);
}
pRateInfo
->
firstKey
=
pRateInfo
->
lastKey
;
pRateInfo
->
firstKey
=
pRateInfo
->
lastKey
;
pRateInfo
->
firstValue
=
pRateInfo
->
lastValue
;
pRateInfo
->
firstValue
=
pRateInfo
->
lastValue
;
...
...
src/client/src/tscLocalMerge.c
浏览文件 @
7314a0dd
...
@@ -913,7 +913,7 @@ static void doFillResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, bool doneO
...
@@ -913,7 +913,7 @@ static void doFillResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, bool doneO
}
}
while
(
1
)
{
while
(
1
)
{
int64_t
newRows
=
taos
Generate
DataBlock
(
pFillInfo
,
pResPages
,
pLocalReducer
->
resColModel
->
capacity
);
int64_t
newRows
=
taos
FillResult
DataBlock
(
pFillInfo
,
pResPages
,
pLocalReducer
->
resColModel
->
capacity
);
if
(
pQueryInfo
->
limit
.
offset
<
newRows
)
{
if
(
pQueryInfo
->
limit
.
offset
<
newRows
)
{
newRows
-=
pQueryInfo
->
limit
.
offset
;
newRows
-=
pQueryInfo
->
limit
.
offset
;
...
@@ -942,7 +942,7 @@ static void doFillResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, bool doneO
...
@@ -942,7 +942,7 @@ static void doFillResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, bool doneO
}
}
// all output in current group are completed
// all output in current group are completed
int32_t
totalRemainRows
=
(
int32_t
)
get
FilledNumOfRes
(
pFillInfo
,
actualETime
,
pLocalReducer
->
resColModel
->
capacity
);
int32_t
totalRemainRows
=
(
int32_t
)
get
NumOfResWithFill
(
pFillInfo
,
actualETime
,
pLocalReducer
->
resColModel
->
capacity
);
if
(
totalRemainRows
<=
0
)
{
if
(
totalRemainRows
<=
0
)
{
break
;
break
;
}
}
...
@@ -1291,7 +1291,7 @@ static bool doBuildFilledResultForGroup(SSqlObj *pSql) {
...
@@ -1291,7 +1291,7 @@ static bool doBuildFilledResultForGroup(SSqlObj *pSql) {
int64_t
etime
=
*
(
int64_t
*
)(
pFinalDataBuf
->
data
+
TSDB_KEYSIZE
*
(
pFillInfo
->
numOfRows
-
1
));
int64_t
etime
=
*
(
int64_t
*
)(
pFinalDataBuf
->
data
+
TSDB_KEYSIZE
*
(
pFillInfo
->
numOfRows
-
1
));
// the first column must be the timestamp column
// the first column must be the timestamp column
int32_t
rows
=
(
int32_t
)
getFilledNumOfRes
(
pFillInfo
,
etime
,
pLocalReducer
->
resColModel
->
capacity
);
int32_t
rows
=
(
int32_t
)
getNumOfResWithFill
(
pFillInfo
,
etime
,
pLocalReducer
->
resColModel
->
capacity
);
if
(
rows
>
0
)
{
// do fill gap
if
(
rows
>
0
)
{
// do fill gap
doFillResult
(
pSql
,
pLocalReducer
,
false
);
doFillResult
(
pSql
,
pLocalReducer
,
false
);
}
}
...
@@ -1320,7 +1320,7 @@ static bool doHandleLastRemainData(SSqlObj *pSql) {
...
@@ -1320,7 +1320,7 @@ static bool doHandleLastRemainData(SSqlObj *pSql) {
((
pRes
->
numOfRowsGroup
<
pQueryInfo
->
limit
.
limit
&&
pQueryInfo
->
limit
.
limit
>
0
)
||
(
pQueryInfo
->
limit
.
limit
<
0
)))
{
((
pRes
->
numOfRowsGroup
<
pQueryInfo
->
limit
.
limit
&&
pQueryInfo
->
limit
.
limit
>
0
)
||
(
pQueryInfo
->
limit
.
limit
<
0
)))
{
int64_t
etime
=
(
pQueryInfo
->
order
.
order
==
TSDB_ORDER_ASC
)
?
pQueryInfo
->
window
.
ekey
:
pQueryInfo
->
window
.
skey
;
int64_t
etime
=
(
pQueryInfo
->
order
.
order
==
TSDB_ORDER_ASC
)
?
pQueryInfo
->
window
.
ekey
:
pQueryInfo
->
window
.
skey
;
int32_t
rows
=
(
int32_t
)
get
FilledNumOfRes
(
pFillInfo
,
etime
,
pLocalReducer
->
resColModel
->
capacity
);
int32_t
rows
=
(
int32_t
)
get
NumOfResWithFill
(
pFillInfo
,
etime
,
pLocalReducer
->
resColModel
->
capacity
);
if
(
rows
>
0
)
{
if
(
rows
>
0
)
{
doFillResult
(
pSql
,
pLocalReducer
,
true
);
doFillResult
(
pSql
,
pLocalReducer
,
true
);
}
}
...
...
src/common/src/ttypes.c
浏览文件 @
7314a0dd
...
@@ -355,32 +355,6 @@ bool isValidDataType(int32_t type) {
...
@@ -355,32 +355,6 @@ bool isValidDataType(int32_t type) {
return
type
>=
TSDB_DATA_TYPE_NULL
&&
type
<=
TSDB_DATA_TYPE_NCHAR
;
return
type
>=
TSDB_DATA_TYPE_NULL
&&
type
<=
TSDB_DATA_TYPE_NCHAR
;
}
}
//bool isNull(const char *val, int32_t type) {
// switch (type) {
// case TSDB_DATA_TYPE_BOOL:
// return *(uint8_t *)val == TSDB_DATA_BOOL_NULL;
// case TSDB_DATA_TYPE_TINYINT:
// return *(uint8_t *)val == TSDB_DATA_TINYINT_NULL;
// case TSDB_DATA_TYPE_SMALLINT:
// return *(uint16_t *)val == TSDB_DATA_SMALLINT_NULL;
// case TSDB_DATA_TYPE_INT:
// return *(uint32_t *)val == TSDB_DATA_INT_NULL;
// case TSDB_DATA_TYPE_BIGINT:
// case TSDB_DATA_TYPE_TIMESTAMP:
// return *(uint64_t *)val == TSDB_DATA_BIGINT_NULL;
// case TSDB_DATA_TYPE_FLOAT:
// return *(uint32_t *)val == TSDB_DATA_FLOAT_NULL;
// case TSDB_DATA_TYPE_DOUBLE:
// return *(uint64_t *)val == TSDB_DATA_DOUBLE_NULL;
// case TSDB_DATA_TYPE_NCHAR:
// return *(uint32_t*) varDataVal(val) == TSDB_DATA_NCHAR_NULL;
// case TSDB_DATA_TYPE_BINARY:
// return *(uint8_t *) varDataVal(val) == TSDB_DATA_BINARY_NULL;
// default:
// return false;
// };
//}
void
setVardataNull
(
char
*
val
,
int32_t
type
)
{
void
setVardataNull
(
char
*
val
,
int32_t
type
)
{
if
(
type
==
TSDB_DATA_TYPE_BINARY
)
{
if
(
type
==
TSDB_DATA_TYPE_BINARY
)
{
varDataSetLen
(
val
,
sizeof
(
int8_t
));
varDataSetLen
(
val
,
sizeof
(
int8_t
));
...
@@ -433,14 +407,10 @@ void setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems) {
...
@@ -433,14 +407,10 @@ void setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems) {
*
(
uint64_t
*
)(
val
+
i
*
tDataTypeDesc
[
type
].
nSize
)
=
TSDB_DATA_DOUBLE_NULL
;
*
(
uint64_t
*
)(
val
+
i
*
tDataTypeDesc
[
type
].
nSize
)
=
TSDB_DATA_DOUBLE_NULL
;
}
}
break
;
break
;
case
TSDB_DATA_TYPE_NCHAR
:
// todo : without length?
case
TSDB_DATA_TYPE_NCHAR
:
for
(
int32_t
i
=
0
;
i
<
numOfElems
;
++
i
)
{
*
(
uint32_t
*
)(
val
+
i
*
bytes
)
=
TSDB_DATA_NCHAR_NULL
;
}
break
;
case
TSDB_DATA_TYPE_BINARY
:
case
TSDB_DATA_TYPE_BINARY
:
for
(
int32_t
i
=
0
;
i
<
numOfElems
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
numOfElems
;
++
i
)
{
*
(
uint8_t
*
)(
val
+
i
*
bytes
)
=
TSDB_DATA_BINARY_NULL
;
setVardataNull
(
val
+
i
*
bytes
,
type
)
;
}
}
break
;
break
;
default:
{
default:
{
...
...
src/inc/taosdef.h
浏览文件 @
7314a0dd
...
@@ -61,6 +61,7 @@ typedef struct tstr {
...
@@ -61,6 +61,7 @@ typedef struct tstr {
// Bytes for each type.
// Bytes for each type.
extern
const
int32_t
TYPE_BYTES
[
11
];
extern
const
int32_t
TYPE_BYTES
[
11
];
// TODO: replace and remove code below
// TODO: replace and remove code below
#define CHAR_BYTES sizeof(char)
#define CHAR_BYTES sizeof(char)
#define SHORT_BYTES sizeof(int16_t)
#define SHORT_BYTES sizeof(int16_t)
...
@@ -68,6 +69,15 @@ extern const int32_t TYPE_BYTES[11];
...
@@ -68,6 +69,15 @@ extern const int32_t TYPE_BYTES[11];
#define LONG_BYTES sizeof(int64_t)
#define LONG_BYTES sizeof(int64_t)
#define FLOAT_BYTES sizeof(float)
#define FLOAT_BYTES sizeof(float)
#define DOUBLE_BYTES sizeof(double)
#define DOUBLE_BYTES sizeof(double)
#define POINTER_BYTES sizeof(void *) // 8 by default assert(sizeof(ptrdiff_t) == sizseof(void*)
#define TSDB_KEYSIZE sizeof(TSKEY)
#if LINUX
#define TSDB_NCHAR_SIZE sizeof(wchar_t)
#else
#define TSDB_NCHAR_SIZE sizeof(int32_t)
#endif
// NULL definition
// NULL definition
#define TSDB_DATA_BOOL_NULL 0x02
#define TSDB_DATA_BOOL_NULL 0x02
...
@@ -101,10 +111,12 @@ extern const int32_t TYPE_BYTES[11];
...
@@ -101,10 +111,12 @@ extern const int32_t TYPE_BYTES[11];
#define TSDB_TIME_PRECISION_MILLI 0
#define TSDB_TIME_PRECISION_MILLI 0
#define TSDB_TIME_PRECISION_MICRO 1
#define TSDB_TIME_PRECISION_MICRO 1
#define TSDB_TIME_PRECISION_NANO 2
#define TSDB_TIME_PRECISION_NANO 2
#define TSDB_TICK_PER_SECOND(precision) ((precision)==TSDB_TIME_PRECISION_MILLI ? 1e3L : ((precision)==TSDB_TIME_PRECISION_MICRO ? 1e6L : 1e9L))
#define TSDB_TIME_PRECISION_MILLI_STR "ms"
#define TSDB_TIME_PRECISION_MILLI_STR "ms"
#define TSDB_TIME_PRECISION_MICRO_STR "us"
#define TSDB_TIME_PRECISION_MICRO_STR "us"
#define TSDB_TIME_PRECISION_NANO_STR "ns"
#define TSDB_TICK_PER_SECOND(precision) ((precision)==TSDB_TIME_PRECISION_MILLI ? 1e3L : ((precision)==TSDB_TIME_PRECISION_MICRO ? 1e6L : 1e9L))
#define T_MEMBER_SIZE(type, member) sizeof(((type *)0)->member)
#define T_MEMBER_SIZE(type, member) sizeof(((type *)0)->member)
#define T_APPEND_MEMBER(dst, ptr, type, member) \
#define T_APPEND_MEMBER(dst, ptr, type, member) \
...
@@ -118,15 +130,6 @@ do { \
...
@@ -118,15 +130,6 @@ do { \
(src) = (void *)((char *)src + sizeof(type));\
(src) = (void *)((char *)src + sizeof(type));\
} while(0)
} while(0)
#define TSDB_KEYSIZE sizeof(TSKEY)
#if LINUX
#define TSDB_NCHAR_SIZE sizeof(wchar_t)
#else
#define TSDB_NCHAR_SIZE 4
#endif
//#define TSDB_CHAR_TERMINATED_SPACE 1
#define GET_INT8_VAL(x) (*(int8_t *)(x))
#define GET_INT8_VAL(x) (*(int8_t *)(x))
#define GET_INT16_VAL(x) (*(int16_t *)(x))
#define GET_INT16_VAL(x) (*(int16_t *)(x))
#define GET_INT32_VAL(x) (*(int32_t *)(x))
#define GET_INT32_VAL(x) (*(int32_t *)(x))
...
@@ -172,7 +175,6 @@ typedef struct tDataTypeDescriptor {
...
@@ -172,7 +175,6 @@ typedef struct tDataTypeDescriptor {
}
tDataTypeDescriptor
;
}
tDataTypeDescriptor
;
extern
tDataTypeDescriptor
tDataTypeDesc
[
11
];
extern
tDataTypeDescriptor
tDataTypeDesc
[
11
];
#define POINTER_BYTES sizeof(void *) // 8 by default assert(sizeof(ptrdiff_t) == sizseof(void*)
bool
isValidDataType
(
int32_t
type
);
bool
isValidDataType
(
int32_t
type
);
//bool isNull(const char *val, int32_t type);
//bool isNull(const char *val, int32_t type);
...
@@ -266,10 +268,6 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf
...
@@ -266,10 +268,6 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void* buf
#define TSDB_AUTH_LEN 16
#define TSDB_AUTH_LEN 16
#define TSDB_KEY_LEN 16
#define TSDB_KEY_LEN 16
#define TSDB_VERSION_LEN 12
#define TSDB_VERSION_LEN 12
#define TSDB_STREET_LEN 64
#define TSDB_CITY_LEN 20
#define TSDB_STATE_LEN 20
#define TSDB_COUNTRY_LEN 20
#define TSDB_LOCALE_LEN 64
#define TSDB_LOCALE_LEN 64
#define TSDB_TIMEZONE_LEN 96
#define TSDB_TIMEZONE_LEN 96
#define TSDB_LABEL_LEN 8
#define TSDB_LABEL_LEN 8
...
...
src/query/inc/qFill.h
浏览文件 @
7314a0dd
...
@@ -28,6 +28,7 @@ typedef struct {
...
@@ -28,6 +28,7 @@ typedef struct {
STColumn
col
;
// column info
STColumn
col
;
// column info
int16_t
functionId
;
// sql function id
int16_t
functionId
;
// sql function id
int16_t
flag
;
// column flag: TAG COLUMN|NORMAL COLUMN
int16_t
flag
;
// column flag: TAG COLUMN|NORMAL COLUMN
int16_t
tagIndex
;
// index of current tag in SFillTagColInfo array list
union
{
int64_t
i
;
double
d
;}
fillVal
;
union
{
int64_t
i
;
double
d
;}
fillVal
;
}
SFillColInfo
;
}
SFillColInfo
;
...
@@ -38,25 +39,28 @@ typedef struct {
...
@@ -38,25 +39,28 @@ typedef struct {
typedef
struct
SFillInfo
{
typedef
struct
SFillInfo
{
TSKEY
start
;
// start timestamp
TSKEY
start
;
// start timestamp
TSKEY
endKey
;
// endKey for fill
TSKEY
end
;
// endKey for fill
TSKEY
currentKey
;
// current active timestamp, the value may be changed during the fill procedure.
int32_t
order
;
// order [TSDB_ORDER_ASC|TSDB_ORDER_DESC]
int32_t
order
;
// order [TSDB_ORDER_ASC|TSDB_ORDER_DESC]
int32_t
fillType
;
// fill type
int32_t
type
;
// fill type
int32_t
numOfRows
;
// number of rows in the input data block
int32_t
numOfRows
;
// number of rows in the input data block
int32_t
rowIdx
;
// rowId
x
int32_t
index
;
// active row inde
x
int32_t
numOfTotal
;
// number of filled rows in one round
int32_t
numOfTotal
;
// number of filled rows in one round
int32_t
numOfCurrent
;
// number of filled rows in current results
int32_t
numOfCurrent
;
// number of filled rows in current results
int32_t
numOfTags
;
// number of tags
int32_t
numOfTags
;
// number of tags
int32_t
numOfCols
;
// number of columns, including the tags columns
int32_t
numOfCols
;
// number of columns, including the tags columns
int32_t
rowSize
;
// size of each row
int32_t
rowSize
;
// size of each row
SFillTagColInfo
*
pTags
;
// tags value for filling gap
SInterval
interval
;
SInterval
interval
;
char
*
prevValues
;
// previous row of data, to generate the interpolation results
char
*
prevValues
;
// previous row of data, to generate the interpolation results
char
*
nextValues
;
// next row of data
char
*
nextValues
;
// next row of data
char
**
pData
;
// original result data block involved in filling data
char
**
pData
;
// original result data block involved in filling data
int32_t
capacityInRows
;
// data buffer size in rows
int32_t
alloc
;
// data buffer size in rows
int8_t
precision
;
// time resoluation
int8_t
precision
;
// time resoluation
SFillColInfo
*
pFillCol
;
// column info for fill operations
SFillColInfo
*
pFillCol
;
// column info for fill operations
SFillTagColInfo
*
pTags
;
// tags value for filling gap
void
*
handle
;
// for dubug purpose
}
SFillInfo
;
}
SFillInfo
;
typedef
struct
SPoint
{
typedef
struct
SPoint
{
...
@@ -74,17 +78,17 @@ void* taosDestroyFillInfo(SFillInfo *pFillInfo);
...
@@ -74,17 +78,17 @@ void* taosDestroyFillInfo(SFillInfo *pFillInfo);
void
taosFillSetStartInfo
(
SFillInfo
*
pFillInfo
,
int32_t
numOfRows
,
TSKEY
endKey
);
void
taosFillSetStartInfo
(
SFillInfo
*
pFillInfo
,
int32_t
numOfRows
,
TSKEY
endKey
);
void
taosFillCopyInputDataFromFilePage
(
SFillInfo
*
pFillInfo
,
tFilePage
**
pInput
);
void
taosFillCopyInputDataFromFilePage
(
SFillInfo
*
pFillInfo
,
const
tFilePage
**
pInput
);
void
taosFillCopyInputDataFromOneFilePage
(
SFillInfo
*
pFillInfo
,
tFilePage
*
pInput
);
void
taosFillCopyInputDataFromOneFilePage
(
SFillInfo
*
pFillInfo
,
const
tFilePage
*
pInput
);
int64_t
get
FilledNumOfRes
(
SFillInfo
*
pFillInfo
,
int64_t
ekey
,
int32_t
maxNumOfRows
);
int64_t
get
NumOfResWithFill
(
SFillInfo
*
pFillInfo
,
int64_t
ekey
,
int32_t
maxNumOfRows
);
int32_t
taosNumOfRemainRows
(
SFillInfo
*
pFillInfo
);
int32_t
taosNumOfRemainRows
(
SFillInfo
*
pFillInfo
);
int32_t
taosGetLinearInterpolationVal
(
int32_t
type
,
SPoint
*
point1
,
SPoint
*
point2
,
SPoint
*
point
);
int32_t
taosGetLinearInterpolationVal
(
int32_t
type
,
SPoint
*
point1
,
SPoint
*
point2
,
SPoint
*
point
);
int64_t
taos
Generate
DataBlock
(
SFillInfo
*
pFillInfo
,
tFilePage
**
output
,
int32_t
capacity
);
int64_t
taos
FillResult
DataBlock
(
SFillInfo
*
pFillInfo
,
tFilePage
**
output
,
int32_t
capacity
);
#ifdef __cplusplus
#ifdef __cplusplus
}
}
...
...
src/query/src/qExecutor.c
浏览文件 @
7314a0dd
...
@@ -4116,7 +4116,7 @@ bool queryHasRemainResForTableQuery(SQueryRuntimeEnv* pRuntimeEnv) {
...
@@ -4116,7 +4116,7 @@ bool queryHasRemainResForTableQuery(SQueryRuntimeEnv* pRuntimeEnv) {
* first result row in the actual result set will fill nothing.
* first result row in the actual result set will fill nothing.
*/
*/
if
(
Q_STATUS_EQUAL
(
pQuery
->
status
,
QUERY_COMPLETED
))
{
if
(
Q_STATUS_EQUAL
(
pQuery
->
status
,
QUERY_COMPLETED
))
{
int32_t
numOfTotal
=
(
int32_t
)
get
FilledNumOfRes
(
pFillInfo
,
pQuery
->
window
.
ekey
,
(
int32_t
)
pQuery
->
rec
.
capacity
);
int32_t
numOfTotal
=
(
int32_t
)
get
NumOfResWithFill
(
pFillInfo
,
pQuery
->
window
.
ekey
,
(
int32_t
)
pQuery
->
rec
.
capacity
);
return
numOfTotal
>
0
;
return
numOfTotal
>
0
;
}
}
...
@@ -4174,7 +4174,7 @@ int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int
...
@@ -4174,7 +4174,7 @@ int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int
SFillInfo
*
pFillInfo
=
pRuntimeEnv
->
pFillInfo
;
SFillInfo
*
pFillInfo
=
pRuntimeEnv
->
pFillInfo
;
while
(
1
)
{
while
(
1
)
{
int32_t
ret
=
(
int32_t
)
taos
Generate
DataBlock
(
pFillInfo
,
(
tFilePage
**
)
pQuery
->
sdata
,
(
int32_t
)
pQuery
->
rec
.
capacity
);
int32_t
ret
=
(
int32_t
)
taos
FillResult
DataBlock
(
pFillInfo
,
(
tFilePage
**
)
pQuery
->
sdata
,
(
int32_t
)
pQuery
->
rec
.
capacity
);
// todo apply limit output function
// todo apply limit output function
/* reached the start position of according to offset value, return immediately */
/* reached the start position of according to offset value, return immediately */
...
@@ -4519,6 +4519,7 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) {
...
@@ -4519,6 +4519,7 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) {
pFillCol
[
i
].
col
.
bytes
=
pExprInfo
->
bytes
;
pFillCol
[
i
].
col
.
bytes
=
pExprInfo
->
bytes
;
pFillCol
[
i
].
col
.
type
=
(
int8_t
)
pExprInfo
->
type
;
pFillCol
[
i
].
col
.
type
=
(
int8_t
)
pExprInfo
->
type
;
pFillCol
[
i
].
col
.
offset
=
offset
;
pFillCol
[
i
].
col
.
offset
=
offset
;
pFillCol
[
i
].
tagIndex
=
-
2
;
pFillCol
[
i
].
flag
=
TSDB_COL_NORMAL
;
// always be ta normal column for table query
pFillCol
[
i
].
flag
=
TSDB_COL_NORMAL
;
// always be ta normal column for table query
pFillCol
[
i
].
functionId
=
pExprInfo
->
base
.
functionId
;
pFillCol
[
i
].
functionId
=
pExprInfo
->
base
.
functionId
;
pFillCol
[
i
].
fillVal
.
i
=
pQuery
->
fillVal
[
i
];
pFillCol
[
i
].
fillVal
.
i
=
pQuery
->
fillVal
[
i
];
...
@@ -4532,7 +4533,6 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) {
...
@@ -4532,7 +4533,6 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) {
int32_t
doInitQInfo
(
SQInfo
*
pQInfo
,
STSBuf
*
pTsBuf
,
void
*
tsdb
,
int32_t
vgId
,
bool
isSTableQuery
)
{
int32_t
doInitQInfo
(
SQInfo
*
pQInfo
,
STSBuf
*
pTsBuf
,
void
*
tsdb
,
int32_t
vgId
,
bool
isSTableQuery
)
{
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pQInfo
->
runtimeEnv
;
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pQInfo
->
runtimeEnv
;
int32_t
code
=
TSDB_CODE_SUCCESS
;
SQuery
*
pQuery
=
pQInfo
->
runtimeEnv
.
pQuery
;
SQuery
*
pQuery
=
pQInfo
->
runtimeEnv
.
pQuery
;
pRuntimeEnv
->
topBotQuery
=
isTopBottomQuery
(
pQuery
);
pRuntimeEnv
->
topBotQuery
=
isTopBottomQuery
(
pQuery
);
...
@@ -4540,7 +4540,7 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
...
@@ -4540,7 +4540,7 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
setScanLimitationByResultBuffer
(
pQuery
);
setScanLimitationByResultBuffer
(
pQuery
);
code
=
setupQueryHandle
(
tsdb
,
pQInfo
,
isSTableQuery
);
int32_t
code
=
setupQueryHandle
(
tsdb
,
pQInfo
,
isSTableQuery
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
return
code
;
}
}
...
@@ -5430,7 +5430,7 @@ static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
...
@@ -5430,7 +5430,7 @@ static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
break
;
break
;
}
else
{
}
else
{
taosFillSetStartInfo
(
pRuntimeEnv
->
pFillInfo
,
(
int32_t
)
pQuery
->
rec
.
rows
,
pQuery
->
window
.
ekey
);
taosFillSetStartInfo
(
pRuntimeEnv
->
pFillInfo
,
(
int32_t
)
pQuery
->
rec
.
rows
,
pQuery
->
window
.
ekey
);
taosFillCopyInputDataFromFilePage
(
pRuntimeEnv
->
pFillInfo
,
(
tFilePage
**
)
pQuery
->
sdata
);
taosFillCopyInputDataFromFilePage
(
pRuntimeEnv
->
pFillInfo
,
(
const
tFilePage
**
)
pQuery
->
sdata
);
numOfFilled
=
0
;
numOfFilled
=
0
;
pQuery
->
rec
.
rows
=
doFillGapsInResults
(
pRuntimeEnv
,
(
tFilePage
**
)
pQuery
->
sdata
,
&
numOfFilled
);
pQuery
->
rec
.
rows
=
doFillGapsInResults
(
pRuntimeEnv
,
(
tFilePage
**
)
pQuery
->
sdata
,
&
numOfFilled
);
...
...
src/query/src/qFill.c
浏览文件 @
7314a0dd
...
@@ -13,17 +13,22 @@
...
@@ -13,17 +13,22 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
*/
#include "qFill.h"
#include "os.h"
#include "os.h"
#include "qExtbuffer.h"
#include "taosdef.h"
#include "taosdef.h"
#include "taosmsg.h"
#include "taosmsg.h"
#include "tsqlfunction.h"
#include "tsqlfunction.h"
#include "ttype.h"
#include "qFill.h"
#include "qExtbuffer.h"
#include "queryLog.h"
#define FILL_IS_ASC_FILL(_f) ((_f)->order == TSDB_ORDER_ASC)
#define FILL_IS_ASC_FILL(_f) ((_f)->order == TSDB_ORDER_ASC)
SFillInfo
*
taosInitFillInfo
(
int32_t
order
,
TSKEY
skey
,
int32_t
numOfTags
,
int32_t
capacity
,
int32_t
numOfCols
,
SFillInfo
*
taosInitFillInfo
(
int32_t
order
,
TSKEY
skey
,
int32_t
numOfTags
,
int32_t
capacity
,
int32_t
numOfCols
,
int64_t
slidingTime
,
int8_t
slidingUnit
,
int8_t
precision
,
int32_t
fillType
,
SFillColInfo
*
pFillCol
)
{
int64_t
slidingTime
,
int8_t
slidingUnit
,
int8_t
precision
,
int32_t
fillType
,
SFillColInfo
*
pCol
)
{
if
(
fillType
==
TSDB_FILL_NONE
)
{
if
(
fillType
==
TSDB_FILL_NONE
)
{
return
NULL
;
return
NULL
;
}
}
...
@@ -33,8 +38,8 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
...
@@ -33,8 +38,8 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
taosResetFillInfo
(
pFillInfo
,
skey
);
taosResetFillInfo
(
pFillInfo
,
skey
);
pFillInfo
->
order
=
order
;
pFillInfo
->
order
=
order
;
pFillInfo
->
fillType
=
fillType
;
pFillInfo
->
type
=
fillType
;
pFillInfo
->
pFillCol
=
pFill
Col
;
pFillInfo
->
pFillCol
=
p
Col
;
pFillInfo
->
numOfTags
=
numOfTags
;
pFillInfo
->
numOfTags
=
numOfTags
;
pFillInfo
->
numOfCols
=
numOfCols
;
pFillInfo
->
numOfCols
=
numOfCols
;
pFillInfo
->
precision
=
precision
;
pFillInfo
->
precision
=
precision
;
...
@@ -47,11 +52,12 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
...
@@ -47,11 +52,12 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
pFillInfo
->
pData
=
malloc
(
POINTER_BYTES
*
numOfCols
);
pFillInfo
->
pData
=
malloc
(
POINTER_BYTES
*
numOfCols
);
if
(
numOfTags
>
0
)
{
if
(
numOfTags
>
0
)
{
pFillInfo
->
pTags
=
calloc
(
pFillInfo
->
numOfTags
,
sizeof
(
SFillTagColInfo
));
pFillInfo
->
pTags
=
calloc
(
pFillInfo
->
numOfTags
,
sizeof
(
SFillTagColInfo
));
for
(
int32_t
i
=
0
;
i
<
numOfTags
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
numOfTags
;
++
i
)
{
pFillInfo
->
pTags
[
i
].
col
.
colId
=
-
2
;
pFillInfo
->
pTags
[
i
].
col
.
colId
=
-
2
;
// TODO
}
}
}
}
// there are no duplicated tags in the SFillTagColInfo list
int32_t
rowsize
=
0
;
int32_t
rowsize
=
0
;
int32_t
k
=
0
;
int32_t
k
=
0
;
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
...
@@ -60,7 +66,7 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
...
@@ -60,7 +66,7 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
if
(
TSDB_COL_IS_TAG
(
pColInfo
->
flag
))
{
if
(
TSDB_COL_IS_TAG
(
pColInfo
->
flag
))
{
bool
exists
=
false
;
bool
exists
=
false
;
for
(
int32_t
j
=
0
;
j
<
k
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
k
;
++
j
)
{
if
(
pFillInfo
->
pTags
[
j
].
col
.
colId
==
pColInfo
->
col
.
colId
)
{
if
(
pFillInfo
->
pTags
[
j
].
col
.
colId
==
pColInfo
->
col
.
colId
)
{
exists
=
true
;
exists
=
true
;
break
;
break
;
...
@@ -70,6 +76,7 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
...
@@ -70,6 +76,7 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
if
(
!
exists
)
{
if
(
!
exists
)
{
pFillInfo
->
pTags
[
k
].
col
.
colId
=
pColInfo
->
col
.
colId
;
pFillInfo
->
pTags
[
k
].
col
.
colId
=
pColInfo
->
col
.
colId
;
pFillInfo
->
pTags
[
k
].
tagVal
=
calloc
(
1
,
pColInfo
->
col
.
bytes
);
pFillInfo
->
pTags
[
k
].
tagVal
=
calloc
(
1
,
pColInfo
->
col
.
bytes
);
pColInfo
->
tagIndex
=
k
;
k
+=
1
;
k
+=
1
;
}
}
...
@@ -78,14 +85,15 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
...
@@ -78,14 +85,15 @@ SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_
}
}
pFillInfo
->
rowSize
=
rowsize
;
pFillInfo
->
rowSize
=
rowsize
;
pFillInfo
->
capacityInRows
=
capacity
;
pFillInfo
->
alloc
=
capacity
;
return
pFillInfo
;
return
pFillInfo
;
}
}
void
taosResetFillInfo
(
SFillInfo
*
pFillInfo
,
TSKEY
startTimestamp
)
{
void
taosResetFillInfo
(
SFillInfo
*
pFillInfo
,
TSKEY
startTimestamp
)
{
pFillInfo
->
start
=
startTimestamp
;
pFillInfo
->
start
=
startTimestamp
;
pFillInfo
->
rowIdx
=
-
1
;
pFillInfo
->
currentKey
=
startTimestamp
;
pFillInfo
->
index
=
-
1
;
pFillInfo
->
numOfRows
=
0
;
pFillInfo
->
numOfRows
=
0
;
pFillInfo
->
numOfCurrent
=
0
;
pFillInfo
->
numOfCurrent
=
0
;
pFillInfo
->
numOfTotal
=
0
;
pFillInfo
->
numOfTotal
=
0
;
...
@@ -112,20 +120,20 @@ void* taosDestroyFillInfo(SFillInfo* pFillInfo) {
...
@@ -112,20 +120,20 @@ void* taosDestroyFillInfo(SFillInfo* pFillInfo) {
}
}
void
taosFillSetStartInfo
(
SFillInfo
*
pFillInfo
,
int32_t
numOfRows
,
TSKEY
endKey
)
{
void
taosFillSetStartInfo
(
SFillInfo
*
pFillInfo
,
int32_t
numOfRows
,
TSKEY
endKey
)
{
if
(
pFillInfo
->
fillT
ype
==
TSDB_FILL_NONE
)
{
if
(
pFillInfo
->
t
ype
==
TSDB_FILL_NONE
)
{
return
;
return
;
}
}
pFillInfo
->
end
Key
=
endKey
;
pFillInfo
->
end
=
endKey
;
if
(
pFillInfo
->
order
!=
TSDB_ORDER_ASC
)
{
if
(
!
FILL_IS_ASC_FILL
(
pFillInfo
)
)
{
pFillInfo
->
end
Key
=
taosTimeTruncate
(
endKey
,
&
pFillInfo
->
interval
,
pFillInfo
->
precision
);
pFillInfo
->
end
=
taosTimeTruncate
(
endKey
,
&
pFillInfo
->
interval
,
pFillInfo
->
precision
);
}
}
pFillInfo
->
rowIdx
=
0
;
pFillInfo
->
index
=
0
;
pFillInfo
->
numOfRows
=
numOfRows
;
pFillInfo
->
numOfRows
=
numOfRows
;
// ensure the space
// ensure the space
if
(
pFillInfo
->
capacityInRows
<
numOfRows
)
{
if
(
pFillInfo
->
alloc
<
numOfRows
)
{
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
char
*
tmp
=
realloc
(
pFillInfo
->
pData
[
i
],
numOfRows
*
pFillInfo
->
pFillCol
[
i
].
col
.
bytes
);
char
*
tmp
=
realloc
(
pFillInfo
->
pData
[
i
],
numOfRows
*
pFillInfo
->
pFillCol
[
i
].
col
.
bytes
);
assert
(
tmp
!=
NULL
);
// todo handle error
assert
(
tmp
!=
NULL
);
// todo handle error
...
@@ -136,42 +144,38 @@ void taosFillSetStartInfo(SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey)
...
@@ -136,42 +144,38 @@ void taosFillSetStartInfo(SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey)
}
}
}
}
void
taosFillCopyInputDataFromFilePage
(
SFillInfo
*
pFillInfo
,
tFilePage
**
pInput
)
{
// copy the data into source data buffer
// copy the data into source data buffer
void
taosFillCopyInputDataFromFilePage
(
SFillInfo
*
pFillInfo
,
const
tFilePage
**
pInput
)
{
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
memcpy
(
pFillInfo
->
pData
[
i
],
pInput
[
i
]
->
data
,
pFillInfo
->
numOfRows
*
pFillInfo
->
pFillCol
[
i
].
col
.
bytes
);
memcpy
(
pFillInfo
->
pData
[
i
],
pInput
[
i
]
->
data
,
pFillInfo
->
numOfRows
*
pFillInfo
->
pFillCol
[
i
].
col
.
bytes
);
}
}
}
}
void
taosFillCopyInputDataFromOneFilePage
(
SFillInfo
*
pFillInfo
,
tFilePage
*
pInput
)
{
void
taosFillCopyInputDataFromOneFilePage
(
SFillInfo
*
pFillInfo
,
const
tFilePage
*
pInput
)
{
assert
(
pFillInfo
->
numOfRows
==
pInput
->
num
);
assert
(
pFillInfo
->
numOfRows
==
pInput
->
num
);
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
char
*
data
=
pInput
->
data
+
pCol
->
col
.
offset
*
pInput
->
num
;
c
onst
c
har
*
data
=
pInput
->
data
+
pCol
->
col
.
offset
*
pInput
->
num
;
memcpy
(
pFillInfo
->
pData
[
i
],
data
,
(
size_t
)(
pInput
->
num
*
pCol
->
col
.
bytes
));
memcpy
(
pFillInfo
->
pData
[
i
],
data
,
(
size_t
)(
pInput
->
num
*
pCol
->
col
.
bytes
));
if
(
TSDB_COL_IS_TAG
(
pCol
->
flag
))
{
// copy the tag value to tag value buffer
if
(
TSDB_COL_IS_TAG
(
pCol
->
flag
))
{
// copy the tag value to tag value buffer
for
(
int32_t
j
=
0
;
j
<
pFillInfo
->
numOfTags
;
++
j
)
{
SFillTagColInfo
*
pTag
=
&
pFillInfo
->
pTags
[
pCol
->
tagIndex
];
SFillTagColInfo
*
pTag
=
&
pFillInfo
->
pTags
[
j
];
assert
(
pTag
->
col
.
colId
==
pCol
->
col
.
colId
);
if
(
pTag
->
col
.
colId
==
pCol
->
col
.
colId
)
{
memcpy
(
pTag
->
tagVal
,
data
,
pCol
->
col
.
bytes
);
memcpy
(
pTag
->
tagVal
,
data
,
pCol
->
col
.
bytes
);
break
;
}
}
}
}
}
}
}
}
int64_t
get
FilledNumOfRes
(
SFillInfo
*
pFillInfo
,
TSKEY
ekey
,
int32_t
maxNumOfRows
)
{
int64_t
get
NumOfResWithFill
(
SFillInfo
*
pFillInfo
,
TSKEY
ekey
,
int32_t
maxNumOfRows
)
{
int64_t
*
tsList
=
(
int64_t
*
)
pFillInfo
->
pData
[
0
];
int64_t
*
tsList
=
(
int64_t
*
)
pFillInfo
->
pData
[
0
];
int32_t
numOfRows
=
taosNumOfRemainRows
(
pFillInfo
);
int32_t
numOfRows
=
taosNumOfRemainRows
(
pFillInfo
);
TSKEY
ekey1
=
ekey
;
TSKEY
ekey1
=
ekey
;
if
(
!
FILL_IS_ASC_FILL
(
pFillInfo
))
{
if
(
!
FILL_IS_ASC_FILL
(
pFillInfo
))
{
pFillInfo
->
end
Key
=
taosTimeTruncate
(
ekey
,
&
pFillInfo
->
interval
,
pFillInfo
->
precision
);
pFillInfo
->
end
=
taosTimeTruncate
(
ekey
,
&
pFillInfo
->
interval
,
pFillInfo
->
precision
);
}
}
int64_t
numOfRes
=
-
1
;
int64_t
numOfRes
=
-
1
;
...
@@ -179,20 +183,20 @@ int64_t getFilledNumOfRes(SFillInfo* pFillInfo, TSKEY ekey, int32_t maxNumOfRows
...
@@ -179,20 +183,20 @@ int64_t getFilledNumOfRes(SFillInfo* pFillInfo, TSKEY ekey, int32_t maxNumOfRows
TSKEY
lastKey
=
tsList
[
pFillInfo
->
numOfRows
-
1
];
TSKEY
lastKey
=
tsList
[
pFillInfo
->
numOfRows
-
1
];
numOfRes
=
taosTimeCountInterval
(
numOfRes
=
taosTimeCountInterval
(
lastKey
,
lastKey
,
pFillInfo
->
start
,
pFillInfo
->
currentKey
,
pFillInfo
->
interval
.
sliding
,
pFillInfo
->
interval
.
sliding
,
pFillInfo
->
interval
.
slidingUnit
,
pFillInfo
->
interval
.
slidingUnit
,
pFillInfo
->
precision
);
pFillInfo
->
precision
);
numOfRes
+=
1
;
numOfRes
+=
1
;
assert
(
numOfRes
>=
numOfRows
);
assert
(
numOfRes
>=
numOfRows
);
}
else
{
// reach the end of data
}
else
{
// reach the end of data
if
((
ekey1
<
pFillInfo
->
start
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
if
((
ekey1
<
pFillInfo
->
currentKey
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
(
ekey1
>
pFillInfo
->
start
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
)))
{
(
ekey1
>
pFillInfo
->
currentKey
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
)))
{
return
0
;
return
0
;
}
}
numOfRes
=
taosTimeCountInterval
(
numOfRes
=
taosTimeCountInterval
(
ekey1
,
ekey1
,
pFillInfo
->
start
,
pFillInfo
->
currentKey
,
pFillInfo
->
interval
.
sliding
,
pFillInfo
->
interval
.
sliding
,
pFillInfo
->
interval
.
slidingUnit
,
pFillInfo
->
interval
.
slidingUnit
,
pFillInfo
->
precision
);
pFillInfo
->
precision
);
...
@@ -203,254 +207,201 @@ int64_t getFilledNumOfRes(SFillInfo* pFillInfo, TSKEY ekey, int32_t maxNumOfRows
...
@@ -203,254 +207,201 @@ int64_t getFilledNumOfRes(SFillInfo* pFillInfo, TSKEY ekey, int32_t maxNumOfRows
}
}
int32_t
taosNumOfRemainRows
(
SFillInfo
*
pFillInfo
)
{
int32_t
taosNumOfRemainRows
(
SFillInfo
*
pFillInfo
)
{
if
(
pFillInfo
->
rowIdx
==
-
1
||
pFillInfo
->
numOfRows
==
0
)
{
if
(
pFillInfo
->
numOfRows
==
0
||
(
pFillInfo
->
numOfRows
>
0
&&
pFillInfo
->
index
>=
pFillInfo
->
numOfRows
)
)
{
return
0
;
return
0
;
}
}
return
pFillInfo
->
numOfRows
-
pFillInfo
->
rowId
x
;
return
pFillInfo
->
numOfRows
-
pFillInfo
->
inde
x
;
}
}
// todo: refactor
#define DO_INTERPOLATION(_v1, _v2, _k1, _k2, _k) ((_v1) + ((_v2) - (_v1)) * (((double)(_k)) - ((double)(_k1))) / (((double)(_k2)) - ((double)(_k1))))
static
double
linearInterpolationImpl
(
double
v1
,
double
v2
,
double
k1
,
double
k2
,
double
k
)
{
return
v1
+
(
v2
-
v1
)
*
(
k
-
k1
)
/
(
k2
-
k1
);
}
int32_t
taosGetLinearInterpolationVal
(
int32_t
type
,
SPoint
*
point1
,
SPoint
*
point2
,
SPoint
*
point
)
{
int32_t
taosGetLinearInterpolationVal
(
int32_t
type
,
SPoint
*
point1
,
SPoint
*
point2
,
SPoint
*
point
)
{
switch
(
type
)
{
double
v1
=
-
1
;
case
TSDB_DATA_TYPE_INT
:
{
double
v2
=
-
1
;
*
(
int32_t
*
)
point
->
val
=
(
int32_t
)
linearInterpolationImpl
(
*
(
int32_t
*
)
point1
->
val
,
*
(
int32_t
*
)
point2
->
val
,
(
double
)
point1
->
key
,
(
double
)
point2
->
key
,
(
double
)
point
->
key
);
GET_TYPED_DATA
(
v1
,
type
,
point1
->
val
);
break
;
GET_TYPED_DATA
(
v2
,
type
,
point2
->
val
);
}
case
TSDB_DATA_TYPE_FLOAT
:
{
double
r
=
DO_INTERPOLATION
(
v1
,
v2
,
point1
->
key
,
point2
->
key
,
point
->
key
);
*
(
float
*
)
point
->
val
=
(
float
)
linearInterpolationImpl
(
*
(
float
*
)
point1
->
val
,
*
(
float
*
)
point2
->
val
,
(
double
)
point1
->
key
,
(
double
)
point2
->
key
,
(
double
)
point
->
key
);
switch
(
type
)
{
break
;
case
TSDB_DATA_TYPE_TINYINT
:
*
(
int8_t
*
)
point
->
val
=
(
int8_t
)
r
;
break
;
};
case
TSDB_DATA_TYPE_SMALLINT
:
*
(
int16_t
*
)
point
->
val
=
(
int16_t
)
r
;
break
;
case
TSDB_DATA_TYPE_DOUBLE
:
{
case
TSDB_DATA_TYPE_INT
:
*
(
int32_t
*
)
point
->
val
=
(
int32_t
)
r
;
break
;
*
(
double
*
)
point
->
val
=
case
TSDB_DATA_TYPE_BIGINT
:
*
(
int64_t
*
)
point
->
val
=
(
int64_t
)
r
;
break
;
linearInterpolationImpl
(
*
(
double
*
)
point1
->
val
,
*
(
double
*
)
point2
->
val
,
(
double
)
point1
->
key
,
(
double
)
point2
->
key
,
(
double
)
point
->
key
);
case
TSDB_DATA_TYPE_DOUBLE
:
*
(
double
*
)
point
->
val
=
(
double
)
r
;
break
;
break
;
case
TSDB_DATA_TYPE_FLOAT
:
*
(
float
*
)
point
->
val
=
(
float
)
r
;
break
;
};
default:
case
TSDB_DATA_TYPE_TIMESTAMP
:
assert
(
0
);
case
TSDB_DATA_TYPE_BIGINT
:
{
*
(
int64_t
*
)
point
->
val
=
(
int64_t
)
linearInterpolationImpl
((
double
)(
*
(
int64_t
*
)
point1
->
val
),
(
double
)(
*
(
int64_t
*
)
point2
->
val
),
(
double
)
point1
->
key
,
(
double
)
point2
->
key
,
(
double
)
point
->
key
);
break
;
};
case
TSDB_DATA_TYPE_SMALLINT
:
{
*
(
int16_t
*
)
point
->
val
=
(
int16_t
)
linearInterpolationImpl
(
*
(
int16_t
*
)
point1
->
val
,
*
(
int16_t
*
)
point2
->
val
,
(
double
)
point1
->
key
,
(
double
)
point2
->
key
,
(
double
)
point
->
key
);
break
;
};
case
TSDB_DATA_TYPE_TINYINT
:
{
*
(
int8_t
*
)
point
->
val
=
(
int8_t
)
linearInterpolationImpl
(
*
(
int8_t
*
)
point1
->
val
,
*
(
int8_t
*
)
point2
->
val
,
(
double
)
point1
->
key
,
(
double
)
point2
->
key
,
(
double
)
point
->
key
);
break
;
};
default:
{
// TODO: Deal with interpolation with bool and strings and timestamp
return
-
1
;
}
}
}
return
0
;
return
TSDB_CODE_SUCCESS
;
}
}
static
void
setTagsValue
(
SFillInfo
*
pFillInfo
,
tFilePage
**
data
,
int32_t
num
)
{
static
void
setTagsValue
(
SFillInfo
*
pFillInfo
,
tFilePage
**
data
,
int32_t
genRows
)
{
for
(
int32_t
j
=
0
;
j
<
pFillInfo
->
numOfCols
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pFillInfo
->
numOfCols
;
++
j
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
j
];
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
j
];
if
(
TSDB_COL_IS_NORMAL_COL
(
pCol
->
flag
))
{
if
(
TSDB_COL_IS_NORMAL_COL
(
pCol
->
flag
))
{
continue
;
continue
;
}
}
char
*
val1
=
elePtrAt
(
data
[
j
]
->
data
,
pCol
->
col
.
bytes
,
num
);
char
*
val1
=
elePtrAt
(
data
[
j
]
->
data
,
pCol
->
col
.
bytes
,
genRows
);
assert
(
pCol
->
tagIndex
>=
0
&&
pCol
->
tagIndex
<
pFillInfo
->
numOfTags
);
SFillTagColInfo
*
pTag
=
&
pFillInfo
->
pTags
[
pCol
->
tagIndex
];
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfTags
;
++
i
)
{
assert
(
pTag
->
col
.
colId
==
pCol
->
col
.
colId
);
SFillTagColInfo
*
pTag
=
&
pFillInfo
->
pTags
[
i
];
if
(
pTag
->
col
.
colId
==
pCol
->
col
.
colId
)
{
assignVal
(
val1
,
pTag
->
tagVal
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
assignVal
(
val1
,
pTag
->
tagVal
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
break
;
}
}
}
}
static
void
setNullValueForRow
(
SFillInfo
*
pFillInfo
,
tFilePage
**
data
,
int32_t
numOfCol
,
int32_t
rowIndex
)
{
// the first are always the timestamp column, so start from the second column.
for
(
int32_t
i
=
1
;
i
<
numOfCol
;
++
i
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
char
*
output
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
rowIndex
);
setNull
(
output
,
pCol
->
col
.
type
,
pCol
->
col
.
bytes
);
}
}
}
}
static
void
doFillResultImpl
(
SFillInfo
*
pFillInfo
,
tFilePage
**
data
,
int32_t
*
num
,
char
**
srcData
,
int64_t
ts
,
static
void
doFillOneRowResult
(
SFillInfo
*
pFillInfo
,
tFilePage
**
data
,
char
**
srcData
,
int64_t
ts
,
bool
outOfBound
)
{
bool
outOfBound
)
{
char
*
prev
=
pFillInfo
->
prevValues
;
char
*
prevValues
=
pFillInfo
->
prevValues
;
char
*
next
=
pFillInfo
->
nextValues
;
char
*
nextValues
=
pFillInfo
->
nextValues
;
SPoint
point1
,
point2
,
point
;
SPoint
point1
,
point2
,
point
;
int32_t
step
=
GET_FORWARD_DIRECTION_FACTOR
(
pFillInfo
->
order
);
int32_t
step
=
GET_FORWARD_DIRECTION_FACTOR
(
pFillInfo
->
order
);
char
*
val
=
elePtrAt
(
data
[
0
]
->
data
,
TSDB_KEYSIZE
,
*
num
);
// set the primary timestamp column value
*
(
TSKEY
*
)
val
=
pFillInfo
->
star
t
;
int32_t
index
=
pFillInfo
->
numOfCurren
t
;
char
*
val
=
elePtrAt
(
data
[
0
]
->
data
,
TSDB_KEYSIZE
,
index
);
int32_t
numOfValCols
=
pFillInfo
->
numOfCols
-
pFillInfo
->
numOfTags
;
*
(
TSKEY
*
)
val
=
pFillInfo
->
currentKey
;
// set the other values
// set the other values
if
(
pFillInfo
->
fillT
ype
==
TSDB_FILL_PREV
)
{
if
(
pFillInfo
->
t
ype
==
TSDB_FILL_PREV
)
{
char
*
p
=
FILL_IS_ASC_FILL
(
pFillInfo
)
?
prev
Values
:
nextValues
;
char
*
p
=
FILL_IS_ASC_FILL
(
pFillInfo
)
?
prev
:
next
;
if
(
p
!=
NULL
)
{
if
(
p
!=
NULL
)
{
for
(
int32_t
i
=
1
;
i
<
numOfVal
Cols
;
++
i
)
{
for
(
int32_t
i
=
1
;
i
<
pFillInfo
->
numOf
Cols
;
++
i
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
if
(
TSDB_COL_IS_TAG
(
pCol
->
flag
))
{
char
*
val1
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
*
num
);
continue
;
if
(
isNull
(
p
+
pCol
->
col
.
offset
,
pCol
->
col
.
type
))
{
if
(
pCol
->
col
.
type
==
TSDB_DATA_TYPE_BINARY
||
pCol
->
col
.
type
==
TSDB_DATA_TYPE_NCHAR
)
{
setVardataNull
(
val1
,
pCol
->
col
.
type
);
}
else
{
setNull
(
val1
,
pCol
->
col
.
type
,
pCol
->
col
.
bytes
);
}
}
else
{
assignVal
(
val1
,
p
+
pCol
->
col
.
offset
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
}
}
}
}
else
{
// no prev value yet, set the value for NULL
for
(
int32_t
i
=
1
;
i
<
numOfValCols
;
++
i
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
char
*
val1
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
*
num
);
char
*
output
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
index
);
if
(
pCol
->
col
.
type
==
TSDB_DATA_TYPE_BINARY
||
pCol
->
col
.
type
==
TSDB_DATA_TYPE_NCHAR
)
{
assignVal
(
output
,
p
+
pCol
->
col
.
offset
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
setVardataNull
(
val1
,
pCol
->
col
.
type
);
}
else
{
setNull
(
val1
,
pCol
->
col
.
type
,
pCol
->
col
.
bytes
);
}
}
}
}
else
{
// no prev value yet, set the value for NULL
setNullValueForRow
(
pFillInfo
,
data
,
pFillInfo
->
numOfCols
,
index
);
}
}
}
else
if
(
pFillInfo
->
type
==
TSDB_FILL_LINEAR
)
{
setTagsValue
(
pFillInfo
,
data
,
*
num
);
}
else
if
(
pFillInfo
->
fillType
==
TSDB_FILL_LINEAR
)
{
// TODO : linear interpolation supports NULL value
// TODO : linear interpolation supports NULL value
if
(
prev
Values
!=
NULL
&&
!
outOfBound
)
{
if
(
prev
!=
NULL
&&
!
outOfBound
)
{
for
(
int32_t
i
=
1
;
i
<
numOfVal
Cols
;
++
i
)
{
for
(
int32_t
i
=
1
;
i
<
pFillInfo
->
numOf
Cols
;
++
i
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
if
(
TSDB_COL_IS_TAG
(
pCol
->
flag
))
{
continue
;
}
int16_t
type
=
pCol
->
col
.
type
;
int16_t
type
=
pCol
->
col
.
type
;
int16_t
bytes
=
pCol
->
col
.
bytes
;
int16_t
bytes
=
pCol
->
col
.
bytes
;
char
*
val1
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
*
num
);
char
*
val1
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
index
);
if
(
type
==
TSDB_DATA_TYPE_BINARY
||
type
==
TSDB_DATA_TYPE_NCHAR
)
{
if
(
type
==
TSDB_DATA_TYPE_BINARY
||
type
==
TSDB_DATA_TYPE_NCHAR
||
type
==
TSDB_DATA_TYPE_BOOL
)
{
setVardataNull
(
val1
,
pCol
->
col
.
type
);
continue
;
}
else
if
(
type
==
TSDB_DATA_TYPE_BOOL
)
{
setNull
(
val1
,
pCol
->
col
.
type
,
bytes
);
setNull
(
val1
,
pCol
->
col
.
type
,
bytes
);
continue
;
continue
;
}
}
point1
=
(
SPoint
){.
key
=
*
(
TSKEY
*
)(
prev
Values
),
.
val
=
prevValues
+
pCol
->
col
.
offset
};
point1
=
(
SPoint
){.
key
=
*
(
TSKEY
*
)(
prev
),
.
val
=
prev
+
pCol
->
col
.
offset
};
point2
=
(
SPoint
){.
key
=
ts
,
.
val
=
srcData
[
i
]
+
pFillInfo
->
rowId
x
*
bytes
};
point2
=
(
SPoint
){.
key
=
ts
,
.
val
=
srcData
[
i
]
+
pFillInfo
->
inde
x
*
bytes
};
point
=
(
SPoint
){.
key
=
pFillInfo
->
start
,
.
val
=
val1
};
point
=
(
SPoint
){.
key
=
pFillInfo
->
start
,
.
val
=
val1
};
taosGetLinearInterpolationVal
(
type
,
&
point1
,
&
point2
,
&
point
);
taosGetLinearInterpolationVal
(
type
,
&
point1
,
&
point2
,
&
point
);
}
}
setTagsValue
(
pFillInfo
,
data
,
*
num
);
}
else
{
}
else
{
for
(
int32_t
i
=
1
;
i
<
numOfValCols
;
++
i
)
{
setNullValueForRow
(
pFillInfo
,
data
,
pFillInfo
->
numOfCols
,
index
);
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
char
*
val1
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
*
num
);
if
(
pCol
->
col
.
type
==
TSDB_DATA_TYPE_BINARY
||
pCol
->
col
.
type
==
TSDB_DATA_TYPE_NCHAR
)
{
setVardataNull
(
val1
,
pCol
->
col
.
type
);
}
else
{
setNull
(
val1
,
pCol
->
col
.
type
,
pCol
->
col
.
bytes
);
}
}
setTagsValue
(
pFillInfo
,
data
,
*
num
);
}
}
}
else
{
/* fill the default value */
}
else
{
/* fill the default value */
for
(
int32_t
i
=
1
;
i
<
numOfVal
Cols
;
++
i
)
{
for
(
int32_t
i
=
1
;
i
<
pFillInfo
->
numOf
Cols
;
++
i
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
if
(
TSDB_COL_IS_TAG
(
pCol
->
flag
))
{
continue
;
}
char
*
val1
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
*
num
);
char
*
val1
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
index
);
assignVal
(
val1
,
(
char
*
)
&
pCol
->
fillVal
.
i
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
assignVal
(
val1
,
(
char
*
)
&
pCol
->
fillVal
.
i
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
}
}
setTagsValue
(
pFillInfo
,
data
,
*
num
);
}
}
pFillInfo
->
start
=
taosTimeAdd
(
pFillInfo
->
start
,
pFillInfo
->
interval
.
sliding
*
step
,
pFillInfo
->
interval
.
slidingUnit
,
pFillInfo
->
precision
);
setTagsValue
(
pFillInfo
,
data
,
index
);
pFillInfo
->
currentKey
=
taosTimeAdd
(
pFillInfo
->
currentKey
,
pFillInfo
->
interval
.
sliding
*
step
,
pFillInfo
->
interval
.
slidingUnit
,
pFillInfo
->
precision
);
pFillInfo
->
numOfCurrent
++
;
pFillInfo
->
numOfCurrent
++
;
(
*
num
)
+=
1
;
}
}
static
void
initBeforeAfterDataBuf
(
SFillInfo
*
pFillInfo
,
char
**
next
Values
)
{
static
void
initBeforeAfterDataBuf
(
SFillInfo
*
pFillInfo
,
char
**
next
)
{
if
(
*
next
Values
!=
NULL
)
{
if
(
*
next
!=
NULL
)
{
return
;
return
;
}
}
*
next
Values
=
calloc
(
1
,
pFillInfo
->
rowSize
);
*
next
=
calloc
(
1
,
pFillInfo
->
rowSize
);
for
(
int
i
=
1
;
i
<
pFillInfo
->
numOfCols
;
i
++
)
{
for
(
int
i
=
1
;
i
<
pFillInfo
->
numOfCols
;
i
++
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
setNull
(
*
next
+
pCol
->
col
.
offset
,
pCol
->
col
.
type
,
pCol
->
col
.
bytes
);
if
(
pCol
->
col
.
type
==
TSDB_DATA_TYPE_BINARY
||
pCol
->
col
.
type
==
TSDB_DATA_TYPE_NCHAR
)
{
setVardataNull
(
*
nextValues
+
pCol
->
col
.
offset
,
pCol
->
col
.
type
);
}
else
{
setNull
(
*
nextValues
+
pCol
->
col
.
offset
,
pCol
->
col
.
type
,
pCol
->
col
.
bytes
);
}
}
}
static
void
copyCurrentRowIntoBuf
(
SFillInfo
*
pFillInfo
,
char
**
srcData
,
int32_t
numOfTags
,
char
*
buf
)
{
int32_t
rowIndex
=
pFillInfo
->
index
;
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
memcpy
(
buf
+
pCol
->
col
.
offset
,
srcData
[
i
]
+
rowIndex
*
pCol
->
col
.
bytes
,
pCol
->
col
.
bytes
);
}
}
}
}
int32_t
generateDataBlockImpl
(
SFillInfo
*
pFillInfo
,
tFilePage
**
data
,
int32_t
numOfRows
,
int32_t
outputRows
,
char
**
srcData
)
{
static
int32_t
fillResultImpl
(
SFillInfo
*
pFillInfo
,
tFilePage
**
data
,
int32_t
outputRows
)
{
int32_t
num
=
0
;
pFillInfo
->
numOfCurrent
=
0
;
pFillInfo
->
numOfCurrent
=
0
;
char
**
prevValues
=
&
pFillInfo
->
prevValues
;
char
**
srcData
=
pFillInfo
->
pData
;
char
**
nextValues
=
&
pFillInfo
->
nextValues
;
char
**
prev
=
&
pFillInfo
->
prevValues
;
char
**
next
=
&
pFillInfo
->
nextValues
;
int32_t
numOfTags
=
pFillInfo
->
numOfTags
;
int32_t
numOfTags
=
pFillInfo
->
numOfTags
;
int32_t
step
=
GET_FORWARD_DIRECTION_FACTOR
(
pFillInfo
->
order
);
int32_t
step
=
GET_FORWARD_DIRECTION_FACTOR
(
pFillInfo
->
order
);
if
(
numOfRows
==
0
)
{
/*
* These data are generated according to fill strategy, since the current timestamp is out of time window of
* real result set. Note that we need to keep the direct previous result rows, to generated the filled data.
*/
while
(
num
<
outputRows
)
{
doFillResultImpl
(
pFillInfo
,
data
,
&
num
,
srcData
,
pFillInfo
->
start
,
true
);
}
pFillInfo
->
numOfTotal
+=
pFillInfo
->
numOfCurrent
;
return
outputRows
;
if
(
FILL_IS_ASC_FILL
(
pFillInfo
))
{
assert
(
pFillInfo
->
currentKey
>=
pFillInfo
->
start
);
}
else
{
}
else
{
while
(
1
)
{
assert
(
pFillInfo
->
currentKey
<=
pFillInfo
->
start
);
int64_t
ts
=
((
int64_t
*
)
pFillInfo
->
pData
[
0
])[
pFillInfo
->
rowIdx
];
}
if
((
pFillInfo
->
start
<
ts
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
while
(
pFillInfo
->
numOfCurrent
<
outputRows
)
{
(
pFillInfo
->
start
>
ts
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
)))
{
int64_t
ts
=
((
int64_t
*
)
pFillInfo
->
pData
[
0
])[
pFillInfo
->
index
];
/* set the next value for interpolation */
initBeforeAfterDataBuf
(
pFillInfo
,
nextValues
);
int32_t
offset
=
pFillInfo
->
rowIdx
;
if
((
pFillInfo
->
currentKey
<
ts
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
-
numOfTags
;
++
i
)
{
(
pFillInfo
->
currentKey
>
ts
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
))
)
{
SFillColInfo
*
pCol
=
&
pFillInfo
->
pFillCol
[
i
];
/* set the next value for interpolation */
memcpy
(
*
nextValues
+
pCol
->
col
.
offset
,
srcData
[
i
]
+
offset
*
pCol
->
col
.
bytes
,
pCol
->
col
.
bytes
);
initBeforeAfterDataBuf
(
pFillInfo
,
next
);
}
copyCurrentRowIntoBuf
(
pFillInfo
,
srcData
,
numOfTags
,
*
next
);
}
}
if
(((
pFillInfo
->
start
<
ts
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
if
(((
pFillInfo
->
currentKey
<
ts
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
(
pFillInfo
->
currentKey
>
ts
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
)))
&&
(
pFillInfo
->
start
>
ts
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
)))
&&
num
<
outputRows
)
{
pFillInfo
->
numOfCurrent
<
outputRows
)
{
while
(((
pFillInfo
->
start
<
ts
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
// fill the gap between two actual input rows
(
pFillInfo
->
start
>
ts
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
)))
&&
num
<
outputRows
)
{
while
(((
pFillInfo
->
currentKey
<
ts
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
doFillResultImpl
(
pFillInfo
,
data
,
&
num
,
srcData
,
ts
,
false
);
(
pFillInfo
->
currentKey
>
ts
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
)))
&&
pFillInfo
->
numOfCurrent
<
outputRows
)
{
doFillOneRowResult
(
pFillInfo
,
data
,
srcData
,
ts
,
false
);
}
}
/* output buffer is full, abort */
// output buffer is full, abort
if
((
num
==
outputRows
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
(
num
<
0
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
))
)
{
if
(
pFillInfo
->
numOfCurrent
==
outputRows
)
{
pFillInfo
->
numOfTotal
+=
pFillInfo
->
numOfCurrent
;
pFillInfo
->
numOfTotal
+=
pFillInfo
->
numOfCurrent
;
return
outputRows
;
return
outputRows
;
}
}
}
else
{
}
else
{
assert
(
pFillInfo
->
start
==
ts
);
assert
(
pFillInfo
->
currentKey
==
ts
);
initBeforeAfterDataBuf
(
pFillInfo
,
prevValues
);
initBeforeAfterDataBuf
(
pFillInfo
,
prev
);
// assign rows to dst buffer
// assign rows to dst buffer
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pFillInfo
->
numOfCols
;
++
i
)
{
...
@@ -459,59 +410,81 @@ int32_t generateDataBlockImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t nu
...
@@ -459,59 +410,81 @@ int32_t generateDataBlockImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t nu
continue
;
continue
;
}
}
char
*
val1
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
num
);
char
*
output
=
elePtrAt
(
data
[
i
]
->
data
,
pCol
->
col
.
bytes
,
pFillInfo
->
numOfCurrent
);
char
*
src
=
elePtrAt
(
srcData
[
i
],
pCol
->
col
.
bytes
,
pFillInfo
->
rowId
x
);
char
*
src
=
elePtrAt
(
srcData
[
i
],
pCol
->
col
.
bytes
,
pFillInfo
->
inde
x
);
if
(
i
==
0
||
if
(
i
==
0
||
(
pCol
->
functionId
!=
TSDB_FUNC_COUNT
&&
!
isNull
(
src
,
pCol
->
col
.
type
))
||
(
pCol
->
functionId
!=
TSDB_FUNC_COUNT
&&
!
isNull
(
src
,
pCol
->
col
.
type
))
||
(
pCol
->
functionId
==
TSDB_FUNC_COUNT
&&
GET_INT64_VAL
(
src
)
!=
0
))
{
(
pCol
->
functionId
==
TSDB_FUNC_COUNT
&&
GET_INT64_VAL
(
src
)
!=
0
))
{
assignVal
(
val1
,
src
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
assignVal
(
output
,
src
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
memcpy
(
*
prevValues
+
pCol
->
col
.
offset
,
src
,
pCol
->
col
.
bytes
);
memcpy
(
*
prev
+
pCol
->
col
.
offset
,
src
,
pCol
->
col
.
bytes
);
}
else
{
// i > 0 and data is null , do interpolation
}
else
{
// i > 0 and data is null , do interpolation
if
(
pFillInfo
->
fillT
ype
==
TSDB_FILL_PREV
)
{
if
(
pFillInfo
->
t
ype
==
TSDB_FILL_PREV
)
{
assignVal
(
val1
,
*
prevValues
+
pCol
->
col
.
offset
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
assignVal
(
output
,
*
prev
+
pCol
->
col
.
offset
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
}
else
if
(
pFillInfo
->
fillT
ype
==
TSDB_FILL_LINEAR
)
{
}
else
if
(
pFillInfo
->
t
ype
==
TSDB_FILL_LINEAR
)
{
assignVal
(
val1
,
src
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
assignVal
(
output
,
src
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
memcpy
(
*
prevValues
+
pCol
->
col
.
offset
,
src
,
pCol
->
col
.
bytes
);
memcpy
(
*
prev
+
pCol
->
col
.
offset
,
src
,
pCol
->
col
.
bytes
);
}
else
{
}
else
{
assignVal
(
val1
,
(
char
*
)
&
pCol
->
fillVal
.
i
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
assignVal
(
output
,
(
char
*
)
&
pCol
->
fillVal
.
i
,
pCol
->
col
.
bytes
,
pCol
->
col
.
type
);
}
}
}
}
}
}
// set the tag value for final result
// set the tag value for final result
setTagsValue
(
pFillInfo
,
data
,
num
);
setTagsValue
(
pFillInfo
,
data
,
pFillInfo
->
numOfCurrent
);
pFillInfo
->
start
=
taosTimeAdd
(
pFillInfo
->
start
,
pFillInfo
->
interval
.
sliding
*
step
,
pFillInfo
->
interval
.
slidingUnit
,
pFillInfo
->
precision
);
pFillInfo
->
rowIdx
+=
1
;
pFillInfo
->
numOfCurrent
+=
1
;
pFillInfo
->
currentKey
=
taosTimeAdd
(
pFillInfo
->
currentKey
,
pFillInfo
->
interval
.
sliding
*
step
,
num
+=
1
;
pFillInfo
->
interval
.
slidingUnit
,
pFillInfo
->
precision
);
pFillInfo
->
index
+=
1
;
pFillInfo
->
numOfCurrent
+=
1
;
}
}
if
((
pFillInfo
->
rowIdx
>=
pFillInfo
->
numOfRows
&&
FILL_IS_ASC_FILL
(
pFillInfo
))
||
if
(
pFillInfo
->
index
>=
pFillInfo
->
numOfRows
||
pFillInfo
->
numOfCurrent
>=
outputRows
)
{
(
pFillInfo
->
rowIdx
<
0
&&
!
FILL_IS_ASC_FILL
(
pFillInfo
))
||
num
>=
outputRows
)
{
if
(
pFillInfo
->
rowIdx
>=
pFillInfo
->
numOfRows
||
pFillInfo
->
rowIdx
<
0
)
{
pFillInfo
->
rowIdx
=
-
1
;
pFillInfo
->
numOfRows
=
0
;
/* the raw data block is exhausted, next value does not exists */
/* the raw data block is exhausted, next value does not exists */
taosTFree
(
*
nextValues
);
if
(
pFillInfo
->
index
>=
pFillInfo
->
numOfRows
)
{
taosTFree
(
*
next
);
}
}
pFillInfo
->
numOfTotal
+=
pFillInfo
->
numOfCurrent
;
pFillInfo
->
numOfTotal
+=
pFillInfo
->
numOfCurrent
;
return
num
;
return
pFillInfo
->
numOfCurrent
;
}
}
}
}
return
pFillInfo
->
numOfCurrent
;
}
static
int64_t
fillExternalResults
(
SFillInfo
*
pFillInfo
,
tFilePage
**
output
,
int64_t
resultCapacity
)
{
/*
* These data are generated according to fill strategy, since the current timestamp is out of the time window of
* real result set. Note that we need to keep the direct previous result rows, to generated the filled data.
*/
pFillInfo
->
numOfCurrent
=
0
;
while
(
pFillInfo
->
numOfCurrent
<
resultCapacity
)
{
doFillOneRowResult
(
pFillInfo
,
output
,
pFillInfo
->
pData
,
pFillInfo
->
start
,
true
);
}
}
pFillInfo
->
numOfTotal
+=
pFillInfo
->
numOfCurrent
;
assert
(
pFillInfo
->
numOfCurrent
==
resultCapacity
);
return
resultCapacity
;
}
}
int64_t
taosGenerateDataBlock
(
SFillInfo
*
pFillInfo
,
tFilePage
**
output
,
int32_t
capacity
)
{
int64_t
taosFillResultDataBlock
(
SFillInfo
*
pFillInfo
,
tFilePage
**
output
,
int32_t
capacity
)
{
int32_t
remain
=
taosNumOfRemainRows
(
pFillInfo
);
// todo use iterator?
int32_t
remain
=
taosNumOfRemainRows
(
pFillInfo
);
int64_t
numOfRes
=
getNumOfResWithFill
(
pFillInfo
,
pFillInfo
->
end
,
capacity
);
assert
(
numOfRes
<=
capacity
);
// no data existed for fill operation now, append result according to the fill strategy
if
(
remain
==
0
)
{
return
fillExternalResults
(
pFillInfo
,
output
,
numOfRes
);
}
fillResultImpl
(
pFillInfo
,
output
,
numOfRes
);
assert
(
numOfRes
==
pFillInfo
->
numOfCurrent
);
int32_t
rows
=
(
int32_t
)
getFilledNumOfRes
(
pFillInfo
,
pFillInfo
->
endKey
,
capacity
);
qDebug
(
"generated fill result, src block:%d, index:%d, startKey:%"
PRId64
", currentKey:%"
PRId64
", current:%d, total:%d"
,
int32_t
numOfRes
=
generateDataBlockImpl
(
pFillInfo
,
output
,
remain
,
rows
,
pFillInfo
->
pData
);
pFillInfo
->
numOfRows
,
pFillInfo
->
index
,
pFillInfo
->
start
,
pFillInfo
->
currentKey
,
pFillInfo
->
numOfCurrent
,
assert
(
numOfRes
==
rows
);
pFillInfo
->
numOfTotal
);
return
numOfRes
;
return
numOfRes
;
}
}
src/util/src/tcompare.c
浏览文件 @
7314a0dd
#include "taosdef.h"
#include "taosdef.h"
#include "tcompare.h"
#include "tcompare.h"
#include "tarray.h"
#include "tarray.h"
#include "tutil.h"
int32_t
compareInt32Val
(
const
void
*
pLeft
,
const
void
*
pRight
)
{
int32_t
compareInt32Val
(
const
void
*
pLeft
,
const
void
*
pRight
)
{
int32_t
left
=
GET_INT32_VAL
(
pLeft
),
right
=
GET_INT32_VAL
(
pRight
);
int32_t
left
=
GET_INT32_VAL
(
pLeft
),
right
=
GET_INT32_VAL
(
pRight
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录