Skip to content

TDengine
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
TDengine
TDengine
提交 d2886dee 编写于 作者: D dapan1121
浏览文件
操作

Merge branch 'develop' into hotfix/TD-4449

上级 3c01a281 f572be8f
显示空白变更内容
内联 并排
Showing with 733 addition and 760 deletion
documentation20/cn/03.architecture/02.replica/docs.md
浏览文件 @ d2886dee
......@@ -107,9 +107,9 @@ TDengine采取的是Master-Slave模式进行同步，与流行的RAFT一致性
![replica-forward.png](page://images/architecture/replica-forward.png)
1. 应用对写请求做基本的合法性检查，通过，则给改请求包打上一个版本号(version, 单调递增）
1. 应用对写请求做基本的合法性检查，通过，则给该请求包打上一个版本号(version, 单调递增）
2. 应用将打上版本号的写请求封装一个WAL Head, 写入WAL(Write Ahead Log)
3. 应用调用API syncForwardToPeer，如多vnode B是slave状态，sync模块将包含WAL Head的数据包通过Forward消息发送给vnode B，否则就不转发。
3. 应用调用API syncForwardToPeer，如果vnode B是slave状态，sync模块将包含WAL Head的数据包通过Forward消息发送给vnode B，否则就不转发。
4. vnode B收到Forward消息后，调用回调函数writeToCache, 交给应用处理
5. vnode B应用在写入成功后，都需要调用syncAckForward通知sync模块已经写入成功。
6. 如果quorum大于1，vnode B需要等待应用的回复确认，收到确认后，vnode B发送Forward Response消息给node A。
......@@ -219,7 +219,7 @@ Arbitrator的程序tarbitrator.c在复制模块的同一目录, 编译整个系
不同之处：
- 选举流程不一样：Raft里任何一个节点是candidate时，主动向其他节点发出vote request, 如果超过半数回答Yes, 这个candidate就成为Leader,开始一个新的term. 而TDengine的实现里，节点上线、离线或角色改变都会触发状态消息在节点组类传播，等节点组里状态稳定一致之后才触发选举流程，因为状态稳定一致，基于同样的状态信息，每个节点做出的决定会是一致的，一旦某个节点符合成为master的条件，无需其他节点认可，它会自动将自己设为master。TDengine里，任何一个节点检测到其他节点或自己的角色发生改变，就会给节点组内其他节点进行广播的。Raft里不存在这样的机制，因此需要投票来解决。
- 选举流程不一样：Raft里任何一个节点是candidate时，主动向其他节点发出vote request，如果超过半数回答Yes，这个candidate就成为Leader，开始一个新的term。而TDengine的实现里，节点上线、离线或角色改变都会触发状态消息在节点组内传播，等节点组里状态稳定一致之后才触发选举流程，因为状态稳定一致，基于同样的状态信息，每个节点做出的决定会是一致的，一旦某个节点符合成为master的条件，无需其他节点认可，它会自动将自己设为master。TDengine里，任何一个节点检测到其他节点或自己的角色发生改变，就会向节点组内其他节点进行广播。Raft里不存在这样的机制，因此需要投票来解决。
- 对WAL的一条记录，Raft用term + index来做唯一标识。但TDengine只用version（类似index)，在TDengine实现里，仅仅用version是完全可行的, 因为TDengine的选举机制，没有term的概念。
如果整个虚拟节点组全部宕机，重启，但不是所有虚拟节点都上线，这个时候TDengine是不会选出master的，因为未上线的节点有可能有最高version的数据。而RAFT协议，只要超过半数上线，就会选出Leader。
......
documentation20/cn/03.architecture/docs.md
浏览文件 @ d2886dee
......@@ -343,7 +343,7 @@ TDengine采用数据驱动的方式让缓存中的数据写入硬盘进行持久
对于采集的数据，一般有保留时长，这个时长由系统配置参数keep决定。超过这个设置天数的数据文件，将被系统自动删除，释放存储空间。
给定days与keep两个参数，一个vnode总的数据文件数为：keep/days。总的数据文件个数不宜过大，也不宜过小。10到100以内合适。基于这个原则，可以设置合理的days。 目前的版本，参数keep可以修改，但对于参数days，一但设置后，不可修改。
给定days与keep两个参数，一个典型工作状态的vnode中总的数据文件数为：`向上取整(keep/days)+1`个。总的数据文件个数不宜过大，也不宜过小。10到100以内合适。基于这个原则，可以设置合理的days。 目前的版本，参数keep可以修改，但对于参数days，一但设置后，不可修改。
在每个数据文件里，一张表的数据是一块一块存储的。一张表可以有一到多个数据文件块。在一个文件块里，数据是列式存储的，占用的是一片连续的存储空间，这样大大提高读取速度。文件块的大小由系统参数maxRows（每块最大记录条数）决定，缺省值为4096。这个值不宜过大，也不宜过小。过大，定位具体时间段的数据的搜索时间会变长，影响读取速度；过小，数据块的索引太大，压缩效率偏低，也影响读取速度。
......
documentation20/cn/08.connector/docs.md
浏览文件 @ d2886dee
......@@ -516,7 +516,7 @@ conn.close()
- _TDengineCursor_ 类
参考python中help(taos.TDengineCursor)。
这个类对应客户端进行的写入、查询操作。在客户端多线程的场景下，这个游标实例必须保持线程独享，不能夸线程共享使用，否则会导致返回结果出现错误。
这个类对应客户端进行的写入、查询操作。在客户端多线程的场景下，这个游标实例必须保持线程独享，不能跨线程共享使用，否则会导致返回结果出现错误。
- _connect_ 方法
......
documentation20/cn/12.taos-sql/docs.md
浏览文件 @ d2886dee
......@@ -37,7 +37,7 @@ taos> DESCRIBE meters;
- Epoch Time：时间戳也可以是一个长整数，表示从 1970-01-01 08:00:00.000 开始的毫秒数
- 时间可以加减，比如 now-2h，表明查询时刻向前推 2 个小时（最近 2 小时）。数字后面的时间单位可以是 u(微秒)、a(毫秒)、s(秒)、m(分)、h(小时)、d(天)、w(周)。 比如 `select * from t1 where ts > now-2w and ts <= now-1w`，表示查询两周前整整一周的数据。在指定降频操作（down sampling）的时间窗口（interval）时，时间单位还可以使用 n(自然月) 和 y(自然年)。
TDengine 缺省的时间戳是毫秒精度，但通过修改配置参数 enableMicrosecond 就可以支持微秒。
TDengine 缺省的时间戳是毫秒精度，但通过在 CREATE DATABASE 时传递的 PRECISION 参数就可以支持微秒。
在TDengine中，普通表的数据模型中可使用以下 10 种数据类型。
......@@ -400,6 +400,7 @@ TDengine 缺省的时间戳是毫秒精度，但通过修改配置参数 enableM
tb2_name (tb2_field1_name, ...) [USING stb2_name TAGS (tag_value2, ...)] VALUES (field1_value1, ...) (field1_value2, ...) ...;
```
以自动建表的方式，同时向表tb1_name和tb2_name中按列分别插入多条记录。
说明：`(tb1_field1_name, ...)`的部分可以省略掉，这样就是使用全列模式写入——也即在 VALUES 部分提供的数据，必须为数据表的每个列都显式地提供数据。全列写入速度会远快于指定列，因此建议尽可能采用全列写入方式，此时空列可以填入NULL。
从 2.0.20.5 版本开始，子表的列名可以不跟在子表名称后面，而是可以放在 TAGS 和 VALUES 之间，例如像下面这样写：
```mysql
INSERT INTO tb1_name [USING stb1_name TAGS (tag_value1, ...)] (tb1_field1_name, ...) VALUES (field1_value1, ...) (field1_value2, ...) ...;
......@@ -423,9 +424,9 @@ Query OK, 1 row(s) in set (0.001029s)
taos> SHOW TABLES;
Query OK, 0 row(s) in set (0.000946s)
taos> INSERT INTO d1001 USING meters TAGS('Beijing.Chaoyang', 2);
taos> INSERT INTO d1001 USING meters TAGS('Beijing.Chaoyang', 2) VALUES('a');
DB error: invalid SQL: keyword VALUES or FILE required
DB error: invalid SQL: 'a' (invalid timestamp) (0.039494s)
taos> SHOW TABLES;
table_name | created_time | columns | stable_name |
......
src/client/inc/tscLocalMerge.h → src/client/inc/tscGlobalmerge.h
浏览文件 @ d2886dee
......@@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_TSCLOCALMERGE_H
#define TDENGINE_TSCLOCALMERGE_H
#ifndef TDENGINE_TSCGLOBALMERGE_H
#define TDENGINE_TSCGLOBALMERGE_H
#ifdef __cplusplus
extern "C" {
......@@ -24,7 +24,7 @@ extern "C" {
#include "qFill.h"
#include "taosmsg.h"
#include "tlosertree.h"
#include "tsclient.h"
#include "qExecutor.h"
#define MAX_NUM_OF_SUBQUERY_RETRY 3
......@@ -38,7 +38,7 @@ typedef struct SLocalDataSource {
tFilePage filePage;
} SLocalDataSource;
typedef struct SLocalMerger {
typedef struct SGlobalMerger {
SLocalDataSource **pLocalDataSrc;
int32_t numOfBuffer;
int32_t numOfCompleted;
......@@ -48,20 +48,22 @@ typedef struct SLocalMerger {
tOrderDescriptor *pDesc;
tExtMemBuffer **pExtMemBuffer; // disk-based buffer
char *buf; // temp buffer
} SLocalMerger;
} SGlobalMerger;
struct SSqlObj;
typedef struct SRetrieveSupport {
tExtMemBuffer ** pExtMemBuffer; // for build loser tree
tOrderDescriptor *pOrderDescriptor;
int32_t subqueryIndex; // index of current vnode in vnode list
SSqlObj * pParentSql;
struct SSqlObj *pParentSql;
tFilePage * localBuffer; // temp buffer, there is a buffer for each vnode to
uint32_t numOfRetry; // record the number of retry times
} SRetrieveSupport;
int32_t tscLocalReducerEnvCreate(SQueryInfo* pQueryInfo, tExtMemBuffer ***pMemBuffer, int32_t numOfSub, tOrderDescriptor **pDesc, uint32_t nBufferSize, int64_t id);
int32_t tscCreateGlobalMergerEnv(SQueryInfo* pQueryInfo, tExtMemBuffer ***pMemBuffer, int32_t numOfSub, tOrderDescriptor **pDesc, uint32_t nBufferSize, int64_t id);
void tscLocalReducerEnvDestroy(tExtMemBuffer **pMemBuffer, tOrderDescriptor *pDesc, int32_t numOfVnodes);
void tscDestroyGlobalMergerEnv(tExtMemBuffer **pMemBuffer, tOrderDescriptor *pDesc, int32_t numOfVnodes);
int32_t saveToBuffer(tExtMemBuffer *pMemoryBuf, tOrderDescriptor *pDesc, tFilePage *pPage, void *data,
int32_t numOfRows, int32_t orderType);
......@@ -71,13 +73,13 @@ int32_t tscFlushTmpBuffer(tExtMemBuffer *pMemoryBuf, tOrderDescriptor *pDesc, tF
/*
* create local reducer to launch the second-stage reduce process at client site
*/
int32_t tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrderDescriptor *pDesc,
SQueryInfo *pQueryInfo, SLocalMerger **pMerger, int64_t id);
int32_t tscCreateGlobalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrderDescriptor *pDesc,
SQueryInfo *pQueryInfo, SGlobalMerger **pMerger, int64_t id);
void tscDestroyLocalMerger(SLocalMerger* pLocalMerger);
void tscDestroyGlobalMerger(SGlobalMerger* pMerger);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_TSCLOCALMERGE_H
#endif // TDENGINE_TSCGLOBALMERGE_H
src/client/inc/tscUtil.h
浏览文件 @ d2886dee
......@@ -20,13 +20,13 @@
extern "C" {
#endif
#include "tsched.h"
#include "exception.h"
#include "os.h"
#include "qExtbuffer.h"
#include "taosdef.h"
#include "tbuffer.h"
#include "tscLocalMerge.h"
#include "tscGlobalmerge.h"
#include "tsched.h"
#include "tsclient.h"
#define UTIL_TABLE_IS_SUPER_TABLE(metaInfo) \
......@@ -110,7 +110,7 @@ void* tscDestroyBlockArrayList(SArray* pDataBlockList);
void* tscDestroyBlockHashTable(SHashObj* pBlockHashTable, bool removeMeta);
int32_t tscCopyDataBlockToPayload(SSqlObj* pSql, STableDataBlocks* pDataBlock);
int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap);
int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBlockMap);
int32_t tscGetDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, int32_t startOffset, int32_t rowSize, SName* pName, STableMeta* pTableMeta,
STableDataBlocks** dataBlocks, SArray* pBlockList);
......@@ -284,6 +284,7 @@ void tscSVgroupInfoCopy(SVgroupInfo* dst, const SVgroupInfo* src);
SSqlObj* createSimpleSubObj(SSqlObj* pSql, __async_cb_func_t fp, void* param, int32_t cmd);
void registerSqlObj(SSqlObj* pSql);
void tscInitResForMerge(SSqlRes* pRes);
SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, __async_cb_func_t fp, void* param, int32_t cmd, SSqlObj* pPrevSql);
void addGroupInfoForSubquery(SSqlObj* pParentObj, SSqlObj* pSql, int32_t subClauseIndex, int32_t tableIndex);
......@@ -328,12 +329,15 @@ SVgroupsInfo* tscVgroupsInfoDup(SVgroupsInfo* pVgroupsInfo);
int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAttr, void* addr);
void tsCreateSQLFunctionCtx(SQueryInfo* pQueryInfo, SQLFunctionCtx* pCtx, SSchema* pSchema);
void* createQInfoFromQueryNode(SQueryInfo* pQueryInfo, SExprInfo* pExprs, STableGroupInfo* pTableGroupInfo, SOperatorInfo* pOperator, char* sql, void* addr, int32_t stage);
void* createQInfoFromQueryNode(SQueryInfo* pQueryInfo, STableGroupInfo* pTableGroupInfo, SOperatorInfo* pOperator, char* sql, void* addr, int32_t stage);
void* malloc_throw(size_t size);
void* calloc_throw(size_t nmemb, size_t size);
char* strdup_throw(const char* str);
bool vgroupInfoIdentical(SNewVgroupInfo *pExisted, SVgroupMsg* src);
SNewVgroupInfo createNewVgroupInfo(SVgroupMsg *pVgroupMsg);
#ifdef __cplusplus
}
#endif
......
src/client/inc/tschemautil.h 已删除 100644 → 0
浏览文件 @ 3c01a281
/*
* 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_TSCHEMAUTIL_H
#define TDENGINE_TSCHEMAUTIL_H
#ifdef __cplusplus
extern "C" {
#endif
#include "taosmsg.h"
#include "tsclient.h"
#include "ttoken.h"
/**
* get the number of tags of this table
* @param pTableMeta
* @return
*/
int32_t tscGetNumOfTags(const STableMeta* pTableMeta);
/**
* get the number of columns of this table
* @param pTableMeta
* @return
*/
int32_t tscGetNumOfColumns(const STableMeta* pTableMeta);
/**
* get the basic info of this table
* @param pTableMeta
* @return
*/
STableComInfo tscGetTableInfo(const STableMeta* pTableMeta);
/**
* get the schema
* @param pTableMeta
* @return
*/
SSchema* tscGetTableSchema(const STableMeta* pTableMeta);
/**
* get the tag schema
* @param pMeta
* @return
*/
SSchema *tscGetTableTagSchema(const STableMeta *pMeta);
/**
* get the column schema according to the column index
* @param pMeta
* @param colIndex
* @return
*/
SSchema *tscGetTableColumnSchema(const STableMeta *pMeta, int32_t colIndex);
/**
* get the column schema according to the column id
* @param pTableMeta
* @param colId
* @return
*/
SSchema* tscGetColumnSchemaById(STableMeta* pTableMeta, int16_t colId);
/**
* create the table meta from the msg
* @param pTableMetaMsg
* @param size size of the table meta
* @return
*/
STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg);
bool vgroupInfoIdentical(SNewVgroupInfo *pExisted, SVgroupMsg* src);
SNewVgroupInfo createNewVgroupInfo(SVgroupMsg *pVgroupMsg);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_TSCHEMAUTIL_H
src/client/inc/tsclient.h
浏览文件 @ d2886dee
......@@ -40,17 +40,9 @@ extern "C" {
// forward declaration
struct SSqlInfo;
struct SLocalMerger;
typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int32_t numOfRows);
typedef struct STableComInfo {
uint8_t numOfTags;
uint8_t precision;
int16_t numOfColumns;
int32_t rowSize;
} STableComInfo;
typedef struct SNewVgroupInfo {
int32_t vgId;
int8_t inUse;
......@@ -66,34 +58,6 @@ typedef struct CChildTableMeta {
uint64_t suid; // super table id
} CChildTableMeta;
typedef struct STableMeta {
int32_t vgId;
STableId id;
uint8_t tableType;
char sTableName[TSDB_TABLE_FNAME_LEN]; // super table name
uint64_t suid; // super table id
int16_t sversion;
int16_t tversion;
STableComInfo tableInfo;
SSchema schema[]; // if the table is TSDB_CHILD_TABLE, schema is acquired by super table meta info
} STableMeta;
typedef struct STableMetaInfo {
STableMeta *pTableMeta; // table meta, cached in client side and acquired by name
uint32_t tableMetaSize;
SVgroupsInfo *vgroupList;
SArray *pVgroupTables; // SArray<SVgroupTableInfo>
/*
* 1. keep the vgroup index during the multi-vnode super table projection query
* 2. keep the vgroup index for multi-vnode insertion
*/
int32_t vgroupIndex;
SName name;
char aliasName[TSDB_TABLE_NAME_LEN]; // alias name of table specified in query sql
SArray *tagColList; // SArray<SColumn*>, involved tag columns
} STableMetaInfo;
typedef struct SColumnIndex {
int16_t tableIndex;
int16_t columnIndex;
......@@ -111,44 +75,6 @@ typedef struct SInternalField {
SExprInfo *pExpr;
} SInternalField;
typedef struct SFieldInfo {
int16_t numOfOutput; // number of column in result
TAOS_FIELD* final;
SArray *internalField; // SArray<SInternalField>
} SFieldInfo;
typedef struct SCond {
uint64_t uid;
int32_t len; // length of tag query condition data
char * cond;
} SCond;
typedef struct SJoinNode {
uint64_t uid;
int16_t tagColId;
SArray* tsJoin;
SArray* tagJoin;
} SJoinNode;
typedef struct SJoinInfo {
bool hasJoin;
SJoinNode *joinTables[TSDB_MAX_JOIN_TABLE_NUM];
} SJoinInfo;
typedef struct STagCond {
// relation between tbname list and query condition, including : TK_AND or TK_OR
int16_t relType;
// tbname query condition, only support tbname query condition on one table
SCond tbnameCond;
// join condition, only support two tables join currently
SJoinInfo joinInfo;
// for different table, the query condition must be seperated
SArray *pCond;
} STagCond;
typedef struct SParamInfo {
int32_t idx;
uint8_t type;
......@@ -191,57 +117,6 @@ typedef struct STableDataBlocks {
SParamInfo *params;
} STableDataBlocks;
typedef struct SQueryInfo {
int16_t command; // the command may be different for each subclause, so keep it seperately.
uint32_t type; // query/insert type
STimeWindow window; // the whole query time window
SInterval interval; // tumble time window
SSessionWindow sessionWindow; // session time window
SGroupbyExpr groupbyExpr; // groupby tags info
SArray * colList; // SArray<SColumn*>
SFieldInfo fieldsInfo;
SArray * exprList; // SArray<SExprInfo*>
SArray * exprList1; // final exprlist in case of arithmetic expression exists
SLimitVal limit;
SLimitVal slimit;
STagCond tagCond;
SOrderVal order;
int16_t fillType; // final result fill type
int16_t numOfTables;
STableMetaInfo **pTableMetaInfo;
struct STSBuf *tsBuf;
int64_t * fillVal; // default value for fill
char * msg; // pointer to the pCmd->payload to keep error message temporarily
int64_t clauseLimit; // limit for current sub clause
int64_t prjOffset; // offset value in the original sql expression, only applied at client side
int64_t vgroupLimit; // table limit in case of super table projection query + global order + limit
int32_t udColumnId; // current user-defined constant output field column id, monotonically decreases from TSDB_UD_COLUMN_INDEX
int16_t resColumnId; // result column id
bool distinctTag; // distinct tag or not
int32_t round; // 0/1/....
int32_t bufLen;
char* buf;
SQInfo* pQInfo; // global merge operator
SQueryAttr* pQueryAttr; // query object
struct SQueryInfo *sibling; // sibling
SArray *pUpstream; // SArray<struct SQueryInfo>
struct SQueryInfo *pDownstream;
int32_t havingFieldNum;
bool stableQuery;
bool groupbyColumn;
bool simpleAgg;
bool arithmeticOnAgg;
bool projectionQuery;
bool hasFilter;
bool onlyTagQuery;
} SQueryInfo;
typedef struct {
STableMeta *pTableMeta;
SVgroupsInfo *pVgroupInfo;
......@@ -255,9 +130,13 @@ typedef struct SInsertStatementParam {
int8_t schemaAttached; // denote if submit block is built with table schema or not
STagData tagData; // NOTE: pTagData->data is used as a variant length array
int32_t batchSize; // for parameter ('?') binding and batch processing
int32_t numOfParams;
char msg[512]; // error message
char *sql; // current sql statement position
uint32_t insertType; // insert data from [file|sql statement| bound statement]
uint64_t objectId; // sql object id
char *sql; // current sql statement position
} SInsertStatementParam;
// TODO extract sql parser supporter
......@@ -266,13 +145,9 @@ typedef struct {
uint8_t msgType;
SInsertStatementParam insertParam;
char reserve1[3]; // fix bus error on arm32
union {
int32_t count;
};
int32_t count; // todo remove it
char * curSql; // current sql, resume position of sql after parsing paused
char reserve2[3]; // fix bus error on arm32
int16_t numOfCols;
char reserve3[2]; // fix bus error on arm32
uint32_t allocSize;
......@@ -283,8 +158,6 @@ typedef struct {
SQueryInfo *pQueryInfo;
SQueryInfo *active; // current active query info
int32_t batchSize; // for parameter ('?') binding and batch processing
int32_t numOfParams;
STagData tagData; // NOTE: pTagData->data is used as a variant length array
int32_t resColumnId;
} SSqlCmd;
......@@ -320,7 +193,7 @@ typedef struct {
TAOS_FIELD* final;
SArithmeticSupport *pArithSup; // support the arithmetic expression calculation on agg functions
struct SLocalMerger *pLocalMerger;
struct SGlobalMerger *pMerger;
} SSqlRes;
typedef struct {
......@@ -447,7 +320,7 @@ void tscSetResRawPtr(SSqlRes* pRes, SQueryInfo* pQueryInfo);
void tscSetResRawPtrRv(SSqlRes* pRes, SQueryInfo* pQueryInfo, SSDataBlock* pBlock);
void handleDownstreamOperator(SSqlObj** pSqlList, int32_t numOfUpstream, SQueryInfo* px, SSqlRes* pOutput);
void destroyTableNameList(SSqlCmd* pCmd);
void destroyTableNameList(SInsertStatementParam* pInsertParam);
void tscResetSqlCmd(SSqlCmd *pCmd, bool removeMeta);
......@@ -479,7 +352,7 @@ void waitForQueryRsp(void *param, TAOS_RES *tres, int code);
void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, __async_cb_func_t fp, void *param, const char *sqlstr, size_t sqlLen);
void tscImportDataFromFile(SSqlObj *pSql);
void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen);
struct SGlobalMerger* tscInitResObjForLocalQuery(int32_t numOfRes, int32_t rowLen, uint64_t id);
bool tscIsUpdateQuery(SSqlObj* pSql);
char* tscGetSqlStr(SSqlObj* pSql);
bool tscIsQueryWithLimit(SSqlObj* pSql);
......@@ -489,7 +362,7 @@ void tscSetBoundColumnInfo(SParsedDataColInfo *pColInfo, SSchema *pSchema, int32
char *tscGetErrorMsgPayload(SSqlCmd *pCmd);
int32_t tscInvalidSQLErrMsg(char *msg, const char *additionalInfo, const char *sql);
int32_t tscInvalidOperationMsg(char *msg, const char *additionalInfo, const char *sql);
int32_t tscSQLSyntaxErrMsg(char* msg, const char* additionalInfo, const char* sql);
int32_t tscValidateSqlInfo(SSqlObj *pSql, struct SSqlInfo *pInfo);
......
src/client/src/tscAsync.c
浏览文件 @ d2886dee
......@@ -22,7 +22,7 @@
#include "tscSubquery.h"
#include "tscUtil.h"
#include "tsched.h"
#include "tschemautil.h"
#include "qTableMeta.h"
#include "tsclient.h"
static void tscAsyncQueryRowsForNextVnode(void *param, TAOS_RES *tres, int numOfRows);
......@@ -58,7 +58,6 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, __async_cb_func_t fp, void* para
strntolower(pSql->sqlstr, sqlstr, (int32_t)sqlLen);
tscDebugL("0x%"PRIx64" SQL: %s", pSql->self, pSql->sqlstr);
pCmd->curSql = pSql->sqlstr;
pCmd->resColumnId = TSDB_RES_COL_ID;
int32_t code = tsParseSql(pSql, true);
......@@ -221,7 +220,7 @@ void taos_fetch_rows_a(TAOS_RES *tres, __async_cb_func_t fp, void *param) {
tscResetForNextRetrieve(pRes);
// handle the sub queries of join query
// handle outer query based on the already retrieved nest query results.
SQueryInfo* pQueryInfo = tscGetQueryInfo(pCmd);
if (pQueryInfo->pUpstream != NULL && taosArrayGetSize(pQueryInfo->pUpstream) > 0) {
SSchedMsg schedMsg = {0};
......
src/client/src/tscLocalMerge.c → src/client/src/tscGlobalmerge.c
浏览文件 @ d2886dee
......@@ -13,15 +13,19 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tscLocalMerge.h"
#include "tscSubquery.h"
#include "os.h"
#include "texpr.h"
#include "tlosertree.h"
#include "tscGlobalmerge.h"
#include "tscSubquery.h"
#include "tscLog.h"
#include "tsclient.h"
#include "qUtil.h"
#define COLMODEL_GET_VAL(data, schema, rowId, colId) \
(data + (schema)->pFields[colId].offset * ((schema)->capacity) + (rowId) * (schema)->pFields[colId].field.bytes)
typedef struct SCompareParam {
SLocalDataSource **pLocalData;
tOrderDescriptor * pDesc;
......@@ -29,9 +33,18 @@ typedef struct SCompareParam {
int32_t groupOrderType;
} SCompareParam;
bool needToMergeRv(SSDataBlock* pBlock, SArray* columnIndex, int32_t index, char **buf);
static bool needToMerge(SSDataBlock* pBlock, SArray* columnIndexList, int32_t index, char **buf) {
int32_t ret = 0;
size_t size = taosArrayGetSize(columnIndexList);
if (size > 0) {
ret = compare_aRv(pBlock, columnIndexList, (int32_t) size, index, buf, TSDB_ORDER_ASC);
}
// if ret == 0, means the result belongs to the same group
return (ret == 0);
}
int32_t treeComparator(const void *pLeft, const void *pRight, void *param) {
static int32_t treeComparator(const void *pLeft, const void *pRight, void *param) {
int32_t pLeftIdx = *(int32_t *)pLeft;
int32_t pRightIdx = *(int32_t *)pRight;
......@@ -57,16 +70,16 @@ int32_t treeComparator(const void *pLeft, const void *pRight, void *param) {
}
}
int32_t tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrderDescriptor *pDesc,
SQueryInfo* pQueryInfo, SLocalMerger **pMerger, int64_t id) {
int32_t tscCreateGlobalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrderDescriptor *pDesc,
SQueryInfo* pQueryInfo, SGlobalMerger **pMerger, int64_t id) {
if (pMemBuffer == NULL) {
tscLocalReducerEnvDestroy(pMemBuffer, pDesc, numOfBuffer);
tscDestroyGlobalMergerEnv(pMemBuffer, pDesc, numOfBuffer);
tscError("0x%"PRIx64" %p pMemBuffer is NULL", id, pMemBuffer);
return TSDB_CODE_TSC_APP_ERROR;
}
if (pDesc->pColumnModel == NULL) {
tscLocalReducerEnvDestroy(pMemBuffer, pDesc, numOfBuffer);
tscDestroyGlobalMergerEnv(pMemBuffer, pDesc, numOfBuffer);
tscError("0x%"PRIx64" no local buffer or intermediate result format model", id);
return TSDB_CODE_TSC_APP_ERROR;
}
......@@ -83,7 +96,7 @@ int32_t tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tO
}
if (numOfFlush == 0 || numOfBuffer == 0) {
tscLocalReducerEnvDestroy(pMemBuffer, pDesc, numOfBuffer);
tscDestroyGlobalMergerEnv(pMemBuffer, pDesc, numOfBuffer);
tscDebug("0x%"PRIx64" no data to retrieve", id);
return TSDB_CODE_SUCCESS;
}
......@@ -92,15 +105,15 @@ int32_t tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tO
tscError("0x%"PRIx64" Invalid value of buffer capacity %d and page size %d ", id, pDesc->pColumnModel->capacity,
pMemBuffer[0]->pageSize);
tscLocalReducerEnvDestroy(pMemBuffer, pDesc, numOfBuffer);
tscDestroyGlobalMergerEnv(pMemBuffer, pDesc, numOfBuffer);
return TSDB_CODE_TSC_APP_ERROR;
}
*pMerger = (SLocalMerger *) calloc(1, sizeof(SLocalMerger));
*pMerger = (SGlobalMerger *) calloc(1, sizeof(SGlobalMerger));
if ((*pMerger) == NULL) {
tscError("0x%"PRIx64" failed to create local merge structure, out of memory", id);
tscLocalReducerEnvDestroy(pMemBuffer, pDesc, numOfBuffer);
tscDestroyGlobalMergerEnv(pMemBuffer, pDesc, numOfBuffer);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
......@@ -160,7 +173,7 @@ int32_t tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tO
// no data actually, no need to merge result.
if (idx == 0) {
tscDebug("0x%"PRIx64" retrieved no data", id);
tscLocalReducerEnvDestroy(pMemBuffer, pDesc, numOfBuffer);
tscDestroyGlobalMergerEnv(pMemBuffer, pDesc, numOfBuffer);
return TSDB_CODE_SUCCESS;
}
......@@ -198,7 +211,7 @@ int32_t tscCreateLocalMerger(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tO
}
// restore the limitation value at the last stage
if (tscOrderedProjectionQueryOnSTable(pQueryInfo, 0)) {
if (pQueryInfo->orderProjectQuery) {
pQueryInfo->limit.limit = pQueryInfo->clauseLimit;
pQueryInfo->limit.offset = pQueryInfo->prjOffset;
}
......@@ -297,28 +310,28 @@ int32_t saveToBuffer(tExtMemBuffer *pMemoryBuf, tOrderDescriptor *pDesc, tFilePa
return 0;
}
void tscDestroyLocalMerger(SLocalMerger* pLocalMerger) {
if (pLocalMerger == NULL) {
void tscDestroyGlobalMerger(SGlobalMerger* pMerger) {
if (pMerger == NULL) {
return;
}
for (int32_t i = 0; i < pLocalMerger->numOfBuffer; ++i) {
tfree(pLocalMerger->pLocalDataSrc[i]);
for (int32_t i = 0; i < pMerger->numOfBuffer; ++i) {
tfree(pMerger->pLocalDataSrc[i]);
}
pLocalMerger->numOfBuffer = 0;
tscLocalReducerEnvDestroy(pLocalMerger->pExtMemBuffer, pLocalMerger->pDesc, pLocalMerger->numOfVnode);
pMerger->numOfBuffer = 0;
tscDestroyGlobalMergerEnv(pMerger->pExtMemBuffer, pMerger->pDesc, pMerger->numOfVnode);
pLocalMerger->numOfCompleted = 0;
pMerger->numOfCompleted = 0;
if (pLocalMerger->pLoserTree) {
tfree(pLocalMerger->pLoserTree->param);
tfree(pLocalMerger->pLoserTree);
if (pMerger->pLoserTree) {
tfree(pMerger->pLoserTree->param);
tfree(pMerger->pLoserTree);
}
tfree(pLocalMerger->buf);
tfree(pLocalMerger->pLocalDataSrc);
free(pLocalMerger);
tfree(pMerger->buf);
tfree(pMerger->pLocalDataSrc);
free(pMerger);
}
static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SQueryInfo* pQueryInfo, SColumnModel *pModel) {
......@@ -329,7 +342,7 @@ static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SQueryInfo*
}
// primary timestamp column is involved in final result
if (pQueryInfo->interval.interval != 0 || tscOrderedProjectionQueryOnSTable(pQueryInfo, 0)) {
if (pQueryInfo->interval.interval != 0 || pQueryInfo->orderProjectQuery) {
numOfGroupByCols++;
}
......@@ -392,7 +405,7 @@ static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SQueryInfo*
}
}
int32_t tscLocalReducerEnvCreate(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBuffer, int32_t numOfSub,
int32_t tscCreateGlobalMergerEnv(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBuffer, int32_t numOfSub,
tOrderDescriptor **pOrderDesc, uint32_t nBufferSizes, int64_t id) {
SSchema *pSchema = NULL;
SColumnModel *pModel = NULL;
......@@ -456,7 +469,7 @@ int32_t tscLocalReducerEnvCreate(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBu
* @param pDesc
* @param numOfVnodes
*/
void tscLocalReducerEnvDestroy(tExtMemBuffer **pMemBuffer, tOrderDescriptor *pDesc, int32_t numOfVnodes) {
void tscDestroyGlobalMergerEnv(tExtMemBuffer **pMemBuffer, tOrderDescriptor *pDesc, int32_t numOfVnodes) {
tOrderDescDestroy(pDesc);
for (int32_t i = 0; i < numOfVnodes; ++i) {
pMemBuffer[i] = destoryExtMemBuffer(pMemBuffer[i]);
......@@ -467,12 +480,12 @@ void tscLocalReducerEnvDestroy(tExtMemBuffer **pMemBuffer, tOrderDescriptor *pDe
/**
*
* @param pLocalMerge
* @param pMerger
* @param pOneInterDataSrc
* @param treeList
* @return the number of remain input source. if ret == 0, all data has been handled
*/
int32_t loadNewDataFromDiskFor(SLocalMerger *pLocalMerge, SLocalDataSource *pOneInterDataSrc,
int32_t loadNewDataFromDiskFor(SGlobalMerger *pMerger, SLocalDataSource *pOneInterDataSrc,
bool *needAdjustLoserTree) {
pOneInterDataSrc->rowIdx = 0;
pOneInterDataSrc->pageId += 1;
......@@ -489,17 +502,17 @@ int32_t loadNewDataFromDiskFor(SLocalMerger *pLocalMerge, SLocalDataSource *pOne
#endif
*needAdjustLoserTree = true;
} else {
pLocalMerge->numOfCompleted += 1;
pMerger->numOfCompleted += 1;
pOneInterDataSrc->rowIdx = -1;
pOneInterDataSrc->pageId = -1;
*needAdjustLoserTree = true;
}
return pLocalMerge->numOfBuffer;
return pMerger->numOfBuffer;
}
void adjustLoserTreeFromNewData(SLocalMerger *pLocalMerge, SLocalDataSource *pOneInterDataSrc,
void adjustLoserTreeFromNewData(SGlobalMerger *pMerger, SLocalDataSource *pOneInterDataSrc,
SLoserTreeInfo *pTree) {
/*
* load a new data page into memory for intermediate dataset source,
......@@ -507,7 +520,7 @@ void adjustLoserTreeFromNewData(SLocalMerger *pLocalMerge, SLocalDataSource *pOn
*/
bool needToAdjust = true;
if (pOneInterDataSrc->filePage.num <= pOneInterDataSrc->rowIdx) {
loadNewDataFromDiskFor(pLocalMerge, pOneInterDataSrc, &needToAdjust);
loadNewDataFromDiskFor(pMerger, pOneInterDataSrc, &needToAdjust);
}
/*
......@@ -515,7 +528,7 @@ void adjustLoserTreeFromNewData(SLocalMerger *pLocalMerge, SLocalDataSource *pOn
* if the loser tree is rebuild completed, we do not need to adjust
*/
if (needToAdjust) {
int32_t leafNodeIdx = pTree->pNode[0].index + pLocalMerge->numOfBuffer;
int32_t leafNodeIdx = pTree->pNode[0].index + pMerger->numOfBuffer;
#ifdef _DEBUG_VIEW
printf("before adjust:\t");
......@@ -567,7 +580,7 @@ static void setTagValueForMultipleRows(SQLFunctionCtx* pCtx, int32_t numOfOutput
}
}
static void doExecuteFinalMergeRv(SOperatorInfo* pOperator, int32_t numOfExpr, SSDataBlock* pBlock) {
static void doExecuteFinalMerge(SOperatorInfo* pOperator, int32_t numOfExpr, SSDataBlock* pBlock) {
SMultiwayMergeInfo* pInfo = pOperator->info;
SQLFunctionCtx* pCtx = pInfo->binfo.pCtx;
......@@ -579,7 +592,7 @@ static void doExecuteFinalMergeRv(SOperatorInfo* pOperator, int32_t numOfExpr, S
for(int32_t i = 0; i < pBlock->info.rows; ++i) {
if (pInfo->hasPrev) {
if (needToMergeRv(pBlock, pInfo->orderColumnList, i, pInfo->prevRow)) {
if (needToMerge(pBlock, pInfo->orderColumnList, i, pInfo->prevRow)) {
for (int32_t j = 0; j < numOfExpr; ++j) {
pCtx[j].pInput = add[j] + pCtx[j].inputBytes * i;
}
......@@ -654,45 +667,27 @@ static void doExecuteFinalMergeRv(SOperatorInfo* pOperator, int32_t numOfExpr, S
tfree(add);
}
bool needToMergeRv(SSDataBlock* pBlock, SArray* columnIndexList, int32_t index, char **buf) {
int32_t ret = 0;
size_t size = taosArrayGetSize(columnIndexList);
if (size > 0) {
ret = compare_aRv(pBlock, columnIndexList, (int32_t) size, index, buf, TSDB_ORDER_ASC);
}
// if ret == 0, means the result belongs to the same group
return (ret == 0);
static bool isAllSourcesCompleted(SGlobalMerger *pMerger) {
return (pMerger->numOfBuffer == pMerger->numOfCompleted);
}
static bool isAllSourcesCompleted(SLocalMerger *pLocalMerge) {
return (pLocalMerge->numOfBuffer == pLocalMerge->numOfCompleted);
}
void tscInitResObjForLocalQuery(SSqlObj *pSql, int32_t numOfRes, int32_t rowLen) {
SSqlRes *pRes = &pSql->res;
if (pRes->pLocalMerger != NULL) {
tscDestroyLocalMerger(pRes->pLocalMerger);
pRes->pLocalMerger = NULL;
tscDebug("0x%"PRIx64" free local reducer finished", pSql->self);
SGlobalMerger* tscInitResObjForLocalQuery(int32_t numOfRes, int32_t rowLen, uint64_t id) {
SGlobalMerger *pMerger = calloc(1, sizeof(SGlobalMerger));
if (pMerger == NULL) {
tscDebug("0x%"PRIx64" free local reducer finished", id);
return NULL;
}
pRes->qId = 1; // hack to pass the safety check in fetch_row function
pRes->numOfRows = 0;
pRes->row = 0;
pRes->rspType = 0; // used as a flag to denote if taos_retrieved() has been called yet
pRes->pLocalMerger = (SLocalMerger *)calloc(1, sizeof(SLocalMerger));
/*
* One more byte space is required, since the sprintf function needs one additional space to put '\0' at
* the end of string
*/
size_t size = numOfRes * rowLen + 1;
pRes->pLocalMerger->buf = calloc(1, size);
pRes->data = pRes->pLocalMerger->buf;
pMerger->buf = calloc(1, size);
return pMerger;
}
// todo remove it
int32_t doArithmeticCalculate(SQueryInfo* pQueryInfo, tFilePage* pOutput, int32_t rowSize, int32_t finalRowSize) {
int32_t maxRowSize = MAX(rowSize, finalRowSize);
char* pbuf = calloc(1, (size_t)(pOutput->num * maxRowSize));
......@@ -736,9 +731,6 @@ int32_t doArithmeticCalculate(SQueryInfo* pQueryInfo, tFilePage* pOutput, int32_
return offset;
}
#define COLMODEL_GET_VAL(data, schema, rowId, colId) \
(data + (schema)->pFields[colId].offset * ((schema)->capacity) + (rowId) * (schema)->pFields[colId].field.bytes)
static void appendOneRowToDataBlock(SSDataBlock *pBlock, char *buf, SColumnModel *pModel, int32_t rowIndex,
int32_t maxRows) {
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
......@@ -760,7 +752,7 @@ SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup) {
SMultiwayMergeInfo *pInfo = pOperator->info;
SLocalMerger *pMerger = pInfo->pMerge;
SGlobalMerger *pMerger = pInfo->pMerge;
SLoserTreeInfo *pTree = pMerger->pLoserTree;
pInfo->binfo.pRes->info.rows = 0;
......@@ -844,7 +836,7 @@ SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup) {
return (pInfo->binfo.pRes->info.rows > 0)? pInfo->binfo.pRes:NULL;
}
static bool isSameGroupRv(SArray* orderColumnList, SSDataBlock* pBlock, char** dataCols) {
static bool isSameGroup(SArray* orderColumnList, SSDataBlock* pBlock, char** dataCols) {
int32_t numOfCols = (int32_t) taosArrayGetSize(orderColumnList);
for (int32_t i = 0; i < numOfCols; ++i) {
SColIndex *pIndex = taosArrayGet(orderColumnList, i);
......@@ -892,7 +884,7 @@ SSDataBlock* doGlobalAggregate(void* param, bool* newgroup) {
}
}
doExecuteFinalMergeRv(pOperator, pOperator->numOfOutput, pAggInfo->pExistBlock);
doExecuteFinalMerge(pOperator, pOperator->numOfOutput, pAggInfo->pExistBlock);
savePrevOrderColumns(pAggInfo->currentGroupColData, pAggInfo->groupColumnList, pAggInfo->pExistBlock, 0,
&pAggInfo->hasGroupColData);
......@@ -913,7 +905,7 @@ SSDataBlock* doGlobalAggregate(void* param, bool* newgroup) {
}
if (pAggInfo->hasGroupColData) {
bool sameGroup = isSameGroupRv(pAggInfo->groupColumnList, pBlock, pAggInfo->currentGroupColData);
bool sameGroup = isSameGroup(pAggInfo->groupColumnList, pBlock, pAggInfo->currentGroupColData);
if (!sameGroup) {
*newgroup = true;
pAggInfo->hasDataBlockForNewGroup = true;
......@@ -927,7 +919,7 @@ SSDataBlock* doGlobalAggregate(void* param, bool* newgroup) {
setInputDataBlock(pOperator, pAggInfo->binfo.pCtx, pBlock, TSDB_ORDER_ASC);
updateOutputBuf(&pAggInfo->binfo, &pAggInfo->bufCapacity, pBlock->info.rows * pAggInfo->resultRowFactor);
doExecuteFinalMergeRv(pOperator, pOperator->numOfOutput, pBlock);
doExecuteFinalMerge(pOperator, pOperator->numOfOutput, pBlock);
savePrevOrderColumns(pAggInfo->currentGroupColData, pAggInfo->groupColumnList, pBlock, 0, &pAggInfo->hasGroupColData);
handleData = true;
}
......
src/client/src/tscLocal.c
浏览文件 @ d2886dee
......@@ -20,7 +20,7 @@
#include "tname.h"
#include "tscLog.h"
#include "tscUtil.h"
#include "tschemautil.h"
#include "qTableMeta.h"
#include "tsclient.h"
#include "taos.h"
#include "tscSubquery.h"
......@@ -71,7 +71,9 @@ static int32_t tscSetValueToResObj(SSqlObj *pSql, int32_t rowLen) {
numOfRows = numOfRows + tscGetNumOfTags(pMeta);
}
tscInitResObjForLocalQuery(pSql, totalNumOfRows, rowLen);
pSql->res.pMerger = tscInitResObjForLocalQuery(totalNumOfRows, rowLen, pSql->self);
tscInitResForMerge(&pSql->res);
SSchema *pSchema = tscGetTableSchema(pMeta);
for (int32_t i = 0; i < numOfRows; ++i) {
......@@ -433,7 +435,8 @@ static int32_t tscSCreateSetValueToResObj(SSqlObj *pSql, int32_t rowLen, const c
if (strlen(ddl) == 0) {
}
tscInitResObjForLocalQuery(pSql, numOfRows, rowLen);
pSql->res.pMerger = tscInitResObjForLocalQuery(numOfRows, rowLen, pSql->self);
tscInitResForMerge(&pSql->res);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 0);
char* dst = pRes->data + tscFieldInfoGetOffset(pQueryInfo, 0) * numOfRows;
......@@ -882,7 +885,8 @@ void tscSetLocalQueryResult(SSqlObj *pSql, const char *val, const char *columnNa
TAOS_FIELD f = tscCreateField((int8_t)type, columnName, (int16_t)valueLength);
tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
tscInitResObjForLocalQuery(pSql, 1, (int32_t)valueLength);
pSql->res.pMerger = tscInitResObjForLocalQuery(1, (int32_t)valueLength, pSql->self);
tscInitResForMerge(&pSql->res);
SInternalField* pInfo = tscFieldInfoGetInternalField(&pQueryInfo->fieldsInfo, 0);
pInfo->pExpr = taosArrayGetP(pQueryInfo->exprList, 0);
......
src/client/src/tscParseInsert.c
浏览文件 @ d2886dee
......@@ -23,7 +23,7 @@
#include "ttype.h"
#include "hash.h"
#include "tscUtil.h"
#include "tschemautil.h"
#include "qTableMeta.h"
#include "tsclient.h"
#include "ttokendef.h"
#include "taosdef.h"
......@@ -38,8 +38,9 @@ enum {
TSDB_USE_CLI_TS = 1,
};
static int32_t tscAllocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int32_t * numOfRows);
static int32_t parseBoundColumns(SSqlCmd* pCmd, SParsedDataColInfo* pColInfo, SSchema* pSchema, char* str, char** end);
static int32_t tscAllocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int32_t *numOfRows);
static int32_t parseBoundColumns(SInsertStatementParam *pInsertParam, SParsedDataColInfo *pColInfo, SSchema *pSchema,
char *str, char **end);
static int32_t tscToDouble(SStrToken *pToken, double *value, char **endPtr) {
errno = 0;
......@@ -71,7 +72,7 @@ int tsParseTime(SStrToken *pToken, int64_t *time, char **next, char *error, int1
} else {
// strptime("2001-11-12 18:31:01", "%Y-%m-%d %H:%M:%S", &tm);
if (taosParseTime(pToken->z, time, pToken->n, timePrec, tsDaylight) != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(error, "invalid timestamp format", pToken->z);
return tscInvalidOperationMsg(error, "invalid timestamp format", pToken->z);
}
return TSDB_CODE_SUCCESS;
......@@ -103,7 +104,7 @@ int tsParseTime(SStrToken *pToken, int64_t *time, char **next, char *error, int1
pTokenEnd += index;
if (valueToken.n < 2) {
return tscInvalidSQLErrMsg(error, "value expected in timestamp", sToken.z);
return tscInvalidOperationMsg(error, "value expected in timestamp", sToken.z);
}
if (parseAbsoluteDuration(valueToken.z, valueToken.n, &interval) != TSDB_CODE_SUCCESS) {
......@@ -138,7 +139,7 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
char *endptr = NULL;
if (IS_NUMERIC_TYPE(pSchema->type) && pToken->n == 0) {
return tscInvalidSQLErrMsg(msg, "invalid numeric data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid numeric data", pToken->z);
}
switch (pSchema->type) {
......@@ -161,7 +162,7 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
double dv = strtod(pToken->z, NULL);
*(uint8_t *)payload = (int8_t)((dv == 0) ? TSDB_FALSE : TSDB_TRUE);
} else {
return tscInvalidSQLErrMsg(msg, "invalid bool data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid bool data", pToken->z);
}
}
break;
......@@ -173,9 +174,9 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
ret = tStrToInteger(pToken->z, pToken->type, pToken->n, &iv, true);
if (ret != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid tinyint data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid tinyint data", pToken->z);
} else if (!IS_VALID_TINYINT(iv)) {
return tscInvalidSQLErrMsg(msg, "data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "data overflow", pToken->z);
}
*((uint8_t *)payload) = (uint8_t)iv;
......@@ -189,9 +190,9 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
ret = tStrToInteger(pToken->z, pToken->type, pToken->n, &iv, false);
if (ret != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid unsigned tinyint data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid unsigned tinyint data", pToken->z);
} else if (!IS_VALID_UTINYINT(iv)) {
return tscInvalidSQLErrMsg(msg, "unsigned tinyint data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "unsigned tinyint data overflow", pToken->z);
}
*((uint8_t *)payload) = (uint8_t)iv;
......@@ -205,9 +206,9 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
ret = tStrToInteger(pToken->z, pToken->type, pToken->n, &iv, true);
if (ret != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid smallint data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid smallint data", pToken->z);
} else if (!IS_VALID_SMALLINT(iv)) {
return tscInvalidSQLErrMsg(msg, "smallint data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "smallint data overflow", pToken->z);
}
*((int16_t *)payload) = (int16_t)iv;
......@@ -221,9 +222,9 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
ret = tStrToInteger(pToken->z, pToken->type, pToken->n, &iv, false);
if (ret != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid unsigned smallint data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid unsigned smallint data", pToken->z);
} else if (!IS_VALID_USMALLINT(iv)) {
return tscInvalidSQLErrMsg(msg, "unsigned smallint data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "unsigned smallint data overflow", pToken->z);
}
*((uint16_t *)payload) = (uint16_t)iv;
......@@ -237,9 +238,9 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
ret = tStrToInteger(pToken->z, pToken->type, pToken->n, &iv, true);
if (ret != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid int data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid int data", pToken->z);
} else if (!IS_VALID_INT(iv)) {
return tscInvalidSQLErrMsg(msg, "int data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "int data overflow", pToken->z);
}
*((int32_t *)payload) = (int32_t)iv;
......@@ -253,9 +254,9 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
ret = tStrToInteger(pToken->z, pToken->type, pToken->n, &iv, false);
if (ret != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid unsigned int data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid unsigned int data", pToken->z);
} else if (!IS_VALID_UINT(iv)) {
return tscInvalidSQLErrMsg(msg, "unsigned int data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "unsigned int data overflow", pToken->z);
}
*((uint32_t *)payload) = (uint32_t)iv;
......@@ -269,9 +270,9 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
ret = tStrToInteger(pToken->z, pToken->type, pToken->n, &iv, true);
if (ret != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid bigint data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid bigint data", pToken->z);
} else if (!IS_VALID_BIGINT(iv)) {
return tscInvalidSQLErrMsg(msg, "bigint data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "bigint data overflow", pToken->z);
}
*((int64_t *)payload) = iv;
......@@ -284,9 +285,9 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
ret = tStrToInteger(pToken->z, pToken->type, pToken->n, &iv, false);
if (ret != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid unsigned bigint data", pToken->z);
return tscInvalidOperationMsg(msg, "invalid unsigned bigint data", pToken->z);
} else if (!IS_VALID_UBIGINT((uint64_t)iv)) {
return tscInvalidSQLErrMsg(msg, "unsigned bigint data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "unsigned bigint data overflow", pToken->z);
}
*((uint64_t *)payload) = iv;
......@@ -299,11 +300,11 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
double dv;
if (TK_ILLEGAL == tscToDouble(pToken, &dv, &endptr)) {
return tscInvalidSQLErrMsg(msg, "illegal float data", pToken->z);
return tscInvalidOperationMsg(msg, "illegal float data", pToken->z);
}
if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || dv > FLT_MAX || dv < -FLT_MAX || isinf(dv) || isnan(dv)) {
return tscInvalidSQLErrMsg(msg, "illegal float data", pToken->z);
return tscInvalidOperationMsg(msg, "illegal float data", pToken->z);
}
// *((float *)payload) = (float)dv;
......@@ -317,11 +318,11 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
double dv;
if (TK_ILLEGAL == tscToDouble(pToken, &dv, &endptr)) {
return tscInvalidSQLErrMsg(msg, "illegal double data", pToken->z);
return tscInvalidOperationMsg(msg, "illegal double data", pToken->z);
}
if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || isinf(dv) || isnan(dv)) {
return tscInvalidSQLErrMsg(msg, "illegal double data", pToken->z);
return tscInvalidOperationMsg(msg, "illegal double data", pToken->z);
}
*((double *)payload) = dv;
......@@ -334,7 +335,7 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
setVardataNull(payload, TSDB_DATA_TYPE_BINARY);
} else { // too long values will return invalid sql, not be truncated automatically
if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) { //todo refactor
return tscInvalidSQLErrMsg(msg, "string data overflow", pToken->z);
return tscInvalidOperationMsg(msg, "string data overflow", pToken->z);
}
STR_WITH_SIZE_TO_VARSTR(payload, pToken->z, pToken->n);
......@@ -351,7 +352,7 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
if (!taosMbsToUcs4(pToken->z, pToken->n, varDataVal(payload), pSchema->bytes - VARSTR_HEADER_SIZE, &output)) {
char buf[512] = {0};
snprintf(buf, tListLen(buf), "%s", strerror(errno));
return tscInvalidSQLErrMsg(msg, buf, pToken->z);
return tscInvalidOperationMsg(msg, buf, pToken->z);
}
varDataSetLen(payload, output);
......@@ -368,7 +369,7 @@ int32_t tsParseOneColumn(SSchema *pSchema, SStrToken *pToken, char *payload, cha
} else {
int64_t temp;
if (tsParseTime(pToken, &temp, str, msg, timePrec) != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(msg, "invalid timestamp", pToken->z);
return tscInvalidOperationMsg(msg, "invalid timestamp", pToken->z);
}
*((int64_t *)payload) = temp;
......@@ -417,8 +418,8 @@ int32_t tsCheckTimestamp(STableDataBlocks *pDataBlocks, const char *start) {
return TSDB_CODE_SUCCESS;
}
int tsParseOneRow(char **str, STableDataBlocks *pDataBlocks, SSqlCmd *pCmd, int16_t timePrec, int32_t *len,
char *tmpTokenBuf) {
int tsParseOneRow(char **str, STableDataBlocks *pDataBlocks, int16_t timePrec, int32_t *len,
char *tmpTokenBuf, SInsertStatementParam* pInsertParam) {
int32_t index = 0;
SStrToken sToken = {0};
char *payload = pDataBlocks->pData + pDataBlocks->size;
......@@ -441,8 +442,8 @@ int tsParseOneRow(char **str, STableDataBlocks *pDataBlocks, SSqlCmd *pCmd, int1
*str += index;
if (sToken.type == TK_QUESTION) {
if (pCmd->insertParam.insertType != TSDB_QUERY_TYPE_STMT_INSERT) {
return tscSQLSyntaxErrMsg(pCmd->payload, "? only allowed in binding insertion", *str);
if (pInsertParam->insertType != TSDB_QUERY_TYPE_STMT_INSERT) {
return tscSQLSyntaxErrMsg(pInsertParam->msg, "? only allowed in binding insertion", *str);
}
uint32_t offset = (uint32_t)(start - pDataBlocks->pData);
......@@ -450,14 +451,14 @@ int tsParseOneRow(char **str, STableDataBlocks *pDataBlocks, SSqlCmd *pCmd, int1
continue;
}
strcpy(pCmd->payload, "client out of memory");
strcpy(pInsertParam->msg, "client out of memory");
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
int16_t type = sToken.type;
if ((type != TK_NOW && type != TK_INTEGER && type != TK_STRING && type != TK_FLOAT && type != TK_BOOL &&
type != TK_NULL && type != TK_HEX && type != TK_OCT && type != TK_BIN) || (sToken.n == 0) || (type == TK_RP)) {
return tscSQLSyntaxErrMsg(pCmd->payload, "invalid data or symbol", sToken.z);
return tscSQLSyntaxErrMsg(pInsertParam->msg, "invalid data or symbol", sToken.z);
}
// Remove quotation marks
......@@ -487,13 +488,13 @@ int tsParseOneRow(char **str, STableDataBlocks *pDataBlocks, SSqlCmd *pCmd, int1
}
bool isPrimaryKey = (colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX);
int32_t ret = tsParseOneColumn(pSchema, &sToken, start, pCmd->payload, str, isPrimaryKey, timePrec);
int32_t ret = tsParseOneColumn(pSchema, &sToken, start, pInsertParam->msg, str, isPrimaryKey, timePrec);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
}
if (isPrimaryKey && tsCheckTimestamp(pDataBlocks, start) != TSDB_CODE_SUCCESS) {
tscInvalidSQLErrMsg(pCmd->payload, "client time/server time can not be mixed up", sToken.z);
tscInvalidOperationMsg(pInsertParam->msg, "client time/server time can not be mixed up", sToken.z);
return TSDB_CODE_TSC_INVALID_TIME_STAMP;
}
}
......@@ -536,7 +537,8 @@ static int32_t rowDataCompar(const void *lhs, const void *rhs) {
}
}
int32_t tsParseValues(char **str, STableDataBlocks *pDataBlock, int maxRows, SSqlCmd* pCmd, int32_t* numOfRows, char *tmpTokenBuf) {
int32_t tsParseValues(char **str, STableDataBlocks *pDataBlock, int maxRows, SInsertStatementParam *pInsertParam,
int32_t* numOfRows, char *tmpTokenBuf) {
int32_t index = 0;
int32_t code = 0;
......@@ -559,7 +561,7 @@ int32_t tsParseValues(char **str, STableDataBlocks *pDataBlock, int maxRows, SSq
int32_t tSize;
code = tscAllocateMemIfNeed(pDataBlock, tinfo.rowSize, &tSize);
if (code != TSDB_CODE_SUCCESS) { //TODO pass the correct error code to client
strcpy(pCmd->payload, "client out of memory");
strcpy(pInsertParam->msg, "client out of memory");
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
......@@ -568,7 +570,7 @@ int32_t tsParseValues(char **str, STableDataBlocks *pDataBlock, int maxRows, SSq
}
int32_t len = 0;
code = tsParseOneRow(str, pDataBlock, pCmd, precision, &len, tmpTokenBuf);
code = tsParseOneRow(str, pDataBlock, precision, &len, tmpTokenBuf, pInsertParam);
if (code != TSDB_CODE_SUCCESS) { // error message has been set in tsParseOneRow, return directly
return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
}
......@@ -578,7 +580,7 @@ int32_t tsParseValues(char **str, STableDataBlocks *pDataBlock, int maxRows, SSq
index = 0;
sToken = tStrGetToken(*str, &index, false);
if (sToken.n == 0 || sToken.type != TK_RP) {
tscSQLSyntaxErrMsg(pCmd->payload, ") expected", *str);
tscSQLSyntaxErrMsg(pInsertParam->msg, ") expected", *str);
code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
return code;
}
......@@ -589,7 +591,7 @@ int32_t tsParseValues(char **str, STableDataBlocks *pDataBlock, int maxRows, SSq
}
if ((*numOfRows) <= 0) {
strcpy(pCmd->payload, "no any data points");
strcpy(pInsertParam->msg, "no any data points");
return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
} else {
return TSDB_CODE_SUCCESS;
......@@ -699,7 +701,7 @@ void tscSortRemoveDataBlockDupRows(STableDataBlocks *dataBuf) {
dataBuf->prevTS = INT64_MIN;
}
static int32_t doParseInsertStatement(SSqlCmd* pCmd, char **str, STableDataBlocks* dataBuf, int32_t *totalNum) {
static int32_t doParseInsertStatement(SInsertStatementParam *pInsertParam, char **str, STableDataBlocks* dataBuf, int32_t *totalNum) {
STableComInfo tinfo = tscGetTableInfo(dataBuf->pTableMeta);
int32_t maxNumOfRows;
......@@ -712,7 +714,7 @@ static int32_t doParseInsertStatement(SSqlCmd* pCmd, char **str, STableDataBlock
char tmpTokenBuf[16*1024] = {0}; // used for deleting Escape character: \\, \', \"
int32_t numOfRows = 0;
code = tsParseValues(str, dataBuf, maxNumOfRows, pCmd, &numOfRows, tmpTokenBuf);
code = tsParseValues(str, dataBuf, maxNumOfRows, pInsertParam, &numOfRows, tmpTokenBuf);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
......@@ -720,7 +722,7 @@ static int32_t doParseInsertStatement(SSqlCmd* pCmd, char **str, STableDataBlock
for (uint32_t i = 0; i < dataBuf->numOfParams; ++i) {
SParamInfo *param = dataBuf->params + i;
if (param->idx == -1) {
param->idx = pCmd->numOfParams++;
param->idx = pInsertParam->numOfParams++;
param->offset -= sizeof(SSubmitBlk);
}
}
......@@ -728,7 +730,7 @@ static int32_t doParseInsertStatement(SSqlCmd* pCmd, char **str, STableDataBlock
SSubmitBlk *pBlocks = (SSubmitBlk *)(dataBuf->pData);
code = tsSetBlockInfo(pBlocks, dataBuf->pTableMeta, numOfRows);
if (code != TSDB_CODE_SUCCESS) {
tscInvalidSQLErrMsg(pCmd->payload, "too many rows in sql, total number of rows should be less than 32767", *str);
tscInvalidOperationMsg(pInsertParam->msg, "too many rows in sql, total number of rows should be less than 32767", *str);
return code;
}
......@@ -748,6 +750,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
SSqlCmd * pCmd = &pSql->cmd;
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd);
SInsertStatementParam* pInsertParam = &pCmd->insertParam;
char *sql = *sqlstr;
......@@ -784,7 +787,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
}
if (numOfColList == 0 && (*boundColumn) != NULL) {
return TSDB_CODE_TSC_INVALID_OPERATION;
return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
}
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, TABLE_INDEX);
......@@ -805,8 +808,8 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
return code;
}
tNameExtractFullName(&pSTableMetaInfo->name, pCmd->tagData.name);
pCmd->tagData.dataLen = 0;
tNameExtractFullName(&pSTableMetaInfo->name, pInsertParam->tagData.name);
pInsertParam->tagData.dataLen = 0;
code = tscGetTableMeta(pSql, pSTableMetaInfo);
if (code != TSDB_CODE_SUCCESS) {
......@@ -814,7 +817,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
}
if (!UTIL_TABLE_IS_SUPER_TABLE(pSTableMetaInfo)) {
return tscInvalidSQLErrMsg(pCmd->payload, "create table only from super table is allowed", sToken.z);
return tscInvalidOperationMsg(pInsertParam->msg, "create table only from super table is allowed", sToken.z);
}
SSchema *pTagSchema = tscGetTableTagSchema(pSTableMetaInfo->pTableMeta);
......@@ -827,7 +830,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
sToken = tStrGetToken(sql, &index, false);
if (sToken.type != TK_TAGS && sToken.type != TK_LP) {
tscDestroyBoundColumnInfo(&spd);
return tscInvalidSQLErrMsg(pCmd->payload, "keyword TAGS expected", sToken.z);
return tscSQLSyntaxErrMsg(pInsertParam->msg, "keyword TAGS expected", sToken.z);
}
// parse the bound tags column
......@@ -837,7 +840,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
* tags(tagVal1, tagVal2, ..., tagValn) values(v1, v2,... vn);
*/
char* end = NULL;
code = parseBoundColumns(pCmd, &spd, pTagSchema, sql, &end);
code = parseBoundColumns(pInsertParam, &spd, pTagSchema, sql, &end);
if (code != TSDB_CODE_SUCCESS) {
tscDestroyBoundColumnInfo(&spd);
return code;
......@@ -858,7 +861,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
if (sToken.type != TK_LP) {
tscDestroyBoundColumnInfo(&spd);
return tscInvalidSQLErrMsg(pCmd->payload, "( is expected", sToken.z);
return tscSQLSyntaxErrMsg(pInsertParam->msg, "( is expected", sToken.z);
}
SKVRowBuilder kvRowBuilder = {0};
......@@ -877,7 +880,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
if (TK_ILLEGAL == sToken.type) {
tdDestroyKVRowBuilder(&kvRowBuilder);
tscDestroyBoundColumnInfo(&spd);
return TSDB_CODE_TSC_INVALID_OPERATION;
return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
}
if (sToken.n == 0 || sToken.type == TK_RP) {
......@@ -891,7 +894,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
}
char tagVal[TSDB_MAX_TAGS_LEN];
code = tsParseOneColumn(pSchema, &sToken, tagVal, pCmd->payload, &sql, false, tinfo.precision);
code = tsParseOneColumn(pSchema, &sToken, tagVal, pInsertParam->msg, &sql, false, tinfo.precision);
if (code != TSDB_CODE_SUCCESS) {
tdDestroyKVRowBuilder(&kvRowBuilder);
tscDestroyBoundColumnInfo(&spd);
......@@ -906,29 +909,29 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
tdDestroyKVRowBuilder(&kvRowBuilder);
if (row == NULL) {
return tscInvalidSQLErrMsg(pCmd->payload, "tag value expected", NULL);
return tscSQLSyntaxErrMsg(pInsertParam->msg, "tag value expected", NULL);
}
tdSortKVRowByColIdx(row);
pCmd->tagData.dataLen = kvRowLen(row);
if (pCmd->tagData.dataLen <= 0){
return tscInvalidSQLErrMsg(pCmd->payload, "tag value expected", NULL);
pInsertParam->tagData.dataLen = kvRowLen(row);
if (pInsertParam->tagData.dataLen <= 0){
return tscSQLSyntaxErrMsg(pInsertParam->msg, "tag value expected", NULL);
}
char* pTag = realloc(pCmd->tagData.data, pCmd->tagData.dataLen);
char* pTag = realloc(pInsertParam->tagData.data, pInsertParam->tagData.dataLen);
if (pTag == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
kvRowCpy(pTag, row);
free(row);
pCmd->tagData.data = pTag;
pInsertParam->tagData.data = pTag;
index = 0;
sToken = tStrGetToken(sql, &index, false);
sql += index;
if (sToken.n == 0 || sToken.type != TK_RP) {
return tscSQLSyntaxErrMsg(pCmd->payload, ") expected", sToken.z);
return tscSQLSyntaxErrMsg(pInsertParam->msg, ") expected", sToken.z);
}
/* parse columns after super table tags values.
......@@ -941,7 +944,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
int numOfColsAfterTags = 0;
if (sToken.type == TK_LP) {
if (*boundColumn != NULL) {
return tscSQLSyntaxErrMsg(pCmd->payload, "bind columns again", sToken.z);
return tscSQLSyntaxErrMsg(pInsertParam->msg, "bind columns again", sToken.z);
} else {
*boundColumn = &sToken.z[0];
}
......@@ -959,7 +962,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
}
if (numOfColsAfterTags == 0 && (*boundColumn) != NULL) {
return TSDB_CODE_TSC_INVALID_OPERATION;
return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
}
sToken = tStrGetToken(sql, &index, false);
......@@ -968,7 +971,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
sql = sToken.z;
if (tscValidateName(&tableToken) != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(pCmd->payload, "invalid table name", *sqlstr);
return tscInvalidOperationMsg(pInsertParam->msg, "invalid table name", *sqlstr);
}
int32_t ret = tscSetTableFullName(&pTableMetaInfo->name, &tableToken, pSql);
......@@ -977,7 +980,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
}
if (sql == NULL) {
return TSDB_CODE_TSC_INVALID_OPERATION;
return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
}
code = tscGetTableMetaEx(pSql, pTableMetaInfo, true);
......@@ -986,20 +989,18 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql, char** boundC
}
} else {
sql = sToken.z;
if (sql == NULL) {
return TSDB_CODE_TSC_INVALID_OPERATION;
if (sToken.z == NULL) {
return tscSQLSyntaxErrMsg(pInsertParam->msg, "", sql);
}
sql = sToken.z;
code = tscGetTableMetaEx(pSql, pTableMetaInfo, false);
if (pCmd->curSql == NULL) {
if (pInsertParam->sql == NULL) {
assert(code == TSDB_CODE_TSC_ACTION_IN_PROGRESS);
}
}
*sqlstr = sql;
return code;
}
......@@ -1013,21 +1014,21 @@ int validateTableName(char *tblName, int len, SStrToken* psTblToken) {
return tscValidateName(psTblToken);
}
static int32_t validateDataSource(SSqlCmd *pCmd, int32_t type, const char *sql) {
uint32_t *insertType = &pCmd->insertParam.insertType;
static int32_t validateDataSource(SInsertStatementParam *pInsertParam, int32_t type, const char *sql) {
uint32_t *insertType = &pInsertParam->insertType;
if (*insertType == TSDB_QUERY_TYPE_STMT_INSERT && type == TSDB_QUERY_TYPE_INSERT) {
return TSDB_CODE_SUCCESS;
}
if ((*insertType) != 0 && (*insertType) != type) {
return tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES and FILE are not allowed to mixed up", sql);
return tscSQLSyntaxErrMsg(pInsertParam->msg, "keyword VALUES and FILE are not allowed to mixed up", sql);
}
*insertType = type;
return TSDB_CODE_SUCCESS;
}
static int32_t parseBoundColumns(SSqlCmd* pCmd, SParsedDataColInfo* pColInfo, SSchema* pSchema,
static int32_t parseBoundColumns(SInsertStatementParam *pInsertParam, SParsedDataColInfo* pColInfo, SSchema* pSchema,
char* str, char **end) {
pColInfo->numOfBound = 0;
......@@ -1043,7 +1044,7 @@ static int32_t parseBoundColumns(SSqlCmd* pCmd, SParsedDataColInfo* pColInfo, SS
str += index;
if (sToken.type != TK_LP) {
code = tscInvalidSQLErrMsg(pCmd->payload, "( is expected", sToken.z);
code = tscSQLSyntaxErrMsg(pInsertParam->msg, "( is expected", sToken.z);
goto _clean;
}
......@@ -1070,7 +1071,7 @@ static int32_t parseBoundColumns(SSqlCmd* pCmd, SParsedDataColInfo* pColInfo, SS
for (int32_t t = 0; t < pColInfo->numOfCols; ++t) {
if (strncmp(sToken.z, pSchema[t].name, sToken.n) == 0 && strlen(pSchema[t].name) == sToken.n) {
if (pColInfo->cols[t].hasVal == true) {
code = tscInvalidSQLErrMsg(pCmd->payload, "duplicated column name", sToken.z);
code = tscInvalidOperationMsg(pInsertParam->msg, "duplicated column name", sToken.z);
goto _clean;
}
......@@ -1083,7 +1084,7 @@ static int32_t parseBoundColumns(SSqlCmd* pCmd, SParsedDataColInfo* pColInfo, SS
}
if (!findColumnIndex) {
code = tscInvalidSQLErrMsg(pCmd->payload, "invalid column/tag name", sToken.z);
code = tscInvalidOperationMsg(pInsertParam->msg, "invalid column/tag name", sToken.z);
goto _clean;
}
}
......@@ -1092,10 +1093,25 @@ static int32_t parseBoundColumns(SSqlCmd* pCmd, SParsedDataColInfo* pColInfo, SS
return TSDB_CODE_SUCCESS;
_clean:
pCmd->curSql = NULL;
pInsertParam->sql = NULL;
return code;
}
static int32_t getFileFullPath(SStrToken* pToken, char* output) {
char path[PATH_MAX] = {0};
strncpy(path, pToken->z, pToken->n);
strdequote(path);
wordexp_t full_path;
if (wordexp(path, &full_path, 0) != 0) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
tstrncpy(output, full_path.we_wordv[0], PATH_MAX);
wordfree(&full_path);
return TSDB_CODE_SUCCESS;
}
/**
* parse insert sql
* @param pSql
......@@ -1103,7 +1119,9 @@ static int32_t parseBoundColumns(SSqlCmd* pCmd, SParsedDataColInfo* pColInfo, SS
*/
int tsParseInsertSql(SSqlObj *pSql) {
SSqlCmd *pCmd = &pSql->cmd;
char* str = pCmd->curSql;
SInsertStatementParam* pInsertParam = &pCmd->insertParam;
char* str = pInsertParam->sql;
int32_t totalNum = 0;
int32_t code = TSDB_CODE_SUCCESS;
......@@ -1118,21 +1136,17 @@ int tsParseInsertSql(SSqlObj *pSql) {
return code;
}
if ((code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) != TSDB_CODE_SUCCESS) {
return code;
}
if (NULL == pCmd->insertParam.pTableBlockHashList) {
pCmd->insertParam.pTableBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
if (NULL == pCmd->insertParam.pTableBlockHashList) {
if (NULL == pInsertParam->pTableBlockHashList) {
pInsertParam->pTableBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
if (NULL == pInsertParam->pTableBlockHashList) {
code = TSDB_CODE_TSC_OUT_OF_MEMORY;
goto _clean;
}
} else {
str = pCmd->curSql;
str = pInsertParam->sql;
}
tscDebug("0x%"PRIx64" create data block list hashList:%p", pSql->self, pCmd->insertParam.pTableBlockHashList);
tscDebug("0x%"PRIx64" create data block list hashList:%p", pSql->self, pInsertParam->pTableBlockHashList);
while (1) {
int32_t index = 0;
......@@ -1144,7 +1158,7 @@ int tsParseInsertSql(SSqlObj *pSql) {
* if the data is from the data file, no data has been generated yet. So, there no data to
* merge or submit, save the file path and parse the file in other routines.
*/
if (TSDB_QUERY_HAS_TYPE(pCmd->insertParam.insertType, TSDB_QUERY_TYPE_FILE_INSERT)) {
if (TSDB_QUERY_HAS_TYPE(pInsertParam->insertType, TSDB_QUERY_TYPE_FILE_INSERT)) {
goto _clean;
}
......@@ -1160,13 +1174,13 @@ int tsParseInsertSql(SSqlObj *pSql) {
}
}
pCmd->curSql = sToken.z;
pInsertParam->sql = sToken.z;
char buf[TSDB_TABLE_FNAME_LEN];
SStrToken sTblToken;
sTblToken.z = buf;
// Check if the table name available or not
if (validateTableName(sToken.z, sToken.n, &sTblToken) != TSDB_CODE_SUCCESS) {
code = tscInvalidSQLErrMsg(pCmd->payload, "table name invalid", sToken.z);
code = tscInvalidOperationMsg(pInsertParam->msg, "table name invalid", sToken.z);
goto _clean;
}
......@@ -1185,12 +1199,12 @@ int tsParseInsertSql(SSqlObj *pSql) {
}
tscError("0x%"PRIx64" async insert parse error, code:%s", pSql->self, tstrerror(code));
pCmd->curSql = NULL;
pInsertParam->sql = NULL;
goto _clean;
}
if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
code = tscInvalidSQLErrMsg(pCmd->payload, "insert data into super table is not supported", NULL);
code = tscInvalidOperationMsg(pInsertParam->msg, "insert data into super table is not supported", NULL);
goto _clean;
}
......@@ -1199,58 +1213,49 @@ int tsParseInsertSql(SSqlObj *pSql) {
str += index;
if (sToken.n == 0 || (sToken.type != TK_FILE && sToken.type != TK_VALUES)) {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES or FILE required", sToken.z);
code = tscSQLSyntaxErrMsg(pInsertParam->msg, "keyword VALUES or FILE required", sToken.z);
goto _clean;
}
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
if (sToken.type == TK_FILE) {
if (validateDataSource(pCmd, TSDB_QUERY_TYPE_FILE_INSERT, sToken.z) != TSDB_CODE_SUCCESS) {
if (validateDataSource(pInsertParam, TSDB_QUERY_TYPE_FILE_INSERT, sToken.z) != TSDB_CODE_SUCCESS) {
goto _clean;
}
index = 0;
sToken = tStrGetToken(str, &index, false);
if (sToken.type != TK_STRING && sToken.type != TK_ID) {
code = tscInvalidSQLErrMsg(pCmd->payload, "file path is required following keyword FILE", sToken.z);
code = tscSQLSyntaxErrMsg(pInsertParam->msg, "file path is required following keyword FILE", sToken.z);
goto _clean;
}
str += index;
if (sToken.n == 0) {
code = tscInvalidSQLErrMsg(pCmd->payload, "file path is required following keyword FILE", sToken.z);
code = tscSQLSyntaxErrMsg(pInsertParam->msg, "file path is required following keyword FILE", sToken.z);
goto _clean;
}
strncpy(pCmd->payload, sToken.z, sToken.n);
strdequote(pCmd->payload);
// todo refactor extract method
wordexp_t full_path;
if (wordexp(pCmd->payload, &full_path, 0) != 0) {
code = tscInvalidSQLErrMsg(pCmd->payload, "invalid filename", sToken.z);
code = getFileFullPath(&sToken, pCmd->payload);
if (code != TSDB_CODE_SUCCESS) {
tscInvalidOperationMsg(pInsertParam->msg, "invalid filename", sToken.z);
goto _clean;
}
tstrncpy(pCmd->payload, full_path.we_wordv[0], pCmd->allocSize);
wordfree(&full_path);
} else {
if (bindedColumns == NULL) {
STableMeta *pTableMeta = pTableMetaInfo->pTableMeta;
if (validateDataSource(pCmd, TSDB_QUERY_TYPE_INSERT, sToken.z) != TSDB_CODE_SUCCESS) {
if (validateDataSource(pInsertParam, TSDB_QUERY_TYPE_INSERT, sToken.z) != TSDB_CODE_SUCCESS) {
goto _clean;
}
STableDataBlocks *dataBuf = NULL;
int32_t ret = tscGetDataBlockFromList(pCmd->insertParam.pTableBlockHashList, pTableMeta->id.uid, TSDB_DEFAULT_PAYLOAD_SIZE,
int32_t ret = tscGetDataBlockFromList(pInsertParam->pTableBlockHashList, pTableMeta->id.uid, TSDB_DEFAULT_PAYLOAD_SIZE,
sizeof(SSubmitBlk), tinfo.rowSize, &pTableMetaInfo->name, pTableMeta,
&dataBuf, NULL);
if (ret != TSDB_CODE_SUCCESS) {
goto _clean;
}
code = doParseInsertStatement(pCmd, &str, dataBuf, &totalNum);
code = doParseInsertStatement(pInsertParam, &str, dataBuf, &totalNum);
if (code != TSDB_CODE_SUCCESS) {
goto _clean;
}
......@@ -1258,12 +1263,12 @@ int tsParseInsertSql(SSqlObj *pSql) {
// insert into tablename(col1, col2,..., coln) values(v1, v2,... vn);
STableMeta *pTableMeta = tscGetTableMetaInfoFromCmd(pCmd, 0)->pTableMeta;
if (validateDataSource(pCmd, TSDB_QUERY_TYPE_INSERT, sToken.z) != TSDB_CODE_SUCCESS) {
if (validateDataSource(pInsertParam, TSDB_QUERY_TYPE_INSERT, sToken.z) != TSDB_CODE_SUCCESS) {
goto _clean;
}
STableDataBlocks *dataBuf = NULL;
int32_t ret = tscGetDataBlockFromList(pCmd->insertParam.pTableBlockHashList, pTableMeta->id.uid, TSDB_DEFAULT_PAYLOAD_SIZE,
int32_t ret = tscGetDataBlockFromList(pInsertParam->pTableBlockHashList, pTableMeta->id.uid, TSDB_DEFAULT_PAYLOAD_SIZE,
sizeof(SSubmitBlk), tinfo.rowSize, &pTableMetaInfo->name, pTableMeta,
&dataBuf, NULL);
if (ret != TSDB_CODE_SUCCESS) {
......@@ -1271,22 +1276,22 @@ int tsParseInsertSql(SSqlObj *pSql) {
}
SSchema *pSchema = tscGetTableSchema(pTableMeta);
code = parseBoundColumns(pCmd, &dataBuf->boundColumnInfo, pSchema, bindedColumns, NULL);
code = parseBoundColumns(pInsertParam, &dataBuf->boundColumnInfo, pSchema, bindedColumns, NULL);
if (code != TSDB_CODE_SUCCESS) {
goto _clean;
}
if (dataBuf->boundColumnInfo.cols[0].hasVal == false) {
code = tscInvalidSQLErrMsg(pCmd->payload, "primary timestamp column can not be null", NULL);
code = tscInvalidOperationMsg(pInsertParam->msg, "primary timestamp column can not be null", NULL);
goto _clean;
}
if (sToken.type != TK_VALUES) {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES is expected", sToken.z);
code = tscSQLSyntaxErrMsg(pInsertParam->msg, "keyword VALUES is expected", sToken.z);
goto _clean;
}
code = doParseInsertStatement(pCmd, &str, dataBuf, &totalNum);
code = doParseInsertStatement(pInsertParam, &str, dataBuf, &totalNum);
if (code != TSDB_CODE_SUCCESS) {
goto _clean;
}
......@@ -1295,13 +1300,13 @@ int tsParseInsertSql(SSqlObj *pSql) {
}
// we need to keep the data blocks if there are parameters in the sql
if (pCmd->numOfParams > 0) {
if (pInsertParam->numOfParams > 0) {
goto _clean;
}
// merge according to vgId
if (!TSDB_QUERY_HAS_TYPE(pCmd->insertParam.insertType, TSDB_QUERY_TYPE_STMT_INSERT) && taosHashGetSize(pCmd->insertParam.pTableBlockHashList) > 0) {
if ((code = tscMergeTableDataBlocks(pSql, true)) != TSDB_CODE_SUCCESS) {
if (!TSDB_QUERY_HAS_TYPE(pInsertParam->insertType, TSDB_QUERY_TYPE_STMT_INSERT) && taosHashGetSize(pInsertParam->pTableBlockHashList) > 0) {
if ((code = tscMergeTableDataBlocks(pInsertParam, true)) != TSDB_CODE_SUCCESS) {
goto _clean;
}
}
......@@ -1310,7 +1315,7 @@ int tsParseInsertSql(SSqlObj *pSql) {
goto _clean;
_clean:
pCmd->curSql = NULL;
pInsertParam->sql = NULL;
return code;
}
......@@ -1327,16 +1332,17 @@ int tsInsertInitialCheck(SSqlObj *pSql) {
pCmd->count = 0;
pCmd->command = TSDB_SQL_INSERT;
SInsertStatementParam* pInsertParam = &pCmd->insertParam;
SQueryInfo *pQueryInfo = tscGetQueryInfoS(pCmd);
TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_INSERT);
sToken = tStrGetToken(pSql->sqlstr, &index, false);
if (sToken.type != TK_INTO) {
return tscInvalidSQLErrMsg(pCmd->payload, "keyword INTO is expected", sToken.z);
return tscSQLSyntaxErrMsg(pInsertParam->msg, "keyword INTO is expected", sToken.z);
}
pCmd->curSql = sToken.z + sToken.n;
pInsertParam->sql = sToken.z + sToken.n;
return TSDB_CODE_SUCCESS;
}
......@@ -1345,7 +1351,7 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
SSqlCmd* pCmd = &pSql->cmd;
if (!initial) {
tscDebug("0x%"PRIx64" resume to parse sql: %s", pSql->self, pCmd->curSql);
tscDebug("0x%"PRIx64" resume to parse sql: %s", pSql->self, pCmd->insertParam.sql);
}
ret = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
......@@ -1355,12 +1361,11 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
if (tscIsInsertData(pSql->sqlstr)) {
if (initial && ((ret = tsInsertInitialCheck(pSql)) != TSDB_CODE_SUCCESS)) {
strncpy(pCmd->payload, pCmd->insertParam.msg, TSDB_DEFAULT_PAYLOAD_SIZE);
return ret;
}
ret = tsParseInsertSql(pSql);
assert(ret == TSDB_CODE_SUCCESS || ret == TSDB_CODE_TSC_ACTION_IN_PROGRESS || ret == TSDB_CODE_TSC_SQL_SYNTAX_ERROR || ret == TSDB_CODE_TSC_INVALID_OPERATION);
if (pSql->parseRetry < 1 && (ret == TSDB_CODE_TSC_SQL_SYNTAX_ERROR || ret == TSDB_CODE_TSC_INVALID_OPERATION)) {
tscDebug("0x%"PRIx64 " parse insert sql statement failed, code:%s, clear meta cache and retry ", pSql->self, tstrerror(ret));
......@@ -1371,6 +1376,10 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
ret = tsParseInsertSql(pSql);
}
}
if (ret != TSDB_CODE_SUCCESS) {
strncpy(pCmd->payload, pCmd->insertParam.msg, TSDB_DEFAULT_PAYLOAD_SIZE);
}
} else {
SSqlInfo sqlInfo = qSqlParse(pSql->sqlstr);
ret = tscValidateSqlInfo(pSql, &sqlInfo);
......@@ -1395,29 +1404,25 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
return ret;
}
static int doPackSendDataBlock(SSqlObj *pSql, int32_t numOfRows, STableDataBlocks *pTableDataBlocks) {
static int doPackSendDataBlock(SSqlObj* pSql, SInsertStatementParam *pInsertParam, STableMeta* pTableMeta, int32_t numOfRows, STableDataBlocks *pTableDataBlocks) {
int32_t code = TSDB_CODE_SUCCESS;
SSqlCmd *pCmd = &pSql->cmd;
pSql->res.numOfRows = 0;
STableMeta *pTableMeta = tscGetTableMetaInfoFromCmd(pCmd, 0)->pTableMeta;
SSubmitBlk *pBlocks = (SSubmitBlk *)(pTableDataBlocks->pData);
code = tsSetBlockInfo(pBlocks, pTableMeta, numOfRows);
if (code != TSDB_CODE_SUCCESS) {
return tscInvalidSQLErrMsg(pCmd->payload, "too many rows in sql, total number of rows should be less than 32767", NULL);
return tscInvalidOperationMsg(pInsertParam->msg, "too many rows in sql, total number of rows should be less than 32767", NULL);
}
if ((code = tscMergeTableDataBlocks(pSql, true)) != TSDB_CODE_SUCCESS) {
if ((code = tscMergeTableDataBlocks(pInsertParam, true)) != TSDB_CODE_SUCCESS) {
return code;
}
STableDataBlocks *pDataBlock = taosArrayGetP(pCmd->insertParam.pDataBlocks, 0);
STableDataBlocks *pDataBlock = taosArrayGetP(pInsertParam->pDataBlocks, 0);
if ((code = tscCopyDataBlockToPayload(pSql, pDataBlock)) != TSDB_CODE_SUCCESS) {
return code;
}
return tscBuildAndSendRequest(pSql, NULL);
return TSDB_CODE_SUCCESS;
}
typedef struct SImportFileSupport {
......@@ -1466,13 +1471,14 @@ static void parseFileSendDataBlock(void *param, TAOS_RES *tres, int32_t numOfRow
STableMeta * pTableMeta = pTableMetaInfo->pTableMeta;
STableComInfo tinfo = tscGetTableInfo(pTableMeta);
destroyTableNameList(pCmd);
SInsertStatementParam* pInsertParam = &pCmd->insertParam;
destroyTableNameList(pInsertParam);
pCmd->insertParam.pDataBlocks = tscDestroyBlockArrayList(pCmd->insertParam.pDataBlocks);
pInsertParam->pDataBlocks = tscDestroyBlockArrayList(pInsertParam->pDataBlocks);
if (pCmd->insertParam.pTableBlockHashList == NULL) {
pCmd->insertParam.pTableBlockHashList = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
if (pCmd->insertParam.pTableBlockHashList == NULL) {
if (pInsertParam->pTableBlockHashList == NULL) {
pInsertParam->pTableBlockHashList = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
if (pInsertParam->pTableBlockHashList == NULL) {
code = TSDB_CODE_TSC_OUT_OF_MEMORY;
goto _error;
}
......@@ -1480,7 +1486,7 @@ static void parseFileSendDataBlock(void *param, TAOS_RES *tres, int32_t numOfRow
STableDataBlocks *pTableDataBlock = NULL;
int32_t ret =
tscGetDataBlockFromList(pCmd->insertParam.pTableBlockHashList, pTableMeta->id.uid, TSDB_PAYLOAD_SIZE, sizeof(SSubmitBlk),
tscGetDataBlockFromList(pInsertParam->pTableBlockHashList, pTableMeta->id.uid, TSDB_PAYLOAD_SIZE, sizeof(SSubmitBlk),
tinfo.rowSize, &pTableMetaInfo->name, pTableMeta, &pTableDataBlock, NULL);
if (ret != TSDB_CODE_SUCCESS) {
pParentSql->res.code = TSDB_CODE_TSC_OUT_OF_MEMORY;
......@@ -1507,7 +1513,7 @@ static void parseFileSendDataBlock(void *param, TAOS_RES *tres, int32_t numOfRow
strtolower(line, line);
int32_t len = 0;
code = tsParseOneRow(&lineptr, pTableDataBlock, pCmd, tinfo.precision, &len, tokenBuf);
code = tsParseOneRow(&lineptr, pTableDataBlock, tinfo.precision, &len, tokenBuf, pInsertParam);
if (code != TSDB_CODE_SUCCESS || pTableDataBlock->numOfParams > 0) {
pSql->res.code = code;
break;
......@@ -1526,12 +1532,14 @@ static void parseFileSendDataBlock(void *param, TAOS_RES *tres, int32_t numOfRow
pParentSql->res.code = code;
if (code == TSDB_CODE_SUCCESS) {
if (count > 0) {
code = doPackSendDataBlock(pSql, count, pTableDataBlock);
if (code == TSDB_CODE_SUCCESS) {
return;
} else {
pSql->res.numOfRows = 0;
code = doPackSendDataBlock(pSql, pInsertParam, pTableMeta, count, pTableDataBlock);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
tscBuildAndSendRequest(pSql, NULL);
return;
} else {
taos_free_result(pSql);
tfree(pSupporter);
......@@ -1562,7 +1570,8 @@ void tscImportDataFromFile(SSqlObj *pSql) {
return;
}
assert(TSDB_QUERY_HAS_TYPE(pCmd->insertParam.insertType, TSDB_QUERY_TYPE_FILE_INSERT) && strlen(pCmd->payload) != 0);
SInsertStatementParam* pInsertParam = &pCmd->insertParam;
assert(TSDB_QUERY_HAS_TYPE(pInsertParam->insertType, TSDB_QUERY_TYPE_FILE_INSERT) && strlen(pCmd->payload) != 0);
pCmd->active = pCmd->pQueryInfo;
SImportFileSupport *pSupporter = calloc(1, sizeof(SImportFileSupport));
......
src/client/src/tscPrepare.c
浏览文件 @ d2886dee
......@@ -1085,7 +1085,7 @@ static int insertStmtExecute(STscStmt* stmt) {
fillTablesColumnsNull(stmt->pSql);
int code = tscMergeTableDataBlocks(stmt->pSql, false);
int code = tscMergeTableDataBlocks(&stmt->pSql->cmd.insertParam, false);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
......@@ -1185,7 +1185,7 @@ static int insertBatchStmtExecute(STscStmt* pStmt) {
fillTablesColumnsNull(pStmt->pSql);
if ((code = tscMergeTableDataBlocks(pStmt->pSql, false)) != TSDB_CODE_SUCCESS) {
if ((code = tscMergeTableDataBlocks(&pStmt->pSql->cmd.insertParam, false)) != TSDB_CODE_SUCCESS) {
return code;
}
......@@ -1213,7 +1213,7 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
}
int32_t index = 0;
SStrToken sToken = tStrGetToken(pCmd->curSql, &index, false);
SStrToken sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n == 0) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
......@@ -1232,7 +1232,7 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
pStmt->mtb.tagSet = true;
sToken = tStrGetToken(pCmd->curSql, &index, false);
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n > 0 && sToken.type == TK_VALUES) {
return TSDB_CODE_SUCCESS;
}
......@@ -1241,18 +1241,18 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
sToken = tStrGetToken(pCmd->curSql, &index, false);
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n <= 0 || ((sToken.type != TK_ID) && (sToken.type != TK_STRING))) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
pStmt->mtb.stbname = sToken;
sToken = tStrGetToken(pCmd->curSql, &index, false);
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n <= 0 || sToken.type != TK_TAGS) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
sToken = tStrGetToken(pCmd->curSql, &index, false);
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n <= 0 || sToken.type != TK_LP) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
......@@ -1262,7 +1262,7 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
int32_t loopCont = 1;
while (loopCont) {
sToken = tStrGetToken(pCmd->curSql, &index, false);
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n <= 0) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
......@@ -1285,7 +1285,7 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
sToken = tStrGetToken(pCmd->curSql, &index, false);
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n <= 0 || (sToken.type != TK_VALUES && sToken.type != TK_LP)) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
......@@ -1468,7 +1468,7 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
if (tscIsInsertData(pSql->sqlstr)) {
pStmt->isInsert = true;
pSql->cmd.numOfParams = 0;
pSql->cmd.insertParam.numOfParams = 0;
pSql->cmd.batchSize = 0;
registerSqlObj(pSql);
......@@ -1565,7 +1565,7 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
tscDebug("0x%"PRIx64" SQL: %s", pSql->self, pSql->sqlstr);
pSql->cmd.numOfParams = 0;
pSql->cmd.insertParam.numOfParams = 0;
pSql->cmd.batchSize = 0;
if (taosHashGetSize(pCmd->insertParam.pTableBlockHashList) > 0) {
......@@ -1852,7 +1852,7 @@ int taos_stmt_num_params(TAOS_STMT *stmt, int *nums) {
if (pStmt->isInsert) {
SSqlObj* pSql = pStmt->pSql;
SSqlCmd *pCmd = &pSql->cmd;
*nums = pCmd->numOfParams;
*nums = pCmd->insertParam.numOfParams;
return TSDB_CODE_SUCCESS;
} else {
SNormalStmt* normal = &pStmt->normal;
......
src/client/src/tscSQLParser.c
浏览文件 @ d2886dee
......@@ -28,7 +28,7 @@
#include "tname.h"
#include "tscLog.h"
#include "tscUtil.h"
#include "tschemautil.h"
#include "qTableMeta.h"
#include "tsclient.h"
#include "tstrbuild.h"
#include "ttoken.h"
......@@ -195,7 +195,7 @@ static bool validateDebugFlag(int32_t v) {
* is not needed in the final error message.
*/
static int32_t invalidOperationMsg(char* dstBuffer, const char* errMsg) {
return tscInvalidSQLErrMsg(dstBuffer, errMsg, NULL);
return tscInvalidOperationMsg(dstBuffer, errMsg, NULL);
}
static int setColumnFilterInfoForTimestamp(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tVariant* pVar) {
......@@ -6603,7 +6603,7 @@ int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo) {
if (!findColumnIndex) {
tdDestroyKVRowBuilder(&kvRowBuilder);
return tscInvalidSQLErrMsg(pCmd->payload, "invalid tag name", sToken->z);
return tscInvalidOperationMsg(pCmd->payload, "invalid tag name", sToken->z);
}
}
} else {
......@@ -7456,6 +7456,7 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf
return TSDB_CODE_TSC_INVALID_OPERATION;
}
// validate the query filter condition info
if (pSqlNode->pWhere != NULL) {
if (validateWhereNode(pQueryInfo, &pSqlNode->pWhere, pSql) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_TSC_INVALID_OPERATION;
......@@ -7467,6 +7468,16 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf
pQueryInfo->window.ekey = pQueryInfo->window.ekey / 1000;
}
}
// validate the interval info
if (validateIntervalNode(pSql, pQueryInfo, pSqlNode) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_TSC_INVALID_OPERATION;
} else {
if (isTimeWindowQuery(pQueryInfo) &&
(validateFunctionsInIntervalOrGroupbyQuery(pCmd, pQueryInfo) != TSDB_CODE_SUCCESS)) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
}
} else {
pQueryInfo->command = TSDB_SQL_SELECT;
......@@ -7612,6 +7623,7 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf
pQueryInfo->groupbyColumn = tscGroupbyColumn(pQueryInfo);
pQueryInfo->arithmeticOnAgg = tsIsArithmeticQueryOnAggResult(pQueryInfo);
pQueryInfo->orderProjectQuery = tscOrderedProjectionQueryOnSTable(pQueryInfo, 0);
SExprInfo** p = NULL;
int32_t numOfExpr = 0;
......
src/client/src/tscServer.c
浏览文件 @ d2886dee
......@@ -15,15 +15,15 @@
#include "os.h"
#include "tcmdtype.h"
#include "tlockfree.h"
#include "trpc.h"
#include "tscLocalMerge.h"
#include "tscGlobalmerge.h"
#include "tscLog.h"
#include "tscProfile.h"
#include "tscUtil.h"
#include "tschemautil.h"
#include "qTableMeta.h"
#include "tsclient.h"
#include "ttimer.h"
#include "tlockfree.h"
#include "qPlan.h"
int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo) = {0};
......@@ -1598,7 +1598,7 @@ int tscProcessRetrieveLocalMergeRsp(SSqlObj *pSql) {
return code;
}
if (pRes->pLocalMerger == NULL) { // no result from subquery, so abort here directly.
if (pRes->pMerger == NULL) { // no result from subquery, so abort here directly.
(*pSql->fp)(pSql->param, pSql, pRes->numOfRows);
return code;
}
......@@ -1615,15 +1615,7 @@ int tscProcessRetrieveLocalMergeRsp(SSqlObj *pSql) {
taosArrayPush(group, &tableKeyInfo);
taosArrayPush(tableGroupInfo.pGroupList, &group);
// todo remove it
SExprInfo* list = calloc(tscNumOfExprs(pQueryInfo), sizeof(SExprInfo));
for(int32_t i = 0; i < tscNumOfExprs(pQueryInfo); ++i) {
SExprInfo* pExprInfo = tscExprGet(pQueryInfo, i);
list[i] = *pExprInfo;
}
pQueryInfo->pQInfo = createQInfoFromQueryNode(pQueryInfo, list, &tableGroupInfo, NULL, NULL, pRes->pLocalMerger, MERGE_STAGE);
tfree(list);
pQueryInfo->pQInfo = createQInfoFromQueryNode(pQueryInfo, &tableGroupInfo, NULL, NULL, pRes->pMerger, MERGE_STAGE);
}
uint64_t localQueryId = 0;
......@@ -2402,8 +2394,8 @@ static int32_t getTableMetaFromMnode(SSqlObj *pSql, STableMetaInfo *pTableMetaIn
char *pMsg = (char *)pInfoMsg + sizeof(STableInfoMsg);
// tag data exists
if (autocreate && pSql->cmd.tagData.dataLen != 0) {
pMsg = serializeTagData(&pSql->cmd.tagData, pMsg);
if (autocreate && pSql->cmd.insertParam.tagData.dataLen != 0) {
pMsg = serializeTagData(&pSql->cmd.insertParam.tagData, pMsg);
}
pNew->cmd.payloadLen = (int32_t)(pMsg - (char*)pInfoMsg);
......
src/client/src/tscStream.c
浏览文件 @ d2886dee
......@@ -13,7 +13,6 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <tschemautil.h>
#include "os.h"
#include "taosmsg.h"
#include "tscLog.h"
......
src/client/src/tscSubquery.c
浏览文件 @ d2886dee
......@@ -21,7 +21,7 @@
#include "tcompare.h"
#include "tscLog.h"
#include "tscSubquery.h"
#include "tschemautil.h"
#include "qTableMeta.h"
#include "tsclient.h"
#include "qUtil.h"
#include "qPlan.h"
......@@ -2446,7 +2446,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
assert(pState->numOfSub > 0);
int32_t ret = tscLocalReducerEnvCreate(pQueryInfo, &pMemoryBuf, pSql->subState.numOfSub, &pDesc, nBufferSize, pSql->self);
int32_t ret = tscCreateGlobalMergerEnv(pQueryInfo, &pMemoryBuf, pSql->subState.numOfSub, &pDesc, nBufferSize, pSql->self);
if (ret != 0) {
pRes->code = ret;
tscAsyncResultOnError(pSql);
......@@ -2459,7 +2459,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
if (pSql->pSubs == NULL) {
tfree(pSql->pSubs);
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
tscLocalReducerEnvDestroy(pMemoryBuf, pDesc,pState->numOfSub);
tscDestroyGlobalMergerEnv(pMemoryBuf, pDesc,pState->numOfSub);
tscAsyncResultOnError(pSql);
return ret;
......@@ -2526,13 +2526,13 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
tscError("0x%"PRIx64" failed to prepare subquery structure and launch subqueries", pSql->self);
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
tscLocalReducerEnvDestroy(pMemoryBuf, pDesc, pState->numOfSub);
tscDestroyGlobalMergerEnv(pMemoryBuf, pDesc, pState->numOfSub);
doCleanupSubqueries(pSql, i);
return pRes->code; // free all allocated resource
}
if (pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED) {
tscLocalReducerEnvDestroy(pMemoryBuf, pDesc, pState->numOfSub);
tscDestroyGlobalMergerEnv(pMemoryBuf, pDesc, pState->numOfSub);
doCleanupSubqueries(pSql, i);
return pRes->code;
}
......@@ -2697,7 +2697,7 @@ void tscHandleSubqueryError(SRetrieveSupport *trsupport, SSqlObj *pSql, int numO
tstrerror(pParentSql->res.code));
// release allocated resource
tscLocalReducerEnvDestroy(trsupport->pExtMemBuffer, trsupport->pOrderDescriptor,
tscDestroyGlobalMergerEnv(trsupport->pExtMemBuffer, trsupport->pOrderDescriptor,
pState->numOfSub);
tscFreeRetrieveSup(pSql);
......@@ -2772,7 +2772,7 @@ static void tscAllDataRetrievedFromDnode(SRetrieveSupport *trsupport, SSqlObj* p
SQueryInfo *pPQueryInfo = tscGetQueryInfo(&pParentSql->cmd);
tscClearInterpInfo(pPQueryInfo);
code = tscCreateLocalMerger(trsupport->pExtMemBuffer, pState->numOfSub, pDesc, pPQueryInfo, &pParentSql->res.pLocalMerger, pParentSql->self);
code = tscCreateGlobalMerger(trsupport->pExtMemBuffer, pState->numOfSub, pDesc, pPQueryInfo, &pParentSql->res.pMerger, pParentSql->self);
pParentSql->res.code = code;
if (code == TSDB_CODE_SUCCESS && trsupport->pExtMemBuffer == NULL) {
......@@ -3499,9 +3499,8 @@ static UNUSED_FUNC bool tscHasRemainDataInSubqueryResultSet(SSqlObj *pSql) {
return hasData;
}
// todo remove pExprs
void* createQInfoFromQueryNode(SQueryInfo* pQueryInfo, SExprInfo* pExprs, STableGroupInfo* pTableGroupInfo,
SOperatorInfo* pSourceOperator, char* sql, void* merger, int32_t stage) {
void* createQInfoFromQueryNode(SQueryInfo* pQueryInfo, STableGroupInfo* pTableGroupInfo, SOperatorInfo* pSourceOperator,
char* sql, void* merger, int32_t stage) {
assert(pQueryInfo != NULL);
SQInfo *pQInfo = (SQInfo *)calloc(1, sizeof(SQInfo));
if (pQInfo == NULL) {
......
src/client/src/tscUtil.c
浏览文件 @ d2886dee
......@@ -14,18 +14,18 @@
*/
#include "tscUtil.h"
#include "tsched.h"
#include "hash.h"
#include "os.h"
#include "taosmsg.h"
#include "texpr.h"
#include "tkey.h"
#include "tmd5.h"
#include "tscLocalMerge.h"
#include "tscGlobalmerge.h"
#include "tscLog.h"
#include "tscProfile.h"
#include "tscSubquery.h"
#include "tschemautil.h"
#include "tsched.h"
#include "qTableMeta.h"
#include "tsclient.h"
#include "ttimer.h"
#include "ttokendef.h"
......@@ -708,6 +708,7 @@ static SColumnInfo* extractColumnInfoFromResult(SArray* pTableCols) {
typedef struct SDummyInputInfo {
SSDataBlock *block;
STableQueryInfo *pTableQueryInfo;
SSqlObj *pSql; // refactor: remove it
int32_t numOfFilterCols;
SSingleColumnFilterInfo *pFilterInfo;
......@@ -726,9 +727,10 @@ typedef struct SJoinOperatorInfo {
SRspResultInfo resultInfo; // todo refactor, add this info for each operator
} SJoinOperatorInfo;
static void doSetupSDataBlock(SSqlRes* pRes, SSDataBlock* pBlock) {
static void doSetupSDataBlock(SSqlRes* pRes, SSDataBlock* pBlock, SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols) {
int32_t offset = 0;
char* pData = pRes->data;
for(int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
if (pData != NULL) {
......@@ -740,6 +742,26 @@ static void doSetupSDataBlock(SSqlRes* pRes, SSDataBlock* pBlock) {
offset += pColData->info.bytes;
}
// filter data if needed
if (numOfFilterCols > 0) {
doSetFilterColumnInfo(pFilterInfo, numOfFilterCols, pBlock);
int8_t* p = calloc(pBlock->info.rows, sizeof(int8_t));
bool all = doFilterDataBlock(pFilterInfo, numOfFilterCols, pBlock->info.rows, p);
if (!all) {
doCompactSDataBlock(pBlock, pBlock->info.rows, p);
}
tfree(p);
}
// todo refactor: extract method
// set the timestamp range of current result data block
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, 0);
if (pColData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
pBlock->info.window.skey = ((int64_t*)pColData->pData)[0];
pBlock->info.window.ekey = ((int64_t*)pColData->pData)[pBlock->info.rows-1];
}
pRes->numOfRows = 0;
}
......@@ -755,22 +777,11 @@ SSDataBlock* doGetDataBlock(void* param, bool* newgroup) {
SSqlRes* pRes = &pSql->res;
SSDataBlock* pBlock = pInput->block;
pOperator->pRuntimeEnv->current = pInput->pTableQueryInfo;
pBlock->info.rows = pRes->numOfRows;
if (pRes->numOfRows != 0) {
doSetupSDataBlock(pRes, pBlock);
if (pInput->numOfFilterCols > 0) {
doSetFilterColumnInfo(pInput->pFilterInfo, pInput->numOfFilterCols, pBlock);
int8_t* p = calloc(pBlock->info.rows, sizeof(int8_t));
bool all = doFilterDataBlock(pInput->pFilterInfo, pInput->numOfFilterCols, pBlock->info.rows, p);
if (!all) {
doCompactSDataBlock(pBlock, pBlock->info.rows, p);
}
tfree(p);
}
doSetupSDataBlock(pRes, pBlock, pInput->pFilterInfo, pInput->numOfFilterCols);
*newgroup = false;
return pBlock;
}
......@@ -785,7 +796,7 @@ SSDataBlock* doGetDataBlock(void* param, bool* newgroup) {
}
pBlock->info.rows = pRes->numOfRows;
doSetupSDataBlock(pRes, pBlock);
doSetupSDataBlock(pRes, pBlock, pInput->pFilterInfo, pInput->numOfFilterCols);
*newgroup = false;
return pBlock;
}
......@@ -942,11 +953,14 @@ static void destroyDummyInputOperator(void* param, int32_t numOfOutput) {
// todo this operator servers as the adapter for Operator tree and SqlRes result, remove it later
SOperatorInfo* createDummyInputOperator(SSqlObj* pSql, SSchema* pSchema, int32_t numOfCols, SSingleColumnFilterInfo* pFilterInfo, int32_t numOfFilterCols) {
assert(numOfCols > 0);
STimeWindow win = {.skey = INT64_MIN, .ekey = INT64_MAX};
SDummyInputInfo* pInfo = calloc(1, sizeof(SDummyInputInfo));
pInfo->pSql = pSql;
pInfo->pFilterInfo = pFilterInfo;
pInfo->numOfFilterCols = numOfFilterCols;
pInfo->pTableQueryInfo = createTmpTableQueryInfo(win);
pInfo->block = calloc(numOfCols, sizeof(SSDataBlock));
pInfo->block->info.numOfCols = numOfCols;
......@@ -1032,20 +1046,31 @@ void convertQueryResult(SSqlRes* pRes, SQueryInfo* pQueryInfo) {
pRes->completed = (pRes->numOfRows == 0);
}
static void createInputDataFilterInfo(SQueryInfo* px, int32_t numOfCol1, int32_t* numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo) {
SColumnInfo* tableCols = calloc(numOfCol1, sizeof(SColumnInfo));
for(int32_t i = 0; i < numOfCol1; ++i) {
SColumn* pCol = taosArrayGetP(px->colList, i);
if (pCol->info.flist.numOfFilters > 0) {
(*numOfFilterCols) += 1;
}
tableCols[i] = pCol->info;
}
if ((*numOfFilterCols) > 0) {
doCreateFilterInfo(tableCols, numOfCol1, (*numOfFilterCols), pFilterInfo, 0);
}
tfree(tableCols);
}
void handleDownstreamOperator(SSqlObj** pSqlObjList, int32_t numOfUpstream, SQueryInfo* px, SSqlRes* pOutput) {
// handle the following query process
if (px->pQInfo == NULL) {
SColumnInfo* pColumnInfo = extractColumnInfoFromResult(px->colList);
int32_t numOfOutput = (int32_t)tscNumOfExprs(px);
int32_t numOfCols = (int32_t)taosArrayGetSize(px->colList);
SQueriedTableInfo info = {
.colList = pColumnInfo,
.numOfCols = numOfCols,
};
SSchema* pSchema = tscGetTableSchema(px->pTableMetaInfo[0]->pTableMeta);
STableMeta* pTableMeta = tscGetMetaInfo(px, 0)->pTableMeta;
SSchema* pSchema = tscGetTableSchema(pTableMeta);
STableGroupInfo tableGroupInfo = {
.numOfTables = 1,
......@@ -1062,23 +1087,11 @@ void handleDownstreamOperator(SSqlObj** pSqlObjList, int32_t numOfUpstream, SQue
taosArrayPush(tableGroupInfo.pGroupList, &group);
// if it is a join query, create join operator here
int32_t numOfCol1 = px->pTableMetaInfo[0]->pTableMeta->tableInfo.numOfColumns;
int32_t numOfCol1 = pTableMeta->tableInfo.numOfColumns;
int32_t numOfFilterCols = 0;
SColumnInfo* tableCols = calloc(numOfCol1, sizeof(SColumnInfo));
for(int32_t i = 0; i < numOfCol1; ++i) {
SColumn* pCol = taosArrayGetP(px->colList, i);
if (pCol->info.flist.numOfFilters > 0) {
numOfFilterCols += 1;
}
tableCols[i] = pCol->info;
}
SSingleColumnFilterInfo* pFilterInfo = NULL;
if (numOfFilterCols > 0) {
doCreateFilterInfo(tableCols, numOfCol1, numOfFilterCols, &pFilterInfo, 0);
}
createInputDataFilterInfo(px, numOfCol1, &numOfFilterCols, &pFilterInfo);
SOperatorInfo* pSourceOperator = createDummyInputOperator(pSqlObjList[0], pSchema, numOfCol1, pFilterInfo, numOfFilterCols);
......@@ -1094,24 +1107,14 @@ void handleDownstreamOperator(SSqlObj** pSqlObjList, int32_t numOfUpstream, SQue
int32_t offset = pSourceOperator->numOfOutput;
for(int32_t i = 1; i < px->numOfTables; ++i) {
SSchema* pSchema1 = tscGetTableSchema(px->pTableMetaInfo[i]->pTableMeta);
int32_t n = px->pTableMetaInfo[i]->pTableMeta->tableInfo.numOfColumns;
STableMeta* pTableMeta1 = tscGetMetaInfo(px, i)->pTableMeta;
int32_t numOfFilterCols1 = 0;
SColumnInfo* tableCols1 = calloc(numOfCol1, sizeof(SColumnInfo));
for(int32_t j = 0; j < numOfCol1; ++j) {
SColumn* pCol = taosArrayGetP(px->colList, j);
if (pCol->info.flist.numOfFilters > 0) {
numOfFilterCols += 1;
}
tableCols1[j] = pCol->info;
}
SSchema* pSchema1 = tscGetTableSchema(pTableMeta1);
int32_t n = pTableMeta1->tableInfo.numOfColumns;
int32_t numOfFilterCols1 = 0;
SSingleColumnFilterInfo* pFilterInfo1 = NULL;
if (numOfFilterCols1 > 0) {
doCreateFilterInfo(tableCols1, numOfCol1, numOfFilterCols1, &pFilterInfo1, 0);
}
createInputDataFilterInfo(px, numOfCol1, &numOfFilterCols1, &pFilterInfo1);
p[i] = createDummyInputOperator(pSqlObjList[i], pSchema1, n, pFilterInfo1, numOfFilterCols1);
memcpy(&schema[offset], pSchema1, n * sizeof(SSchema));
......@@ -1126,23 +1129,12 @@ void handleDownstreamOperator(SSqlObj** pSqlObjList, int32_t numOfUpstream, SQue
memcpy(schema, pSchema, numOfCol1*sizeof(SSchema));
}
SExprInfo* exprInfo = NULL;
/*int32_t code = */ createQueryFunc(&info, numOfOutput, &exprInfo, px->exprList->pData, NULL, px->type, NULL);
for(int32_t i = 0; i < numOfOutput; ++i) {
SExprInfo* pex = taosArrayGetP(px->exprList, i);
int32_t colId = pex->base.colInfo.colId;
for(int32_t j = 0; j < pSourceOperator->numOfOutput; ++j) {
if (colId == schema[j].colId) {
pex->base.colInfo.colIndex = j;
break;
}
}
}
px->pQInfo = createQInfoFromQueryNode(px, exprInfo, &tableGroupInfo, pSourceOperator, NULL, NULL, MASTER_SCAN);
px->pQInfo = createQInfoFromQueryNode(px, &tableGroupInfo, pSourceOperator, NULL, NULL, MASTER_SCAN);
tfree(pColumnInfo);
tfree(schema);
tfree(exprInfo);
// set the pRuntimeEnv for pSourceOperator
pSourceOperator->pRuntimeEnv = &px->pQInfo->runtimeEnv;
}
uint64_t qId = 0;
......@@ -1219,31 +1211,34 @@ void tscFreeQueryInfo(SSqlCmd* pCmd, bool removeMeta) {
pCmd->active = NULL;
}
void destroyTableNameList(SSqlCmd* pCmd) {
if (pCmd->insertParam.numOfTables == 0) {
assert(pCmd->insertParam.pTableNameList == NULL);
void destroyTableNameList(SInsertStatementParam* pInsertParam) {
if (pInsertParam->numOfTables == 0) {
assert(pInsertParam->pTableNameList == NULL);
return;
}
for(int32_t i = 0; i < pCmd->insertParam.numOfTables; ++i) {
tfree(pCmd->insertParam.pTableNameList[i]);
for(int32_t i = 0; i < pInsertParam->numOfTables; ++i) {
tfree(pInsertParam->pTableNameList[i]);
}
pCmd->insertParam.numOfTables = 0;
tfree(pCmd->insertParam.pTableNameList);
pInsertParam->numOfTables = 0;
tfree(pInsertParam->pTableNameList);
}
void tscResetSqlCmd(SSqlCmd* pCmd, bool clearCachedMeta) {
pCmd->command = 0;
pCmd->numOfCols = 0;
pCmd->count = 0;
pCmd->curSql = NULL;
pCmd->msgType = 0;
destroyTableNameList(pCmd);
pCmd->insertParam.sql = NULL;
destroyTableNameList(&pCmd->insertParam);
pCmd->insertParam.pTableBlockHashList = tscDestroyBlockHashTable(pCmd->insertParam.pTableBlockHashList, clearCachedMeta);
pCmd->insertParam.pDataBlocks = tscDestroyBlockArrayList(pCmd->insertParam.pDataBlocks);
tfree(pCmd->insertParam.tagData.data);
pCmd->insertParam.tagData.dataLen = 0;
tscFreeQueryInfo(pCmd, clearCachedMeta);
if (pCmd->pTableMetaMap != NULL) {
......@@ -1262,8 +1257,8 @@ void tscResetSqlCmd(SSqlCmd* pCmd, bool clearCachedMeta) {
void tscFreeSqlResult(SSqlObj* pSql) {
SSqlRes* pRes = &pSql->res;
tscDestroyLocalMerger(pRes->pLocalMerger);
pRes->pLocalMerger = NULL;
tscDestroyGlobalMerger(pRes->pMerger);
pRes->pMerger = NULL;
tscDestroyResPointerInfo(pRes);
memset(&pSql->res, 0, sizeof(SSqlRes));
......@@ -1356,9 +1351,6 @@ void tscFreeSqlObj(SSqlObj* pSql) {
tscFreeSqlResult(pSql);
tscResetSqlCmd(pCmd, false);
tfree(pCmd->tagData.data);
pCmd->tagData.dataLen = 0;
memset(pCmd->payload, 0, (size_t)pCmd->allocSize);
tfree(pCmd->payload);
pCmd->allocSize = 0;
......@@ -1659,37 +1651,36 @@ static int32_t getRowExpandSize(STableMeta* pTableMeta) {
return result;
}
static void extractTableNameList(SSqlCmd* pCmd, bool freeBlockMap) {
pCmd->insertParam.numOfTables = (int32_t) taosHashGetSize(pCmd->insertParam.pTableBlockHashList);
if (pCmd->insertParam.pTableNameList == NULL) {
pCmd->insertParam.pTableNameList = calloc(pCmd->insertParam.numOfTables, POINTER_BYTES);
static void extractTableNameList(SInsertStatementParam *pInsertParam, bool freeBlockMap) {
pInsertParam->numOfTables = (int32_t) taosHashGetSize(pInsertParam->pTableBlockHashList);
if (pInsertParam->pTableNameList == NULL) {
pInsertParam->pTableNameList = calloc(pInsertParam->numOfTables, POINTER_BYTES);
} else {
memset(pCmd->insertParam.pTableNameList, 0, pCmd->insertParam.numOfTables * POINTER_BYTES);
memset(pInsertParam->pTableNameList, 0, pInsertParam->numOfTables * POINTER_BYTES);
}
STableDataBlocks **p1 = taosHashIterate(pCmd->insertParam.pTableBlockHashList, NULL);
STableDataBlocks **p1 = taosHashIterate(pInsertParam->pTableBlockHashList, NULL);
int32_t i = 0;
while(p1) {
STableDataBlocks* pBlocks = *p1;
tfree(pCmd->insertParam.pTableNameList[i]);
tfree(pInsertParam->pTableNameList[i]);
pCmd->insertParam.pTableNameList[i++] = tNameDup(&pBlocks->tableName);
p1 = taosHashIterate(pCmd->insertParam.pTableBlockHashList, p1);
pInsertParam->pTableNameList[i++] = tNameDup(&pBlocks->tableName);
p1 = taosHashIterate(pInsertParam->pTableBlockHashList, p1);
}
if (freeBlockMap) {
pCmd->insertParam.pTableBlockHashList = tscDestroyBlockHashTable(pCmd->insertParam.pTableBlockHashList, false);
pInsertParam->pTableBlockHashList = tscDestroyBlockHashTable(pInsertParam->pTableBlockHashList, false);
}
}
int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap) {
int32_t tscMergeTableDataBlocks(SInsertStatementParam *pInsertParam, bool freeBlockMap) {
const int INSERT_HEAD_SIZE = sizeof(SMsgDesc) + sizeof(SSubmitMsg);
SSqlCmd* pCmd = &pSql->cmd;
void* pVnodeDataBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
SArray* pVnodeDataBlockList = taosArrayInit(8, POINTER_BYTES);
STableDataBlocks** p = taosHashIterate(pCmd->insertParam.pTableBlockHashList, NULL);
STableDataBlocks** p = taosHashIterate(pInsertParam->pTableBlockHashList, NULL);
STableDataBlocks* pOneTableBlock = *p;
while(pOneTableBlock) {
......@@ -1702,7 +1693,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap) {
int32_t ret = tscGetDataBlockFromList(pVnodeDataBlockHashList, pOneTableBlock->vgId, TSDB_PAYLOAD_SIZE,
INSERT_HEAD_SIZE, 0, &pOneTableBlock->tableName, pOneTableBlock->pTableMeta, &dataBuf, pVnodeDataBlockList);
if (ret != TSDB_CODE_SUCCESS) {
tscError("0x%"PRIx64" failed to prepare the data block buffer for merging table data, code:%d", pSql->self, ret);
tscError("0x%"PRIx64" failed to prepare the data block buffer for merging table data, code:%d", pInsertParam->objectId, ret);
taosHashCleanup(pVnodeDataBlockHashList);
tscDestroyBlockArrayList(pVnodeDataBlockList);
return ret;
......@@ -1720,7 +1711,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap) {
dataBuf->pData = tmp;
memset(dataBuf->pData + dataBuf->size, 0, dataBuf->nAllocSize - dataBuf->size);
} else { // failed to allocate memory, free already allocated memory and return error code
tscError("0x%"PRIx64" failed to allocate memory for merging submit block, size:%d", pSql->self, dataBuf->nAllocSize);
tscError("0x%"PRIx64" failed to allocate memory for merging submit block, size:%d", pInsertParam->objectId, dataBuf->nAllocSize);
taosHashCleanup(pVnodeDataBlockHashList);
tscDestroyBlockArrayList(pVnodeDataBlockList);
......@@ -1733,7 +1724,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap) {
tscSortRemoveDataBlockDupRows(pOneTableBlock);
char* ekey = (char*)pBlocks->data + pOneTableBlock->rowSize*(pBlocks->numOfRows-1);
tscDebug("0x%"PRIx64" name:%s, name:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64, pSql->self, tNameGetTableName(&pOneTableBlock->tableName),
tscDebug("0x%"PRIx64" name:%s, name:%d rows:%d sversion:%d skey:%" PRId64 ", ekey:%" PRId64, pInsertParam->objectId, tNameGetTableName(&pOneTableBlock->tableName),
pBlocks->tid, pBlocks->numOfRows, pBlocks->sversion, GET_INT64_VAL(pBlocks->data), GET_INT64_VAL(ekey));
int32_t len = pBlocks->numOfRows * (pOneTableBlock->rowSize + expandSize) + sizeof(STColumn) * tscGetNumOfColumns(pOneTableBlock->pTableMeta);
......@@ -1745,7 +1736,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap) {
pBlocks->schemaLen = 0;
// erase the empty space reserved for binary data
int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock, pCmd->insertParam.schemaAttached);
int32_t finalLen = trimDataBlock(dataBuf->pData + dataBuf->size, pOneTableBlock, pInsertParam->schemaAttached);
assert(finalLen <= len);
dataBuf->size += (finalLen + sizeof(SSubmitBlk));
......@@ -1757,10 +1748,10 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap) {
pBlocks->numOfRows = 0;
}else {
tscDebug("0x%"PRIx64" table %s data block is empty", pSql->self, pOneTableBlock->tableName.tname);
tscDebug("0x%"PRIx64" table %s data block is empty", pInsertParam->objectId, pOneTableBlock->tableName.tname);
}
p = taosHashIterate(pCmd->insertParam.pTableBlockHashList, p);
p = taosHashIterate(pInsertParam->pTableBlockHashList, p);
if (p == NULL) {
break;
}
......@@ -1768,10 +1759,10 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, bool freeBlockMap) {
pOneTableBlock = *p;
}
extractTableNameList(pCmd, freeBlockMap);
extractTableNameList(pInsertParam, freeBlockMap);
// free the table data blocks;
pCmd->insertParam.pDataBlocks = pVnodeDataBlockList;
pInsertParam->pDataBlocks = pVnodeDataBlockList;
taosHashCleanup(pVnodeDataBlockHashList);
return TSDB_CODE_SUCCESS;
......@@ -3098,6 +3089,15 @@ void tscResetForNextRetrieve(SSqlRes* pRes) {
pRes->numOfRows = 0;
}
void tscInitResForMerge(SSqlRes* pRes) {
pRes->qId = 1; // hack to pass the safety check in fetch_row function
pRes->rspType = 0; // used as a flag to denote if taos_retrieved() has been called yet
tscResetForNextRetrieve(pRes);
assert(pRes->pMerger != NULL);
pRes->data = pRes->pMerger->buf;
}
void registerSqlObj(SSqlObj* pSql) {
taosAcquireRef(tscRefId, pSql->pTscObj->rid);
pSql->self = taosAddRef(tscObjRef, pSql);
......@@ -3119,14 +3119,6 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, __async_cb_func_t fp, void* param, in
SSqlCmd* pCmd = &pNew->cmd;
pCmd->command = cmd;
int32_t code = copyTagData(&pNew->cmd.tagData, &pSql->cmd.tagData);
if (code != TSDB_CODE_SUCCESS) {
tscError("0x%"PRIx64" new subquery failed, unable to malloc tag data, tableIndex:%d", pSql->self, 0);
free(pNew);
return NULL;
}
if (tscAddQueryInfo(pCmd) != TSDB_CODE_SUCCESS) {
#ifdef __APPLE__
// to satisfy later tsem_destroy in taos_free_result
......@@ -3222,8 +3214,6 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, __async_cb_func_t
pnCmd->insertParam.numOfTables = 0;
pnCmd->insertParam.pTableNameList = NULL;
pnCmd->insertParam.pTableBlockHashList = NULL;
pnCmd->tagData.data = NULL;
pnCmd->tagData.dataLen = 0;
if (tscAddQueryInfo(pnCmd) != TSDB_CODE_SUCCESS) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
......@@ -3389,7 +3379,11 @@ void doExecuteQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
tscHandleMasterSTableQuery(pSql);
tscUnlockByThread(&pSql->squeryLock);
} else if (TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_INSERT)) {
if (TSDB_QUERY_HAS_TYPE(pSql->cmd.insertParam.insertType, TSDB_QUERY_TYPE_FILE_INSERT)) {
tscImportDataFromFile(pSql);
} else {
tscHandleMultivnodeInsert(pSql);
}
} else if (pSql->cmd.command > TSDB_SQL_LOCAL) {
tscProcessLocalCmd(pSql);
} else { // send request to server directly
......@@ -3607,10 +3601,10 @@ int32_t tscSQLSyntaxErrMsg(char* msg, const char* additionalInfo, const char* s
return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
}
int32_t tscInvalidSQLErrMsg(char* msg, const char* additionalInfo, const char* sql) {
const char* msgFormat1 = "invalid SQL: %s";
const char* msgFormat2 = "invalid SQL: \'%s\' (%s)";
const char* msgFormat3 = "invalid SQL: \'%s\'";
int32_t tscInvalidOperationMsg(char* msg, const char* additionalInfo, const char* sql) {
const char* msgFormat1 = "invalid operation: %s";
const char* msgFormat2 = "invalid operation: \'%s\' (%s)";
const char* msgFormat3 = "invalid operation: \'%s\'";
const int32_t BACKWARD_CHAR_STEP = 0;
......@@ -4410,3 +4404,38 @@ int tscTransferTableNameList(SSqlObj *pSql, const char *pNameList, int32_t lengt
return TSDB_CODE_SUCCESS;
}
bool vgroupInfoIdentical(SNewVgroupInfo *pExisted, SVgroupMsg* src) {
assert(pExisted != NULL && src != NULL);
if (pExisted->numOfEps != src->numOfEps) {
return false;
}
for(int32_t i = 0; i < pExisted->numOfEps; ++i) {
if (pExisted->ep[i].port != src->epAddr[i].port) {
return false;
}
if (strncmp(pExisted->ep[i].fqdn, src->epAddr[i].fqdn, tListLen(pExisted->ep[i].fqdn)) != 0) {
return false;
}
}
return true;
}
SNewVgroupInfo createNewVgroupInfo(SVgroupMsg *pVgroupMsg) {
assert(pVgroupMsg != NULL);
SNewVgroupInfo info = {0};
info.numOfEps = pVgroupMsg->numOfEps;
info.vgId = pVgroupMsg->vgId;
info.inUse = 0; // 0 is the default value of inUse in case of multiple replica
assert(info.numOfEps >= 1 && info.vgId >= 1);
for(int32_t i = 0; i < pVgroupMsg->numOfEps; ++i) {
tstrncpy(info.ep[i].fqdn, pVgroupMsg->epAddr[i].fqdn, TSDB_FQDN_LEN);
info.ep[i].port = pVgroupMsg->epAddr[i].port;
}
return info;
}
\ No newline at end of file
go @ 050667e5
比较 8ce6d865...050667e5
Subproject commit 8ce6d86558afc8c0b50c10f990fd2b4270cf06fc
Subproject commit 050667e5b4d0eafa5387e4283e713559b421203f
grafanaplugin @ 32e2c97a
比较 3530c6df...32e2c97a
Subproject commit 3530c6df097134a410bacec6b3cd013ef38a61aa
Subproject commit 32e2c97a4cf7bedaa99f5d6dd8cb036e7f4470df
src/query/inc/qExecutor.h
浏览文件 @ d2886dee
......@@ -22,6 +22,7 @@
#include "qFill.h"
#include "qResultbuf.h"
#include "qSqlparser.h"
#include "qTableMeta.h"
#include "qTsbuf.h"
#include "query.h"
#include "taosdef.h"
......@@ -70,14 +71,6 @@ typedef struct SResultRowPool {
SArray* pData; // SArray<void*>
} SResultRowPool;
typedef struct SGroupbyExpr {
int16_t tableIndex;
SArray* columnInfo; // SArray<SColIndex>, group by columns information
int16_t numOfGroupCols; // todo remove it
int16_t orderIndex; // order by column index
int16_t orderType; // order by type: asc/desc
} SGroupbyExpr;
typedef struct SResultRow {
int32_t pageId; // pageId & rowId is the position of current result in disk-based output buffer
int32_t offset:29; // row index in buffer page
......@@ -216,7 +209,7 @@ typedef struct SQueryAttr {
int32_t intermediateResultRowSize; // intermediate result row size, in case of top-k query.
int32_t maxTableColumnWidth;
int32_t tagLen; // tag value length of current query
SGroupbyExpr* pGroupbyExpr;
SGroupbyExpr *pGroupbyExpr;
SExprInfo* pExpr1;
SExprInfo* pExpr2;
......@@ -475,10 +468,10 @@ typedef struct SDistinctOperatorInfo {
int64_t outputCapacity;
} SDistinctOperatorInfo;
struct SLocalMerger;
struct SGlobalMerger;
typedef struct SMultiwayMergeInfo {
struct SLocalMerger *pMerge;
struct SGlobalMerger *pMerge;
SOptrBasicInfo binfo;
int32_t bufCapacity;
int64_t seed;
......@@ -559,6 +552,8 @@ int32_t createFilterInfo(SQueryAttr* pQueryAttr, uint64_t qId);
void freeColumnFilterInfo(SColumnFilterInfo* pFilter, int32_t numOfFilters);
STableQueryInfo *createTableQueryInfo(SQueryAttr* pQueryAttr, void* pTable, bool groupbyColumn, STimeWindow win, void* buf);
STableQueryInfo* createTmpTableQueryInfo(STimeWindow win);
int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg);
bool isQueryKilled(SQInfo *pQInfo);
......
src/query/inc/qPlan.h
浏览文件 @ d2886dee
......@@ -16,6 +16,8 @@
#ifndef TDENGINE_QPLAN_H
#define TDENGINE_QPLAN_H
#include "qExecutor.h"
struct SQueryInfo;
typedef struct SQueryNodeBasicInfo {
......
src/query/inc/qSqlparser.h
浏览文件 @ d2886dee
......@@ -258,7 +258,6 @@ SArray *tVariantListInsert(SArray *pList, tVariant *pVar, uint8_t sortOrder, int
SArray *tVariantListAppendToken(SArray *pList, SStrToken *pAliasToken, uint8_t sortOrder);
SRelationInfo *setTableNameList(SRelationInfo* pFromInfo, SStrToken *pName, SStrToken* pAlias);
//SRelationInfo *setSubquery(SRelationInfo* pFromInfo, SRelElementPair* p);
void *destroyRelationInfo(SRelationInfo* pFromInfo);
SRelationInfo *addSubqueryElem(SRelationInfo* pRelationInfo, SArray* pSub, SStrToken* pAlias);
......
src/query/inc/qTableMeta.h 0 → 100644
浏览文件 @ d2886dee
#ifndef TDENGINE_QTABLEUTIL_H
#define TDENGINE_QTABLEUTIL_H
#include "tsdb.h" //todo tsdb should not be here
#include "qSqlparser.h"
typedef struct SFieldInfo {
int16_t numOfOutput; // number of column in result
TAOS_FIELD* final;
SArray *internalField; // SArray<SInternalField>
} SFieldInfo;
typedef struct SCond {
uint64_t uid;
int32_t len; // length of tag query condition data
char * cond;
} SCond;
typedef struct SJoinNode {
uint64_t uid;
int16_t tagColId;
SArray* tsJoin;
SArray* tagJoin;
} SJoinNode;
typedef struct SJoinInfo {
bool hasJoin;
SJoinNode *joinTables[TSDB_MAX_JOIN_TABLE_NUM];
} SJoinInfo;
typedef struct STagCond {
// relation between tbname list and query condition, including : TK_AND or TK_OR
int16_t relType;
// tbname query condition, only support tbname query condition on one table
SCond tbnameCond;
// join condition, only support two tables join currently
SJoinInfo joinInfo;
// for different table, the query condition must be seperated
SArray *pCond;
} STagCond;
typedef struct SGroupbyExpr {
int16_t tableIndex;
SArray* columnInfo; // SArray<SColIndex>, group by columns information
int16_t numOfGroupCols; // todo remove it
int16_t orderIndex; // order by column index
int16_t orderType; // order by type: asc/desc
} SGroupbyExpr;
typedef struct STableComInfo {
uint8_t numOfTags;
uint8_t precision;
int16_t numOfColumns;
int32_t rowSize;
} STableComInfo;
typedef struct STableMeta {
int32_t vgId;
STableId id;
uint8_t tableType;
char sTableName[TSDB_TABLE_FNAME_LEN]; // super table name
uint64_t suid; // super table id
int16_t sversion;
int16_t tversion;
STableComInfo tableInfo;
SSchema schema[]; // if the table is TSDB_CHILD_TABLE, schema is acquired by super table meta info
} STableMeta;
typedef struct STableMetaInfo {
STableMeta *pTableMeta; // table meta, cached in client side and acquired by name
uint32_t tableMetaSize;
SVgroupsInfo *vgroupList;
SArray *pVgroupTables; // SArray<SVgroupTableInfo>
/*
* 1. keep the vgroup index during the multi-vnode super table projection query
* 2. keep the vgroup index for multi-vnode insertion
*/
int32_t vgroupIndex;
SName name;
char aliasName[TSDB_TABLE_NAME_LEN]; // alias name of table specified in query sql
SArray *tagColList; // SArray<SColumn*>, involved tag columns
} STableMetaInfo;
struct SQInfo; // global merge operator
struct SQueryAttr; // query object
typedef struct SQueryInfo {
int16_t command; // the command may be different for each subclause, so keep it seperately.
uint32_t type; // query/insert type
STimeWindow window; // the whole query time window
SInterval interval; // tumble time window
SSessionWindow sessionWindow; // session time window
SGroupbyExpr groupbyExpr; // groupby tags info
SArray * colList; // SArray<SColumn*>
SFieldInfo fieldsInfo;
SArray * exprList; // SArray<SExprInfo*>
SArray * exprList1; // final exprlist in case of arithmetic expression exists
SLimitVal limit;
SLimitVal slimit;
STagCond tagCond;
SOrderVal order;
int16_t fillType; // final result fill type
int16_t numOfTables;
STableMetaInfo **pTableMetaInfo;
struct STSBuf *tsBuf;
int64_t * fillVal; // default value for fill
char * msg; // pointer to the pCmd->payload to keep error message temporarily
int64_t clauseLimit; // limit for current sub clause
int64_t prjOffset; // offset value in the original sql expression, only applied at client side
int64_t vgroupLimit; // table limit in case of super table projection query + global order + limit
int32_t udColumnId; // current user-defined constant output field column id, monotonically decreases from TSDB_UD_COLUMN_INDEX
int16_t resColumnId; // result column id
bool distinctTag; // distinct tag or not
int32_t round; // 0/1/....
int32_t bufLen;
char* buf;
struct SQInfo* pQInfo; // global merge operator
struct SQueryAttr* pQueryAttr; // query object
struct SQueryInfo *sibling; // sibling
SArray *pUpstream; // SArray<struct SQueryInfo>
struct SQueryInfo *pDownstream;
int32_t havingFieldNum;
bool stableQuery;
bool groupbyColumn;
bool simpleAgg;
bool arithmeticOnAgg;
bool projectionQuery;
bool hasFilter;
bool onlyTagQuery;
bool orderProjectQuery;
} SQueryInfo;
/**
* get the number of tags of this table
* @param pTableMeta
* @return
*/
int32_t tscGetNumOfTags(const STableMeta* pTableMeta);
/**
* get the number of columns of this table
* @param pTableMeta
* @return
*/
int32_t tscGetNumOfColumns(const STableMeta* pTableMeta);
/**
* get the basic info of this table
* @param pTableMeta
* @return
*/
STableComInfo tscGetTableInfo(const STableMeta* pTableMeta);
/**
* get the schema
* @param pTableMeta
* @return
*/
SSchema* tscGetTableSchema(const STableMeta* pTableMeta);
/**
* get the tag schema
* @param pMeta
* @return
*/
SSchema *tscGetTableTagSchema(const STableMeta *pMeta);
/**
* get the column schema according to the column index
* @param pMeta
* @param colIndex
* @return
*/
SSchema *tscGetTableColumnSchema(const STableMeta *pMeta, int32_t colIndex);
/**
* get the column schema according to the column id
* @param pTableMeta
* @param colId
* @return
*/
SSchema* tscGetColumnSchemaById(STableMeta* pTableMeta, int16_t colId);
/**
* create the table meta from the msg
* @param pTableMetaMsg
* @param size size of the table meta
* @return
*/
STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg);
#endif // TDENGINE_QTABLEUTIL_H
src/query/src/qExecutor.c
浏览文件 @ d2886dee
......@@ -1718,7 +1718,10 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
case OP_TimeWindow: {
pRuntimeEnv->proot =
createTimeIntervalOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput);
int32_t opType = pRuntimeEnv->proot->upstream[0]->operatorType;
if (opType != OP_DummyInput && opType != OP_Join) {
setTableScanFilterOperatorInfo(pRuntimeEnv->proot->upstream[0]->info, pRuntimeEnv->proot);
}
break;
}
case OP_Groupby: {
......@@ -3267,6 +3270,25 @@ STableQueryInfo *createTableQueryInfo(SQueryAttr* pQueryAttr, void* pTable, bool
return pTableQueryInfo;
}
STableQueryInfo* createTmpTableQueryInfo(STimeWindow win) {
STableQueryInfo* pTableQueryInfo = calloc(1, sizeof(STableQueryInfo));
pTableQueryInfo->win = win;
pTableQueryInfo->lastKey = win.skey;
pTableQueryInfo->pTable = NULL;
pTableQueryInfo->cur.vgroupIndex = -1;
// set more initial size of interval/groupby query
int32_t initialSize = 16;
int32_t code = initResultRowInfo(&pTableQueryInfo->resInfo, initialSize, TSDB_DATA_TYPE_INT);
if (code != TSDB_CODE_SUCCESS) {
return NULL;
}
return pTableQueryInfo;
}
void destroyTableQueryInfoImpl(STableQueryInfo *pTableQueryInfo) {
if (pTableQueryInfo == NULL) {
return;
......
src/query/src/qPlan.c
浏览文件 @ d2886dee
#include "os.h"
#include "tschemautil.h"
#include "qTableMeta.h"
#include "qPlan.h"
#include "qExecutor.h"
#include "qUtil.h"
......
src/client/src/tscSchemaUtil.c → src/query/src/qTableMeta.c
浏览文件 @ d2886dee
/*
* 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 "taosmsg.h"
#include "tschemautil.h"
#include "qTableMeta.h"
#include "ttokendef.h"
#include "taosdef.h"
#include "tutil.h"
#include "tsclient.h"
int32_t tscGetNumOfTags(const STableMeta* pTableMeta) {
assert(pTableMeta != NULL);
......@@ -120,58 +105,3 @@ STableMeta* tscCreateTableMetaFromMsg(STableMetaMsg* pTableMetaMsg) {
return pTableMeta;
}
bool vgroupInfoIdentical(SNewVgroupInfo *pExisted, SVgroupMsg* src) {
assert(pExisted != NULL && src != NULL);
if (pExisted->numOfEps != src->numOfEps) {
return false;
}
for(int32_t i = 0; i < pExisted->numOfEps; ++i) {
if (pExisted->ep[i].port != src->epAddr[i].port) {
return false;
}
if (strncmp(pExisted->ep[i].fqdn, src->epAddr[i].fqdn, tListLen(pExisted->ep[i].fqdn)) != 0) {
return false;
}
}
return true;
}
SNewVgroupInfo createNewVgroupInfo(SVgroupMsg *pVgroupMsg) {
assert(pVgroupMsg != NULL);
SNewVgroupInfo info = {0};
info.numOfEps = pVgroupMsg->numOfEps;
info.vgId = pVgroupMsg->vgId;
info.inUse = 0; // 0 is the default value of inUse in case of multiple replica
assert(info.numOfEps >= 1 && info.vgId >= 1);
for(int32_t i = 0; i < pVgroupMsg->numOfEps; ++i) {
tstrncpy(info.ep[i].fqdn, pVgroupMsg->epAddr[i].fqdn, TSDB_FQDN_LEN);
info.ep[i].port = pVgroupMsg->epAddr[i].port;
}
return info;
}
// todo refactor
UNUSED_FUNC static FORCE_INLINE char* skipSegments(char* input, char delim, int32_t num) {
for (int32_t i = 0; i < num; ++i) {
while (*input != 0 && *input++ != delim) {
};
}
return input;
}
UNUSED_FUNC static FORCE_INLINE size_t copy(char* dst, const char* src, char delimiter) {
size_t len = 0;
while (*src != delimiter && *src != 0) {
*dst++ = *src++;
len++;
}
return len;
}
tests/script/general/parser/import_file.sim
浏览文件 @ d2886dee
......@@ -18,6 +18,7 @@ system general/parser/gendata.sh
sql create table tbx (ts TIMESTAMP, collect_area NCHAR(12), device_id BINARY(16), imsi BINARY(16), imei BINARY(16), mdn BINARY(10), net_type BINARY(4), mno NCHAR(4), province NCHAR(10), city NCHAR(16), alarm BINARY(2))
print ====== create tables success, starting import data
sql import into tbx file '~/data.sql'
sql import into tbx file '~/data.sql'
sql select count(*) from tbx
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或