Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
XKZJ_WCL
TDengine
提交
f1b6c002
T
TDengine
项目概览
XKZJ_WCL
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
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,发现更多精彩内容 >>
提交
f1b6c002
编写于
7月 10, 2020
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[td-225] opt query perf
上级
709a007c
变更
12
隐藏空白更改
内联
并排
Showing
12 changed file
with
266 addition
and
218 deletion
+266
-218
cmake/platform.inc
cmake/platform.inc
+1
-1
src/client/inc/tscUtil.h
src/client/inc/tscUtil.h
+3
-1
src/client/inc/tsclient.h
src/client/inc/tsclient.h
+38
-1
src/client/src/tscFunctionImpl.c
src/client/src/tscFunctionImpl.c
+8
-4
src/client/src/tscSql.c
src/client/src/tscSql.c
+1
-1
src/client/src/tscSubquery.c
src/client/src/tscSubquery.c
+17
-15
src/client/src/tscUtil.c
src/client/src/tscUtil.c
+38
-40
src/common/src/ttypes.c
src/common/src/ttypes.c
+25
-25
src/inc/taosdef.h
src/inc/taosdef.h
+26
-1
src/query/src/qExecutor.c
src/query/src/qExecutor.c
+56
-77
src/tsdb/src/tsdbRead.c
src/tsdb/src/tsdbRead.c
+52
-51
src/util/inc/tarray.h
src/util/inc/tarray.h
+1
-1
未找到文件。
cmake/platform.inc
浏览文件 @
f1b6c002
...
...
@@ -109,7 +109,7 @@ IF (TD_LINUX_64)
IF
(
$
{
CMAKE_CXX_COMPILER_ID
}
MATCHES
"Clang"
)
SET
(
COMMON_FLAGS
"-std=gnu99 -Wall -Werror -Wno-missing-braces -fPIC -g3 -gdwarf-2 -msse4.2 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE"
)
ELSE
()
SET
(
COMMON_FLAGS
"-std=gnu99 -Wall -Werror -fPIC -malign-double -g3 -gdwarf-2 -malign-stringops -msse4.2 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE"
)
SET
(
COMMON_FLAGS
"-std=gnu99 -Wall -Werror -
pg -
fPIC -malign-double -g3 -gdwarf-2 -malign-stringops -msse4.2 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE"
)
ENDIF
()
ELSE
()
SET
(
COMMON_FLAGS
"-std=gnu99 -Wall -Werror -fPIC -g -fsigned-char -fpack-struct=8 -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE"
)
...
...
src/client/inc/tscUtil.h
浏览文件 @
f1b6c002
...
...
@@ -160,7 +160,9 @@ void tscFieldInfoUpdateOffsetForInterResult(SQueryInfo* pQueryInfo);
int16_t
tscFieldInfoGetOffset
(
SQueryInfo
*
pQueryInfo
,
int32_t
index
);
void
tscFieldInfoClear
(
SFieldInfo
*
pFieldInfo
);
int32_t
tscNumOfFields
(
SQueryInfo
*
pQueryInfo
);
static
FORCE_INLINE
int32_t
tscNumOfFields
(
SQueryInfo
*
pQueryInfo
)
{
return
pQueryInfo
->
fieldsInfo
.
numOfOutput
;
}
int32_t
tscFieldInfoCompare
(
const
SFieldInfo
*
pFieldInfo1
,
const
SFieldInfo
*
pFieldInfo2
);
void
addExprParams
(
SSqlExpr
*
pExpr
,
char
*
argument
,
int32_t
type
,
int32_t
bytes
,
int16_t
tableIndex
);
...
...
src/client/inc/tsclient.h
浏览文件 @
f1b6c002
...
...
@@ -411,7 +411,44 @@ char *tscGetErrorMsgPayload(SSqlCmd *pCmd);
int32_t
tscInvalidSQLErrMsg
(
char
*
msg
,
const
char
*
additionalInfo
,
const
char
*
sql
);
int32_t
tscToSQLCmd
(
SSqlObj
*
pSql
,
struct
SSqlInfo
*
pInfo
);
void
tscGetResultColumnChr
(
SSqlRes
*
pRes
,
SFieldInfo
*
pFieldInfo
,
int32_t
column
);
//void tscGetResultColumnChr(SSqlRes *pRes, SFieldInfo* pFieldInfo, int32_t column);
static
FORCE_INLINE
void
tscGetResultColumnChr
(
SSqlRes
*
pRes
,
SFieldInfo
*
pFieldInfo
,
int32_t
columnIndex
)
{
SFieldSupInfo
*
pInfo
=
TARRAY_GET_ELEM
(
pFieldInfo
->
pSupportInfo
,
columnIndex
);
assert
(
pInfo
->
pSqlExpr
!=
NULL
);
int32_t
type
=
pInfo
->
pSqlExpr
->
resType
;
int32_t
bytes
=
pInfo
->
pSqlExpr
->
resBytes
;
char
*
pData
=
pRes
->
data
+
pInfo
->
pSqlExpr
->
offset
*
pRes
->
numOfRows
+
bytes
*
pRes
->
row
;
if
(
type
==
TSDB_DATA_TYPE_NCHAR
||
type
==
TSDB_DATA_TYPE_BINARY
)
{
int32_t
realLen
=
varDataLen
(
pData
);
assert
(
realLen
<=
bytes
-
VARSTR_HEADER_SIZE
);
if
(
isNull
(
pData
,
type
))
{
pRes
->
tsrow
[
columnIndex
]
=
NULL
;
}
else
{
pRes
->
tsrow
[
columnIndex
]
=
((
tstr
*
)
pData
)
->
data
;
}
if
(
realLen
<
pInfo
->
pSqlExpr
->
resBytes
-
VARSTR_HEADER_SIZE
)
{
// todo refactor
*
(
pData
+
realLen
+
VARSTR_HEADER_SIZE
)
=
0
;
}
pRes
->
length
[
columnIndex
]
=
realLen
;
}
else
{
assert
(
bytes
==
tDataTypeDesc
[
type
].
nSize
);
if
(
isNull
(
pData
,
type
))
{
pRes
->
tsrow
[
columnIndex
]
=
NULL
;
}
else
{
pRes
->
tsrow
[
columnIndex
]
=
pData
;
}
pRes
->
length
[
columnIndex
]
=
bytes
;
}
}
extern
void
*
tscCacheHandle
;
extern
void
*
tscTmr
;
...
...
src/client/src/tscFunctionImpl.c
浏览文件 @
f1b6c002
...
...
@@ -2952,10 +2952,14 @@ static void tag_project_function(SQLFunctionCtx *pCtx) {
INC_INIT_VAL
(
pCtx
,
pCtx
->
size
);
assert
(
pCtx
->
inputBytes
==
pCtx
->
outputBytes
);
for
(
int32_t
i
=
0
;
i
<
pCtx
->
size
;
++
i
)
{
tVariantDump
(
&
pCtx
->
tag
,
pCtx
->
aOutputBuf
,
pCtx
->
outputType
,
true
);
tVariantDump
(
&
pCtx
->
tag
,
pCtx
->
aOutputBuf
,
pCtx
->
outputType
,
true
);
char
*
data
=
pCtx
->
aOutputBuf
;
pCtx
->
aOutputBuf
+=
pCtx
->
outputBytes
;
// directly copy from the first one
for
(
int32_t
i
=
1
;
i
<
pCtx
->
size
;
++
i
)
{
memmove
(
pCtx
->
aOutputBuf
,
data
,
pCtx
->
outputBytes
);
pCtx
->
aOutputBuf
+=
pCtx
->
outputBytes
;
}
}
...
...
src/client/src/tscSql.c
浏览文件 @
f1b6c002
...
...
@@ -403,7 +403,7 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
taos_fetch_rows_a
(
res
,
waitForRetrieveRsp
,
pSql
->
pTscObj
);
sem_wait
(
&
pSql
->
rspSem
);
}
return
doSetResultRowData
(
pSql
,
true
);
}
...
...
src/client/src/tscSubquery.c
浏览文件 @
f1b6c002
...
...
@@ -2108,9 +2108,9 @@ static char *getArithemicInputSrc(void *param, const char *name, int32_t colId)
void
**
doSetResultRowData
(
SSqlObj
*
pSql
,
bool
finalResult
)
{
SSqlCmd
*
pCmd
=
&
pSql
->
cmd
;
SSqlRes
*
pRes
=
&
pSql
->
res
;
assert
(
pRes
->
row
>=
0
&&
pRes
->
row
<=
pRes
->
numOfRows
);
if
(
pCmd
->
command
==
TSDB_SQL_TABLE_JOIN_RETRIEVE
)
{
if
(
pRes
->
completed
)
{
tfree
(
pRes
->
tsrow
);
...
...
@@ -2118,29 +2118,31 @@ void **doSetResultRowData(SSqlObj *pSql, bool finalResult) {
return
pRes
->
tsrow
;
}
if
(
pRes
->
row
>=
pRes
->
numOfRows
)
{
// all the results has returned to invoker
tfree
(
pRes
->
tsrow
);
return
pRes
->
tsrow
;
}
SQueryInfo
*
pQueryInfo
=
tscGetQueryInfoDetail
(
pCmd
,
pCmd
->
clauseIndex
);
size_t
size
=
tscNumOfFields
(
pQueryInfo
);
for
(
int
i
=
0
;
i
<
size
;
++
i
)
{
SFieldSupInfo
*
pSup
=
tscFieldInfoGetSupp
(
&
pQueryInfo
->
fields
Info
,
i
);
SFieldSupInfo
*
pSup
=
TARRAY_GET_ELEM
(
pQueryInfo
->
fieldsInfo
.
pSupport
Info
,
i
);
if
(
pSup
->
pSqlExpr
!=
NULL
)
{
tscGetResultColumnChr
(
pRes
,
&
pQueryInfo
->
fieldsInfo
,
i
);
}
// primary key column cannot be null in interval query, no need to check
if
(
i
==
0
&&
pQueryInfo
->
intervalTime
>
0
)
{
continue
;
}
TAOS_FIELD
*
pField
=
tscFieldInfoGetField
(
&
pQueryInfo
->
fieldsInfo
,
i
);
transferNcharData
(
pSql
,
i
,
pField
);
TAOS_FIELD
*
pField
=
TARRAY_GET_ELEM
(
pQueryInfo
->
fieldsInfo
.
pFields
,
i
);
if
(
pRes
->
tsrow
[
i
]
!=
NULL
&&
pField
->
type
==
TSDB_DATA_TYPE_NCHAR
)
{
transferNcharData
(
pSql
,
i
,
pField
);
}
// calculate the result from several other columns
if
(
pSup
->
pArithExprInfo
!=
NULL
)
{
if
(
pRes
->
pArithSup
==
NULL
)
{
...
...
@@ -2150,10 +2152,10 @@ void **doSetResultRowData(SSqlObj *pSql, bool finalResult) {
sas
->
numOfCols
=
tscSqlExprNumOfExprs
(
pQueryInfo
);
sas
->
exprList
=
pQueryInfo
->
exprList
;
sas
->
data
=
calloc
(
sas
->
numOfCols
,
POINTER_BYTES
);
pRes
->
pArithSup
=
sas
;
}
if
(
pRes
->
buffer
[
i
]
==
NULL
)
{
TAOS_FIELD
*
field
=
tscFieldInfoGetField
(
&
pQueryInfo
->
fieldsInfo
,
i
);
pRes
->
buffer
[
i
]
=
malloc
(
field
->
bytes
);
...
...
@@ -2163,13 +2165,13 @@ void **doSetResultRowData(SSqlObj *pSql, bool finalResult) {
SSqlExpr
*
pExpr
=
tscSqlExprGet
(
pQueryInfo
,
k
);
pRes
->
pArithSup
->
data
[
k
]
=
(
pRes
->
data
+
pRes
->
numOfRows
*
pExpr
->
offset
)
+
pRes
->
row
*
pExpr
->
resBytes
;
}
tExprTreeCalcTraverse
(
pRes
->
pArithSup
->
pArithExpr
->
pExpr
,
1
,
pRes
->
buffer
[
i
],
pRes
->
pArithSup
,
TSDB_ORDER_ASC
,
getArithemicInputSrc
);
pRes
->
tsrow
[
i
]
=
pRes
->
buffer
[
i
];
}
}
pRes
->
row
++
;
// index increase one-step
return
pRes
->
tsrow
;
}
...
...
src/client/src/tscUtil.c
浏览文件 @
f1b6c002
...
...
@@ -794,7 +794,7 @@ SFieldSupInfo* tscFieldInfoAppend(SFieldInfo* pFieldInfo, TAOS_FIELD* pField) {
}
SFieldSupInfo
*
tscFieldInfoGetSupp
(
SFieldInfo
*
pFieldInfo
,
int32_t
index
)
{
return
taosArrayGet
(
pFieldInfo
->
pSupportInfo
,
index
);
return
TARRAY_GET_ELEM
(
pFieldInfo
->
pSupportInfo
,
index
);
}
SFieldSupInfo
*
tscFieldInfoInsert
(
SFieldInfo
*
pFieldInfo
,
int32_t
index
,
TAOS_FIELD
*
field
)
{
...
...
@@ -858,11 +858,9 @@ void tscFieldInfoCopy(SFieldInfo* dst, const SFieldInfo* src) {
}
TAOS_FIELD
*
tscFieldInfoGetField
(
SFieldInfo
*
pFieldInfo
,
int32_t
index
)
{
return
taosArrayGet
(
pFieldInfo
->
pFields
,
index
);
return
TARRAY_GET_ELEM
(
pFieldInfo
->
pFields
,
index
);
}
int32_t
tscNumOfFields
(
SQueryInfo
*
pQueryInfo
)
{
return
pQueryInfo
->
fieldsInfo
.
numOfOutput
;
}
int16_t
tscFieldInfoGetOffset
(
SQueryInfo
*
pQueryInfo
,
int32_t
index
)
{
SFieldSupInfo
*
pInfo
=
tscFieldInfoGetSupp
(
&
pQueryInfo
->
fieldsInfo
,
index
);
assert
(
pInfo
!=
NULL
);
...
...
@@ -2086,42 +2084,42 @@ void tscTryQueryNextClause(SSqlObj* pSql, void (*queryFp)()) {
}
}
void
tscGetResultColumnChr
(
SSqlRes
*
pRes
,
SFieldInfo
*
pFieldInfo
,
int32_t
columnIndex
)
{
SFieldSupInfo
*
pInfo
=
taosArrayGet
(
pFieldInfo
->
pSupportInfo
,
columnIndex
);
assert
(
pInfo
->
pSqlExpr
!=
NULL
);
int32_t
type
=
pInfo
->
pSqlExpr
->
resType
;
int32_t
bytes
=
pInfo
->
pSqlExpr
->
resBytes
;
char
*
pData
=
pRes
->
data
+
pInfo
->
pSqlExpr
->
offset
*
pRes
->
numOfRows
+
bytes
*
pRes
->
row
;
if
(
type
==
TSDB_DATA_TYPE_NCHAR
||
type
==
TSDB_DATA_TYPE_BINARY
)
{
int32_t
realLen
=
varDataLen
(
pData
);
assert
(
realLen
<=
bytes
-
VARSTR_HEADER_SIZE
);
if
(
isNull
(
pData
,
type
))
{
pRes
->
tsrow
[
columnIndex
]
=
NULL
;
}
else
{
pRes
->
tsrow
[
columnIndex
]
=
((
tstr
*
)
pData
)
->
data
;
}
if
(
realLen
<
pInfo
->
pSqlExpr
->
resBytes
-
VARSTR_HEADER_SIZE
)
{
// todo refactor
*
(
pData
+
realLen
+
VARSTR_HEADER_SIZE
)
=
0
;
}
pRes
->
length
[
columnIndex
]
=
realLen
;
}
else
{
assert
(
bytes
==
tDataTypeDesc
[
type
].
nSize
);
if
(
isNull
(
pData
,
type
))
{
pRes
->
tsrow
[
columnIndex
]
=
NULL
;
}
else
{
pRes
->
tsrow
[
columnIndex
]
=
pData
;
}
pRes
->
length
[
columnIndex
]
=
bytes
;
}
}
//
void tscGetResultColumnChr(SSqlRes* pRes, SFieldInfo* pFieldInfo, int32_t columnIndex) {
// SFieldSupInfo* pInfo = TARRAY_GET_ELEM
(pFieldInfo->pSupportInfo, columnIndex);
//
assert(pInfo->pSqlExpr != NULL);
//
//
int32_t type = pInfo->pSqlExpr->resType;
//
int32_t bytes = pInfo->pSqlExpr->resBytes;
//
//
char* pData = pRes->data + pInfo->pSqlExpr->offset * pRes->numOfRows + bytes * pRes->row;
//
//
if (type == TSDB_DATA_TYPE_NCHAR || type == TSDB_DATA_TYPE_BINARY) {
//
int32_t realLen = varDataLen(pData);
//
assert(realLen <= bytes - VARSTR_HEADER_SIZE);
//
//
if (isNull(pData, type)) {
//
pRes->tsrow[columnIndex] = NULL;
//
} else {
//
pRes->tsrow[columnIndex] = ((tstr*)pData)->data;
//
}
//
//
if (realLen < pInfo->pSqlExpr->resBytes - VARSTR_HEADER_SIZE) { // todo refactor
//
*(pData + realLen + VARSTR_HEADER_SIZE) = 0;
//
}
//
//
pRes->length[columnIndex] = realLen;
//
} else {
//
assert(bytes == tDataTypeDesc[type].nSize);
//
//
if (isNull(pData, type)) {
//
pRes->tsrow[columnIndex] = NULL;
//
} else {
//
pRes->tsrow[columnIndex] = pData;
//
}
//
//
pRes->length[columnIndex] = bytes;
//
}
//
}
void
*
malloc_throw
(
size_t
size
)
{
void
*
p
=
malloc
(
size
);
...
...
src/common/src/ttypes.c
浏览文件 @
f1b6c002
...
...
@@ -367,31 +367,31 @@ bool isValidDataType(int32_t type) {
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
;
};
}
//
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
)
{
if
(
type
==
TSDB_DATA_TYPE_BINARY
)
{
...
...
src/inc/taosdef.h
浏览文件 @
f1b6c002
...
...
@@ -160,7 +160,32 @@ extern tDataTypeDescriptor tDataTypeDesc[11];
#define POINTER_BYTES sizeof(void *) // 8 by default assert(sizeof(ptrdiff_t) == sizseof(void*)
bool
isValidDataType
(
int32_t
type
);
bool
isNull
(
const
char
*
val
,
int32_t
type
);
//bool isNull(const char *val, int32_t type);
static
inline
__attribute__
((
always_inline
))
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
setNull
(
char
*
val
,
int32_t
type
,
int32_t
bytes
);
...
...
src/query/src/qExecutor.c
浏览文件 @
f1b6c002
...
...
@@ -44,7 +44,7 @@
#define SET_MASTER_SCAN_FLAG(runtime) ((runtime)->scanFlag = MASTER_SCAN)
#define SET_REVERSE_SCAN_FLAG(runtime) ((runtime)->scanFlag = REVERSE_SCAN)
#define GET_QINFO_ADDR(x) ((
void
*)((char *)(x)-offsetof(SQInfo, runtimeEnv)))
#define GET_QINFO_ADDR(x) ((
SQInfo
*)((char *)(x)-offsetof(SQInfo, runtimeEnv)))
#define GET_COL_DATA_POS(query, index, step) ((query)->pos + (index) * (step))
#define SWITCH_ORDER(n) (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))
...
...
@@ -351,27 +351,6 @@ static bool hasTagValOutput(SQuery* pQuery) {
return
false
;
}
static
SDataStatis
*
getStatisInfo
(
SQuery
*
pQuery
,
SDataStatis
*
pStatis
,
int32_t
numOfCols
,
int32_t
index
)
{
// for a tag column, no corresponding field info
SColIndex
*
pColIndex
=
&
pQuery
->
pSelectExpr
[
index
].
base
.
colInfo
;
if
(
TSDB_COL_IS_TAG
(
pColIndex
->
flag
))
{
return
NULL
;
}
/*
* Choose the right column field info by field id, since the file block may be out of date,
* which means the newest table schema is not equalled to the schema of this block.
* TODO: speedup by using bsearch
*/
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
if
(
pColIndex
->
colId
==
pStatis
[
i
].
colId
)
{
return
&
pStatis
[
i
];
}
}
return
NULL
;
}
/**
* @param pQuery
* @param col
...
...
@@ -380,19 +359,14 @@ static SDataStatis *getStatisInfo(SQuery *pQuery, SDataStatis *pStatis, int32_t
* @param pColStatis
* @return
*/
static
bool
hasNullValue
(
SQuery
*
pQuery
,
int32_t
col
,
int32_t
numOfCols
,
SDataStatis
*
pStatis
,
SDataStatis
**
pColStatis
)
{
SColIndex
*
pColIndex
=
&
pQuery
->
pSelectExpr
[
col
].
base
.
colInfo
;
if
(
TSDB_COL_IS_TAG
(
pColIndex
->
flag
))
{
return
false
;
}
// query on primary timestamp column, not null value at all
if
(
pColIndex
->
colId
==
PRIMARYKEY_TIMESTAMP_COL_INDEX
)
{
static
bool
hasNullValue
(
SColIndex
*
pColIndex
,
SDataStatis
*
pStatis
,
SDataStatis
**
pColStatis
)
{
if
(
TSDB_COL_IS_TAG
(
pColIndex
->
flag
)
||
pColIndex
->
colId
==
PRIMARYKEY_TIMESTAMP_COL_INDEX
)
{
return
false
;
}
if
(
pStatis
!=
NULL
)
{
*
pColStatis
=
getStatisInfo
(
pQuery
,
pStatis
,
numOfCols
,
col
);
*
pColStatis
=
&
pStatis
[
pColIndex
->
colIndex
];
assert
((
*
pColStatis
)
->
colId
==
pColIndex
->
colId
);
}
else
{
*
pColStatis
=
NULL
;
}
...
...
@@ -842,8 +816,8 @@ static char *getDataBlock(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas
if
(
pDataBlock
==
NULL
)
{
return
NULL
;
}
char
*
dataBlock
=
NULL
;
char
*
dataBlock
=
NULL
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQLFunctionCtx
*
pCtx
=
pRuntimeEnv
->
pCtx
;
...
...
@@ -887,10 +861,14 @@ static char *getDataBlock(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas
}
else
{
// other type of query function
SColIndex
*
pCol
=
&
pQuery
->
pSelectExpr
[
col
].
base
.
colInfo
;
if
(
TSDB_COL_IS_TAG
(
pCol
->
flag
)
||
pDataBlock
==
NULL
)
{
if
(
TSDB_COL_IS_TAG
(
pCol
->
flag
))
{
dataBlock
=
NULL
;
}
else
{
dataBlock
=
getDataBlockImpl
(
pDataBlock
,
pCol
->
colId
);
SColIndex
*
pColIndex
=
&
pQuery
->
pSelectExpr
[
col
].
base
.
colInfo
;
SColumnInfoData
*
p
=
taosArrayGet
(
pDataBlock
,
pColIndex
->
colIndex
);
assert
(
p
->
info
.
colId
==
pColIndex
->
colId
);
dataBlock
=
p
->
pData
;
}
}
...
...
@@ -1365,7 +1343,7 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void* inputData, TSKEY
int32_t
colId
=
pQuery
->
pSelectExpr
[
colIndex
].
base
.
colInfo
.
colId
;
SDataStatis
*
tpField
=
NULL
;
pCtx
->
hasNull
=
hasNullValue
(
pQuery
,
colIndex
,
pBlockInfo
->
numOfCols
,
pStatis
,
&
tpField
);
pCtx
->
hasNull
=
hasNullValue
(
&
pQuery
->
pSelectExpr
[
colIndex
].
base
.
colInfo
,
pStatis
,
&
tpField
);
pCtx
->
aInputElemBuf
=
inputData
;
if
(
tpField
!=
NULL
)
{
...
...
@@ -1622,19 +1600,18 @@ static void teardownQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv) {
pRuntimeEnv
->
pTSBuf
=
tsBufDestory
(
pRuntimeEnv
->
pTSBuf
);
}
static
bool
isQueryKilled
(
SQInfo
*
pQInfo
)
{
return
(
pQInfo
->
code
==
TSDB_CODE_TSC_QUERY_CANCELLED
);
}
#define IS_QUERY_KILLED(_q) ((_q)->code == TSDB_CODE_TSC_QUERY_CANCELLED)
static
void
setQueryKilled
(
SQInfo
*
pQInfo
)
{
pQInfo
->
code
=
TSDB_CODE_TSC_QUERY_CANCELLED
;
}
static
bool
isFixedOutputQuery
(
SQuery
*
pQuery
)
{
if
(
pQuery
->
intervalTime
!=
0
)
{
static
bool
isFixedOutputQuery
(
SQueryRuntimeEnv
*
pRuntimeEnv
)
{
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
if
(
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
{
return
false
;
}
// Note:top/bottom query is fixed output query
if
(
isTopBottomQuery
(
pQuery
)
||
isGroupbyNormalCol
(
pQuery
->
pGroupbyExpr
)
)
{
if
(
pRuntimeEnv
->
topBotQuery
||
pRuntimeEnv
->
groupbyNormalCol
)
{
return
true
;
}
...
...
@@ -2054,7 +2031,7 @@ SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle,
// check if this data block is required to load
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SSqlFuncMsg
*
pSqlFunc
=
&
pQuery
->
pSelectExpr
[
i
].
base
;
int32_t
functionId
=
pSqlFunc
->
functionId
;
int32_t
colId
=
pSqlFunc
->
colInfo
.
colId
;
r
|=
aAggs
[
functionId
].
dataReqFunc
(
&
pRuntimeEnv
->
pCtx
[
i
],
pQuery
->
window
.
skey
,
pQuery
->
window
.
ekey
,
colId
);
...
...
@@ -2066,8 +2043,7 @@ SArray *loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle,
}
if
(
r
==
BLK_DATA_NO_NEEDED
)
{
qDebug
(
"QInfo:%p data block discard, brange:%"
PRId64
"-%"
PRId64
", rows:%d"
,
GET_QINFO_ADDR
(
pRuntimeEnv
),
pBlockInfo
->
window
.
skey
,
pBlockInfo
->
window
.
ekey
,
pBlockInfo
->
rows
);
qDebug
(
"QInfo:%p data block discard, rows:%d"
,
GET_QINFO_ADDR
(
pRuntimeEnv
),
pBlockInfo
->
rows
);
pRuntimeEnv
->
summary
.
discardBlocks
+=
1
;
}
else
if
(
r
==
BLK_DATA_STATIS_NEEDED
)
{
if
(
tsdbRetrieveDataBlockStatisInfo
(
pQueryHandle
,
pStatis
)
!=
TSDB_CODE_SUCCESS
)
{
...
...
@@ -2199,7 +2175,7 @@ static void ensureOutputBufferSimple(SQueryRuntimeEnv* pRuntimeEnv, int32_t capa
static
void
ensureOutputBuffer
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SDataBlockInfo
*
pBlockInfo
)
{
// in case of prj/diff query, ensure the output buffer is sufficient to accommodate the results of current block
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
if
(
!
QUERY_IS_INTERVAL_QUERY
(
pQuery
)
&&
!
pRuntimeEnv
->
groupbyNormalCol
&&
!
isFixedOutputQuery
(
p
Query
))
{
if
(
!
QUERY_IS_INTERVAL_QUERY
(
pQuery
)
&&
!
pRuntimeEnv
->
groupbyNormalCol
&&
!
isFixedOutputQuery
(
p
RuntimeEnv
))
{
SResultRec
*
pRec
=
&
pQuery
->
rec
;
if
(
pQuery
->
rec
.
capacity
-
pQuery
->
rec
.
rows
<
pBlockInfo
->
rows
)
{
...
...
@@ -2249,7 +2225,7 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
SDataBlockInfo
blockInfo
=
SDATA_BLOCK_INITIALIZER
;
while
(
tsdbNextDataBlock
(
pQueryHandle
))
{
summary
->
totalBlocks
+=
1
;
if
(
isQueryKilled
(
GET_QINFO_ADDR
(
pRuntimeEnv
)))
{
if
(
IS_QUERY_KILLED
(
GET_QINFO_ADDR
(
pRuntimeEnv
)))
{
return
0
;
}
...
...
@@ -3304,7 +3280,7 @@ void scanOneTableDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start) {
cond
.
twindow
.
skey
,
cond
.
twindow
.
ekey
);
// check if query is killed or not
if
(
isQueryKilled
(
pQInfo
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
))
{
return
;
}
}
...
...
@@ -3695,23 +3671,24 @@ void copyFromWindowResToSData(SQInfo *pQInfo, SWindowResInfo *pResultInfo) {
assert
(
pQuery
->
rec
.
rows
<=
pQuery
->
rec
.
capacity
);
}
static
void
updateWindowResNumOfRes
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
STableQueryInfo
*
pTableQueryInfo
)
{
static
void
updateWindowResNumOfRes
(
SQueryRuntimeEnv
*
pRuntimeEnv
)
{
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
// update the number of result for each, only update the number of rows for the corresponding window result.
if
(
!
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
{
for
(
int32_t
i
=
0
;
i
<
pRuntimeEnv
->
windowResInfo
.
size
;
++
i
)
{
SWindowResult
*
pResult
=
&
pRuntimeEnv
->
windowResInfo
.
pResult
[
i
];
if
(
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
{
return
;
}
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int32_t
functionId
=
pRuntimeEnv
->
pCtx
[
j
].
functionId
;
if
(
functionId
==
TSDB_FUNC_TS
||
functionId
==
TSDB_FUNC_TAG
||
functionId
==
TSDB_FUNC_TAGPRJ
)
{
continue
;
}
for
(
int32_t
i
=
0
;
i
<
pRuntimeEnv
->
windowResInfo
.
size
;
++
i
)
{
SWindowResult
*
pResult
=
&
pRuntimeEnv
->
windowResInfo
.
pResult
[
i
];
pResult
->
numOfRows
=
MAX
(
pResult
->
numOfRows
,
pResult
->
resultInfo
[
j
].
numOfRes
);
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int32_t
functionId
=
pRuntimeEnv
->
pCtx
[
j
].
functionId
;
if
(
functionId
==
TSDB_FUNC_TS
||
functionId
==
TSDB_FUNC_TAG
||
functionId
==
TSDB_FUNC_TAGPRJ
)
{
continue
;
}
pResult
->
numOfRows
=
MAX
(
pResult
->
numOfRows
,
pResult
->
resultInfo
[
j
].
numOfRes
);
}
}
}
...
...
@@ -3729,8 +3706,6 @@ void stableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *
}
else
{
blockwiseApplyFunctions
(
pRuntimeEnv
,
pStatis
,
pDataBlockInfo
,
pWindowResInfo
,
searchFn
,
pDataBlock
);
}
updateWindowResNumOfRes
(
pRuntimeEnv
,
pTableQueryInfo
);
}
bool
queryHasRemainResults
(
SQueryRuntimeEnv
*
pRuntimeEnv
)
{
...
...
@@ -3950,7 +3925,7 @@ void skipBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
SDataBlockInfo
blockInfo
=
SDATA_BLOCK_INITIALIZER
;
while
(
tsdbNextDataBlock
(
pQueryHandle
))
{
if
(
isQueryKilled
(
GET_QINFO_ADDR
(
pRuntimeEnv
)))
{
if
(
IS_QUERY_KILLED
(
GET_QINFO_ADDR
(
pRuntimeEnv
)))
{
return
;
}
...
...
@@ -4099,7 +4074,7 @@ static void setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery) {
return
;
}
if
(
isSTableQuery
&&
(
!
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
&&
(
!
isFixedOutputQuery
(
p
Query
)))
{
if
(
isSTableQuery
&&
(
!
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
&&
(
!
isFixedOutputQuery
(
p
RuntimeEnv
)))
{
return
;
}
...
...
@@ -4115,7 +4090,7 @@ static void setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery) {
&&
(
cond
.
order
==
TSDB_ORDER_ASC
)
&&
(
!
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
&&
(
!
isGroupbyNormalCol
(
pQuery
->
pGroupbyExpr
))
&&
(
!
isFixedOutputQuery
(
p
Query
))
&&
(
!
isFixedOutputQuery
(
p
RuntimeEnv
))
)
{
SArray
*
pa
=
GET_TABLEGROUP
(
pQInfo
,
0
);
STableQueryInfo
*
pCheckInfo
=
taosArrayGetP
(
pa
,
0
);
...
...
@@ -4267,7 +4242,7 @@ static int64_t scanMultiTableDataBlocks(SQInfo *pQInfo) {
while
(
tsdbNextDataBlock
(
pQueryHandle
))
{
summary
->
totalBlocks
+=
1
;
if
(
isQueryKilled
(
pQInfo
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
))
{
break
;
}
...
...
@@ -4304,6 +4279,8 @@ static int64_t scanMultiTableDataBlocks(SQInfo *pQInfo) {
pQInfo
,
blockInfo
.
uid
,
blockInfo
.
tid
,
blockInfo
.
window
.
skey
,
blockInfo
.
window
.
ekey
,
blockInfo
.
rows
,
pQuery
->
current
->
lastKey
);
}
updateWindowResNumOfRes
(
pRuntimeEnv
);
int64_t
et
=
taosGetTimestampMs
();
return
et
-
st
;
}
...
...
@@ -4316,7 +4293,9 @@ static bool multiTableMultioutputHelper(SQInfo *pQInfo, int32_t index) {
SArray
*
group
=
GET_TABLEGROUP
(
pQInfo
,
0
);
STableQueryInfo
*
pCheckInfo
=
taosArrayGetP
(
group
,
index
);
setTagVal
(
pRuntimeEnv
,
pCheckInfo
->
pTable
,
pQInfo
->
tsdb
);
if
(
pRuntimeEnv
->
hasTagResults
||
pRuntimeEnv
->
pTSBuf
!=
NULL
)
{
setTagVal
(
pRuntimeEnv
,
pCheckInfo
->
pTable
,
pQInfo
->
tsdb
);
}
STableId
*
id
=
TSDB_TABLEID
(
pCheckInfo
->
pTable
);
qDebug
(
"QInfo:%p query on (%d): uid:%"
PRIu64
", tid:%d, qrange:%"
PRId64
"-%"
PRId64
,
pQInfo
,
index
,
...
...
@@ -4547,7 +4526,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
1
==
taosArrayGetSize
(
pQInfo
->
tableqinfoGroupInfo
.
pGroupList
));
while
(
pQInfo
->
tableIndex
<
pQInfo
->
tableqinfoGroupInfo
.
numOfTables
)
{
if
(
isQueryKilled
(
pQInfo
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
))
{
return
;
}
...
...
@@ -4742,7 +4721,7 @@ static void multiTableQueryProcess(SQInfo *pQInfo) {
qDebug
(
"QInfo:%p master scan completed, elapsed time: %"
PRId64
"ms, reverse scan start"
,
pQInfo
,
el
);
// query error occurred or query is killed, abort current execution
if
(
pQInfo
->
code
!=
TSDB_CODE_SUCCESS
||
isQueryKilled
(
pQInfo
))
{
if
(
pQInfo
->
code
!=
TSDB_CODE_SUCCESS
||
IS_QUERY_KILLED
(
pQInfo
))
{
qDebug
(
"QInfo:%p query killed or error occurred, code:%s, abort"
,
pQInfo
,
tstrerror
(
pQInfo
->
code
));
return
;
}
...
...
@@ -4764,7 +4743,7 @@ static void multiTableQueryProcess(SQInfo *pQInfo) {
setQueryStatus
(
pQuery
,
QUERY_COMPLETED
);
if
(
pQInfo
->
code
!=
TSDB_CODE_SUCCESS
||
isQueryKilled
(
pQInfo
))
{
if
(
pQInfo
->
code
!=
TSDB_CODE_SUCCESS
||
IS_QUERY_KILLED
(
pQInfo
))
{
qDebug
(
"QInfo:%p query killed or error occurred, code:%s, abort"
,
pQInfo
,
tstrerror
(
pQInfo
->
code
));
return
;
}
...
...
@@ -4804,7 +4783,7 @@ static void tableFixedOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo)
scanOneTableDataBlocks
(
pRuntimeEnv
,
pTableInfo
->
lastKey
);
finalizeQueryResult
(
pRuntimeEnv
);
if
(
isQueryKilled
(
pQInfo
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
))
{
return
;
}
...
...
@@ -4837,7 +4816,7 @@ static void tableMultiOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo)
scanOneTableDataBlocks
(
pRuntimeEnv
,
pQuery
->
current
->
lastKey
);
finalizeQueryResult
(
pRuntimeEnv
);
if
(
isQueryKilled
(
pQInfo
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
))
{
return
;
}
...
...
@@ -4885,7 +4864,7 @@ static void tableIntervalProcessImpl(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start)
while
(
1
)
{
scanOneTableDataBlocks
(
pRuntimeEnv
,
start
);
if
(
isQueryKilled
(
GET_QINFO_ADDR
(
pRuntimeEnv
)))
{
if
(
IS_QUERY_KILLED
(
GET_QINFO_ADDR
(
pRuntimeEnv
)))
{
return
;
}
...
...
@@ -5022,7 +5001,7 @@ static void tableQueryImpl(SQInfo *pQInfo) {
// group by normal column, sliding window query, interval query are handled by interval query processor
if
(
QUERY_IS_INTERVAL_QUERY
(
pQuery
)
||
pRuntimeEnv
->
groupbyNormalCol
)
{
// interval (down sampling operation)
tableIntervalProcess
(
pQInfo
,
item
);
}
else
if
(
isFixedOutputQuery
(
p
Query
))
{
}
else
if
(
isFixedOutputQuery
(
p
RuntimeEnv
))
{
tableFixedOutputProcess
(
pQInfo
,
item
);
}
else
{
// diff/add/multiply/subtract/division
assert
(
pQuery
->
checkBuffer
==
1
);
...
...
@@ -5042,7 +5021,7 @@ static void stableQueryImpl(SQInfo *pQInfo) {
int64_t
st
=
taosGetTimestampUs
();
if
(
QUERY_IS_INTERVAL_QUERY
(
pQuery
)
||
(
isFixedOutputQuery
(
p
Query
)
&&
(
!
isPointInterpoQuery
(
pQuery
))
&&
!
pRuntimeEnv
->
groupbyNormalCol
&&
(
isFixedOutputQuery
(
p
RuntimeEnv
)
&&
(
!
isPointInterpoQuery
(
pQuery
))
&&
!
pRuntimeEnv
->
groupbyNormalCol
&&
!
isFirstLastRowQuery
(
pQuery
)))
{
multiTableQueryProcess
(
pQInfo
);
}
else
{
...
...
@@ -6173,7 +6152,7 @@ void qTableQuery(qinfo_t qinfo) {
return
;
}
if
(
isQueryKilled
(
pQInfo
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
))
{
qDebug
(
"QInfo:%p it is already killed, abort"
,
pQInfo
);
sem_post
(
&
pQInfo
->
dataReady
);
...
...
@@ -6214,7 +6193,7 @@ void qTableQuery(qinfo_t qinfo) {
}
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
if
(
isQueryKilled
(
pQInfo
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
))
{
qDebug
(
"QInfo:%p query is killed"
,
pQInfo
);
}
else
if
(
pQuery
->
rec
.
rows
==
0
)
{
qDebug
(
"QInfo:%p over, %zu tables queried, %"
PRId64
" rows are returned"
,
pQInfo
,
pQInfo
->
tableqinfoGroupInfo
.
numOfTables
,
pQuery
->
rec
.
total
);
...
...
@@ -6235,7 +6214,7 @@ int32_t qRetrieveQueryResultInfo(qinfo_t qinfo) {
}
SQuery
*
pQuery
=
pQInfo
->
runtimeEnv
.
pQuery
;
if
(
isQueryKilled
(
pQInfo
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
))
{
qDebug
(
"QInfo:%p query is killed, code:%d"
,
pQInfo
,
pQInfo
->
code
);
return
pQInfo
->
code
;
}
...
...
@@ -6310,7 +6289,7 @@ int32_t qDumpRetrieveResult(qinfo_t qinfo, SRetrieveTableRsp **pRsp, int32_t *co
code
=
pQInfo
->
code
;
}
if
(
isQueryKilled
(
pQInfo
)
||
Q_STATUS_EQUAL
(
pQuery
->
status
,
QUERY_OVER
))
{
if
(
IS_QUERY_KILLED
(
pQInfo
)
||
Q_STATUS_EQUAL
(
pQuery
->
status
,
QUERY_OVER
))
{
(
*
pRsp
)
->
completed
=
1
;
// notify no more result to client
}
...
...
src/tsdb/src/tsdbRead.c
浏览文件 @
f1b6c002
...
...
@@ -110,9 +110,10 @@ typedef struct STsdbQueryHandle {
SFileGroupIter
fileIter
;
SRWHelper
rhelper
;
STableBlockInfo
*
pDataBlockInfo
;
int32_t
allocSize
;
// allocated data block size
SMemTable
*
mem
;
// mem-table
SMemTable
*
imem
;
// imem-table, acquired from snapshot
SArray
*
defaultLoadColumn
;
// default load column
SDataBlockLoadInfo
dataBlockLoadInfo
;
/* record current block load information */
SLoadCompBlockInfo
compBlockLoadInfo
;
/* record current compblock information in SQuery */
}
STsdbQueryHandle
;
...
...
@@ -136,6 +137,34 @@ static void tsdbInitCompBlockLoadInfo(SLoadCompBlockInfo* pCompBlockLoadInfo) {
pCompBlockLoadInfo
->
fileId
=
-
1
;
}
static
SArray
*
getColumnIdList
(
STsdbQueryHandle
*
pQueryHandle
)
{
size_t
numOfCols
=
QH_GET_NUM_OF_COLS
(
pQueryHandle
);
assert
(
numOfCols
<=
TSDB_MAX_COLUMNS
);
SArray
*
pIdList
=
taosArrayInit
(
numOfCols
,
sizeof
(
int16_t
));
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
SColumnInfoData
*
pCol
=
taosArrayGet
(
pQueryHandle
->
pColumns
,
i
);
taosArrayPush
(
pIdList
,
&
pCol
->
info
.
colId
);
}
return
pIdList
;
}
static
SArray
*
getDefaultLoadColumns
(
STsdbQueryHandle
*
pQueryHandle
,
bool
loadTS
)
{
SArray
*
pLocalIdList
=
getColumnIdList
(
pQueryHandle
);
// check if the primary time stamp column needs to load
int16_t
colId
=
*
(
int16_t
*
)
taosArrayGet
(
pLocalIdList
,
0
);
// the primary timestamp column does not be included in the the specified load column list, add it
if
(
loadTS
&&
colId
!=
0
)
{
int16_t
columnId
=
0
;
taosArrayInsert
(
pLocalIdList
,
0
,
&
columnId
);
}
return
pLocalIdList
;
}
TsdbQueryHandleT
*
tsdbQueryTables
(
TSDB_REPO_T
*
tsdb
,
STsdbQueryCond
*
pCond
,
STableGroupInfo
*
groupList
,
void
*
qinfo
)
{
STsdbQueryHandle
*
pQueryHandle
=
calloc
(
1
,
sizeof
(
STsdbQueryHandle
));
pQueryHandle
->
order
=
pCond
->
order
;
...
...
@@ -148,7 +177,8 @@ TsdbQueryHandleT* tsdbQueryTables(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, STab
pQueryHandle
->
activeIndex
=
0
;
// current active table index
pQueryHandle
->
qinfo
=
qinfo
;
pQueryHandle
->
outputCapacity
=
((
STsdbRepo
*
)
tsdb
)
->
config
.
maxRowsPerFileBlock
;
pQueryHandle
->
allocSize
=
0
;
tsdbInitReadHelper
(
&
pQueryHandle
->
rhelper
,
(
STsdbRepo
*
)
tsdb
);
tsdbTakeMemSnapshot
(
pQueryHandle
->
pTsdb
,
&
pQueryHandle
->
mem
,
&
pQueryHandle
->
imem
);
...
...
@@ -195,7 +225,9 @@ TsdbQueryHandleT* tsdbQueryTables(TSDB_REPO_T* tsdb, STsdbQueryCond* pCond, STab
taosArrayPush
(
pQueryHandle
->
pTableCheckInfo
,
&
info
);
}
}
pQueryHandle
->
defaultLoadColumn
=
getDefaultLoadColumns
(
pQueryHandle
,
true
);
tsdbDebug
(
"%p total numOfTable:%zu in query"
,
pQueryHandle
,
taosArrayGetSize
(
pQueryHandle
->
pTableCheckInfo
));
tsdbInitDataBlockLoadInfo
(
&
pQueryHandle
->
dataBlockLoadInfo
);
...
...
@@ -546,33 +578,7 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
.tid = (_checkInfo)->tableId.tid, \
.uid = (_checkInfo)->tableId.uid})
static
SArray
*
getColumnIdList
(
STsdbQueryHandle
*
pQueryHandle
)
{
size_t
numOfCols
=
QH_GET_NUM_OF_COLS
(
pQueryHandle
);
assert
(
numOfCols
<=
TSDB_MAX_COLUMNS
);
SArray
*
pIdList
=
taosArrayInit
(
numOfCols
,
sizeof
(
int16_t
));
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
SColumnInfoData
*
pCol
=
taosArrayGet
(
pQueryHandle
->
pColumns
,
i
);
taosArrayPush
(
pIdList
,
&
pCol
->
info
.
colId
);
}
return
pIdList
;
}
static
SArray
*
getDefaultLoadColumns
(
STsdbQueryHandle
*
pQueryHandle
,
bool
loadTS
)
{
SArray
*
pLocalIdList
=
getColumnIdList
(
pQueryHandle
);
// check if the primary time stamp column needs to load
int16_t
colId
=
*
(
int16_t
*
)
taosArrayGet
(
pLocalIdList
,
0
);
// the primary timestamp column does not be included in the the specified load column list, add it
if
(
loadTS
&&
colId
!=
0
)
{
int16_t
columnId
=
0
;
taosArrayInsert
(
pLocalIdList
,
0
,
&
columnId
);
}
return
pLocalIdList
;
}
static
bool
doLoadFileDataBlock
(
STsdbQueryHandle
*
pQueryHandle
,
SCompBlock
*
pBlock
,
STableCheckInfo
*
pCheckInfo
)
{
STsdbRepo
*
pRepo
=
pQueryHandle
->
pTsdb
;
...
...
@@ -584,8 +590,6 @@ static bool doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlo
data
->
uid
=
pCheckInfo
->
pTableObj
->
tableId
.
uid
;
bool
blockLoaded
=
false
;
SArray
*
sa
=
getDefaultLoadColumns
(
pQueryHandle
,
true
);
int64_t
st
=
taosGetTimestampUs
();
if
(
pCheckInfo
->
pDataCols
==
NULL
)
{
...
...
@@ -613,7 +617,6 @@ static bool doLoadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlo
assert
(
pCols
->
numOfRows
!=
0
&&
pCols
->
numOfRows
<=
pBlock
->
numOfRows
);
pBlock
->
numOfRows
=
pCols
->
numOfRows
;
taosArrayDestroy
(
sa
);
tfree
(
data
);
int64_t
et
=
taosGetTimestampUs
()
-
st
;
...
...
@@ -656,12 +659,8 @@ static void handleDataMergeIfNeeded(STsdbQueryHandle* pQueryHandle, SCompBlock*
return
;
}
SArray
*
sa
=
getDefaultLoadColumns
(
pQueryHandle
,
true
);
doLoadFileDataBlock
(
pQueryHandle
,
pBlock
,
pCheckInfo
);
doMergeTwoLevelData
(
pQueryHandle
,
pCheckInfo
,
pBlock
,
sa
);
taosArrayDestroy
(
sa
);
doMergeTwoLevelData
(
pQueryHandle
,
pCheckInfo
,
pBlock
,
pQueryHandle
->
defaultLoadColumn
);
}
else
{
/*
* no data in cache, only load data from file
...
...
@@ -681,14 +680,12 @@ static void handleDataMergeIfNeeded(STsdbQueryHandle* pQueryHandle, SCompBlock*
}
static
bool
loadFileDataBlock
(
STsdbQueryHandle
*
pQueryHandle
,
SCompBlock
*
pBlock
,
STableCheckInfo
*
pCheckInfo
)
{
SArray
*
sa
=
getDefaultLoadColumns
(
pQueryHandle
,
true
);
SQueryFilePos
*
cur
=
&
pQueryHandle
->
cur
;
if
(
ASCENDING_TRAVERSE
(
pQueryHandle
->
order
))
{
// query ended in current block
if
(
pQueryHandle
->
window
.
ekey
<
pBlock
->
keyLast
||
pCheckInfo
->
lastKey
>
pBlock
->
keyFirst
)
{
if
(
!
doLoadFileDataBlock
(
pQueryHandle
,
pBlock
,
pCheckInfo
))
{
taosArrayDestroy
(
sa
);
return
false
;
}
...
...
@@ -702,7 +699,7 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
cur
->
pos
=
0
;
}
doMergeTwoLevelData
(
pQueryHandle
,
pCheckInfo
,
pBlock
,
sa
);
doMergeTwoLevelData
(
pQueryHandle
,
pCheckInfo
,
pBlock
,
pQueryHandle
->
defaultLoadColumn
);
}
else
{
// the whole block is loaded in to buffer
handleDataMergeIfNeeded
(
pQueryHandle
,
pBlock
,
pCheckInfo
);
}
...
...
@@ -719,13 +716,12 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
cur
->
pos
=
pBlock
->
numOfRows
-
1
;
}
doMergeTwoLevelData
(
pQueryHandle
,
pCheckInfo
,
pBlock
,
sa
);
doMergeTwoLevelData
(
pQueryHandle
,
pCheckInfo
,
pBlock
,
pQueryHandle
->
defaultLoadColumn
);
}
else
{
handleDataMergeIfNeeded
(
pQueryHandle
,
pBlock
,
pCheckInfo
);
}
}
taosArrayDestroy
(
sa
);
return
pQueryHandle
->
realNumOfRows
>
0
;
}
...
...
@@ -1250,13 +1246,19 @@ static int32_t dataBlockOrderCompar(const void* pLeft, const void* pRight, void*
}
static
int32_t
createDataBlocksInfo
(
STsdbQueryHandle
*
pQueryHandle
,
int32_t
numOfBlocks
,
int32_t
*
numOfAllocBlocks
)
{
char
*
tmp
=
realloc
(
pQueryHandle
->
pDataBlockInfo
,
sizeof
(
STableBlockInfo
)
*
numOfBlocks
);
if
(
tmp
==
NULL
)
{
return
TSDB_CODE_TDB_OUT_OF_MEMORY
;
size_t
size
=
sizeof
(
STableBlockInfo
)
*
numOfBlocks
;
if
(
pQueryHandle
->
allocSize
<
size
)
{
pQueryHandle
->
allocSize
=
size
;
char
*
tmp
=
realloc
(
pQueryHandle
->
pDataBlockInfo
,
pQueryHandle
->
allocSize
);
if
(
tmp
==
NULL
)
{
return
TSDB_CODE_TDB_OUT_OF_MEMORY
;
}
pQueryHandle
->
pDataBlockInfo
=
(
STableBlockInfo
*
)
tmp
;
}
pQueryHandle
->
pDataBlockInfo
=
(
STableBlockInfo
*
)
tmp
;
memset
(
pQueryHandle
->
pDataBlockInfo
,
0
,
sizeof
(
STableBlockInfo
)
*
numOfBlocks
);
memset
(
pQueryHandle
->
pDataBlockInfo
,
0
,
size
);
*
numOfAllocBlocks
=
numOfBlocks
;
int32_t
numOfTables
=
taosArrayGetSize
(
pQueryHandle
->
pTableCheckInfo
);
...
...
@@ -1492,9 +1494,8 @@ bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) {
return
false
;
}
SArray
*
sa
=
getDefaultLoadColumns
(
pQueryHandle
,
true
);
/*SDataBlockInfo* pBlockInfo =*/
tsdbRetrieveDataBlockInfo
(
pHandle
,
&
blockInfo
);
/*SArray *pDataBlock = */
tsdbRetrieveDataBlock
(
pHandle
,
sa
);
/*SArray *pDataBlock = */
tsdbRetrieveDataBlock
(
pHandle
,
pQueryHandle
->
defaultLoadColumn
);
if
(
pQueryHandle
->
cur
.
win
.
ekey
==
pQueryHandle
->
window
.
skey
)
{
// data already retrieve, discard other data rows and return
...
...
@@ -1508,7 +1509,6 @@ bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) {
pQueryHandle
->
window
=
pQueryHandle
->
cur
.
win
;
pQueryHandle
->
cur
.
rows
=
1
;
pQueryHandle
->
type
=
TSDB_QUERY_TYPE_EXTERNAL
;
taosArrayDestroy
(
sa
);
return
true
;
}
else
{
STsdbQueryHandle
*
pSecQueryHandle
=
calloc
(
1
,
sizeof
(
STsdbQueryHandle
));
...
...
@@ -1565,7 +1565,7 @@ bool tsdbNextDataBlock(TsdbQueryHandleT* pHandle) {
assert
(
ret
);
/*SDataBlockInfo* pBlockInfo =*/
tsdbRetrieveDataBlockInfo
((
void
*
)
pSecQueryHandle
,
&
blockInfo
);
/*SArray *pDataBlock = */
tsdbRetrieveDataBlock
((
void
*
)
pSecQueryHandle
,
sa
);
/*SArray *pDataBlock = */
tsdbRetrieveDataBlock
((
void
*
)
pSecQueryHandle
,
pSecQueryHandle
->
defaultLoadColumn
);
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
SColumnInfoData
*
pCol
=
taosArrayGet
(
pQueryHandle
->
pColumns
,
i
);
...
...
@@ -2333,6 +2333,7 @@ void tsdbCleanupQueryHandle(TsdbQueryHandleT queryHandle) {
}
taosArrayDestroy
(
pQueryHandle
->
pColumns
);
taosArrayDestroy
(
pQueryHandle
->
defaultLoadColumn
);
tfree
(
pQueryHandle
->
pDataBlockInfo
);
tfree
(
pQueryHandle
->
statis
);
...
...
src/util/inc/tarray.h
浏览文件 @
f1b6c002
...
...
@@ -23,7 +23,7 @@ extern "C" {
#include "os.h"
#define TARRAY_MIN_SIZE 8
#define TARRAY_GET_ELEM(array, index) ((
array)->pData + (index) * (array)->elemSize
)
#define TARRAY_GET_ELEM(array, index) ((
void*)((array)->pData + (index) * (array)->elemSize)
)
typedef
struct
SArray
{
size_t
size
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录