Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
codyshao
TDengine
提交
7e12d2f9
T
TDengine
项目概览
codyshao
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
2
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,发现更多精彩内容 >>
未验证
提交
7e12d2f9
编写于
4月 16, 2020
作者:
S
slguan
提交者:
GitHub
4月 16, 2020
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #1620 from taosdata/feature/query
Feature/query
上级
49bfa0dd
6a504d36
变更
34
展开全部
隐藏空白更改
内联
并排
Showing
34 changed file
with
881 addition
and
575 deletion
+881
-575
src/client/inc/tscUtil.h
src/client/inc/tscUtil.h
+3
-2
src/client/inc/tsclient.h
src/client/inc/tsclient.h
+4
-6
src/client/src/tscAsync.c
src/client/src/tscAsync.c
+2
-2
src/client/src/tscPrepare.c
src/client/src/tscPrepare.c
+2
-2
src/client/src/tscSQLParser.c
src/client/src/tscSQLParser.c
+14
-13
src/client/src/tscSecondaryMerge.c
src/client/src/tscSecondaryMerge.c
+1
-1
src/client/src/tscServer.c
src/client/src/tscServer.c
+46
-40
src/client/src/tscSql.c
src/client/src/tscSql.c
+3
-4
src/client/src/tscStream.c
src/client/src/tscStream.c
+0
-1
src/client/src/tscSub.c
src/client/src/tscSub.c
+1
-1
src/client/src/tscSubquery.c
src/client/src/tscSubquery.c
+64
-70
src/client/src/tscUtil.c
src/client/src/tscUtil.c
+21
-15
src/inc/taosmsg.h
src/inc/taosmsg.h
+12
-2
src/mnode/src/mgmtTable.c
src/mnode/src/mgmtTable.c
+20
-6
src/query/inc/qextbuffer.h
src/query/inc/qextbuffer.h
+13
-17
src/query/inc/queryExecutor.h
src/query/inc/queryExecutor.h
+2
-2
src/query/src/qast.c
src/query/src/qast.c
+1
-1
src/query/src/qextbuffer.c
src/query/src/qextbuffer.c
+5
-5
src/query/src/queryExecutor.c
src/query/src/queryExecutor.c
+117
-113
src/tsdb/inc/tsdb.h
src/tsdb/inc/tsdb.h
+9
-4
src/tsdb/inc/tsdbMain.h
src/tsdb/inc/tsdbMain.h
+1
-0
src/tsdb/src/tsdbFile.c
src/tsdb/src/tsdbFile.c
+2
-1
src/tsdb/src/tsdbMain.c
src/tsdb/src/tsdbMain.c
+2
-1
src/tsdb/src/tsdbMeta.c
src/tsdb/src/tsdbMeta.c
+24
-8
src/tsdb/src/tsdbRead.c
src/tsdb/src/tsdbRead.c
+208
-160
src/util/inc/talgo.h
src/util/inc/talgo.h
+58
-0
src/util/inc/tskiplist.h
src/util/inc/tskiplist.h
+1
-1
src/util/inc/tutil.h
src/util/inc/tutil.h
+0
-8
src/util/src/hash.c
src/util/src/hash.c
+0
-2
src/util/src/talgo.c
src/util/src/talgo.c
+227
-0
src/util/src/tskiplist.c
src/util/src/tskiplist.c
+1
-1
src/util/src/tutil.c
src/util/src/tutil.c
+0
-69
src/util/tests/skiplistTest.cpp
src/util/tests/skiplistTest.cpp
+6
-6
tests/examples/c/demo.c
tests/examples/c/demo.c
+11
-11
未找到文件。
src/client/inc/tscUtil.h
浏览文件 @
7e12d2f9
...
...
@@ -200,8 +200,9 @@ int32_t tscGetQueryInfoDetailSafely(SSqlCmd *pCmd, int32_t subClauseIndex, SQuer
STableMetaInfo
*
tscGetMeterMetaInfoByUid
(
SQueryInfo
*
pQueryInfo
,
uint64_t
uid
,
int32_t
*
index
);
void
tscClearMeterMetaInfo
(
STableMetaInfo
*
pTableMetaInfo
,
bool
removeFromCache
);
STableMetaInfo
*
tscAddTableMetaInfo
(
SQueryInfo
*
pQueryInfo
,
const
char
*
name
,
STableMeta
*
pTableMeta
,
SArray
*
vgroupList
,
int16_t
numOfTags
,
int16_t
*
tags
);
STableMetaInfo
*
tscAddTableMetaInfo
(
SQueryInfo
*
pQueryInfo
,
const
char
*
name
,
STableMeta
*
pTableMeta
,
SVgroupsInfo
*
vgroupList
,
int16_t
numOfTags
,
int16_t
*
tags
);
STableMetaInfo
*
tscAddEmptyMetaInfo
(
SQueryInfo
*
pQueryInfo
);
int32_t
tscAddSubqueryInfo
(
SSqlCmd
*
pCmd
);
void
tscFreeSubqueryInfo
(
SSqlCmd
*
pCmd
);
...
...
src/client/inc/tsclient.h
浏览文件 @
7e12d2f9
...
...
@@ -40,6 +40,8 @@ extern "C" {
// forward declaration
struct
SSqlInfo
;
typedef
SCMSTableVgroupRspMsg
SVgroupsInfo
;
typedef
struct
SSqlGroupbyExpr
{
int16_t
tableIndex
;
int16_t
numOfGroupCols
;
...
...
@@ -70,14 +72,12 @@ typedef struct STableMeta {
typedef
struct
STableMetaInfo
{
STableMeta
*
pTableMeta
;
// table meta, cached in client side and acquried by name
// SSuperTableMeta *pMetricMeta; // metricmeta
SArray
*
vgroupIdList
;
SVgroupsInfo
*
vgroupList
;
/*
* 1. keep the vnode index during the multi-vnode super table projection query
* 2. keep the vnode index for multi-vnode insertion
*/
int32_t
v
nodeIndex
;
int32_t
d
nodeIndex
;
char
name
[
TSDB_TABLE_ID_LEN
];
// (super) table name
int16_t
numOfTags
;
// total required tags in query, including groupby tags
int16_t
tagColumnIndex
[
TSDB_MAX_TAGS
];
// clause + tag projection
...
...
@@ -210,7 +210,6 @@ typedef struct STableDataBlocks {
}
STableDataBlocks
;
typedef
struct
SDataBlockList
{
int32_t
idx
;
uint32_t
nSize
;
uint32_t
nAlloc
;
STableDataBlocks
**
pData
;
...
...
@@ -257,7 +256,6 @@ typedef struct {
union
{
bool
existsCheck
;
// check if the table exists or not
bool
inStream
;
// denote if current sql is executed in stream or not
bool
autoCreated
;
// if the table is missing, on-the-fly create it. during getmeterMeta
int8_t
dataSourceType
;
// load data from file or not
};
...
...
src/client/src/tscAsync.c
浏览文件 @
7e12d2f9
...
...
@@ -443,12 +443,12 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
if
((
pQueryInfo
->
type
&
TSDB_QUERY_TYPE_STABLE_SUBQUERY
)
==
TSDB_QUERY_TYPE_STABLE_SUBQUERY
)
{
STableMetaInfo
*
pTableMetaInfo
=
tscGetMetaInfo
(
pQueryInfo
,
0
);
assert
((
tscGetNumOfTags
(
pTableMetaInfo
->
pTableMeta
)
!=
0
)
&&
pTableMetaInfo
->
v
nodeIndex
>=
0
&&
pSql
->
param
!=
NULL
);
assert
((
tscGetNumOfTags
(
pTableMetaInfo
->
pTableMeta
)
!=
0
)
&&
pTableMetaInfo
->
d
nodeIndex
>=
0
&&
pSql
->
param
!=
NULL
);
SRetrieveSupport
*
trs
=
(
SRetrieveSupport
*
)
pSql
->
param
;
SSqlObj
*
pParObj
=
trs
->
pParentSqlObj
;
assert
(
pParObj
->
signature
==
pParObj
&&
trs
->
subqueryIndex
==
pTableMetaInfo
->
v
nodeIndex
&&
assert
(
pParObj
->
signature
==
pParObj
&&
trs
->
subqueryIndex
==
pTableMetaInfo
->
d
nodeIndex
&&
tscGetNumOfTags
(
pTableMetaInfo
->
pTableMeta
)
!=
0
);
tscTrace
(
"%p get metricMeta during super table query successfully"
,
pSql
);
...
...
src/client/src/tscPrepare.c
浏览文件 @
7e12d2f9
...
...
@@ -408,7 +408,7 @@ static int insertStmtReset(STscStmt* pStmt) {
pCmd
->
batchSize
=
0
;
STableMetaInfo
*
pTableMetaInfo
=
tscGetTableMetaInfoFromCmd
(
pCmd
,
pCmd
->
clauseIndex
,
0
);
pTableMetaInfo
->
v
nodeIndex
=
0
;
pTableMetaInfo
->
d
nodeIndex
=
0
;
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -438,7 +438,7 @@ static int insertStmtExecute(STscStmt* stmt) {
}
// set the next sent data vnode index in data block arraylist
pTableMetaInfo
->
v
nodeIndex
=
1
;
pTableMetaInfo
->
d
nodeIndex
=
1
;
}
else
{
pCmd
->
pDataBlocks
=
tscDestroyBlockArrayList
(
pCmd
->
pDataBlocks
);
}
...
...
src/client/src/tscSQLParser.c
浏览文件 @
7e12d2f9
...
...
@@ -2470,7 +2470,7 @@ int32_t parseGroupbyClause(SQueryInfo* pQueryInfo, tVariantList* pList, SSqlCmd*
const
char
*
msg8
=
"not allowed column type for group by"
;
const
char
*
msg9
=
"tags not allowed for table query"
;
// todo : handle two
meter
situation
// todo : handle two
tables
situation
STableMetaInfo
*
pTableMetaInfo
=
NULL
;
if
(
pList
==
NULL
)
{
...
...
@@ -2493,7 +2493,6 @@ int32_t parseGroupbyClause(SQueryInfo* pQueryInfo, tVariantList* pList, SSqlCmd*
SSQLToken
token
=
{
pVar
->
nLen
,
pVar
->
nType
,
pVar
->
pz
};
SColumnIndex
index
=
COLUMN_INDEX_INITIALIZER
;
if
(
getColumnIndexByName
(
&
token
,
pQueryInfo
,
&
index
)
!=
TSDB_CODE_SUCCESS
)
{
return
invalidSqlErrMsg
(
pQueryInfo
->
msg
,
msg2
);
}
...
...
@@ -2523,13 +2522,13 @@ int32_t parseGroupbyClause(SQueryInfo* pQueryInfo, tVariantList* pList, SSqlCmd*
return
invalidSqlErrMsg
(
pQueryInfo
->
msg
,
msg9
);
}
int32_t
relIndex
=
index
.
columnIndex
;
if
(
index
.
columnIndex
!=
TSDB_TBNAME_COLUMN_INDEX
)
{
relIndex
-=
tscGetNumOfColumns
(
pTableMeta
);
}
//
int32_t relIndex = index.columnIndex;
//
if (index.columnIndex != TSDB_TBNAME_COLUMN_INDEX) {
//
relIndex -= tscGetNumOfColumns(pTableMeta);
//
}
pQueryInfo
->
groupbyExpr
.
columnInfo
[
i
]
=
(
SColIndex
){.
colIndex
=
rel
Index
,
.
flag
=
TSDB_COL_TAG
,
.
colId
=
pSchema
->
colId
};
// relIndex;
(
SColIndex
){.
colIndex
=
index
.
column
Index
,
.
flag
=
TSDB_COL_TAG
,
.
colId
=
pSchema
->
colId
};
// relIndex;
addRequiredTagColumn
(
pQueryInfo
,
pQueryInfo
->
groupbyExpr
.
columnInfo
[
i
].
colIndex
,
index
.
tableIndex
);
}
else
{
// check if the column type is valid, here only support the bool/tinyint/smallint/bigint group by
...
...
@@ -5095,9 +5094,8 @@ static int32_t doAddGroupbyColumnsOnDemand(SQueryInfo* pQueryInfo) {
bytes
=
TSDB_TABLE_NAME_LEN
;
name
=
TSQL_TBNAME_L
;
}
else
{
colIndex
=
(
TSDB_COL_IS_TAG
(
pColIndex
->
flag
))
?
tscGetNumOfColumns
(
pTableMetaInfo
->
pTableMeta
)
+
pColIndex
->
colIndex
:
pColIndex
->
colIndex
;
// colIndex = (TSDB_COL_IS_TAG(pColIndex->flag)) ? tscGetNumOfColumns(pTableMetaInfo->pTableMeta) + pColIndex->colIndex
// : pColIndex->colIndex;
type
=
pSchema
[
colIndex
].
type
;
bytes
=
pSchema
[
colIndex
].
bytes
;
name
=
pSchema
[
colIndex
].
name
;
...
...
@@ -5108,11 +5106,14 @@ static int32_t doAddGroupbyColumnsOnDemand(SQueryInfo* pQueryInfo) {
SSqlExpr
*
pExpr
=
tscSqlExprInsert
(
pQueryInfo
,
pQueryInfo
->
exprsInfo
.
numOfExprs
,
TSDB_FUNC_TAG
,
&
index
,
type
,
bytes
,
bytes
);
memset
(
pExpr
->
aliasName
,
0
,
tListLen
(
pExpr
->
aliasName
));
strncpy
(
pExpr
->
aliasName
,
name
,
TSDB_COL_NAME_LEN
);
pExpr
->
colInfo
.
flag
=
TSDB_COL_TAG
;
// NOTE: tag column does not add to source column list
SColumnList
ids
=
{
0
}
;
SColumnList
ids
=
getColumnList
(
1
,
0
,
pColIndex
->
colIndex
)
;
insertResultField
(
pQueryInfo
,
pQueryInfo
->
exprsInfo
.
numOfExprs
-
1
,
&
ids
,
bytes
,
type
,
name
,
pExpr
);
}
else
{
// if this query is "group by" normal column, interval is not allowed
...
...
@@ -5693,7 +5694,7 @@ int32_t doCheckForQuery(SSqlObj* pSql, SQuerySQL* pQuerySql, int32_t index) {
assert
(
pQueryInfo
->
numOfTables
==
pQuerySql
->
from
->
nExpr
);
if
(
UTIL_TABLE_IS_SUPERTABLE
(
pTableMetaInfo
))
{
int32_t
code
=
tscGetSTableVgroupInfo
(
pSql
,
index
);
code
=
tscGetSTableVgroupInfo
(
pSql
,
index
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
...
...
src/client/src/tscSecondaryMerge.c
浏览文件 @
7e12d2f9
...
...
@@ -636,7 +636,7 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
pModel
=
createColumnModel
(
pSchema
,
pQueryInfo
->
exprsInfo
.
numOfExprs
,
capacity
);
size_t
numOfSubs
=
taosArrayGetSize
(
pTableMetaInfo
->
vgroupIdList
)
;
size_t
numOfSubs
=
pTableMetaInfo
->
vgroupList
->
numOfDnodes
;
for
(
int32_t
i
=
0
;
i
<
numOfSubs
;
++
i
)
{
(
*
pMemBuffer
)[
i
]
=
createExtMemBuffer
(
nBufferSizes
,
rlen
,
pModel
);
(
*
pMemBuffer
)[
i
]
->
flushModel
=
MULTIPLE_APPEND_MODEL
;
...
...
src/client/src/tscServer.c
浏览文件 @
7e12d2f9
...
...
@@ -413,7 +413,7 @@ int tscProcessSql(SSqlObj *pSql) {
type
=
pQueryInfo
->
type
;
// for hearbeat, numOfTables == 0;
// for hear
t
beat, numOfTables == 0;
assert
((
pQueryInfo
->
numOfTables
==
0
&&
pQueryInfo
->
command
==
TSDB_SQL_HB
)
||
pQueryInfo
->
numOfTables
>
0
);
}
...
...
@@ -424,19 +424,6 @@ int tscProcessSql(SSqlObj *pSql) {
pSql
->
res
.
code
=
TSDB_CODE_OTHERS
;
return
pSql
->
res
.
code
;
}
// temp
// pSql->ipList = tscMgmtIpList;
// if (UTIL_TABLE_IS_NOMRAL_TABLE(pTableMetaInfo)) {
// pSql->index = pTableMetaInfo->pTableMeta->index;
// } else { // it must be the parent SSqlObj for super table query
// if ((pQueryInfo->type & TSDB_QUERY_TYPE_SUBQUERY) != 0) {
// int32_t idx = pTableMetaInfo->vnodeIndex;
//
// SVnodeSidList *pSidList = tscGetVnodeSidList(pTableMetaInfo->pMetricMeta, idx);
// pSql->index = pSidList->index;
// }
// }
}
else
if
(
pSql
->
cmd
.
command
<
TSDB_SQL_LOCAL
)
{
pSql
->
ipList
=
tscMgmtIpList
;
}
else
{
// local handler
...
...
@@ -522,8 +509,17 @@ int tscBuildRetrieveMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pRetrieveMsg
->
free
=
htons
(
pQueryInfo
->
type
);
pMsg
+=
sizeof
(
pQueryInfo
->
type
);
STableMeta
*
pTableMeta
=
pQueryInfo
->
pTableMetaInfo
[
0
]
->
pTableMeta
;
pRetrieveMsg
->
header
.
vgId
=
htonl
(
pTableMeta
->
vgId
);
// todo valid the vgroupId at the client side
if
(
UTIL_TABLE_IS_SUPERTABLE
(
pQueryInfo
->
pTableMetaInfo
[
0
]))
{
SVgroupsInfo
*
pVgroupInfo
=
pQueryInfo
->
pTableMetaInfo
[
0
]
->
vgroupList
;
assert
(
pVgroupInfo
->
dnodeVgroups
->
numOfVgroups
==
1
);
// todo fix me
pRetrieveMsg
->
header
.
vgId
=
htonl
(
pVgroupInfo
->
dnodeVgroups
[
0
].
vgId
[
0
]);
}
else
{
STableMeta
*
pTableMeta
=
pQueryInfo
->
pTableMetaInfo
[
0
]
->
pTableMeta
;
pRetrieveMsg
->
header
.
vgId
=
htonl
(
pTableMeta
->
vgId
);
}
pMsg
+=
sizeof
(
SRetrieveTableMsg
);
pRetrieveMsg
->
header
.
contLen
=
htonl
(
pSql
->
cmd
.
payloadLen
);
...
...
@@ -584,7 +580,7 @@ static int32_t tscEstimateQueryMsgSize(SSqlCmd *pCmd, int32_t clauseIndex) {
#if 0
SSuperTableMeta *pMetricMeta = pTableMetaInfo->pMetricMeta;
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->
v
nodeIndex);
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->
d
nodeIndex);
int32_t meterInfoSize = (pMetricMeta->tagLen + sizeof(STableIdInfo)) * pVnodeSidList->numOfSids;
int32_t outputColumnSize = pQueryInfo->exprsInfo.numOfExprs * sizeof(SSqlFuncExprMsg);
...
...
@@ -655,21 +651,22 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg
->
head
.
vgId
=
htonl
(
pTableMeta
->
vgId
);
tscTrace
(
"%p queried tables:%d, table id: %s"
,
pSql
,
1
,
pTableMetaInfo
->
name
);
}
else
{
// query super table
if
(
pTableMetaInfo
->
vnodeIndex
<
0
)
{
tscError
(
"%p error vnodeIdx:%d"
,
pSql
,
pTableMetaInfo
->
vnodeIndex
);
if
(
pTableMetaInfo
->
dnodeIndex
<
0
)
{
tscError
(
"%p error vnodeIdx:%d"
,
pSql
,
pTableMetaInfo
->
dnodeIndex
);
return
-
1
;
}
pSql
->
ipList
.
numOfIps
=
taosArrayGetSize
(
pTableMetaInfo
->
vgroupIdList
);
pSql
->
ipList
.
numOfIps
=
1
;
// todo fix me
pSql
->
ipList
.
port
=
tsDnodeShellPort
;
pSql
->
ipList
.
inUse
=
0
;
for
(
int32_t
i
=
0
;
i
<
pSql
->
ipList
.
numOfIps
;
++
i
)
{
pSql
->
ipList
.
ip
[
i
]
=
*
(
uint32_t
*
)
taosArrayGet
(
pTableMetaInfo
->
vgroupIdList
,
i
)
;
}
// todo extract method
STableDnodeVgroupInfo
*
pVgroupInfo
=
&
pTableMetaInfo
->
vgroupList
->
dnodeVgroups
[
pTableMetaInfo
->
dnodeIndex
]
;
pSql
->
ipList
.
ip
[
0
]
=
pVgroupInfo
->
ipAddr
.
ip
;
#if 0
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->
v
nodeIndex);
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pTableMetaInfo->
d
nodeIndex);
uint32_t vnodeId = pVnodeSidList->vpeerDesc[pVnodeSidList->index].vnode;
numOfTables = pVnodeSidList->numOfSids;
...
...
@@ -679,9 +676,10 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
#endif
uint32_t
vnodeId
=
1
;
tscTrace
(
"%p query on vid:%d, number of tables:%d"
,
pSql
,
vnodeId
,
numOfTables
);
pQueryMsg
->
head
.
vgId
=
htonl
(
vnodeId
);
tscTrace
(
"%p query on super table, numOfVgroup:%d, dnodeIndex:%d"
,
pSql
,
pVgroupInfo
->
numOfVgroups
,
pTableMetaInfo
->
dnodeIndex
);
pQueryMsg
->
head
.
vgId
=
htonl
(
pVgroupInfo
->
vgId
[
0
]);
numOfTables
=
1
;
}
...
...
@@ -859,7 +857,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
int32_t
numOfBlocks
=
0
;
if
(
pQueryInfo
->
tsBuf
!=
NULL
)
{
STSVnodeBlockInfo
*
pBlockInfo
=
tsBufGetVnodeBlockInfo
(
pQueryInfo
->
tsBuf
,
pTableMetaInfo
->
v
nodeIndex
);
STSVnodeBlockInfo
*
pBlockInfo
=
tsBufGetVnodeBlockInfo
(
pQueryInfo
->
tsBuf
,
pTableMetaInfo
->
d
nodeIndex
);
assert
(
QUERY_IS_JOIN_QUERY
(
pQueryInfo
->
type
)
&&
pBlockInfo
!=
NULL
);
// this query should not be sent
// todo refactor
...
...
@@ -1851,7 +1849,6 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) {
}
for
(
int
i
=
0
;
i
<
TSDB_VNODES_SUPPORT
;
++
i
)
{
pMetaMsg
->
vpeerDesc
[
i
].
vgId
=
htonl
(
pMetaMsg
->
vpeerDesc
[
i
].
vgId
);
pMetaMsg
->
vpeerDesc
[
i
].
ip
=
htonl
(
pMetaMsg
->
vpeerDesc
[
i
].
ip
);
pMetaMsg
->
vpeerDesc
[
i
].
dnodeId
=
htonl
(
pMetaMsg
->
vpeerDesc
[
i
].
dnodeId
);
}
...
...
@@ -2116,21 +2113,30 @@ _error_clean:
free(sizes);
free(metricMetaList);
#endif
SSqlRes
*
pRes
=
&
pSql
->
res
;
SCMSTableVgroupRspMsg
*
pStableVgroup
=
(
SCMSTableVgroupRspMsg
*
)
p
Sql
->
res
.
pRsp
;
SCMSTableVgroupRspMsg
*
pStableVgroup
=
(
SCMSTableVgroupRspMsg
*
)
p
Res
->
pRsp
;
pStableVgroup
->
numOfDnodes
=
htonl
(
pStableVgroup
->
numOfDnodes
);
SSqlObj
*
pparent
=
pSql
->
param
;
assert
(
pparent
!=
NULL
);
// master sqlObj locates in param
SSqlObj
*
parent
=
pSql
->
param
;
assert
(
parent
!=
NULL
);
SSqlCmd
*
pCmd
=
&
p
p
arent
->
cmd
;
SSqlCmd
*
pCmd
=
&
parent
->
cmd
;
STableMetaInfo
*
pInfo
=
tscGetTableMetaInfoFromCmd
(
pCmd
,
pCmd
->
clauseIndex
,
0
);
pInfo
->
vgroupIdList
=
taosArrayInit
(
pStableVgroup
->
numOfDnodes
,
sizeof
(
int32_t
));
// todo opt performance
for
(
int32_t
i
=
0
;
i
<
pStableVgroup
->
numOfDnodes
;
++
i
)
{
int32_t
ip
=
htonl
(
pStableVgroup
->
dnodeIps
[
i
]);
taosArrayPush
(
pInfo
->
vgroupIdList
,
&
ip
);
pInfo
->
vgroupList
=
malloc
(
pRes
->
rspLen
);
memcpy
(
pInfo
->
vgroupList
,
pStableVgroup
,
pRes
->
rspLen
);
for
(
int32_t
i
=
0
;
i
<
pInfo
->
vgroupList
->
numOfDnodes
;
++
i
)
{
STableDnodeVgroupInfo
*
pVgroups
=
&
pInfo
->
vgroupList
->
dnodeVgroups
[
i
];
pVgroups
->
numOfVgroups
=
htonl
(
pVgroups
->
numOfVgroups
);
pVgroups
->
ipAddr
.
ip
=
htonl
(
pVgroups
->
ipAddr
.
ip
);
pVgroups
->
ipAddr
.
port
=
htons
(
pVgroups
->
ipAddr
.
port
);
for
(
int32_t
j
=
0
;
j
<
pVgroups
->
numOfVgroups
;
++
j
)
{
pVgroups
->
vgId
[
j
]
=
htonl
(
pVgroups
->
vgId
[
j
]);
}
}
return
pSql
->
res
.
code
;
...
...
@@ -2492,7 +2498,7 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, int32_t clauseIndex) {
// bool required = false;
SQueryInfo
*
pQueryInfo
=
tscGetQueryInfoDetail
(
pCmd
,
clauseIndex
);
if
(
pQueryInfo
->
pTableMetaInfo
[
0
]
->
vgroup
Id
List
!=
NULL
)
{
if
(
pQueryInfo
->
pTableMetaInfo
[
0
]
->
vgroupList
!=
NULL
)
{
return
TSDB_CODE_SUCCESS
;
}
...
...
src/client/src/tscSql.c
浏览文件 @
7e12d2f9
...
...
@@ -852,10 +852,9 @@ void taos_free_result_imp(TAOS_RES *res, int keepCmd) {
}
}
else
{
// if no free resource msg is sent to vnode, we free this object immediately.
bool
free
=
tscShouldFreeAsyncSqlObj
(
pSql
);
if
(
free
)
{
assert
(
pRes
->
numOfRows
==
0
||
(
pCmd
->
command
>
TSDB_SQL_LOCAL
));
STscObj
*
pTscObj
=
pSql
->
pTscObj
;
if
(
pTscObj
->
pSql
!=
pSql
)
{
tscFreeSqlObj
(
pSql
);
tscTrace
(
"%p sql result is freed by app"
,
pSql
);
}
else
{
...
...
src/client/src/tscStream.c
浏览文件 @
7e12d2f9
...
...
@@ -517,7 +517,6 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
return
NULL
;
}
pSql
->
cmd
.
inStream
=
1
;
// 1 means sql in stream, allowed the sliding clause.
pRes
->
code
=
tscToSQLCmd
(
pSql
,
&
SQLInfo
);
SQLInfoDestroy
(
&
SQLInfo
);
...
...
src/client/src/tscSub.c
浏览文件 @
7e12d2f9
...
...
@@ -382,7 +382,7 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
pSql
->
cmd
.
command
=
TSDB_SQL_SELECT
;
pQueryInfo
->
type
=
type
;
tscGetTableMetaInfoFromCmd
(
&
pSql
->
cmd
,
0
,
0
)
->
v
nodeIndex
=
0
;
tscGetTableMetaInfoFromCmd
(
&
pSql
->
cmd
,
0
,
0
)
->
d
nodeIndex
=
0
;
}
tscDoQuery
(
pSql
);
...
...
src/client/src/tscSubquery.c
浏览文件 @
7e12d2f9
...
...
@@ -341,8 +341,8 @@ int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql) {
tscPrintSelectClause
(
pNew
,
0
);
tscTrace
(
"%p subquery:%p tableIndex:%d,
vnodeId
x:%d, type:%d, exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s"
,
pSql
,
pNew
,
0
,
pTableMetaInfo
->
v
nodeIndex
,
pNewQueryInfo
->
type
,
tscTrace
(
"%p subquery:%p tableIndex:%d,
dnodeInde
x:%d, type:%d, exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s"
,
pSql
,
pNew
,
0
,
pTableMetaInfo
->
d
nodeIndex
,
pNewQueryInfo
->
type
,
pNewQueryInfo
->
exprsInfo
.
numOfExprs
,
pNewQueryInfo
->
colList
.
numOfCols
,
pNewQueryInfo
->
fieldsInfo
.
numOfOutputCols
,
pNewQueryInfo
->
pTableMetaInfo
[
0
]
->
name
);
}
...
...
@@ -457,7 +457,7 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) {
assert
(
pQueryInfo
->
numOfTables
==
1
);
// for subquery, only one metermetaInfo
STableMetaInfo
*
pTableMetaInfo
=
tscGetMetaInfo
(
pQueryInfo
,
0
);
tsBufMerge
(
pSupporter
->
pTSBuf
,
pBuf
,
pTableMetaInfo
->
v
nodeIndex
);
tsBufMerge
(
pSupporter
->
pTSBuf
,
pBuf
,
pTableMetaInfo
->
d
nodeIndex
);
tsBufDestory
(
pBuf
);
}
...
...
@@ -478,9 +478,9 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) {
// for projection query, need to try next vnode
// int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
int32_t
totalVnode
=
0
;
if
((
++
pTableMetaInfo
->
v
nodeIndex
)
<
totalVnode
)
{
if
((
++
pTableMetaInfo
->
d
nodeIndex
)
<
totalVnode
)
{
tscTrace
(
"%p current vnode:%d exhausted, try next:%d. total vnode:%d. current numOfRes:%d"
,
pSql
,
pTableMetaInfo
->
vnodeIndex
-
1
,
pTableMetaInfo
->
v
nodeIndex
,
totalVnode
,
pRes
->
numOfTotal
);
pTableMetaInfo
->
dnodeIndex
-
1
,
pTableMetaInfo
->
d
nodeIndex
,
totalVnode
,
pRes
->
numOfTotal
);
pSql
->
cmd
.
command
=
TSDB_SQL_SELECT
;
pSql
->
fp
=
tscJoinQueryCallback
;
...
...
@@ -542,7 +542,7 @@ static void joinRetrieveCallback(void* param, TAOS_RES* tres, int numOfRows) {
assert
(
pQueryInfo
->
numOfTables
==
1
);
// for projection query, need to try next vnode if current vnode is exhausted
// if ((++pTableMetaInfo->
v
nodeIndex) < pTableMetaInfo->pMetricMeta->numOfVnodes) {
// if ((++pTableMetaInfo->
d
nodeIndex) < pTableMetaInfo->pMetricMeta->numOfVnodes) {
// pSupporter->pState->numOfCompleted = 0;
// pSupporter->pState->numOfTotal = 1;
//
...
...
@@ -609,7 +609,7 @@ void tscFetchDatablockFromSubquery(SSqlObj* pSql) {
// STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
if
(
tscNonOrderedProjectionQueryOnSTable
(
pQueryInfo
,
0
))
{
// if (pRes->row >= pRes->numOfRows && pTableMetaInfo->
v
nodeIndex < pTableMetaInfo->pMetricMeta->numOfVnodes &&
// if (pRes->row >= pRes->numOfRows && pTableMetaInfo->
d
nodeIndex < pTableMetaInfo->pMetricMeta->numOfVnodes &&
// (!tscHasReachLimitation(pQueryInfo, pRes))) {
// numOfFetch++;
// }
...
...
@@ -647,8 +647,8 @@ void tscFetchDatablockFromSubquery(SSqlObj* pSql) {
STableMetaInfo
*
pTableMetaInfo
=
tscGetMetaInfo
(
pQueryInfo
,
0
);
if
(
pRes1
->
row
>=
pRes1
->
numOfRows
)
{
tscTrace
(
"%p subquery:%p retrieve data from vnode, subquery:%d,
v
nodeIndex:%d"
,
pSql
,
pSql1
,
pSupporter
->
subqueryIndex
,
pTableMetaInfo
->
v
nodeIndex
);
tscTrace
(
"%p subquery:%p retrieve data from vnode, subquery:%d,
d
nodeIndex:%d"
,
pSql
,
pSql1
,
pSupporter
->
subqueryIndex
,
pTableMetaInfo
->
d
nodeIndex
);
tscResetForNextRetrieve
(
pRes1
);
pSql1
->
fp
=
joinRetrieveCallback
;
...
...
@@ -785,11 +785,11 @@ void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code) {
STableMetaInfo
*
pTableMetaInfo
=
tscGetMetaInfo
(
pQueryInfo
,
0
);
/**
* if the query is a continue query (
v
nodeIndex > 0 for projection query) for next vnode, do the retrieval of
* if the query is a continue query (
d
nodeIndex > 0 for projection query) for next vnode, do the retrieval of
* data instead of returning to its invoker
*/
if
(
pTableMetaInfo
->
v
nodeIndex
>
0
&&
tscNonOrderedProjectionQueryOnSTable
(
pQueryInfo
,
0
))
{
// assert(pTableMetaInfo->
v
nodeIndex < pTableMetaInfo->pMetricMeta->numOfVnodes);
if
(
pTableMetaInfo
->
d
nodeIndex
>
0
&&
tscNonOrderedProjectionQueryOnSTable
(
pQueryInfo
,
0
))
{
// assert(pTableMetaInfo->
d
nodeIndex < pTableMetaInfo->pMetricMeta->numOfVnodes);
pSupporter
->
pState
->
numOfCompleted
=
0
;
// reset the record value
pSql
->
fp
=
joinRetrieveCallback
;
// continue retrieve data
...
...
@@ -897,14 +897,14 @@ int32_t tscLaunchJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSubquerySu
tscTrace
(
"%p subquery:%p tableIndex:%d, vnodeIdx:%d, type:%d, transfer to ts_comp query to retrieve timestamps, "
"exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s"
,
pSql
,
pNew
,
tableIndex
,
pTableMetaInfo
->
v
nodeIndex
,
pNewQueryInfo
->
type
,
pSql
,
pNew
,
tableIndex
,
pTableMetaInfo
->
d
nodeIndex
,
pNewQueryInfo
->
type
,
pNewQueryInfo
->
exprsInfo
.
numOfExprs
,
pNewQueryInfo
->
colList
.
numOfCols
,
pNewQueryInfo
->
fieldsInfo
.
numOfOutputCols
,
pNewQueryInfo
->
pTableMetaInfo
[
0
]
->
name
);
tscPrintSelectClause
(
pNew
,
0
);
tscTrace
(
"%p subquery:%p tableIndex:%d, vnodeIdx:%d, type:%d, transfer to ts_comp query to retrieve timestamps, "
"exprInfo:%d, colList:%d, fieldsInfo:%d, name:%s"
,
pSql
,
pNew
,
tableIndex
,
pTableMetaInfo
->
v
nodeIndex
,
pNewQueryInfo
->
type
,
pSql
,
pNew
,
tableIndex
,
pTableMetaInfo
->
d
nodeIndex
,
pNewQueryInfo
->
type
,
pNewQueryInfo
->
exprsInfo
.
numOfExprs
,
pNewQueryInfo
->
colList
.
numOfCols
,
pNewQueryInfo
->
fieldsInfo
.
numOfOutputCols
,
pNewQueryInfo
->
pTableMetaInfo
[
0
]
->
name
);
tscPrintSelectClause
(
pNew
,
0
);
...
...
@@ -1005,7 +1005,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
SQueryInfo
*
pQueryInfo
=
tscGetQueryInfoDetail
(
pCmd
,
pCmd
->
clauseIndex
);
STableMetaInfo
*
pTableMetaInfo
=
tscGetMetaInfo
(
pQueryInfo
,
0
);
pSql
->
numOfSubs
=
taosArrayGetSize
(
pTableMetaInfo
->
vgroupIdList
)
;
pSql
->
numOfSubs
=
pTableMetaInfo
->
vgroupList
->
numOfDnodes
;
assert
(
pSql
->
numOfSubs
>
0
);
int32_t
ret
=
tscLocalReducerEnvCreate
(
pSql
,
&
pMemoryBuf
,
&
pDesc
,
&
pModel
,
nBufferSize
);
...
...
@@ -1111,6 +1111,7 @@ static void tscFreeSubSqlObj(SRetrieveSupport *trsupport, SSqlObj *pSql) {
}
static
void
tscRetrieveFromDnodeCallBack
(
void
*
param
,
TAOS_RES
*
tres
,
int
numOfRows
);
static
void
tscHandleSubqueryError
(
SRetrieveSupport
*
trsupport
,
SSqlObj
*
pSql
,
int
numOfRows
);
static
void
tscAbortFurtherRetryRetrieval
(
SRetrieveSupport
*
trsupport
,
TAOS_RES
*
tres
,
int32_t
errCode
)
{
// set no disk space error info
...
...
@@ -1132,10 +1133,10 @@ static void tscAbortFurtherRetryRetrieval(SRetrieveSupport *trsupport, TAOS_RES
pthread_mutex_unlock
(
&
trsupport
->
queryMutex
);
tsc
RetrieveFromDnodeCallBack
(
trsupport
,
tres
,
trsupport
->
pState
->
code
);
tsc
HandleSubqueryError
(
trsupport
,
tres
,
trsupport
->
pState
->
code
);
}
static
void
tscHandleSubRetrieval
Error
(
SRetrieveSupport
*
trsupport
,
SSqlObj
*
pSql
,
int
numOfRows
)
{
void
tscHandleSubquery
Error
(
SRetrieveSupport
*
trsupport
,
SSqlObj
*
pSql
,
int
numOfRows
)
{
SSqlObj
*
pPObj
=
trsupport
->
pParentSqlObj
;
int32_t
subqueryIndex
=
trsupport
->
subqueryIndex
;
...
...
@@ -1144,9 +1145,9 @@ static void tscHandleSubRetrievalError(SRetrieveSupport *trsupport, SSqlObj *pSq
assert
(
pState
->
numOfCompleted
<
pState
->
numOfTotal
&&
pState
->
numOfCompleted
>=
0
&&
pPObj
->
numOfSubs
==
pState
->
numOfTotal
);
/
* retrieved in subquery failed. OR query cancelled in retrieve phase. */
/
/ retrieved in subquery failed. OR query cancelled in retrieve phase.
if
(
pState
->
code
==
TSDB_CODE_SUCCESS
&&
pPObj
->
res
.
code
!=
TSDB_CODE_SUCCESS
)
{
pState
->
code
=
-
(
int
)
pPObj
->
res
.
code
;
pState
->
code
=
pPObj
->
res
.
code
;
/*
* kill current sub-query connection, which may retrieve data from vnodes;
...
...
@@ -1179,7 +1180,7 @@ static void tscHandleSubRetrievalError(SRetrieveSupport *trsupport, SSqlObj *pSq
SSqlObj
*
pNew
=
tscCreateSqlObjForSubquery
(
trsupport
->
pParentSqlObj
,
trsupport
,
pSql
);
if
(
pNew
==
NULL
)
{
tscError
(
"%p sub:%p failed to create new subquery sql
o
bj due to out of memory, abort retry"
,
tscError
(
"%p sub:%p failed to create new subquery sql
O
bj due to out of memory, abort retry"
,
trsupport
->
pParentSqlObj
,
pSql
);
pState
->
code
=
TSDB_CODE_CLI_OUT_OF_MEMORY
;
...
...
@@ -1235,10 +1236,13 @@ static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* p
SSubqueryState
*
pState
=
trsupport
->
pState
;
SQueryInfo
*
pQueryInfo
=
tscGetQueryInfoDetail
(
&
pSql
->
cmd
,
0
);
STableMetaInfo
*
pTableMetaInfo
=
pQueryInfo
->
pTableMetaInfo
[
0
];
// data in from current vnode is stored in cache and disk
// uint32_t numOfRowsFromSubquery = trsupport->pExtMemBuffer[idx]->numOfTotalElems + trsupport->localBuffer->numOfElems;
// tscTrace("%p sub:%p all data retrieved from ip:%u,vid:%d, numOfRows:%d, orderOfSub:%d", pPObj, pSql, pSvd->ip,
// pSvd->vnode, numOfRowsFromSubquery, idx);
uint32_t
numOfRowsFromSubquery
=
trsupport
->
pExtMemBuffer
[
idx
]
->
numOfTotalElems
+
trsupport
->
localBuffer
->
numOfElems
;
tscTrace
(
"%p sub:%p all data retrieved from ip:%u,vgId:%d, numOfRows:%d, orderOfSub:%d"
,
pPObj
,
pSql
,
pTableMetaInfo
->
vgroupList
->
dnodeVgroups
[
0
].
ipAddr
.
ip
,
pTableMetaInfo
->
vgroupList
->
dnodeVgroups
[
0
].
vgId
[
0
],
numOfRowsFromSubquery
,
idx
);
tColModelCompact
(
pDesc
->
pColumnModel
,
trsupport
->
localBuffer
,
pDesc
->
pColumnModel
->
capacity
);
...
...
@@ -1326,7 +1330,7 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
pthread_mutex_lock
(
&
trsupport
->
queryMutex
);
if
(
numOfRows
<
0
||
pState
->
code
<
0
||
pPObj
->
res
.
code
!=
TSDB_CODE_SUCCESS
)
{
return
tscHandleSub
Retrieval
Error
(
trsupport
,
pSql
,
numOfRows
);
return
tscHandleSub
query
Error
(
trsupport
,
pSql
,
numOfRows
);
}
SSqlRes
*
pRes
=
&
pSql
->
res
;
...
...
@@ -1352,7 +1356,6 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
return
;
}
#ifdef _DEBUG_VIEW
printf
(
"received data from vnode: %d rows
\n
"
,
pRes
->
numOfRows
);
SSrcColumnInfo
colInfo
[
256
]
=
{
0
};
...
...
@@ -1360,6 +1363,7 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
tscGetSrcColumnInfo
(
colInfo
,
pQueryInfo
);
tColModelDisplayEx
(
pDesc
->
pColumnModel
,
pRes
->
data
,
pRes
->
numOfRows
,
pRes
->
numOfRows
,
colInfo
);
#endif
if
(
tsTotalTmpDirGB
!=
0
&&
tsAvailTmpDirGB
<
tsMinimalTmpDirGB
)
{
tscError
(
"%p sub:%p client disk space remain %.3f GB, need at least %.3f GB, stop query"
,
pPObj
,
pSql
,
tsAvailTmpDirGB
,
tsMinimalTmpDirGB
);
...
...
@@ -1371,8 +1375,11 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
pRes
->
numOfRows
,
pQueryInfo
->
groupbyExpr
.
orderType
);
if
(
ret
<
0
)
{
// set no disk space error info, and abort retry
tscAbortFurtherRetryRetrieval
(
trsupport
,
tres
,
TSDB_CODE_CLI_NO_DISKSPACE
);
}
else
if
(
pRes
->
completed
)
{
tscAllDataRetrievedFromDnode
(
trsupport
,
pSql
);
return
;
}
else
{
// continue fetch data from dnode
pthread_mutex_unlock
(
&
trsupport
->
queryMutex
);
taos_fetch_rows_a
(
tres
,
tscRetrieveFromDnodeCallBack
,
param
);
...
...
@@ -1380,6 +1387,7 @@ static void tscRetrieveFromDnodeCallBack(void *param, TAOS_RES *tres, int numOfR
}
else
{
// all data has been retrieved to client
tscAllDataRetrievedFromDnode
(
trsupport
,
pSql
);
}
pthread_mutex_unlock
(
&
trsupport
->
queryMutex
);
}
...
...
@@ -1393,9 +1401,9 @@ static SSqlObj *tscCreateSqlObjForSubquery(SSqlObj *pSql, SRetrieveSupport *trsu
assert
(
pQueryInfo
->
numOfTables
==
1
&&
pNew
->
cmd
.
numOfClause
==
1
);
// launch subquery for each vnode, so the subquery index equals to the
v
nodeIndex.
// launch subquery for each vnode, so the subquery index equals to the
d
nodeIndex.
STableMetaInfo
*
pTableMetaInfo
=
tscGetMetaInfo
(
pQueryInfo
,
table_index
);
pTableMetaInfo
->
v
nodeIndex
=
trsupport
->
subqueryIndex
;
pTableMetaInfo
->
d
nodeIndex
=
trsupport
->
subqueryIndex
;
pSql
->
pSubs
[
trsupport
->
subqueryIndex
]
=
pNew
;
}
...
...
@@ -1404,37 +1412,34 @@ static SSqlObj *tscCreateSqlObjForSubquery(SSqlObj *pSql, SRetrieveSupport *trsu
}
void
tscRetrieveDataRes
(
void
*
param
,
TAOS_RES
*
tres
,
int
code
)
{
SRetrieveSupport
*
trsupport
=
(
SRetrieveSupport
*
)
param
;
SRetrieveSupport
*
trsupport
=
(
SRetrieveSupport
*
)
param
;
SSqlObj
*
pParentSql
=
trsupport
->
pParentSqlObj
;
SSqlObj
*
pSql
=
(
SSqlObj
*
)
tres
;
// STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, 0, 0);
assert
(
pSql
->
cmd
.
numOfClause
==
1
&&
pSql
->
cmd
.
pQueryInfo
[
0
]
->
numOfTables
==
1
);
SSqlObj
*
pSql
=
(
SSqlObj
*
)
tres
;
// int32_t idx = pTableMetaInfo->vnodeIndex;
SQueryInfo
*
pQueryInfo
=
tscGetQueryInfoDetail
(
&
pSql
->
cmd
,
0
);
assert
(
pSql
->
cmd
.
numOfClause
==
1
&&
pQueryInfo
->
numOfTables
==
1
);
SVnodeSidList
*
vnodeInfo
=
NULL
;
SVnodeDesc
*
pSvd
=
NULL
;
// if (pTableMetaInfo->pMetricMeta != NULL) {
// vnodeInfo = tscGetVnodeSidList(pTableMetaInfo->pMetricMeta, idx);
// pSvd = &vnodeInfo->vpeerDesc[vnodeInfo->index];
// }
STableMetaInfo
*
pTableMetaInfo
=
tscGetTableMetaInfoFromCmd
(
&
pSql
->
cmd
,
0
,
0
);
STableDnodeVgroupInfo
*
pVgroupInfo
=
&
pTableMetaInfo
->
vgroupList
->
dnodeVgroups
[
0
];
SSubqueryState
*
pState
=
trsupport
->
pState
;
assert
(
pState
->
numOfCompleted
<
pState
->
numOfTotal
&&
pState
->
numOfCompleted
>=
0
&&
pParentSql
->
numOfSubs
==
pState
->
numOfTotal
);
if
(
pParentSql
->
res
.
code
!=
TSDB_CODE_SUCCESS
||
pState
->
code
!=
TSDB_CODE_SUCCESS
)
{
// metric query is killed, Note: code must be less than 0
// stable query is killed, abort further retry
trsupport
->
numOfRetry
=
MAX_NUM_OF_SUBQUERY_RETRY
;
if
(
pParentSql
->
res
.
code
!=
TSDB_CODE_SUCCESS
)
{
code
=
-
(
int
)(
pParentSql
->
res
.
code
)
;
code
=
pParentSql
->
res
.
code
;
}
else
{
code
=
pState
->
code
;
}
tscTrace
(
"%p query cancelled or failed, sub:%p, orderOfSub:%d abort, code:%d"
,
pParentSql
,
pSql
,
trsupport
->
subqueryIndex
,
code
);
tscTrace
(
"%p query cancelled or failed, sub:%p, orderOfSub:%d abort, code:%s"
,
pParentSql
,
pSql
,
trsupport
->
subqueryIndex
,
tstrerror
(
code
));
}
/*
...
...
@@ -1442,51 +1447,40 @@ void tscRetrieveDataRes(void *param, TAOS_RES *tres, int code) {
* than this one are actually not necessary, we simply call the tscRetrieveFromDnodeCallBack
* function to abort current and remain retrieve process.
*
* NOTE: threadsafe is required.
* NOTE: thread
safe is required.
*/
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
if
(
trsupport
->
numOfRetry
++
>=
MAX_NUM_OF_SUBQUERY_RETRY
)
{
tscTrace
(
"%p sub:%p reach the max retry
count,
set global code:%d"
,
pParentSql
,
pSql
,
code
);
tscTrace
(
"%p sub:%p reach the max retry
times,
set global code:%d"
,
pParentSql
,
pSql
,
code
);
atomic_val_compare_exchange_32
(
&
pState
->
code
,
0
,
code
);
}
else
{
// does not reach the maximum retry
count
, go on
}
else
{
// does not reach the maximum retry
time
, go on
tscTrace
(
"%p sub:%p failed code:%s, retry:%d"
,
pParentSql
,
pSql
,
tstrerror
(
code
),
trsupport
->
numOfRetry
);
SSqlObj
*
pNew
=
tscCreateSqlObjForSubquery
(
pParentSql
,
trsupport
,
pSql
);
if
(
pNew
==
NULL
)
{
tscError
(
"%p sub:%p failed to create new subquery due to out of memory, abort retry, v
i
d:%d, orderOfSub:%d"
,
trsupport
->
pParentSqlObj
,
pSql
,
p
Svd
!=
NULL
?
pSvd
->
vgId
:
-
1
,
trsupport
->
subqueryIndex
);
tscError
(
"%p sub:%p failed to create new subquery due to out of memory, abort retry, v
gI
d:%d, orderOfSub:%d"
,
trsupport
->
pParentSqlObj
,
pSql
,
p
VgroupInfo
->
vgId
[
0
]
,
trsupport
->
subqueryIndex
);
pState
->
code
=
-
TSDB_CODE_CLI_OUT_OF_MEMORY
;
pState
->
code
=
TSDB_CODE_CLI_OUT_OF_MEMORY
;
trsupport
->
numOfRetry
=
MAX_NUM_OF_SUBQUERY_RETRY
;
}
else
{
// SQueryInfo *pNewQueryInfo = tscGetQueryInfoDetail(&pNew->cmd, 0);
// assert(pNewQueryInfo->pTableMetaInfo[0]->pTableMeta != NULL && pNewQueryInfo->pTableMetaInfo[0]->pMetricMeta != NULL);
SQueryInfo
*
pNewQueryInfo
=
tscGetQueryInfoDetail
(
&
pNew
->
cmd
,
0
);
assert
(
pNewQueryInfo
->
pTableMetaInfo
[
0
]
->
pTableMeta
!=
NULL
);
tscProcessSql
(
pNew
);
return
;
}
}
}
if
(
pState
->
code
!=
TSDB_CODE_SUCCESS
)
{
// failed, abort
if
(
vnodeInfo
!=
NULL
)
{
tscTrace
(
"%p sub:%p query failed,ip:%u,vid:%d,orderOfSub:%d,global code:%d"
,
pParentSql
,
pSql
,
vnodeInfo
->
vpeerDesc
[
vnodeInfo
->
index
].
ip
,
vnodeInfo
->
vpeerDesc
[
vnodeInfo
->
index
].
vgId
,
trsupport
->
subqueryIndex
,
pState
->
code
);
}
else
{
tscTrace
(
"%p sub:%p query failed,orderOfSub:%d,global code:%d"
,
pParentSql
,
pSql
,
trsupport
->
subqueryIndex
,
pState
->
code
);
}
tscRetrieveFromDnodeCallBack
(
param
,
tres
,
pState
->
code
);
if
(
pState
->
code
!=
TSDB_CODE_SUCCESS
)
{
// at least one peer subquery failed, abort current query
tscTrace
(
"%p sub:%p query failed,ip:%u,vgId:%d,orderOfSub:%d,global code:%d"
,
pParentSql
,
pSql
,
pVgroupInfo
->
ipAddr
.
ip
,
pVgroupInfo
->
vgId
[
0
],
trsupport
->
subqueryIndex
,
pState
->
code
);
tscHandleSubqueryError
(
param
,
tres
,
pState
->
code
);
}
else
{
// success, proceed to retrieve data from dnode
if
(
vnodeInfo
!=
NULL
)
{
tscTrace
(
"%p sub:%p query complete,ip:%u,vid:%d,orderOfSub:%d,retrieve data"
,
trsupport
->
pParentSqlObj
,
pSql
,
vnodeInfo
->
vpeerDesc
[
vnodeInfo
->
index
].
ip
,
vnodeInfo
->
vpeerDesc
[
vnodeInfo
->
index
].
vgId
,
trsupport
->
subqueryIndex
);
}
else
{
tscTrace
(
"%p sub:%p query complete, orderOfSub:%d,retrieve data"
,
trsupport
->
pParentSqlObj
,
pSql
,
trsupport
->
subqueryIndex
);
}
tscTrace
(
"%p sub:%p query complete, ip:%u, vgId:%d, orderOfSub:%d,retrieve data"
,
trsupport
->
pParentSqlObj
,
pSql
,
pVgroupInfo
->
ipAddr
.
ip
,
pVgroupInfo
->
vgId
[
0
],
trsupport
->
subqueryIndex
);
taos_fetch_rows_a
(
tres
,
tscRetrieveFromDnodeCallBack
,
param
);
}
...
...
src/client/src/tscUtil.c
浏览文件 @
7e12d2f9
...
...
@@ -1178,8 +1178,9 @@ SSqlExpr* tscSqlExprInsert(SQueryInfo* pQueryInfo, int32_t index, int16_t functi
pExprInfo
->
pExprs
[
index
]
=
pExpr
;
pExpr
->
functionId
=
functionId
;
int16_t
numOfCols
=
tscGetNumOfColumns
(
pTableMetaInfo
->
pTableMeta
);
// set the correct column index
if
(
pColIndex
->
columnIndex
==
TSDB_TBNAME_COLUMN_INDEX
)
{
pExpr
->
colInfo
.
colId
=
TSDB_TBNAME_COLUMN_INDEX
;
...
...
@@ -1190,7 +1191,6 @@ SSqlExpr* tscSqlExprInsert(SQueryInfo* pQueryInfo, int32_t index, int16_t functi
// tag columns require the column index revised.
if
(
pColIndex
->
columnIndex
>=
numOfCols
)
{
pColIndex
->
columnIndex
-=
numOfCols
;
pExpr
->
colInfo
.
flag
=
TSDB_COL_TAG
;
}
else
{
if
(
pColIndex
->
columnIndex
!=
TSDB_TBNAME_COLUMN_INDEX
)
{
...
...
@@ -1916,7 +1916,7 @@ void tscFreeSubqueryInfo(SSqlCmd* pCmd) {
}
STableMetaInfo
*
tscAddTableMetaInfo
(
SQueryInfo
*
pQueryInfo
,
const
char
*
name
,
STableMeta
*
pTableMeta
,
S
Array
*
vgroupList
,
int16_t
numOfTags
,
int16_t
*
tags
)
{
S
VgroupsInfo
*
vgroupList
,
int16_t
numOfTags
,
int16_t
*
tags
)
{
void
*
pAlloc
=
realloc
(
pQueryInfo
->
pTableMetaInfo
,
(
pQueryInfo
->
numOfTables
+
1
)
*
POINTER_BYTES
);
if
(
pAlloc
==
NULL
)
{
return
NULL
;
...
...
@@ -1937,7 +1937,12 @@ STableMetaInfo* tscAddTableMetaInfo(SQueryInfo* pQueryInfo, const char* name, ST
pTableMetaInfo
->
numOfTags
=
numOfTags
;
if
(
vgroupList
!=
NULL
)
{
pTableMetaInfo
->
vgroupIdList
=
taosArrayClone
(
vgroupList
);
assert
(
vgroupList
->
numOfDnodes
==
1
);
// todo fix me
size_t
size
=
sizeof
(
SVgroupsInfo
)
+
(
sizeof
(
STableDnodeVgroupInfo
)
+
vgroupList
->
dnodeVgroups
[
0
].
numOfVgroups
*
sizeof
(
int32_t
))
*
vgroupList
->
numOfDnodes
;
pTableMetaInfo
->
vgroupList
=
malloc
(
size
);
memcpy
(
pTableMetaInfo
->
vgroupList
,
vgroupList
,
size
);
}
if
(
tags
!=
NULL
)
{
...
...
@@ -1952,7 +1957,7 @@ STableMetaInfo* tscAddEmptyMetaInfo(SQueryInfo* pQueryInfo) {
return
tscAddTableMetaInfo
(
pQueryInfo
,
NULL
,
NULL
,
NULL
,
0
,
NULL
);
}
void
doRemove
Meter
MetaInfo
(
SQueryInfo
*
pQueryInfo
,
int32_t
index
,
bool
removeFromCache
)
{
void
doRemove
Table
MetaInfo
(
SQueryInfo
*
pQueryInfo
,
int32_t
index
,
bool
removeFromCache
)
{
if
(
index
<
0
||
index
>=
pQueryInfo
->
numOfTables
)
{
return
;
}
...
...
@@ -1975,7 +1980,7 @@ void tscRemoveAllMeterMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool
int32_t
index
=
pQueryInfo
->
numOfTables
;
while
(
index
>=
0
)
{
doRemove
Meter
MetaInfo
(
pQueryInfo
,
--
index
,
removeFromCache
);
doRemove
Table
MetaInfo
(
pQueryInfo
,
--
index
,
removeFromCache
);
}
tfree
(
pQueryInfo
->
pTableMetaInfo
);
...
...
@@ -1987,6 +1992,7 @@ void tscClearMeterMetaInfo(STableMetaInfo* pTableMetaInfo, bool removeFromCache)
}
taosCacheRelease
(
tscCacheHandle
,
(
void
**
)
&
(
pTableMetaInfo
->
pTableMeta
),
removeFromCache
);
tfree
(
pTableMetaInfo
->
vgroupList
);
// taosCacheRelease(tscCacheHandle, (void**)&(pTableMetaInfo->pMetricMeta), removeFromCache);
}
...
...
@@ -2014,7 +2020,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
pNew
->
sqlstr
=
strdup
(
pSql
->
sqlstr
);
if
(
pNew
->
sqlstr
==
NULL
)
{
tscError
(
"%p new subquery failed, tableIndex:%d,
vnodeIndex:%d"
,
pSql
,
tableIndex
,
pTableMetaInfo
->
v
nodeIndex
);
tscError
(
"%p new subquery failed, tableIndex:%d,
dnodeIndex:%d"
,
pSql
,
tableIndex
,
pTableMetaInfo
->
d
nodeIndex
);
free
(
pNew
);
return
NULL
;
...
...
@@ -2058,7 +2064,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
}
if
(
tscAllocPayload
(
pnCmd
,
TSDB_DEFAULT_PAYLOAD_SIZE
)
!=
TSDB_CODE_SUCCESS
)
{
tscError
(
"%p new subquery failed, tableIndex:%d,
vnodeIndex:%d"
,
pSql
,
tableIndex
,
pTableMetaInfo
->
v
nodeIndex
);
tscError
(
"%p new subquery failed, tableIndex:%d,
dnodeIndex:%d"
,
pSql
,
tableIndex
,
pTableMetaInfo
->
d
nodeIndex
);
tscFreeSqlObj
(
pNew
);
return
NULL
;
}
...
...
@@ -2128,7 +2134,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
// pMetricMeta = taosCacheAcquireByName(tscCacheHandle, key);
// }
pFinalInfo
=
tscAddTableMetaInfo
(
pNewQueryInfo
,
name
,
pTableMeta
,
pTableMetaInfo
->
vgroup
Id
List
,
pTableMetaInfo
->
numOfTags
,
pFinalInfo
=
tscAddTableMetaInfo
(
pNewQueryInfo
,
name
,
pTableMeta
,
pTableMetaInfo
->
vgroupList
,
pTableMetaInfo
->
numOfTags
,
pTableMetaInfo
->
tagColumnIndex
);
}
else
{
// transfer the ownership of pTableMeta/pMetricMeta to the newly create sql object.
// STableMetaInfo* pPrevInfo = tscGetTableMetaInfoFromCmd(&pPrevSql->cmd, pPrevSql->cmd.clauseIndex, 0);
...
...
@@ -2149,13 +2155,13 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
tscTrace
(
"%p new subquery: %p, tableIndex:%d, vnodeIdx:%d, type:%d, exprInfo:%d, colList:%d,"
"fieldInfo:%d, name:%s, qrang:%"
PRId64
" - %"
PRId64
" order:%d, limit:%"
PRId64
,
pSql
,
pNew
,
tableIndex
,
pTableMetaInfo
->
v
nodeIndex
,
pNewQueryInfo
->
type
,
pNewQueryInfo
->
exprsInfo
.
numOfExprs
,
pSql
,
pNew
,
tableIndex
,
pTableMetaInfo
->
d
nodeIndex
,
pNewQueryInfo
->
type
,
pNewQueryInfo
->
exprsInfo
.
numOfExprs
,
pNewQueryInfo
->
colList
.
numOfCols
,
pNewQueryInfo
->
fieldsInfo
.
numOfOutputCols
,
pFinalInfo
->
name
,
pNewQueryInfo
->
stime
,
pNewQueryInfo
->
etime
,
pNewQueryInfo
->
order
.
order
,
pNewQueryInfo
->
limit
.
limit
);
tscPrintSelectClause
(
pNew
,
0
);
}
else
{
tscTrace
(
"%p new sub insertion: %p, vnodeIdx:%d"
,
pSql
,
pNew
,
pTableMetaInfo
->
v
nodeIndex
);
tscTrace
(
"%p new sub insertion: %p, vnodeIdx:%d"
,
pSql
,
pNew
,
pTableMetaInfo
->
d
nodeIndex
);
}
return
pNew
;
...
...
@@ -2252,7 +2258,7 @@ bool hasMoreVnodesToTry(SSqlObj* pSql) {
// int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
// return pRes->numOfRows == 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) &&
// (!tscHasReachLimitation(pQueryInfo, pRes)) && (pTableMetaInfo->
v
nodeIndex < totalVnode - 1);
// (!tscHasReachLimitation(pQueryInfo, pRes)) && (pTableMetaInfo->
d
nodeIndex < totalVnode - 1);
}
void
tscTryQueryNextVnode
(
SSqlObj
*
pSql
,
__async_cb_func_t
fp
)
{
...
...
@@ -2271,9 +2277,9 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
int32_t
totalVnode
=
0
;
// int32_t totalVnode = pTableMetaInfo->pMetricMeta->numOfVnodes;
while
(
++
pTableMetaInfo
->
v
nodeIndex
<
totalVnode
)
{
while
(
++
pTableMetaInfo
->
d
nodeIndex
<
totalVnode
)
{
tscTrace
(
"%p current vnode:%d exhausted, try next:%d. total vnode:%d. current numOfRes:%d"
,
pSql
,
pTableMetaInfo
->
vnodeIndex
-
1
,
pTableMetaInfo
->
v
nodeIndex
,
totalVnode
,
pRes
->
numOfTotalInCurrentClause
);
pTableMetaInfo
->
dnodeIndex
-
1
,
pTableMetaInfo
->
d
nodeIndex
,
totalVnode
,
pRes
->
numOfTotalInCurrentClause
);
/*
* update the limit and offset value for the query on the next vnode,
...
...
@@ -2292,7 +2298,7 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
assert
((
pRes
->
offset
>=
0
&&
pRes
->
numOfRows
==
0
)
||
(
pRes
->
offset
==
0
&&
pRes
->
numOfRows
>=
0
));
tscTrace
(
"%p new query to next vnode, vnode index:%d, limit:%"
PRId64
", offset:%"
PRId64
", glimit:%"
PRId64
,
pSql
,
pTableMetaInfo
->
v
nodeIndex
,
pQueryInfo
->
limit
.
limit
,
pQueryInfo
->
limit
.
offset
,
pQueryInfo
->
clauseLimit
);
pTableMetaInfo
->
d
nodeIndex
,
pQueryInfo
->
limit
.
limit
,
pQueryInfo
->
limit
.
offset
,
pQueryInfo
->
clauseLimit
);
/*
* For project query with super table join, the numOfSub is equalled to the number of all subqueries.
...
...
src/inc/taosmsg.h
浏览文件 @
7e12d2f9
...
...
@@ -188,6 +188,11 @@ extern char *taosMsg[];
#pragma pack(push, 1)
typedef
struct
{
uint32_t
ip
;
uint16_t
port
;
}
SIpAddr
;
typedef
struct
{
int32_t
numOfVnodes
;
}
SMsgDesc
;
...
...
@@ -469,7 +474,6 @@ typedef struct {
int16_t
numOfGroupCols
;
// num of group by columns
int16_t
orderByIdx
;
int16_t
orderType
;
// used in group by xx order by xxx
uint64_t
groupbyTagIds
;
int64_t
limit
;
int64_t
offset
;
uint16_t
queryType
;
// denote another query process
...
...
@@ -629,9 +633,15 @@ typedef struct SCMSTableVgroupMsg {
char
tableId
[
TSDB_TABLE_ID_LEN
];
}
SCMSTableVgroupMsg
;
typedef
struct
{
SIpAddr
ipAddr
;
int32_t
numOfVgroups
;
int32_t
vgId
[];
}
STableDnodeVgroupInfo
;
typedef
struct
{
int32_t
numOfDnodes
;
uint32_t
dnodeI
ps
[];
STableDnodeVgroupInfo
dnodeVgrou
ps
[];
}
SCMSTableVgroupRspMsg
;
typedef
struct
{
...
...
src/mnode/src/mgmtTable.c
浏览文件 @
7e12d2f9
...
...
@@ -1117,12 +1117,26 @@ static void mgmtProcessSuperTableVgroupMsg(SQueuedMsg *pMsg) {
if
(
pRsp
==
NULL
)
{
mgmtSendSimpleResp
(
pMsg
->
thandle
,
TSDB_CODE_INVALID_TABLE
);
return
;
}
pRsp
->
numOfDnodes
=
htonl
(
1
);
pRsp
->
dnodeIps
[
0
]
=
htonl
(
inet_addr
(
tsPrivateIp
));
}
int32_t
numOfVgroups
=
1
;
int32_t
numOfDnodes
=
1
;
pRsp
->
numOfDnodes
=
htonl
(
numOfDnodes
);
STableDnodeVgroupInfo
*
pVgroupInfo
=
pRsp
->
dnodeVgroups
;
pVgroupInfo
->
ipAddr
.
ip
=
htonl
(
inet_addr
(
tsPrivateIp
));
pVgroupInfo
->
ipAddr
.
port
=
htons
(
0
);
// todo fix it
pVgroupInfo
->
numOfVgroups
=
htonl
(
numOfVgroups
);
// todo fix it
int32_t
*
vgIdList
=
pVgroupInfo
->
vgId
;
for
(
int32_t
i
=
0
;
i
<
numOfVgroups
;
++
i
)
{
vgIdList
[
i
]
=
htonl
(
2
);
// todo fix it
}
assert
(
numOfDnodes
==
1
);
// this size is valid only when numOfDnodes equals 1
int32_t
msgLen
=
sizeof
(
SCMSTableVgroupRspMsg
)
+
sizeof
(
STableDnodeVgroupInfo
)
+
numOfVgroups
*
sizeof
(
int32_t
);
int32_t
msgLen
=
sizeof
(
SSuperTableObj
)
+
htonl
(
pRsp
->
numOfDnodes
)
*
sizeof
(
int32_t
);
SRpcMsg
rpcRsp
=
{
0
};
rpcRsp
.
handle
=
pMsg
->
thandle
;
rpcRsp
.
pCont
=
pRsp
;
...
...
@@ -1524,7 +1538,7 @@ static int32_t mgmtDoGetChildTableMeta(SQueuedMsg *pMsg, STableMetaMsg *pMeta) {
}
else
{
pMeta
->
vpeerDesc
[
i
].
ip
=
htonl
(
pVgroup
->
vnodeGid
[
i
].
privateIp
);
}
pMeta
->
vpeerDesc
[
i
].
vgId
=
htonl
(
pVgroup
->
vgId
);
//
pMeta->vpeerDesc[i].vgId = htonl(pVgroup->vgId);
pMeta
->
vpeerDesc
[
i
].
dnodeId
=
htonl
(
pVgroup
->
vnodeGid
[
i
].
dnodeId
);
}
pMeta
->
numOfVpeers
=
pVgroup
->
numOfVnodes
;
...
...
src/query/inc/qextbuffer.h
浏览文件 @
7e12d2f9
...
...
@@ -19,9 +19,14 @@
extern
"C"
{
#endif
#include "os.h"
#include "taosmsg.h"
#include "tarray.h"
#include "tutil.h"
#include "dataformat.h"
#include "talgo.h"
#define DEFAULT_PAGE_SIZE 16384 // 16k larger than the SHistoInfo
#define MIN_BUFFER_SIZE (1 << 19)
...
...
@@ -55,12 +60,12 @@ typedef struct tFlushoutData {
tFlushoutInfo
*
pFlushoutInfo
;
}
tFlushoutData
;
typedef
struct
SFileInfo
{
typedef
struct
S
Ext
FileInfo
{
uint32_t
nFileSize
;
// in pages
uint32_t
pageSize
;
uint32_t
numOfElemsInFile
;
tFlushoutData
flushoutData
;
}
SFileInfo
;
}
S
Ext
FileInfo
;
typedef
struct
tFilePage
{
uint64_t
numOfElems
;
...
...
@@ -109,26 +114,17 @@ typedef struct tExtMemBuffer {
char
*
path
;
FILE
*
file
;
SFileInfo
fileMeta
;
S
Ext
FileInfo
fileMeta
;
SColumnModel
*
pColumnModel
;
EXT_BUFFER_FLUSH_MODEL
flushModel
;
}
tExtMemBuffer
;
typedef
struct
tTagSchema
{
struct
SSchema
*
pSchema
;
int32_t
numOfCols
;
int32_t
colOffset
[];
}
tTagSchema
;
typedef
struct
tSidSet
{
int32_t
numOfSids
;
int32_t
numOfSubSet
;
STableIdInfo
**
pTableIdList
;
int32_t
*
starterPos
;
// position of each subgroup, generated according to
SColumnModel
*
pColumnModel
;
SColumnOrderInfo
orderIdx
;
}
tSidSet
;
//typedef struct tTagSchema {
// struct SSchema *pSchema;
// int32_t numOfCols;
// int32_t colOffset[];
//} tTagSchema;
/**
*
...
...
src/query/inc/queryExecutor.h
浏览文件 @
7e12d2f9
...
...
@@ -39,7 +39,7 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int
typedef
struct
SSqlGroupbyExpr
{
int16_t
tableIndex
;
int16_t
numOfGroupCols
;
SColIndex
columnInfo
[
TSDB_MAX_TAGS
];
// group by columns information
SColIndex
*
columnInfo
;
// group by columns information
int16_t
orderIndex
;
// order by column index
int16_t
orderType
;
// order by type: asc/desc
}
SSqlGroupbyExpr
;
...
...
@@ -171,7 +171,7 @@ typedef struct SQInfo {
int32_t
pointsInterpo
;
int32_t
code
;
// error code to returned to client
sem_t
dataReady
;
SArray
*
pTable
IdList
;
// table id list
SArray
*
pTable
List
;
// table id list
void
*
tsdb
;
SQueryRuntimeEnv
runtimeEnv
;
...
...
src/query/src/qast.c
浏览文件 @
7e12d2f9
...
...
@@ -869,7 +869,7 @@ void tExprTreeTraverse(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, S
if
(
pQueryInfo
->
colIndex
==
0
&&
pQueryInfo
->
optr
!=
TSDB_RELATION_LIKE
)
{
SQueryCond
cond
=
{
0
};
/*int32_t ret = */
setQueryCond
(
pQueryInfo
,
&
cond
);
/*int32_t ret = */
setQueryCond
(
pQueryInfo
,
&
cond
);
tQueryOnSkipList
(
pSkipList
,
&
cond
,
pQueryInfo
->
q
.
nType
,
result
);
}
else
{
/* Brutal force scan the whole skip list to find the appropriate result,
...
...
src/query/src/qextbuffer.c
浏览文件 @
7e12d2f9
...
...
@@ -43,7 +43,7 @@ tExtMemBuffer* createExtMemBuffer(int32_t inMemSize, int32_t elemSize, SColumnMo
pMemBuffer
->
path
=
strdup
(
name
);
pTrace
(
"create tmp file:%s"
,
pMemBuffer
->
path
);
SFileInfo
*
pFMeta
=
&
pMemBuffer
->
fileMeta
;
S
Ext
FileInfo
*
pFMeta
=
&
pMemBuffer
->
fileMeta
;
pFMeta
->
pageSize
=
DEFAULT_PAGE_SIZE
;
...
...
@@ -63,7 +63,7 @@ void* destoryExtMemBuffer(tExtMemBuffer *pMemBuffer) {
}
// release flush out info link
SFileInfo
*
pFileMeta
=
&
pMemBuffer
->
fileMeta
;
S
Ext
FileInfo
*
pFileMeta
=
&
pMemBuffer
->
fileMeta
;
if
(
pFileMeta
->
flushoutData
.
nAllocSize
!=
0
&&
pFileMeta
->
flushoutData
.
pFlushoutInfo
!=
NULL
)
{
tfree
(
pFileMeta
->
flushoutData
.
pFlushoutInfo
);
}
...
...
@@ -97,7 +97,7 @@ void* destoryExtMemBuffer(tExtMemBuffer *pMemBuffer) {
/*
* alloc more memory for flush out info entries.
*/
static
bool
allocFlushoutInfoEntries
(
SFileInfo
*
pFileMeta
)
{
static
bool
allocFlushoutInfoEntries
(
S
Ext
FileInfo
*
pFileMeta
)
{
pFileMeta
->
flushoutData
.
nAllocSize
=
pFileMeta
->
flushoutData
.
nAllocSize
<<
1
;
tFlushoutInfo
*
tmp
=
(
tFlushoutInfo
*
)
realloc
(
pFileMeta
->
flushoutData
.
pFlushoutInfo
,
...
...
@@ -208,7 +208,7 @@ int16_t tExtMemBufferPut(tExtMemBuffer *pMemBuffer, void *data, int32_t numOfRow
}
static
bool
tExtMemBufferUpdateFlushoutInfo
(
tExtMemBuffer
*
pMemBuffer
)
{
SFileInfo
*
pFileMeta
=
&
pMemBuffer
->
fileMeta
;
S
Ext
FileInfo
*
pFileMeta
=
&
pMemBuffer
->
fileMeta
;
if
(
pMemBuffer
->
flushModel
==
MULTIPLE_APPEND_MODEL
)
{
if
(
pFileMeta
->
flushoutData
.
nLength
==
pFileMeta
->
flushoutData
.
nAllocSize
&&
!
allocFlushoutInfoEntries
(
pFileMeta
))
{
...
...
@@ -238,7 +238,7 @@ static bool tExtMemBufferUpdateFlushoutInfo(tExtMemBuffer *pMemBuffer) {
}
static
void
tExtMemBufferClearFlushoutInfo
(
tExtMemBuffer
*
pMemBuffer
)
{
SFileInfo
*
pFileMeta
=
&
pMemBuffer
->
fileMeta
;
S
Ext
FileInfo
*
pFileMeta
=
&
pMemBuffer
->
fileMeta
;
pFileMeta
->
flushoutData
.
nLength
=
0
;
memset
(
pFileMeta
->
flushoutData
.
pFlushoutInfo
,
0
,
sizeof
(
tFlushoutInfo
)
*
pFileMeta
->
flushoutData
.
nAllocSize
);
...
...
src/query/src/queryExecutor.c
浏览文件 @
7e12d2f9
此差异已折叠。
点击以展开。
src/tsdb/inc/tsdb.h
浏览文件 @
7e12d2f9
...
...
@@ -181,8 +181,10 @@ int32_t tsdbInsertData(tsdb_repo_t *pRepo, SSubmitMsg *pMsg);
typedef
void
*
tsdb_query_handle_t
;
// Use void to hide implementation details
// typedef struct {
// } SColumnFilterInfo;
typedef
struct
STableGroupList
{
// qualified table object list in group
SArray
*
pGroupList
;
int32_t
numOfTables
;
}
STableGroupList
;
// query condition to build vnode iterator
typedef
struct
STsdbQueryCond
{
...
...
@@ -233,7 +235,7 @@ typedef void *tsdbpos_t;
* @param pTableList table sid list
* @return
*/
tsdb_query_handle_t
*
tsdbQuery
ByTableId
(
tsdb_repo_t
*
tsdb
,
STsdbQueryCond
*
pCond
,
SArray
*
idList
,
SArray
*
pColumnInfo
);
tsdb_query_handle_t
*
tsdbQuery
Tables
(
tsdb_repo_t
*
tsdb
,
STsdbQueryCond
*
pCond
,
SArray
*
idList
,
SArray
*
pColumnInfo
);
/**
* move to next block
...
...
@@ -335,7 +337,10 @@ SArray *tsdbGetTableList(tsdb_query_handle_t *pQueryHandle);
* @param pTagCond. tag query condition
*
*/
int32_t
tsdbQueryTags
(
tsdb_repo_t
*
tsdb
,
int64_t
uid
,
const
char
*
pTagCond
,
size_t
len
,
SArray
*
list
);
int32_t
tsdbQueryTags
(
tsdb_repo_t
*
tsdb
,
int64_t
uid
,
const
char
*
pTagCond
,
size_t
len
,
SArray
**
pGroupList
,
SColIndex
*
pColIndex
,
int32_t
numOfCols
);
int32_t
tsdbGetOneTableGroup
(
tsdb_repo_t
*
tsdb
,
int64_t
uid
,
SArray
**
pGroupList
);
/**
* clean up the query handle
...
...
src/tsdb/inc/tsdbMain.h
浏览文件 @
7e12d2f9
...
...
@@ -100,6 +100,7 @@ typedef struct {
STsdbMeta
*
tsdbInitMeta
(
char
*
rootDir
,
int32_t
maxTables
);
int32_t
tsdbFreeMeta
(
STsdbMeta
*
pMeta
);
STSchema
*
tsdbGetTableSchema
(
STsdbMeta
*
pMeta
,
STable
*
pTable
);
STSchema
*
tsdbGetTableTagSchema
(
STsdbMeta
*
pMeta
,
STable
*
pTable
);
// ---- Operation on STable
#define TSDB_TABLE_ID(pTable) ((pTable)->tableId)
...
...
src/tsdb/src/tsdbFile.c
浏览文件 @
7e12d2f9
...
...
@@ -22,8 +22,9 @@
#include <sys/types.h>
#include <unistd.h>
#include "tutil.h"
#include "tsdbMain.h"
#include "tutil.h"
#include "talgo.h"
const
char
*
tsdbFileSuffix
[]
=
{
".head"
,
// TSDB_FILE_TYPE_HEAD
...
...
src/tsdb/src/tsdbMain.c
浏览文件 @
7e12d2f9
...
...
@@ -15,6 +15,7 @@
// #include "taosdef.h"
// #include "disk.h"
#include "os.h"
#include "talgo.h"
#include "tsdb.h"
#include "tsdbMain.h"
...
...
@@ -704,7 +705,7 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
pTable
->
mem
->
keyLast
=
0
;
}
tSkipList
Rand
NodeInfo
(
pTable
->
mem
->
pData
,
&
level
,
&
headSize
);
tSkipList
New
NodeInfo
(
pTable
->
mem
->
pData
,
&
level
,
&
headSize
);
TSKEY
key
=
dataRowKey
(
row
);
// printf("insert:%lld, size:%d\n", key, pTable->mem->numOfPoints);
...
...
src/tsdb/src/tsdbMeta.c
浏览文件 @
7e12d2f9
...
...
@@ -105,8 +105,9 @@ int tsdbRestoreTable(void *pHandle, void *cont, int contLen) {
if
(
pTable
==
NULL
)
return
-
1
;
if
(
pTable
->
type
==
TSDB_SUPER_TABLE
)
{
pTable
->
pIndex
=
tSkipListCreate
(
TSDB_SUPER_TABLE_SL_LEVEL
,
TSDB_DATA_TYPE_TIMESTAMP
,
sizeof
(
int64_t
),
1
,
0
,
0
,
getTagIndexKey
);
STColumn
*
pColSchema
=
schemaColAt
(
pTable
->
tagSchema
,
0
);
pTable
->
pIndex
=
tSkipListCreate
(
TSDB_SUPER_TABLE_SL_LEVEL
,
pColSchema
->
type
,
pColSchema
->
bytes
,
1
,
0
,
0
,
getTagIndexKey
);
}
tsdbAddTableToMeta
(
pMeta
,
pTable
,
false
);
...
...
@@ -201,6 +202,18 @@ STSchema *tsdbGetTableSchema(STsdbMeta *pMeta, STable *pTable) {
}
}
STSchema
*
tsdbGetTableTagSchema
(
STsdbMeta
*
pMeta
,
STable
*
pTable
)
{
if
(
pTable
->
type
==
TSDB_SUPER_TABLE
)
{
return
pTable
->
tagSchema
;
}
else
if
(
pTable
->
type
==
TSDB_CHILD_TABLE
)
{
STable
*
pSuper
=
tsdbGetTableByUid
(
pMeta
,
pTable
->
superUid
);
if
(
pSuper
==
NULL
)
return
NULL
;
return
pSuper
->
tagSchema
;
}
else
{
return
NULL
;
}
}
int32_t
tsdbCreateTableImpl
(
STsdbMeta
*
pMeta
,
STableCfg
*
pCfg
)
{
if
(
tsdbCheckTableCfg
(
pCfg
)
<
0
)
return
-
1
;
...
...
@@ -222,8 +235,11 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
super
->
schema
=
tdDupSchema
(
pCfg
->
schema
);
super
->
tagSchema
=
tdDupSchema
(
pCfg
->
tagSchema
);
super
->
tagVal
=
tdDataRowDup
(
pCfg
->
tagValues
);
super
->
pIndex
=
tSkipListCreate
(
TSDB_SUPER_TABLE_SL_LEVEL
,
TSDB_DATA_TYPE_TIMESTAMP
,
sizeof
(
int64_t
),
1
,
0
,
0
,
getTagIndexKey
);
// Allow duplicate key, no lock
// index the first tag column
STColumn
*
pColSchema
=
schemaColAt
(
super
->
tagSchema
,
0
);
super
->
pIndex
=
tSkipListCreate
(
TSDB_SUPER_TABLE_SL_LEVEL
,
pColSchema
->
type
,
pColSchema
->
bytes
,
1
,
0
,
0
,
getTagIndexKey
);
// Allow duplicate key, no lock
if
(
super
->
pIndex
==
NULL
)
{
tdFreeSchema
(
super
->
schema
);
...
...
@@ -411,11 +427,11 @@ static int tsdbAddTableIntoIndex(STsdbMeta *pMeta, STable *pTable) {
int32_t
level
=
0
;
int32_t
headSize
=
0
;
// first tag column
STColumn
*
s
=
schemaColAt
(
pSTable
->
tagSchema
,
0
);
tSkipListNewNodeInfo
(
pSTable
->
pIndex
,
&
level
,
&
headSize
);
tSkipListRandNodeInfo
(
pSTable
->
pIndex
,
&
level
,
&
headSize
);
SSkipListNode
*
pNode
=
calloc
(
1
,
headSize
+
s
->
bytes
+
POINTER_BYTES
);
// NOTE: do not allocate the space for key, since in each skip list node, only keep the pointer to pTable, not the
// actual key value, and the key value will be retrieved during query through the pTable and getTagIndexKey function
SSkipListNode
*
pNode
=
calloc
(
1
,
headSize
+
POINTER_BYTES
);
pNode
->
level
=
level
;
SSkipList
*
list
=
pSTable
->
pIndex
;
...
...
src/tsdb/src/tsdbRead.c
浏览文件 @
7e12d2f9
...
...
@@ -15,10 +15,12 @@
#include "os.h"
#include "talgo.h"
#include "tlog.h"
#include "tutil.h"
#include "../../../query/inc/qast.h"
#include "../../../query/inc/qextbuffer.h"
#include "../../../query/inc/tlosertree.h"
#include "../../../query/inc/tsqlfunction.h"
#include "tsdb.h"
...
...
@@ -141,9 +143,8 @@ static void tsdbInitCompBlockLoadInfo(SLoadCompBlockInfo* pCompBlockLoadInfo) {
pCompBlockLoadInfo
->
fileListIndex
=
-
1
;
}
tsdb_query_handle_t
*
tsdbQuery
ByTableId
(
tsdb_repo_t
*
tsdb
,
STsdbQueryCond
*
pCond
,
SArray
*
id
List
,
SArray
*
pColumnInfo
)
{
tsdb_query_handle_t
*
tsdbQuery
Tables
(
tsdb_repo_t
*
tsdb
,
STsdbQueryCond
*
pCond
,
SArray
*
group
List
,
SArray
*
pColumnInfo
)
{
// todo 1. filter not exist table
// todo 2. add the reference count for each table that is involved in query
STsdbQueryHandle
*
pQueryHandle
=
calloc
(
1
,
sizeof
(
STsdbQueryHandle
));
...
...
@@ -156,26 +157,26 @@ tsdb_query_handle_t* tsdbQueryByTableId(tsdb_repo_t* tsdb, STsdbQueryCond* pCond
pQueryHandle
->
isFirstSlot
=
true
;
pQueryHandle
->
cur
.
fid
=
-
1
;
size_t
size
=
taosArrayGetSize
(
id
List
);
size_t
size
=
taosArrayGetSize
(
group
List
);
assert
(
size
>=
1
);
pQueryHandle
->
pTableCheckInfo
=
taosArrayInit
(
size
,
sizeof
(
STableCheckInfo
));
for
(
int32_t
i
=
0
;
i
<
size
;
++
i
)
{
STableId
id
=
*
(
STableId
*
)
taosArrayGet
(
idList
,
i
);
STable
*
pTable
=
tsdbGetTableByUid
(
tsdbGetMeta
(
tsdb
),
id
.
uid
);
if
(
pTable
==
NULL
)
{
dError
(
"%p failed to get table, error uid:%"
PRIu64
,
pQueryHandle
,
id
.
uid
);
continue
;
}
STableCheckInfo
info
=
{
.
lastKey
=
pQueryHandle
->
window
.
skey
,
.
tableId
=
id
,
.
pTableObj
=
pTable
,
};
SArray
*
group
=
*
(
SArray
**
)
taosArrayGet
(
groupList
,
i
);
taosArrayPush
(
pQueryHandle
->
pTableCheckInfo
,
&
info
);
size_t
gsize
=
taosArrayGetSize
(
group
);
for
(
int32_t
j
=
0
;
j
<
gsize
;
++
j
)
{
STable
*
pTable
=
*
(
STable
**
)
taosArrayGet
(
group
,
j
);
assert
(
pTable
!=
NULL
);
STableCheckInfo
info
=
{
.
lastKey
=
pQueryHandle
->
window
.
skey
,
.
tableId
=
pTable
->
tableId
,
.
pTableObj
=
pTable
,
};
taosArrayPush
(
pQueryHandle
->
pTableCheckInfo
,
&
info
);
}
}
dTrace
(
"%p total numOfTable:%d in query"
,
pQueryHandle
,
taosArrayGetSize
(
pQueryHandle
->
pTableCheckInfo
));
...
...
@@ -208,7 +209,8 @@ tsdb_query_handle_t* tsdbQueryByTableId(tsdb_repo_t* tsdb, STsdbQueryCond* pCond
}
static
bool
hasMoreDataInCache
(
STsdbQueryHandle
*
pHandle
)
{
assert
(
pHandle
->
activeIndex
==
0
&&
taosArrayGetSize
(
pHandle
->
pTableCheckInfo
)
==
1
);
size_t
size
=
taosArrayGetSize
(
pHandle
->
pTableCheckInfo
);
assert
(
pHandle
->
activeIndex
<
size
&&
pHandle
->
activeIndex
>=
0
&&
size
>=
1
);
pHandle
->
cur
.
fid
=
-
1
;
STableCheckInfo
*
pCheckInfo
=
taosArrayGet
(
pHandle
->
pTableCheckInfo
,
pHandle
->
activeIndex
);
...
...
@@ -312,7 +314,7 @@ static int32_t getFileCompInfo(STsdbQueryHandle* pQueryHandle, int32_t* numOfBlo
SCompIdx
*
compIndex
=
&
pQueryHandle
->
compIndex
[
pCheckInfo
->
tableId
.
tid
];
if
(
compIndex
->
len
==
0
||
compIndex
->
numOfSuperBlocks
==
0
)
{
// no data block in this file, try next file
assert
(
0
);
continue
;
//no data blocks in the file belongs to pCheckInfo->pTable
}
else
{
if
(
pCheckInfo
->
compSize
<
compIndex
->
len
)
{
assert
(
compIndex
->
len
>
0
);
...
...
@@ -488,61 +490,6 @@ static bool loadFileDataBlock(STsdbQueryHandle* pQueryHandle, SCompBlock* pBlock
return
pQueryHandle
->
realNumOfRows
>
0
;
}
//bool moveToNextBlock(STsdbQueryHandle* pQueryHandle, int32_t step) {
// SQueryFilePos* cur = &pQueryHandle->cur;
//
// if (pQueryHandle->cur.fid >= 0) {
// /*
// * 1. ascending order. The last data block of data file
// * 2. descending order. The first block of file
// */
// STableCheckInfo* pCheckInfo = taosArrayGet(pQueryHandle->pTableCheckInfo, pQueryHandle->activeIndex);
// int32_t tid = pCheckInfo->tableId.tid;
//
// if ((step == QUERY_ASC_FORWARD_STEP &&
// (pQueryHandle->cur.slot == pQueryHandle->compIndex[tid].numOfSuperBlocks - 1)) ||
// (step == QUERY_DESC_FORWARD_STEP && (pQueryHandle->cur.slot == 0))) {
// // temporarily keep the position value, in case of no data qualified when move forwards(backwards)
// // SQueryFilePos save = pQueryHandle->cur;
// pQueryHandle->pFileGroup = tsdbGetFileGroupNext(&pQueryHandle->fileIter);
//
// int32_t fid = -1;
// int32_t numOfBlocks = 0;
//
// if (pQueryHandle->pFileGroup != NULL) {
// if ((fid = getFileCompInfo(pQueryHandle, &numOfBlocks, 1)) < 0) {
// } else {
// cur->slot = (step == QUERY_ASC_FORWARD_STEP) ? 0 : pQueryHandle->numOfBlocks - 1;
// cur->pos = (step == QUERY_ASC_FORWARD_STEP) ? 0 : pQueryHandle->pBlock[cur->slot].numOfPoints - 1;
//
// SCompBlock* pBlock = &pCheckInfo->pCompInfo->blocks[cur->slot];
// cur->fid = pQueryHandle->pFileGroup->fileId;
// assert(cur->pos >= 0 && cur->fid >= 0 && cur->slot >= 0);
//
// if (pBlock->keyFirst > pQueryHandle->window.ekey) { // done
// return false;
// }
//
// return loadFileDataBlock(pQueryHandle, pBlock, pCheckInfo);
// }
// } else { // check data in cache
// pQueryHandle->cur.fid = -1;
// return hasMoreDataInCache(pQueryHandle);
// }
// } else { // next block in the same file
// cur->slot += step;
//
// SCompBlock* pBlock = &pCheckInfo->pCompInfo->blocks[cur->slot];
// cur->pos = (step == QUERY_ASC_FORWARD_STEP) ? 0 : pBlock->numOfPoints - 1;
// return loadFileDataBlock(pQueryHandle, pBlock, pCheckInfo);
// }
// } else { // data in cache
// return hasMoreDataInCache(pQueryHandle);
// }
//
// return false;
//}
static
int
vnodeBinarySearchKey
(
char
*
pValue
,
int
num
,
TSKEY
key
,
int
order
)
{
int
firstPos
,
lastPos
,
midPos
=
-
1
;
int
numOfPoints
;
...
...
@@ -732,71 +679,6 @@ int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
return
midPos
;
}
//static bool getQualifiedDataBlock(STsdbQueryHandle* pQueryHandle, STableCheckInfo* pCheckInfo, int32_t type) {
// STsdbFileH* pFileHandle = tsdbGetFile(pQueryHandle->pTsdb);
// int32_t fid = getFileIdFromKey(pCheckInfo->lastKey);
//
// tsdbInitFileGroupIter(pFileHandle, &pQueryHandle->fileIter, TSDB_FGROUP_ITER_FORWARD);
// tsdbSeekFileGroupIter(&pQueryHandle->fileIter, fid);
// pQueryHandle->pFileGroup = tsdbGetFileGroupNext(&pQueryHandle->fileIter);
//
// SQueryFilePos* cur = &pQueryHandle->cur;
//
// int32_t tid = pCheckInfo->tableId.tid;
// int32_t numOfBlocks = 0;
//
// while (pQueryHandle->pFileGroup != NULL) {
// if (getFileCompInfo(pQueryHandle, &numOfBlocks, 1) != TSDB_CODE_SUCCESS) {
// break;
// }
//
// assert(pCheckInfo->numOfBlocks >= 0);
//
// // no data block in current file, try next
// if (pCheckInfo->numOfBlocks > 0) {
// cur->fid = pQueryHandle->pFileGroup->fileId;
// break;
// }
//
// dTrace("%p no data block in file, fid:%d, tid:%d, try next, %p", pQueryHandle, pQueryHandle->pFileGroup->fileId,
// tid, pQueryHandle->qinfo);
//
// pQueryHandle->pFileGroup = tsdbGetFileGroupNext(&pQueryHandle->fileIter);
// }
//
// if (pCheckInfo->numOfBlocks == 0) {
// return false;
// }
//
// cur->slot = 0; // always start from the first slot
// SCompBlock* pBlock = &pCheckInfo->pCompInfo->blocks[cur->slot];
// return loadFileDataBlock(pQueryHandle, pBlock, pCheckInfo);
//}
//static UNUSED_FUNC bool hasMoreDataForSingleTable(STsdbQueryHandle* pHandle) {
// assert(pHandle->activeIndex == 0 && taosArrayGetSize(pHandle->pTableCheckInfo) == 1);
//
// STsdbFileH* pFileHandle = tsdbGetFile(pHandle->pTsdb);
// STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex);
//
// if (!pCheckInfo->checkFirstFileBlock) {
// pCheckInfo->checkFirstFileBlock = true;
//
// if (pFileHandle != NULL) {
// bool found = getQualifiedDataBlock(pHandle, pCheckInfo, 1);
// if (found) {
// return true;
// }
// }
//
// // no data in file, try cache
// pHandle->cur.fid = -1;
// return hasMoreDataInCache(pHandle);
// } else { // move to next data block in file or in cache
// return moveToNextBlock(pHandle, 1);
// }
//}
static
void
cleanBlockOrderSupporter
(
SBlockOrderSupporter
*
pSupporter
,
int32_t
numOfTables
)
{
tfree
(
pSupporter
->
numOfBlocksPerMeter
);
tfree
(
pSupporter
->
blockIndexArray
);
...
...
@@ -862,23 +744,26 @@ static int32_t createDataBlocksInfo(STsdbQueryHandle* pQueryHandle, int32_t numO
}
int32_t
cnt
=
0
;
int32_t
numOfQual
Meter
s
=
0
;
int32_t
numOfQual
Table
s
=
0
;
for
(
int32_t
j
=
0
;
j
<
numOfTables
;
++
j
)
{
STableCheckInfo
*
pTableCheck
=
(
STableCheckInfo
*
)
taosArrayGet
(
pQueryHandle
->
pTableCheckInfo
,
j
);
if
(
pTableCheck
->
numOfBlocks
<=
0
)
{
continue
;
}
SCompBlock
*
pBlock
=
pTableCheck
->
pCompInfo
->
blocks
;
sup
.
numOfBlocksPerMeter
[
numOfQual
Meter
s
]
=
pTableCheck
->
numOfBlocks
;
sup
.
numOfBlocksPerMeter
[
numOfQual
Table
s
]
=
pTableCheck
->
numOfBlocks
;
char
*
buf
=
calloc
(
1
,
sizeof
(
STableBlockInfo
)
*
pTableCheck
->
numOfBlocks
);
if
(
buf
==
NULL
)
{
cleanBlockOrderSupporter
(
&
sup
,
numOfQual
Meter
s
);
cleanBlockOrderSupporter
(
&
sup
,
numOfQual
Table
s
);
return
TSDB_CODE_SERV_OUT_OF_MEMORY
;
}
sup
.
pDataBlockInfo
[
numOfQual
Meter
s
]
=
(
STableBlockInfo
*
)
buf
;
sup
.
pDataBlockInfo
[
numOfQual
Table
s
]
=
(
STableBlockInfo
*
)
buf
;
for
(
int32_t
k
=
0
;
k
<
pTableCheck
->
numOfBlocks
;
++
k
)
{
STableBlockInfo
*
pBlockInfoEx
=
&
sup
.
pDataBlockInfo
[
numOfQual
Meter
s
][
k
];
STableBlockInfo
*
pBlockInfoEx
=
&
sup
.
pDataBlockInfo
[
numOfQual
Table
s
][
k
];
pBlockInfoEx
->
pBlock
.
compBlock
=
&
pBlock
[
k
];
pBlockInfoEx
->
pBlock
.
fields
=
NULL
;
...
...
@@ -889,13 +774,13 @@ static int32_t createDataBlocksInfo(STsdbQueryHandle* pQueryHandle, int32_t numO
cnt
++
;
}
numOfQual
Meter
s
++
;
numOfQual
Table
s
++
;
}
dTrace
(
"%p create data blocks info struct completed
"
,
pQueryHandle
);
dTrace
(
"%p create data blocks info struct completed
, %d blocks in %d tables"
,
pQueryHandle
,
cnt
,
numOfQualTables
);
assert
(
cnt
<=
numOfBlocks
&&
numOfQual
Meter
s
<=
numOfTables
);
// the pMeterDataInfo[j]->numOfBlocks may be 0
sup
.
numOfTables
=
numOfQual
Meter
s
;
assert
(
cnt
<=
numOfBlocks
&&
numOfQual
Table
s
<=
numOfTables
);
// the pMeterDataInfo[j]->numOfBlocks may be 0
sup
.
numOfTables
=
numOfQual
Table
s
;
SLoserTreeInfo
*
pTree
=
NULL
;
uint8_t
ret
=
tLoserTreeCreate
(
&
pTree
,
sup
.
numOfTables
,
&
sup
,
dataBlockOrderCompar
);
...
...
@@ -1256,11 +1141,11 @@ static int32_t getAllTableIdList(STsdbRepo* tsdb, int64_t uid, SArray* list) {
SSkipListIterator
*
iter
=
tSkipListCreateIter
(
pTable
->
pIndex
);
while
(
tSkipListIterNext
(
iter
))
{
SSkipListNode
*
pNode
=
tSkipListIterGet
(
iter
);
STable
*
t
=
*
(
STable
**
)
SL_GET_NODE_DATA
(
pNode
);
taosArrayPush
(
list
,
&
t
->
tableId
);
STable
*
t
=
*
(
STable
**
)
SL_GET_NODE_DATA
(
pNode
);
taosArrayPush
(
list
,
t
);
}
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -1348,6 +1233,132 @@ void filterPrepare(void* expr, void* param) {
tVariantTypeSetType
(
&
pInfo
->
q
,
pInfo
->
sch
.
type
);
}
int32_t
doCompare
(
const
char
*
f1
,
const
char
*
f2
,
int32_t
type
,
size_t
size
)
{
switch
(
type
)
{
case
TSDB_DATA_TYPE_INT
:
DEFAULT_COMP
(
GET_INT32_VAL
(
f1
),
GET_INT32_VAL
(
f2
));
case
TSDB_DATA_TYPE_DOUBLE
:
DEFAULT_COMP
(
GET_DOUBLE_VAL
(
f1
),
GET_DOUBLE_VAL
(
f2
));
case
TSDB_DATA_TYPE_FLOAT
:
DEFAULT_COMP
(
GET_FLOAT_VAL
(
f1
),
GET_FLOAT_VAL
(
f2
));
case
TSDB_DATA_TYPE_BIGINT
:
DEFAULT_COMP
(
GET_INT64_VAL
(
f1
),
GET_INT64_VAL
(
f2
));
case
TSDB_DATA_TYPE_SMALLINT
:
DEFAULT_COMP
(
GET_INT16_VAL
(
f1
),
GET_INT16_VAL
(
f2
));
case
TSDB_DATA_TYPE_TINYINT
:
case
TSDB_DATA_TYPE_BOOL
:
DEFAULT_COMP
(
GET_INT8_VAL
(
f1
),
GET_INT8_VAL
(
f2
));
case
TSDB_DATA_TYPE_NCHAR
:
{
int32_t
ret
=
wcsncmp
((
wchar_t
*
)
f1
,
(
wchar_t
*
)
f2
,
size
/
TSDB_NCHAR_SIZE
);
if
(
ret
==
0
)
{
return
ret
;
}
return
(
ret
<
0
)
?
-
1
:
1
;
}
default:
{
int32_t
ret
=
strncmp
(
f1
,
f2
,
(
size_t
)
size
);
if
(
ret
==
0
)
{
return
ret
;
}
return
(
ret
<
0
)
?
-
1
:
1
;
}
}
}
typedef
struct
STableGroupSupporter
{
int32_t
numOfCols
;
SColIndex
*
pCols
;
STSchema
*
pTagSchema
;
}
STableGroupSupporter
;
int32_t
tableGroupComparFn
(
const
void
*
p1
,
const
void
*
p2
,
const
void
*
param
)
{
STableGroupSupporter
*
pTableGroupSupp
=
(
STableGroupSupporter
*
)
param
;
STable
*
pTable1
=
*
(
STable
**
)
p1
;
STable
*
pTable2
=
*
(
STable
**
)
p2
;
for
(
int32_t
i
=
0
;
i
<
pTableGroupSupp
->
numOfCols
;
++
i
)
{
SColIndex
*
pColIndex
=
&
pTableGroupSupp
->
pCols
[
i
];
int32_t
colIndex
=
pColIndex
->
colIndex
;
char
*
f1
=
NULL
;
char
*
f2
=
NULL
;
int32_t
type
=
0
;
int32_t
bytes
=
0
;
if
(
colIndex
==
-
1
)
{
// table name, todo fix me
// f1 = s1->tags;
// f2 = s2->tags;
type
=
TSDB_DATA_TYPE_BINARY
;
bytes
=
TSDB_TABLE_NAME_LEN
;
}
else
{
f1
=
dataRowTuple
(
pTable1
->
tagVal
);
f2
=
dataRowTuple
(
pTable2
->
tagVal
);
type
=
schemaColAt
(
pTableGroupSupp
->
pTagSchema
,
colIndex
)
->
type
;
bytes
=
schemaColAt
(
pTableGroupSupp
->
pTagSchema
,
colIndex
)
->
bytes
;
}
int32_t
ret
=
doCompare
(
f1
,
f2
,
type
,
bytes
);
if
(
ret
==
0
)
{
continue
;
}
else
{
return
ret
;
}
}
return
0
;
}
void
createTableGroupImpl
(
SArray
*
pGroups
,
STable
**
pTables
,
size_t
numOfTables
,
STableGroupSupporter
*
pSupp
,
__ext_compar_fn_t
compareFn
)
{
SArray
*
g
=
taosArrayInit
(
16
,
POINTER_BYTES
);
taosArrayPush
(
g
,
&
pTables
[
0
]);
for
(
int32_t
i
=
1
;
i
<
numOfTables
;
++
i
)
{
int32_t
ret
=
compareFn
(
&
pTables
[
i
-
1
],
&
pTables
[
i
],
pSupp
);
assert
(
ret
==
0
||
ret
==
-
1
);
if
(
ret
==
0
)
{
taosArrayPush
(
g
,
&
pTables
[
i
]);
}
else
{
taosArrayPush
(
pGroups
,
&
g
);
// current group is ended, start a new group
g
=
taosArrayInit
(
16
,
POINTER_BYTES
);
taosArrayPush
(
g
,
&
pTables
[
i
]);
}
}
}
SArray
*
createTableGroup
(
SArray
*
pTableList
,
STSchema
*
pTagSchema
,
SColIndex
*
pCols
,
int32_t
numOfOrderCols
)
{
assert
(
pTableList
!=
NULL
&&
taosArrayGetSize
(
pTableList
)
>
0
);
SArray
*
pTableGroup
=
taosArrayInit
(
1
,
POINTER_BYTES
);
size_t
size
=
taosArrayGetSize
(
pTableList
);
if
(
size
==
0
)
{
pTrace
(
"no qualified tables"
);
return
pTableGroup
;
}
if
(
numOfOrderCols
==
0
||
size
==
1
)
{
// no group by tags clause or only one table
taosArrayPush
(
pTableGroup
,
pTableList
);
pTrace
(
"all %d tables belong to one group"
,
size
);
#ifdef _DEBUG_VIEW
tSidSetDisplay
(
pTableGroup
);
#endif
}
else
{
STableGroupSupporter
*
pSupp
=
(
STableGroupSupporter
*
)
calloc
(
1
,
sizeof
(
STableGroupSupporter
));
pSupp
->
numOfCols
=
numOfOrderCols
;
pSupp
->
pTagSchema
=
pTagSchema
;
pSupp
->
pCols
=
pCols
;
taosqsort
(
pTableList
->
pData
,
size
,
POINTER_BYTES
,
pSupp
,
tableGroupComparFn
);
createTableGroupImpl
(
pTableGroup
,
pTableList
->
pData
,
size
,
pSupp
,
tableGroupComparFn
);
#ifdef _DEBUG_VIEW
tSidSetDisplay
(
pTableGroup
);
#endif
tfree
(
pSupp
);
}
return
pTableGroup
;
}
bool
tSkipListNodeFilterCallback
(
const
void
*
pNode
,
void
*
param
)
{
tQueryInfo
*
pInfo
=
(
tQueryInfo
*
)
param
;
...
...
@@ -1419,13 +1430,29 @@ static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr)
return
TSDB_CODE_SUCCESS
;
}
int32_t
tsdbQueryTags
(
tsdb_repo_t
*
tsdb
,
int64_t
uid
,
const
char
*
pTagCond
,
size_t
len
,
SArray
*
res
)
{
if
(
pTagCond
==
NULL
||
len
==
0
)
{
// no condition, all tables created according to this stable are involved
return
getAllTableIdList
(
tsdb
,
uid
,
res
);
}
int32_t
tsdbQueryTags
(
tsdb_repo_t
*
tsdb
,
int64_t
uid
,
const
char
*
pTagCond
,
size_t
len
,
SArray
**
pGroupList
,
SColIndex
*
pColIndex
,
int32_t
numOfCols
)
{
STable
*
pSTable
=
tsdbGetTableByUid
(
tsdbGetMeta
(
tsdb
),
uid
);
assert
(
pSTable
!=
NULL
);
if
(
pSTable
==
NULL
)
{
dError
(
"failed to get stable, uid:%"
PRIu64
,
uid
);
return
TSDB_CODE_INVALID_TABLE_ID
;
}
SArray
*
res
=
taosArrayInit
(
8
,
POINTER_BYTES
);
STSchema
*
pTagSchema
=
tsdbGetTableTagSchema
(
tsdbGetMeta
(
tsdb
),
pSTable
);
if
(
pTagCond
==
NULL
||
len
==
0
)
{
// no tags condition, all tables created according to this stable are involved
int32_t
ret
=
getAllTableIdList
(
tsdb
,
uid
,
res
);
if
(
ret
!=
TSDB_CODE_SUCCESS
)
{
taosArrayDestroy
(
res
);
return
ret
;
}
*
pGroupList
=
createTableGroup
(
res
,
pTagSchema
,
pColIndex
,
numOfCols
);
taosArrayDestroy
(
res
);
return
ret
;
}
tExprNode
*
pExprNode
=
NULL
;
int32_t
ret
=
TSDB_CODE_SUCCESS
;
...
...
@@ -1433,12 +1460,33 @@ int32_t tsdbQueryTags(tsdb_repo_t* tsdb, int64_t uid, const char* pTagCond, size
// failed to build expression, no result, return immediately
if
((
ret
=
exprTreeFromBinary
(
pTagCond
,
len
,
&
pExprNode
)
!=
TSDB_CODE_SUCCESS
)
||
(
pExprNode
==
NULL
))
{
dError
(
"stable:%"
PRIu64
", failed to deserialize expression tree, error exists"
,
uid
);
taosArrayDestroy
(
res
);
return
ret
;
}
return
doQueryTableList
(
pSTable
,
res
,
pExprNode
);
doQueryTableList
(
pSTable
,
res
,
pExprNode
);
*
pGroupList
=
createTableGroup
(
res
,
pTagSchema
,
pColIndex
,
numOfCols
);
taosArrayDestroy
(
res
);
return
ret
;
}
int32_t
tsdbGetOneTableGroup
(
tsdb_repo_t
*
tsdb
,
int64_t
uid
,
SArray
**
pGroupList
)
{
STable
*
pTable
=
tsdbGetTableByUid
(
tsdbGetMeta
(
tsdb
),
uid
);
if
(
pTable
==
NULL
)
{
return
TSDB_CODE_INVALID_TABLE_ID
;
}
//todo assert table type, add the table ref count
*
pGroupList
=
taosArrayInit
(
1
,
POINTER_BYTES
);
SArray
*
group
=
taosArrayInit
(
1
,
POINTER_BYTES
);
taosArrayPush
(
group
,
&
pTable
);
taosArrayPush
(
*
pGroupList
,
&
group
);
return
TSDB_CODE_SUCCESS
;
}
void
tsdbCleanupQueryHandle
(
tsdb_query_handle_t
queryHandle
)
{
STsdbQueryHandle
*
pQueryHandle
=
(
STsdbQueryHandle
*
)
queryHandle
;
if
(
pQueryHandle
==
NULL
)
{
...
...
src/util/inc/talgo.h
0 → 100644
浏览文件 @
7e12d2f9
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_TALGO_H
#define TDENGINE_TALGO_H
#ifdef __cplusplus
extern
"C"
{
#endif
#define TD_EQ 0x1
#define TD_GT 0x2
#define TD_LT 0x4
#define TD_GE (TD_EQ | TD_GT)
#define TD_LE (TD_EQ | TD_LT)
typedef
int32_t
(
*
__ext_compar_fn_t
)(
const
void
*
p1
,
const
void
*
p2
,
const
void
*
param
);
/**
* quick sort, with the compare function requiring additional parameters support
*
* @param src
* @param numOfElem
* @param size
* @param param
* @param comparFn
*/
void
taosqsort
(
void
*
src
,
size_t
numOfElem
,
size_t
size
,
const
void
*
param
,
__ext_compar_fn_t
comparFn
);
/**
* binary search, with range support
*
* @param key
* @param base
* @param nmemb
* @param size
* @param fn
* @param flags
* @return
*/
void
*
taosbsearch
(
const
void
*
key
,
const
void
*
base
,
size_t
nmemb
,
size_t
size
,
__compar_fn_t
fn
,
int
flags
);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_TALGO_H
src/util/inc/tskiplist.h
浏览文件 @
7e12d2f9
...
...
@@ -159,7 +159,7 @@ void *tSkipListDestroy(SSkipList *pSkipList);
* @param level
* @param headSize
*/
void
tSkipList
Rand
NodeInfo
(
SSkipList
*
pSkipList
,
int32_t
*
level
,
int32_t
*
headSize
);
void
tSkipList
New
NodeInfo
(
SSkipList
*
pSkipList
,
int32_t
*
level
,
int32_t
*
headSize
);
/**
* put the skip list node into the skip list.
...
...
src/util/inc/tutil.h
浏览文件 @
7e12d2f9
...
...
@@ -170,14 +170,6 @@ uint32_t ip2uint(const char *const ip_addr);
void
taosSetAllocMode
(
int
mode
,
const
char
*
path
,
bool
autoDump
);
void
taosDumpMemoryLeak
();
#define TD_EQ 0x1
#define TD_GT 0x2
#define TD_LT 0x4
#define TD_GE (TD_EQ | TD_GT)
#define TD_LE (TD_EQ | TD_LT)
void
*
taosbsearch
(
const
void
*
key
,
const
void
*
base
,
size_t
nmemb
,
size_t
size
,
int
(
*
compar
)(
const
void
*
,
const
void
*
),
int
flags
);
#ifdef TAOS_MEM_CHECK
void
*
taos_malloc
(
size_t
size
,
const
char
*
file
,
uint32_t
line
);
...
...
src/util/src/hash.c
浏览文件 @
7e12d2f9
...
...
@@ -461,9 +461,7 @@ void taosHashRemove(SHashObj *pHashObj, const char *key, size_t keyLen) {
pNode
->
next
=
NULL
;
pNode
->
prev
=
NULL
;
pTrace
(
"key:%s %p remove from hash table"
,
pNode
->
key
,
pNode
);
tfree
(
pNode
);
__unlock
(
pHashObj
->
lock
);
}
...
...
src/util/src/talgo.c
0 → 100644
浏览文件 @
7e12d2f9
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "tutil.h"
#include "talgo.h"
#define doswap(__left, __right, __size, __buf) do {\
memcpy((__buf), (__left), (__size));\
memcpy((__left), (__right),(__size));\
memcpy((__right), (__buf), (__size));\
} while (0);
#define elePtrAt(base, size, idx) (void *)((char *)(base) + (size) * (idx))
static
void
median
(
void
*
src
,
size_t
size
,
size_t
s
,
size_t
e
,
const
void
*
param
,
__ext_compar_fn_t
comparFn
,
void
*
buf
)
{
int32_t
mid
=
((
e
-
s
)
>>
1u
)
+
s
;
if
(
comparFn
(
elePtrAt
(
src
,
size
,
mid
),
elePtrAt
(
src
,
size
,
s
),
param
)
==
1
)
{
doswap
(
elePtrAt
(
src
,
size
,
mid
),
elePtrAt
(
src
,
size
,
s
),
size
,
buf
);
}
if
(
comparFn
(
elePtrAt
(
src
,
size
,
mid
),
elePtrAt
(
src
,
size
,
e
),
param
)
==
1
)
{
doswap
(
elePtrAt
(
src
,
size
,
mid
),
elePtrAt
(
src
,
size
,
s
),
size
,
buf
);
doswap
(
elePtrAt
(
src
,
size
,
mid
),
elePtrAt
(
src
,
size
,
e
),
size
,
buf
);
}
else
if
(
comparFn
(
elePtrAt
(
src
,
size
,
s
),
elePtrAt
(
src
,
size
,
e
),
param
)
==
1
)
{
doswap
(
elePtrAt
(
src
,
size
,
s
),
elePtrAt
(
src
,
size
,
e
),
size
,
buf
);
}
assert
(
comparFn
(
elePtrAt
(
src
,
size
,
mid
),
elePtrAt
(
src
,
size
,
s
),
param
)
<=
0
&&
comparFn
(
elePtrAt
(
src
,
size
,
s
),
elePtrAt
(
src
,
size
,
e
),
param
)
<=
0
);
#ifdef _DEBUG_VIEW
tTagsPrints
(
src
[
s
],
pOrderDesc
->
pColumnModel
,
&
pOrderDesc
->
orderIdx
);
tTagsPrints
(
src
[
mid
],
pOrderDesc
->
pColumnModel
,
&
pOrderDesc
->
orderIdx
);
tTagsPrints
(
src
[
e
],
pOrderDesc
->
pColumnModel
,
&
pOrderDesc
->
orderIdx
);
#endif
}
static
void
tInsertSort
(
void
*
src
,
size_t
size
,
int32_t
s
,
int32_t
e
,
const
void
*
param
,
__ext_compar_fn_t
comparFn
,
void
*
buf
)
{
for
(
int32_t
i
=
s
+
1
;
i
<=
e
;
++
i
)
{
for
(
int32_t
j
=
i
;
j
>
s
;
--
j
)
{
if
(
comparFn
(
elePtrAt
(
src
,
size
,
j
),
elePtrAt
(
src
,
size
,
j
-
1
),
param
)
==
-
1
)
{
doswap
(
elePtrAt
(
src
,
size
,
j
),
elePtrAt
(
src
,
size
,
j
-
1
),
size
,
buf
);
}
else
{
break
;
}
}
}
}
static
void
tqsortImpl
(
void
*
src
,
int32_t
start
,
int32_t
end
,
size_t
size
,
const
void
*
param
,
__ext_compar_fn_t
comparFn
,
void
*
buf
)
{
// short array sort, incur another sort procedure instead of quick sort process
const
int32_t
THRESHOLD_SIZE
=
6
;
if
(
end
-
start
+
1
<=
THRESHOLD_SIZE
)
{
tInsertSort
(
src
,
size
,
start
,
end
,
param
,
comparFn
,
buf
);
return
;
}
median
(
src
,
size
,
start
,
end
,
param
,
comparFn
,
buf
);
int32_t
s
=
start
,
e
=
end
;
int32_t
endRightS
=
end
,
startLeftS
=
start
;
while
(
s
<
e
)
{
while
(
e
>
s
)
{
int32_t
ret
=
comparFn
(
elePtrAt
(
src
,
size
,
e
),
elePtrAt
(
src
,
size
,
s
),
param
);
if
(
ret
<
0
)
{
break
;
}
//move the data that equals to pivotal value to the right end of the list
if
(
ret
==
0
&&
e
!=
endRightS
)
{
doswap
(
elePtrAt
(
src
,
size
,
e
),
elePtrAt
(
src
,
size
,
endRightS
),
size
,
buf
);
endRightS
--
;
}
e
--
;
}
if
(
e
!=
s
)
{
doswap
(
elePtrAt
(
src
,
size
,
e
),
elePtrAt
(
src
,
size
,
s
),
size
,
buf
);
}
while
(
s
<
e
)
{
int32_t
ret
=
comparFn
(
elePtrAt
(
src
,
size
,
s
),
elePtrAt
(
src
,
size
,
e
),
param
);
if
(
ret
>
0
)
{
break
;
}
if
(
ret
==
0
&&
s
!=
startLeftS
)
{
doswap
(
elePtrAt
(
src
,
size
,
s
),
elePtrAt
(
src
,
size
,
startLeftS
),
size
,
buf
);
startLeftS
++
;
}
s
++
;
}
if
(
e
!=
s
)
{
doswap
(
elePtrAt
(
src
,
size
,
s
),
elePtrAt
(
src
,
size
,
e
),
size
,
buf
);
}
}
int32_t
rightPartStart
=
e
+
1
;
if
(
endRightS
!=
end
&&
e
<
end
)
{
int32_t
left
=
rightPartStart
;
int32_t
right
=
end
;
while
(
right
>
endRightS
&&
left
<=
endRightS
)
{
doswap
(
elePtrAt
(
src
,
size
,
left
),
elePtrAt
(
src
,
size
,
right
),
size
,
buf
);
left
++
;
right
--
;
}
rightPartStart
+=
(
end
-
endRightS
);
}
int32_t
leftPartEnd
=
e
-
1
;
if
(
startLeftS
!=
end
&&
s
>
start
)
{
int32_t
left
=
start
;
int32_t
right
=
leftPartEnd
;
while
(
left
<
startLeftS
&&
right
>=
startLeftS
)
{
doswap
(
elePtrAt
(
src
,
size
,
left
),
elePtrAt
(
src
,
size
,
right
),
size
,
buf
);
left
++
;
right
--
;
}
leftPartEnd
-=
(
startLeftS
-
start
);
}
if
(
leftPartEnd
>
start
)
{
tqsortImpl
(
src
,
size
,
start
,
leftPartEnd
,
param
,
comparFn
,
buf
);
}
if
(
rightPartStart
<
end
)
{
tqsortImpl
(
src
,
size
,
rightPartStart
,
end
,
param
,
comparFn
,
buf
);
}
}
void
taosqsort
(
void
*
src
,
size_t
numOfElem
,
size_t
size
,
const
void
*
param
,
__ext_compar_fn_t
comparFn
)
{
char
*
buf
=
calloc
(
1
,
size
);
// prepare the swap buffer
tqsortImpl
(
src
,
0
,
numOfElem
-
1
,
size
,
param
,
comparFn
,
buf
);
tfree
(
buf
);
}
void
*
taosbsearch
(
const
void
*
key
,
const
void
*
base
,
size_t
nmemb
,
size_t
size
,
__compar_fn_t
compar
,
int
flags
)
{
// TODO: need to check the correctness of this function
int
l
=
0
;
int
r
=
nmemb
;
int
idx
=
0
;
int
comparison
;
if
(
flags
==
TD_EQ
)
{
return
bsearch
(
key
,
base
,
nmemb
,
size
,
compar
);
}
else
if
(
flags
==
TD_GE
)
{
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
0
))
<=
0
)
return
elePtrAt
(
base
,
size
,
0
);
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
nmemb
-
1
))
>
0
)
return
NULL
;
while
(
l
<
r
)
{
idx
=
(
l
+
r
)
/
2
;
comparison
=
(
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
idx
));
if
(
comparison
<
0
)
{
r
=
idx
;
}
else
if
(
comparison
>
0
)
{
l
=
idx
+
1
;
}
else
{
return
elePtrAt
(
base
,
size
,
idx
);
}
}
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
idx
))
<
0
)
{
return
elePtrAt
(
base
,
size
,
idx
);
}
else
{
if
(
idx
+
1
>
nmemb
-
1
)
{
return
NULL
;
}
else
{
return
elePtrAt
(
base
,
size
,
idx
+
1
);
}
}
}
else
if
(
flags
==
TD_LE
)
{
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
nmemb
-
1
))
>=
0
)
return
elePtrAt
(
base
,
size
,
nmemb
-
1
);
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
0
))
<
0
)
return
NULL
;
while
(
l
<
r
)
{
idx
=
(
l
+
r
)
/
2
;
comparison
=
(
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
idx
));
if
(
comparison
<
0
)
{
r
=
idx
;
}
else
if
(
comparison
>
0
)
{
l
=
idx
+
1
;
}
else
{
return
elePtrAt
(
base
,
size
,
idx
);
}
}
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
idx
))
>
0
)
{
return
elePtrAt
(
base
,
size
,
idx
);
}
else
{
if
(
idx
==
0
)
{
return
NULL
;
}
else
{
return
elePtrAt
(
base
,
size
,
idx
-
1
);
}
}
}
else
{
assert
(
0
);
return
NULL
;
}
return
NULL
;
}
src/util/src/tskiplist.c
浏览文件 @
7e12d2f9
...
...
@@ -190,7 +190,7 @@ void *tSkipListDestroy(SSkipList *pSkipList) {
return
NULL
;
}
void
tSkipList
Rand
NodeInfo
(
SSkipList
*
pSkipList
,
int32_t
*
level
,
int32_t
*
headSize
)
{
void
tSkipList
New
NodeInfo
(
SSkipList
*
pSkipList
,
int32_t
*
level
,
int32_t
*
headSize
)
{
if
(
pSkipList
==
NULL
)
{
return
;
}
...
...
src/util/src/tutil.c
浏览文件 @
7e12d2f9
...
...
@@ -618,72 +618,3 @@ char *taosCharsetReplace(char *charsetstr) {
return
strdup
(
charsetstr
);
}
#define elePtrAt(base, size, idx) (void *)((char *)(base) + (size) * (idx))
void
*
taosbsearch
(
const
void
*
key
,
const
void
*
base
,
size_t
nmemb
,
size_t
size
,
int
(
*
compar
)(
const
void
*
,
const
void
*
),
int
flags
)
{
// TODO: need to check the correctness of this function
int
l
=
0
;
int
r
=
nmemb
;
int
idx
=
0
;
int
comparison
;
if
(
flags
==
TD_EQ
)
{
return
bsearch
(
key
,
base
,
nmemb
,
size
,
compar
);
}
else
if
(
flags
==
TD_GE
)
{
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
0
))
<=
0
)
return
elePtrAt
(
base
,
size
,
0
);
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
nmemb
-
1
))
>
0
)
return
NULL
;
while
(
l
<
r
)
{
idx
=
(
l
+
r
)
/
2
;
comparison
=
(
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
idx
));
if
(
comparison
<
0
)
{
r
=
idx
;
}
else
if
(
comparison
>
0
)
{
l
=
idx
+
1
;
}
else
{
return
elePtrAt
(
base
,
size
,
idx
);
}
}
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
idx
))
<
0
)
{
return
elePtrAt
(
base
,
size
,
idx
);
}
else
{
if
(
idx
+
1
>
nmemb
-
1
)
{
return
NULL
;
}
else
{
return
elePtrAt
(
base
,
size
,
idx
+
1
);
}
}
}
else
if
(
flags
==
TD_LE
)
{
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
nmemb
-
1
))
>=
0
)
return
elePtrAt
(
base
,
size
,
nmemb
-
1
);
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
0
))
<
0
)
return
NULL
;
while
(
l
<
r
)
{
idx
=
(
l
+
r
)
/
2
;
comparison
=
(
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
idx
));
if
(
comparison
<
0
)
{
r
=
idx
;
}
else
if
(
comparison
>
0
)
{
l
=
idx
+
1
;
}
else
{
return
elePtrAt
(
base
,
size
,
idx
);
}
}
if
((
*
compar
)(
key
,
elePtrAt
(
base
,
size
,
idx
))
>
0
)
{
return
elePtrAt
(
base
,
size
,
idx
);
}
else
{
if
(
idx
==
0
)
{
return
NULL
;
}
else
{
return
elePtrAt
(
base
,
size
,
idx
-
1
);
}
}
}
else
{
assert
(
0
);
return
NULL
;
}
return
NULL
;
}
src/util/tests/skiplistTest.cpp
浏览文件 @
7e12d2f9
...
...
@@ -28,7 +28,7 @@ void doubleSkipListTest() {
int32_t
level
=
0
;
int32_t
size
=
0
;
tSkipList
Rand
NodeInfo
(
pSkipList
,
&
level
,
&
size
);
tSkipList
New
NodeInfo
(
pSkipList
,
&
level
,
&
size
);
auto
d
=
(
SSkipListNode
*
)
calloc
(
1
,
size
+
sizeof
(
double
)
*
2
);
d
->
level
=
level
;
...
...
@@ -81,7 +81,7 @@ void randKeyTest() {
int32_t
level
=
0
;
int32_t
s
=
0
;
tSkipList
Rand
NodeInfo
(
pSkipList
,
&
level
,
&
s
);
tSkipList
New
NodeInfo
(
pSkipList
,
&
level
,
&
s
);
auto
d
=
(
SSkipListNode
*
)
calloc
(
1
,
s
+
sizeof
(
int32_t
)
*
2
);
d
->
level
=
level
;
...
...
@@ -112,7 +112,7 @@ void stringKeySkiplistTest() {
int32_t
level
=
0
;
int32_t
headsize
=
0
;
tSkipList
Rand
NodeInfo
(
pSkipList
,
&
level
,
&
headsize
);
tSkipList
New
NodeInfo
(
pSkipList
,
&
level
,
&
headsize
);
auto
pNode
=
(
SSkipListNode
*
)
calloc
(
1
,
headsize
+
max_key_size
+
sizeof
(
double
));
pNode
->
level
=
level
;
...
...
@@ -124,7 +124,7 @@ void stringKeySkiplistTest() {
tSkipListPut
(
pSkipList
,
pNode
);
tSkipList
Rand
NodeInfo
(
pSkipList
,
&
level
,
&
headsize
);
tSkipList
New
NodeInfo
(
pSkipList
,
&
level
,
&
headsize
);
pNode
=
(
SSkipListNode
*
)
calloc
(
1
,
headsize
+
max_key_size
+
sizeof
(
double
));
pNode
->
level
=
level
;
...
...
@@ -164,7 +164,7 @@ void stringKeySkiplistTest() {
int32_t
total
=
10000
;
for
(
int32_t
i
=
0
;
i
<
total
;
++
i
)
{
int32_t
n
=
sprintf
(
k
,
"abc_%d_%d"
,
i
,
i
);
tSkipList
Rand
NodeInfo
(
pSkipList
,
&
level
,
&
headsize
);
tSkipList
New
NodeInfo
(
pSkipList
,
&
level
,
&
headsize
);
auto
pNode
=
(
SSkipListNode
*
)
calloc
(
1
,
headsize
+
20
+
sizeof
(
double
));
pNode
->
level
=
level
;
...
...
@@ -222,7 +222,7 @@ void skiplistPerformanceTest() {
char
*
p
=
total
;
for
(
int32_t
i
=
0
;
i
<
size
;
++
i
)
{
tSkipList
Rand
NodeInfo
(
pSkipList
,
&
level
,
&
headsize
);
tSkipList
New
NodeInfo
(
pSkipList
,
&
level
,
&
headsize
);
SSkipListNode
*
d
=
(
SSkipListNode
*
)
p
;
p
+=
headsize
+
sizeof
(
double
)
*
2
;
...
...
tests/examples/c/demo.c
浏览文件 @
7e12d2f9
...
...
@@ -75,17 +75,17 @@ int main(int argc, char *argv[]) {
doQuery
(
taos
,
"create database if not exists test"
);
doQuery
(
taos
,
"use test"
);
doQuery
(
taos
,
"create table if not exists tm0 (ts timestamp, k int);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:1', 1);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:2', 2);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:3', 3);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:4', 4);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:5', 5);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:6', 6);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:7', 7);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:8', 8);"
);
doQuery
(
taos
,
"insert into tm0 values('2020-1-1 1:1:9', 9);"
);
doQuery
(
taos
,
"select
* from tm0;
"
);
//
doQuery(taos, "create table if not exists tm0 (ts timestamp, k int);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:1', 1);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:2', 2);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:3', 3);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:4', 4);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:5', 5);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:6', 6);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:7', 7);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:8', 8);");
//
doQuery(taos, "insert into tm0 values('2020-1-1 1:1:9', 9);");
doQuery
(
taos
,
"select
sum(k),count(*) from m1 group by a
"
);
taos_close
(
taos
);
return
0
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录