Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
提交 be9c504e 编写于 作者: P Ping Xiao
浏览文件
操作

Merge branch 'develop' into test

上级 1b6ca91d a47ec20e
展开全部 隐藏空白更改
内联 并排
Showing with 13739 addition and 7614 deletion
.travis.yml
浏览文件 @ be9c504e
......@@ -4,6 +4,13 @@
#
# Build Matrix
#
branches:
only:
- master
- develop
- coverity_scan
- /^.*ci-.*$/
matrix:
- os: linux
dist: bionic
......@@ -260,19 +267,75 @@ matrix:
- cmake .. > /dev/null
- make > /dev/null
# - os: osx
# language: c
# compiler: clang
# env: DESC="mac/clang build"
# git:
# - depth: 1
# addons:
# homebrew:
# - cmake
#
# script:
# - cd ${TRAVIS_BUILD_DIR}
# - mkdir debug
# - cd debug
# - cmake .. > /dev/null
# - make > /dev/null
- os: linux
arch: arm64
dist: bionic
language: c
compiler: clang
env: DESC="linux/clang build"
git:
- depth: 1
addons:
apt:
packages:
- build-essential
- cmake
before_script:
- cd ${TRAVIS_BUILD_DIR}
- mkdir debug
- cd debug
script:
- if [ "${TRAVIS_CPU_ARCH}" == "arm64" ]; then
cmake .. -DCPUTYPE=aarch64 > /dev/null;
else
cmake .. > /dev/null;
fi
- make > /dev/null
- os: linux
arch: arm64
dist: trusty
language: c
git:
- depth: 1
addons:
apt:
packages:
- build-essential
- cmake
env:
- DESC="trusty/gcc-4.8 build"
before_script:
- cd ${TRAVIS_BUILD_DIR}
- mkdir debug
- cd debug
script:
- if [ "${TRAVIS_CPU_ARCH}" == "arm64" ]; then
cmake .. -DCPUTYPE=aarch64 > /dev/null;
else
cmake .. > /dev/null;
fi
- make > /dev/null
# - os: osx
# language: c
# compiler: clang
# env: DESC="mac/clang build"
# git:
# - depth: 1
# addons:
# homebrew:
# - cmake
#
# script:
# - cd ${TRAVIS_BUILD_DIR}
# - mkdir debug
# - cd debug
# - cmake .. > /dev/null
# - make > /dev/null
documentation/webdocs/markdowndocs/Super Table-ch.md
浏览文件 @ be9c504e
......@@ -53,7 +53,7 @@ STable从属于库，一个STable只属于一个库，但一个库可以有一
说明：
1. TAGS列总长度不能超过64k bytes；
1. TAGS列总长度不能超过16k bytes；
2. TAGS列的数据类型不能是timestamp；
3. TAGS列名不能与其他列名相同;
4. TAGS列名不能为预留关键字.
......
documentation/webdocs/markdowndocs/Super Table.md
浏览文件 @ be9c504e
......@@ -22,7 +22,7 @@ New keyword "tags" is introduced, where tag_name is the tag name, and tag_type i
Note：
1. The bytes of all tags together shall be less than 64k
1. The bytes of all tags together shall be less than 16k
2. Tag's data type can not be time stamp
3. Tag name shall be different from the field name
4. Tag name shall not be the same as system keywords
......
documentation/webdocs/markdowndocs/TAOS SQL-ch.md
浏览文件 @ be9c504e
......@@ -63,7 +63,7 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic
| 3 | BIGINT | 8 | 长整型，范围 [-2^63+1, 2^63-1], -2^63用于NULL |
| 4 | FLOAT | 4 | 浮点型，有效位数6-7，范围 [-3.4E38, 3.4E38] |
| 5 | DOUBLE | 8 | 双精度浮点型，有效位数15-16，范围 [-1.7E308, 1.7E308] |
| 6 | BINARY | 自定义 | 用于记录字符串，理论上，最长可以有65526字节，但由于每行数据最多64K字节，实际上限一般小于理论值。 binary仅支持字符串输入，字符串两端使用单引号引用，否则英文全部自动转化为小写。使用时须指定大小，如binary(20)定义了最长为20个字符的字符串，每个字符占1byte的存储空间。如果用户字符串超出20字节将会报错。对于字符串内的单引号，可以用转义字符反斜线加单引号来表示， 即 **\’**。 |
| 6 | BINARY | 自定义 | 用于记录字符串，理论上，最长可以有16374字节，但由于每行数据最多16K字节，实际上限一般小于理论值。 binary仅支持字符串输入，字符串两端使用单引号引用，否则英文全部自动转化为小写。使用时须指定大小，如binary(20)定义了最长为20个字符的字符串，每个字符占1byte的存储空间。如果用户字符串超出20字节将会报错。对于字符串内的单引号，可以用转义字符反斜线加单引号来表示， 即 **\’**。 |
| 7 | SMALLINT | 2 | 短整型， 范围 [-32767, 32767], -32768用于NULL |
| 8 | TINYINT | 1 | 单字节整型，范围 [-127, 127], -128用于NULL |
| 9 | BOOL | 1 | 布尔型，{true, false} |
......@@ -106,7 +106,7 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic
```mysql
CREATE TABLE [IF NOT EXISTS] tb_name (timestamp_field_name TIMESTAMP, field1_name data_type1 [, field2_name data_type2 ...])
```
说明：1）表的第一个字段必须是TIMESTAMP，并且系统自动将其设为主键；2）表的每行长度不能超过64K字节；3）使用数据类型binary或nchar，需指定其最长的字节数，如binary(20)，表示20字节。
说明：1）表的第一个字段必须是TIMESTAMP，并且系统自动将其设为主键；2）表的每行长度不能超过16K字节；3）使用数据类型binary或nchar，需指定其最长的字节数，如binary(20)，表示20字节。
- **删除数据表**
......
documentation/webdocs/markdowndocs/TAOS SQL.md
浏览文件 @ be9c504e
......@@ -39,7 +39,7 @@ The full list of data types is listed below. For string types of data, we will
| 6 | DOUBLE | 8 | A standard nullable double float type with 15-16 significant digits and a range of [-1.7E308, 1.7E308]​ |
| 7 | BOOL | 1 | A nullable boolean type, [**`true`**, **`false`**] |
| 8 | TIMESTAMP | 8 | A nullable timestamp type with the same usage as the primary column timestamp |
| 9 | BINARY(*M*) | *M* | A nullable string type whose length is *M*, error should be threw with exceeded chars, the maximum length of *M* is 65526, but as maximum row size is 64K bytes, the actual upper limit will generally less than 65526. This type of string only supports ASCii encoded chars. |
| 9 | BINARY(*M*) | *M* | A nullable string type whose length is *M*, error should be threw with exceeded chars, the maximum length of *M* is 16374, but as maximum row size is 16K bytes, the actual upper limit will generally less than 16374. This type of string only supports ASCii encoded chars. |
| 10 | NCHAR(*M*) | 4 * *M* | A nullable string type whose length is *M*, error should be threw with exceeded chars. The **`NCHAR`** type supports Unicode encoded chars. |
All the keywords in a SQL statement are case-insensitive, but strings values are case-sensitive and must be quoted by a pair of `'` or `"`. To quote a `'` or a `"` , you can use the escape character `\`.
......@@ -86,7 +86,7 @@ All the keywords in a SQL statement are case-insensitive, but strings values are
1) The first column must be a `timestamp`, and the system will set it as the primary key.
2) The record size is limited to 64k bytes
2) The record size is limited to 16k bytes
3) For `binary` or `nchar` data types, the length must be specified. For example, binary(20) means a binary data type with 20 bytes.
......
packaging/cfg/taos.cfg
浏览文件 @ be9c504e
......@@ -6,10 +6,22 @@
########################################################
# first full-qualified domain name (FQDN) for TDengine system
# first 127.0.0.1
# first hostname1:6030
# second full-qualified domain name (FQDN) for TDengine system, for cluster edition only
# second 127.0.0.1
# second cluster_hostname2:6030
# the arbitrator's full-qualified domain name (FQDN) for TDengine system, for cluster edition only
# arbitrator arbitrator_hostname:6030
# the full-qualified domain name (FQDN) of dnode
# fqdn hostname:6030
# port for MNode connect to Client, default udp[6030-6055] tcp[6030]
# serverPort 6030
# http service port, default tcp[6020]
# httpPort 6020
# set socket type ("udp" and "tcp")
# the server and client should have the same socket type. Otherwise, connect will fail
......@@ -30,42 +42,38 @@
# log file's directory
# logDir /var/log/taos
# http service port, default tcp[6020]
# httpPort 6020
# number of management nodes in the system
# numOfMnodes 3
# port for MNode connect to Client, default udp[6030-6055] tcp[6030]
# serverPort 6030
# optional roles for dnode. 0 - any, 1 - mnode, 2 - dnode
# alternativeRole 0
# number of threads per CPU core
# numOfThreadsPerCore 1
# numOfThreadsPerCore 1.0
# number of vnodes per core in DNode
# numOfVnodesPerCore 8
# the ratio of threads responsible for querying in the total thread
# ratioOfQueryThreads 0.5
# number of total vnodes in DNode
# numOfTotalVnodes 0
# max number of tables per vnode
# maxtablesPerVnode 1000
# cache block size
# cache 16384
# interval of check load balance when the management node is in normal operation
# balanceInterval 300
# row of records in file block
# minRows 100
# maxRows 4096
# interval of DNode report status to MNode, unit is Second, for cluster version only
# statusInterval 1
# number of cache blocks per vnode
# blocks 2
# interval of Shell send HB to MNode, unit is Second
# shellActivityTimer 3
# interval of system monitor
# monitorInterval 60
# the compressed rpc message, option:
# -1 (no compression)
# 0 (all message compressed),
# > 0 (rpc message body which larger than this value will be compressed)
# compressMsgSize -1
# monitorInterval 30
# RPC re-try timer, millisecond
# rpcTimer 300
......@@ -73,38 +81,20 @@
# RPC maximum time for ack, seconds
# rpcMaxTime 600
# commit interval，unit is second
# ctime 3600
# interval of DNode report status to MNode, unit is Second, for cluster version only
# statusInterval 1
# interval of Shell send HB to MNode, unit is Second
# shellActivityTimer 3
# interval of DNode send HB to DNode, unit is Second, for cluster version only
# vnodePeerHBTimer 1
# interval of MNode send HB to MNode, unit is Second, for cluster version only
# mgmtPeerHBTimer 1
# duration of to keep tableMeta kept in Cache, seconds
# tableMetaKeepTimer 7200
# max number of users
# maxUsers 1000
# max number of databases
# maxDbs 1000
# Minimum sliding window time
# minSlidingTime 10
# max number of tables
# maxTables 650000
# Time window minimum
# minIntervalTime 10
# max number of Dnodes, for cluster version only
# maxDnodes 1000
# max length of an SQL
# maxSQLLength 65380
# Max number of VGroups, for cluster version only
# maxVGroups 1000
# Support the maximum number of records allowed for super table time sorting
# maxNumOfOrderedRes 100000
# system locale
# locale en_US.UTF-8
......@@ -121,6 +111,24 @@
# enable/disable async log
# asyncLog 1
# the compressed rpc message, option:
# -1 (no compression)
# 0 (all message compressed),
# > 0 (rpc message body which larger than this value will be compressed)
# compressMsgSize -1
# cache block size (Mbyte)
# cache 16
# number of cache blocks per vnode
# blocks 4
# min row of records in file block
# minRows 100
# max row of records in file block
# maxRows 4096
# enable/disable compression
# comp 1
......@@ -131,31 +139,25 @@
# keep 3650
# number of replications, for cluster version only
# replications 1
# replica 1
# client default database(database should be created)
# defaultDB
# max number of connections from client for dnode
# maxShellConns 5000
# client default username
# defaultUser root
# max numerber of connections to one database
# maxConnections 5000
# client default password
# defaultPass taosdata
# Stop writing logs when the disk size of the log folder is less than this value
# minimalLogDirGB 0.1
# max number of connections from client for mgmt node
# maxShellConns 2000
# Stop writing temporary files when the disk size of the log folder is less than this value
# minimalTmpDirGB 0.1
# max numerber of meter Connections
# maxMeterConnections 10000
# max connection to management node
# maxMgmtConnections 2000
# max connection to Vnode
# maxVnodeConnections 10000
# Stop writing data when the disk size of the log folder is less than this value
# minimalDataDirGB 0.1
# mnode take into account while balance, for cluster version only
# mnodeEqualVnodeNum 4
# mnodeEqualVnodeNum 4
# number of seconds allowed for a dnode to be offline, for cluster version only
# offlineThreshold 864000
......@@ -170,10 +172,10 @@
# mqtt 0
# mqtt uri
# mqttBrokerAddress mqtt://username:password@hostname:1883/taos/
# mqttBrokerAddress mqtt://username:password@hostname:1883/taos/
# mqtt client name
# mqttBrokerClientId taos_mqtt
# mqttBrokerClientId taos_mqtt
# maximum number of rows returned by the restful interface
# restfulRowLimit 10240
......@@ -181,41 +183,17 @@
# number of threads used to process http requests
# httpMaxThreads 2
# pre-allocated number of http sessions
# httpCacheSessions 100
# The minimum time to wait before the first stream execution
# maxFirstStreamCompDelay 10000
# whether to enable HTTP compression transmission
# httpEnableCompress 0
# Retry wait time benchmark
# retryStreamCompDelay 10
# the delayed time for launching each continuous query. 10% of the whole computing time window by default.
# streamCompDelayRatio 0.1
# the max allowed delayed time for launching continuous query. 20ms by default
# tsMaxStreamComputDelay 20000
# whether the telegraf table name contains the number of tags and the number of fields
# telegrafUseFieldNum 0
# enable table names generated by telegraf can be configured
# use /etc/taos/taos.telegraf.cfg as the configuration file, such as
# {
# "metrics": [
# {
# "name" : "system",
# "tbname" : "system_uptime",
# "fields": [
# "uptime"
# ]
# },
# {
# "name": "system",
# "tbname" : "system_uptime_format",
# "fields": [
# "uptime_format"
# ]
# }
# ]
# }
# maxStreamCompDelay 20000
# The following parameter is used to limit the maximum number of lines in log files.
# max number of rows per log filters
......@@ -226,35 +204,62 @@
# 131: output warning and error，135: output info, warning and error to log.
# 199: output debug, info, warning and error to both screen and file
# debug flag for basic utils
# debugFlag 131
# debug flag for meta management messages
# mDebugFlag 135
# debug flag for dnode messages
# dDebugFlag 131
# debug flag for TDengine SDB
# sDebugFlag 135
# debug flag for TDengine SDB
# sdbDebugFlag 135
# debug flag for RPC
# rpcDebugFlag 131
# debug flag for basic utils
# debugFlag 131
# debug flag for TAOS TIMER
# tmrDebugFlag 131
# debug flag for TDengine client
# cDebugFlag 131
# debug flag for query
# qDebugflag 131
# debug flag for JNI
# jniDebugflag 131
# debug flag for ODBC
# odbcDebugflag 131
# debug flag for storage
# uDebugflag 131
# debug flag for http server
# httpDebugFlag 131
# debug flag for system monitor
# debug flag for mqtt
# mqttDebugFlag 131
# debug flag for monitor
# monitorDebugFlag 131
#debug flag for mqtt client
# mqttDebugFlag 131
# debug flag for query
# qDebugflag 131
# debug flag for TAOS TIMER
# tmrDebugFlag 131
# debug flag for http server
# tsdbDebugFlag 131
# Record the SQL through restful interface
# httpEnableRecordSql 0
# Record the SQL in taos client
# tscEnableRecordSql 0
# if generate core file when service crash
# enableCoreFile 1
# The maximum display width of binary and nchar fields in the shell. The parts exceeding this limit will be hidden
# maxBinaryDisplayWidth 30
\ No newline at end of file
src/client/inc/tscSecondaryMerge.h → src/client/inc/tscLocalMerge.h
浏览文件 @ be9c504e
......@@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_TSCSECONARYMERGE_H
#define TDENGINE_TSCSECONARYMERGE_H
#ifndef TDENGINE_TSCLOCALMERGE_H
#define TDENGINE_TSCLOCALMERGE_H
#ifdef __cplusplus
extern "C" {
......@@ -27,14 +27,7 @@ extern "C" {
#include "tsclient.h"
#define MAX_NUM_OF_SUBQUERY_RETRY 3
/*
* @version 0.1
* @date 2018/01/05
* @author liaohj
* management of client-side reducer for metric query
*/
struct SQLFunctionCtx;
typedef struct SLocalDataSource {
......@@ -60,7 +53,6 @@ typedef struct SLocalReducer {
char * prevRowOfInput;
tFilePage * pResultBuf;
int32_t nResultBufSize;
// char * pBufForInterpo; // intermediate buffer for interpolation
tFilePage * pTempBuffer;
struct SQLFunctionCtx *pCtx;
int32_t rowSize; // size of each intermediate result.
......@@ -81,13 +73,8 @@ typedef struct SLocalReducer {
} SLocalReducer;
typedef struct SSubqueryState {
/*
* the number of completed retrieval subquery, once this value equals to numOfVnodes,
* all retrieval are completed.Local merge is launched.
*/
int32_t numOfCompleted;
int32_t numOfTotal; // number of total sub-queries
int32_t code; // code from subqueries
int32_t numOfRemain; // the number of remain unfinished subquery
int32_t numOfTotal; // the number of total sub-queries
uint64_t numOfRetrievedRows; // total number of points in this query
} SSubqueryState;
......@@ -128,4 +115,4 @@ int32_t tscDoLocalMerge(SSqlObj *pSql);
}
#endif
#endif // TDENGINE_TSCSECONARYMERGE_H
#endif // TDENGINE_TSCLOCALMERGE_H
src/client/inc/tscSubquery.h
浏览文件 @ be9c504e
......@@ -26,11 +26,9 @@ extern "C" {
void tscFetchDatablockFromSubquery(SSqlObj* pSql);
void tscSetupOutputColumnIndex(SSqlObj* pSql);
int32_t tscLaunchSecondPhaseSubqueries(SSqlObj* pSql);
void tscJoinQueryCallback(void* param, TAOS_RES* tres, int code);
SJoinSupporter* tscCreateJoinSupporter(SSqlObj* pSql, SSubqueryState* pState, int32_t index);
void tscDestroyJoinSupporter(SJoinSupporter* pSupporter);
int32_t tscHandleMasterJoinQuery(SSqlObj* pSql);
......
src/client/inc/tscUtil.h
浏览文件 @ be9c504e
......@@ -28,7 +28,7 @@ extern "C" {
#include "exception.h"
#include "qextbuffer.h"
#include "taosdef.h"
#include "tscSecondaryMerge.h"
#include "tscLocalMerge.h"
#include "tsclient.h"
#define UTIL_TABLE_IS_SUPER_TABLE(metaInfo) \
......@@ -64,7 +64,8 @@ typedef struct SJoinSupporter {
SSubqueryState* pState;
SSqlObj* pObj; // parent SqlObj
int32_t subqueryIndex; // index of sub query
int64_t interval; // interval time
int64_t intervalTime; // interval time
int64_t slidingTime; // sliding time
SLimitVal limit; // limit info
uint64_t uid; // query meter uid
SArray* colList; // previous query information, no need to use this attribute, and the corresponding attribution
......@@ -122,15 +123,13 @@ bool tscNonOrderedProjectionQueryOnSTable(SQueryInfo *pQueryInfo, int32_t tableI
bool tscOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex);
bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex);
bool tscProjectionQueryOnTable(SQueryInfo* pQueryInfo);
bool tscIsProjectionQuery(SQueryInfo* pQueryInfo);
bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex);
bool tscQueryOnSTable(SSqlCmd* pCmd);
bool tscQueryTags(SQueryInfo* pQueryInfo);
bool tscIsSelectivityWithTagQuery(SSqlCmd* pCmd);
void tscAddSpecialColumnForSelect(SQueryInfo* pQueryInfo, int32_t outputColIndex, int16_t functionId, SColumnIndex* pIndex,
SSchema* pColSchema, int16_t isTag);
SSchema* pColSchema, int16_t colType);
int32_t tscSetTableFullName(STableMetaInfo* pTableMetaInfo, SSQLToken* pzTableName, SSqlObj* pSql);
void tscClearInterpInfo(SQueryInfo* pQueryInfo);
......@@ -139,7 +138,7 @@ bool tscIsInsertData(char* sqlstr);
/* use for keep current db info temporarily, for handle table with db prefix */
// todo remove it
void tscGetDBInfoFromMeterId(char* tableId, char* db);
void tscGetDBInfoFromTableFullName(char* tableId, char* db);
int tscAllocPayload(SSqlCmd* pCmd, int size);
......@@ -248,12 +247,14 @@ void tscDoQuery(SSqlObj* pSql);
* @param pPrevSql
* @return
*/
SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cmd);
SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void* param, int32_t cmd, SSqlObj* pPrevSql);
void addGroupInfoForSubquery(SSqlObj* pParentObj, SSqlObj* pSql, int32_t subClauseIndex, int32_t tableIndex);
void doAddGroupColumnForSubquery(SQueryInfo* pQueryInfo, int32_t tagIndex);
int16_t tscGetJoinTagColIndexByUid(STagCond* pTagCond, uint64_t uid);
int16_t tscGetJoinTagColIdByUid(STagCond* pTagCond, uint64_t uid);
void tscPrintSelectClause(SSqlObj* pSql, int32_t subClauseIndex);
......
src/client/inc/tschemautil.h
浏览文件 @ be9c504e
......@@ -64,12 +64,20 @@ SSchema* tscGetTableSchema(const STableMeta* pTableMeta);
SSchema *tscGetTableTagSchema(const STableMeta *pMeta);
/**
*
* get the column schema according to the column index
* @param pMeta
* @param startCol
* @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 *tscGetTableColumnSchema(const STableMeta *pMeta, int32_t startCol);
SSchema* tscGetTableColumnSchemaById(STableMeta* pTableMeta, int16_t colId);
/**
* check if the schema is valid or not, including following aspects:
......
src/client/inc/tsclient.h
浏览文件 @ be9c504e
......@@ -85,7 +85,7 @@ typedef struct SSqlExpr {
int16_t functionId; // function id in aAgg array
int16_t resType; // return value type
int16_t resBytes; // length of return value
int32_t interBytes; // inter result buffer size
int32_t interBytes; // inter result buffer size
int16_t numOfParams; // argument value of each function
tVariant param[3]; // parameters are not more than 3
int32_t offset; // sub result column value of arithmetic expression.
......@@ -123,7 +123,7 @@ typedef struct SCond {
typedef struct SJoinNode {
char tableId[TSDB_TABLE_ID_LEN];
uint64_t uid;
int16_t tagCol;
int16_t tagColId;
} SJoinNode;
typedef struct SJoinInfo {
......@@ -155,20 +155,19 @@ typedef struct SParamInfo {
} SParamInfo;
typedef struct STableDataBlocks {
char tableId[TSDB_TABLE_ID_LEN];
int8_t tsSource; // where does the UNIX timestamp come from, server or client
bool ordered; // if current rows are ordered or not
int64_t vgId; // virtual group id
int64_t prevTS; // previous timestamp, recorded to decide if the records array is ts ascending
int32_t numOfTables; // number of tables in current submit block
int32_t rowSize; // row size for current table
char tableId[TSDB_TABLE_ID_LEN];
int8_t tsSource; // where does the UNIX timestamp come from, server or client
bool ordered; // if current rows are ordered or not
int64_t vgId; // virtual group id
int64_t prevTS; // previous timestamp, recorded to decide if the records array is ts ascending
int32_t numOfTables; // number of tables in current submit block
int32_t rowSize; // row size for current table
uint32_t nAllocSize;
uint32_t headerSize; // header for metadata (submit metadata)
uint32_t headerSize; // header for table info (uid, tid, submit metadata)
uint32_t size;
/*
* the metermeta for current table, the metermeta will be used during submit stage, keep a ref
* the table meta of table, the table meta will be used during submit, keep a ref
* to avoid it to be removed from cache
*/
STableMeta *pTableMeta;
......@@ -191,32 +190,28 @@ typedef struct SDataBlockList { // todo remove
} SDataBlockList;
typedef struct SQueryInfo {
int16_t command; // the command may be different for each subclause, so keep it seperately.
uint32_t type; // query/insert/import type
int16_t command; // the command may be different for each subclause, so keep it seperately.
uint32_t type; // query/insert/import type
char slidingTimeUnit;
STimeWindow window;
int64_t intervalTime; // aggregation time interval
int64_t slidingTime; // sliding window in mseconds
SSqlGroupbyExpr groupbyExpr; // group by tags info
SArray * colList; // SArray<SColumn*>
SArray * colList; // SArray<SColumn*>
SFieldInfo fieldsInfo;
SArray * exprList; // SArray<SSqlExpr*>
SArray * exprList; // SArray<SSqlExpr*>
SLimitVal limit;
SLimitVal slimit;
STagCond tagCond;
SOrderVal order;
int16_t fillType; // final result fill type
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
// offset value in the original sql expression, NOT sent to virtual node, only applied at client side
int64_t prjOffset;
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
} SQueryInfo;
typedef struct {
......@@ -299,11 +294,12 @@ typedef struct STscObj {
} STscObj;
typedef struct SSqlObj {
void * signature;
STscObj *pTscObj;
void (*fp)();
void (*fetchFp)();
void * param;
void *signature;
STscObj *pTscObj;
void *pRpcCtx;
void (*fp)();
void (*fetchFp)();
void *param;
int64_t stime;
uint32_t queryId;
void * pStream;
......@@ -312,8 +308,7 @@ typedef struct SSqlObj {
char retry;
char maxRetry;
SRpcIpSet ipList;
char freed : 4;
char listed : 4;
char listed;
tsem_t rspSem;
SSqlCmd cmd;
SSqlRes res;
......@@ -353,7 +348,7 @@ typedef struct SSqlStream {
int32_t tscInitRpc(const char *user, const char *secret, void** pDnodeConn);
void tscInitMsgsFp();
int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion);
int tsParseSql(SSqlObj *pSql, bool initial);
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet);
int tscProcessSql(SSqlObj *pSql);
......@@ -409,6 +404,7 @@ void tscKillSTableQuery(SSqlObj *pSql);
void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen);
bool tscIsUpdateQuery(SSqlObj* pSql);
bool tscHasReachLimitation(SQueryInfo *pQueryInfo, SSqlRes *pRes);
bool tscResultsetFetchCompleted(TAOS_RES *result);
char *tscGetErrorMsgPayload(SSqlCmd *pCmd);
......@@ -431,7 +427,7 @@ extern int (*tscBuildMsg[TSDB_SQL_MAX])(SSqlObj *pSql, SSqlInfo *pInfo);
typedef void (*__async_cb_func_t)(void *param, TAOS_RES *tres, int numOfRows);
int32_t tscCompareTidTags(const void* p1, const void* p2);
void tscBuildVgroupTableInfo(STableMetaInfo* pTableMetaInfo, SArray* tables);
void tscBuildVgroupTableInfo(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, SArray* tables);
#ifdef __cplusplus
}
......
src/client/src/TSDBJNIConnector.c
浏览文件 @ be9c504e
......@@ -151,8 +151,8 @@ JNIEXPORT void JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_dumpMemoryLeakImp
JNIEXPORT void JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_initImp(JNIEnv *env, jobject jobj, jstring jconfigDir) {
if (jconfigDir != NULL) {
const char *confDir = (*env)->GetStringUTFChars(env, jconfigDir, NULL);
if (confDir && strlen(configDir) != 0) {
strcpy(configDir, confDir);
if (confDir && strlen(confDir) != 0) {
tstrncpy(configDir, confDir, TSDB_FILENAME_LEN);
}
(*env)->ReleaseStringUTFChars(env, jconfigDir, confDir);
}
......@@ -192,7 +192,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_setOptions(JNIEnv
const char *tz1 = (*env)->GetStringUTFChars(env, optionValue, NULL);
if (tz1 && strlen(tz1) != 0) {
res = taos_options(TSDB_OPTION_TIMEZONE, tz1);
jniTrace("set timezone to %s, result:%d", timezone, res);
jniTrace("set timezone to %s, result:%d", tz1, res);
} else {
jniTrace("input timezone is empty");
}
......@@ -227,12 +227,10 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_connectImp(JNIEn
}
if (user == NULL) {
jniTrace("jobj:%p, user is null, use tsDefaultUser", jobj);
user = tsDefaultUser;
jniTrace("jobj:%p, user is null, use default user %s", jobj, TSDB_DEFAULT_USER);
}
if (pass == NULL) {
jniTrace("jobj:%p, pass is null, use tsDefaultPass", jobj);
pass = tsDefaultPass;
jniTrace("jobj:%p, pass is null, use default password", jobj);
}
/*
......@@ -252,8 +250,8 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_connectImp(JNIEn
if (host != NULL) (*env)->ReleaseStringUTFChars(env, jhost, host);
if (dbname != NULL) (*env)->ReleaseStringUTFChars(env, jdbName, dbname);
if (user != NULL && user != tsDefaultUser) (*env)->ReleaseStringUTFChars(env, juser, user);
if (pass != NULL && pass != tsDefaultPass) (*env)->ReleaseStringUTFChars(env, jpass, pass);
if (user != NULL) (*env)->ReleaseStringUTFChars(env, juser, user);
if (pass != NULL) (*env)->ReleaseStringUTFChars(env, jpass, pass);
return ret;
}
......@@ -344,7 +342,7 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_getResultSetImp(
STscObj *pObj = pSql->pTscObj;
if (tscIsUpdateQuery(pSql)) {
taos_free_result(pSql); // free result here
// taos_free_result(pSql); // free result here
jniTrace("jobj:%p, conn:%p, no resultset, %p", jobj, pObj, (void *)tres);
return 0;
} else {
......@@ -385,7 +383,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_getAffectedRowsIm
}
jint ret = taos_affected_rows((SSqlObj *)res);
jniTrace("jobj:%p, conn:%p, sql:%p, affect rows:%d", jobj, tscon, (void *)con, res, ret);
jniTrace("jobj:%p, conn:%p, sql:%p, res: %p, affect rows:%d", jobj, tscon, (void *)con, (void *)res, ret);
return ret;
}
......@@ -411,10 +409,10 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_getSchemaMetaData
// jobject arrayListObj = (*env)->NewObject(env, g_arrayListClass, g_arrayListConstructFp, "");
if (num_fields == 0) {
jniError("jobj:%p, conn:%p, resultset:%p, fields size is %d", jobj, tscon, res, num_fields);
jniError("jobj:%p, conn:%p, resultset:%p, fields size is %d", jobj, tscon, (void *)res, num_fields);
return JNI_NUM_OF_FIELDS_0;
} else {
jniTrace("jobj:%p, conn:%p, resultset:%p, fields size is %d", jobj, tscon, res, num_fields);
jniTrace("jobj:%p, conn:%p, resultset:%p, fields size is %d", jobj, tscon, (void *)res, num_fields);
for (int i = 0; i < num_fields; ++i) {
jobject metadataObj = (*env)->NewObject(env, g_metadataClass, g_metadataConstructFp);
(*env)->SetIntField(env, metadataObj, g_metadataColtypeField, fields[i].type);
......@@ -465,7 +463,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_fetchRowImp(JNIEn
int num_fields = taos_num_fields(result);
if (num_fields == 0) {
jniError("jobj:%p, conn:%p, resultset:%p, fields size is %d", jobj, tscon, res, num_fields);
jniError("jobj:%p, conn:%p, resultset:%p, fields size is %d", jobj, tscon, (void*)res, num_fields);
return JNI_NUM_OF_FIELDS_0;
}
......@@ -473,7 +471,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_fetchRowImp(JNIEn
if (row == NULL) {
int tserrno = taos_errno(result);
if (tserrno == 0) {
jniTrace("jobj:%p, conn:%p, resultset:%p, fields size is %d, fetch row to the end", jobj, tscon, res, num_fields);
jniTrace("jobj:%p, conn:%p, resultset:%p, fields size is %d, fetch row to the end", jobj, tscon, (void*)res, num_fields);
return JNI_FETCH_END;
} else {
jniTrace("jobj:%p, conn:%p, interruptted query", jobj, tscon);
......@@ -571,9 +569,9 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_subscribeImp(JNI
sub = (jlong)tsub;
if (sub == 0) {
jniTrace("jobj:%p, failed to subscribe: topic:%s", jobj, jtopic);
jniTrace("jobj:%p, failed to subscribe: topic:%s", jobj, topic);
} else {
jniTrace("jobj:%p, successfully subscribe: topic: %s", jobj, jtopic);
jniTrace("jobj:%p, successfully subscribe: topic: %s", jobj, topic);
}
if (topic != NULL) (*env)->ReleaseStringUTFChars(env, jtopic, topic);
......@@ -583,7 +581,7 @@ JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_subscribeImp(JNI
}
JNIEXPORT jlong JNICALL Java_com_taosdata_jdbc_TSDBJNIConnector_consumeImp(JNIEnv *env, jobject jobj, jlong sub) {
jniTrace("jobj:%p, in TSDBJNIConnector_consumeImp, sub:%ld", jobj, sub);
jniTrace("jobj:%p, in TSDBJNIConnector_consumeImp, sub:%lld", jobj, sub);
jniGetGlobalMethod(env);
TAOS_SUB *tsub = (TAOS_SUB *)sub;
......
src/client/src/tscAsync.c
浏览文件 @ be9c504e
......@@ -19,9 +19,8 @@
#include "tnote.h"
#include "trpc.h"
#include "tscLog.h"
#include "tscProfile.h"
#include "tscSubquery.h"
#include "tscSecondaryMerge.h"
#include "tscLocalMerge.h"
#include "tscUtil.h"
#include "tsched.h"
#include "tschemautil.h"
......@@ -40,38 +39,26 @@ static void tscProcessAsyncRetrieveImpl(void *param, TAOS_RES *tres, int numOfRo
static void tscAsyncFetchRowsProxy(void *param, TAOS_RES *tres, int numOfRows);
static void tscAsyncFetchSingleRowProxy(void *param, TAOS_RES *tres, int numOfRows);
int doAsyncParseSql(SSqlObj* pSql) {
SSqlCmd* pCmd = &pSql->cmd;
SSqlRes* pRes = &pSql->res;
int32_t code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE);
if (code != TSDB_CODE_SUCCESS) {
tscError("failed to malloc payload");
tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_TSC_OUT_OF_MEMORY);
return code;
}
pRes->qhandle = 0;
pRes->numOfRows = 1;
tscDump("%p SQL: %s", pSql, pSql->sqlstr);
return tsParseSql(pSql, true);
}
void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const char* sqlstr, size_t sqlLen) {
pSql->signature = pSql;
pSql->param = param;
pSql->pTscObj = pObj;
pSql->maxRetry = TSDB_MAX_REPLICA_NUM;
pSql->fp = fp;
pSql->sqlstr = calloc(1, sqlLen + 1);
if (pSql->sqlstr == NULL) {
tscError("%p failed to malloc sql string buffer", pSql);
tscQueueAsyncError(pSql->fp, pSql->param, TSDB_CODE_TSC_OUT_OF_MEMORY);
return;
}
strtolower(pSql->sqlstr, sqlstr);
int32_t code = doAsyncParseSql(pSql);
tscDump("%p SQL: %s", pSql, pSql->sqlstr);
pSql->cmd.curSql = pSql->sqlstr;
int32_t code = tsParseSql(pSql, true);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) return;
if (code != TSDB_CODE_SUCCESS) {
......@@ -211,7 +198,8 @@ void taos_fetch_rows_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, int), voi
if (pRes->qhandle == 0) {
tscError("qhandle is NULL");
tscQueueAsyncError(fp, param, TSDB_CODE_TSC_INVALID_QHANDLE);
pRes->code = TSDB_CODE_TSC_INVALID_QHANDLE;
tscQueueAsyncRes(pSql);
return;
}
......@@ -285,7 +273,7 @@ void taos_fetch_row_a(TAOS_RES *taosa, void (*fp)(void *, TAOS_RES *, TAOS_ROW),
tscProcessSql(pSql);
} else {
SSchedMsg schedMsg;
SSchedMsg schedMsg = { 0 };
schedMsg.fp = tscProcessFetchRow;
schedMsg.ahandle = pSql;
schedMsg.thandle = pRes->tsrow;
......@@ -387,7 +375,7 @@ void tscQueueAsyncError(void(*fp), void *param, int32_t code) {
int32_t* c = malloc(sizeof(int32_t));
*c = code;
SSchedMsg schedMsg;
SSchedMsg schedMsg = { 0 };
schedMsg.fp = tscProcessAsyncError;
schedMsg.ahandle = fp;
schedMsg.thandle = param;
......@@ -403,7 +391,7 @@ void tscQueueAsyncRes(SSqlObj *pSql) {
tscError("%p add into queued async res, code:%s", pSql, tstrerror(pSql->res.code));
}
SSchedMsg schedMsg;
SSchedMsg schedMsg = { 0 };
schedMsg.fp = tscProcessAsyncRes;
schedMsg.ahandle = pSql;
schedMsg.thandle = (void *)1;
......@@ -420,7 +408,7 @@ void tscProcessAsyncFree(SSchedMsg *pMsg) {
void tscQueueAsyncFreeResult(SSqlObj *pSql) {
tscTrace("%p sqlObj put in queue to async free", pSql);
SSchedMsg schedMsg;
SSchedMsg schedMsg = { 0 };
schedMsg.fp = tscProcessAsyncFree;
schedMsg.ahandle = pSql;
schedMsg.thandle = (void *)1;
......
src/client/src/tscFunctionImpl.c
浏览文件 @ be9c504e
......@@ -153,7 +153,7 @@ typedef struct SRateInfo {
int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionId, int32_t param, int16_t *type,
int16_t *bytes, int32_t *interBytes, int16_t extLength, bool isSuperTable) {
if (!isValidDataType(dataType, dataBytes)) {
if (!isValidDataType(dataType)) {
tscError("Illegal data type %d or data type length %d", dataType, dataBytes);
return TSDB_CODE_TSC_INVALID_SQL;
}
......@@ -1853,26 +1853,14 @@ static void last_row_function(SQLFunctionCtx *pCtx) {
static void last_row_finalizer(SQLFunctionCtx *pCtx) {
// do nothing at the first stage
SResultInfo *pResInfo = GET_RES_INFO(pCtx);
if (pCtx->currentStage == SECONDARY_STAGE_MERGE) {
if (pResInfo->hasResult != DATA_SET_FLAG) {
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
}
return;
}
} else {
if (pResInfo->hasResult != DATA_SET_FLAG) {
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
}
return;
if (pResInfo->hasResult != DATA_SET_FLAG) {
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
}
return;
}
GET_RES_INFO(pCtx)->numOfRes = 1;
......@@ -2989,12 +2977,12 @@ static void tag_project_function_f(SQLFunctionCtx *pCtx, int32_t index) {
*/
static void tag_function(SQLFunctionCtx *pCtx) {
SET_VAL(pCtx, 1, 1);
tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->tag.nType, true);
tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->outputType, true);
}
static void tag_function_f(SQLFunctionCtx *pCtx, int32_t index) {
SET_VAL(pCtx, 1, 1);
tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->tag.nType, true);
tVariantDump(&pCtx->tag, pCtx->aOutputBuf, pCtx->outputType, true);
}
static void copy_function(SQLFunctionCtx *pCtx) {
......@@ -3903,7 +3891,7 @@ static bool ts_comp_function_setup(SQLFunctionCtx *pCtx) {
SResultInfo *pResInfo = GET_RES_INFO(pCtx);
STSCompInfo *pInfo = pResInfo->interResultBuf;
pInfo->pTSBuf = tsBufCreate(false);
pInfo->pTSBuf = tsBufCreate(false, pCtx->order);
pInfo->pTSBuf->tsOrder = pCtx->order;
return true;
}
......@@ -3925,7 +3913,6 @@ static void ts_comp_function(SQLFunctionCtx *pCtx) {
}
SET_VAL(pCtx, pCtx->size, 1);
pResInfo->hasResult = DATA_SET_FLAG;
}
......
src/client/src/tscLocal.c
浏览文件 @ be9c504e
......@@ -22,7 +22,6 @@
#include "taosdef.h"
#include "tscLog.h"
#include "qextbuffer.h"
#include "tscSecondaryMerge.h"
#include "tschemautil.h"
#include "tname.h"
......@@ -183,8 +182,12 @@ static int32_t tscSetValueToResObj(SSqlObj *pSql, int32_t rowLen) {
// type length
int32_t bytes = pSchema[i].bytes;
pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, 2);
if (pSchema[i].type == TSDB_DATA_TYPE_NCHAR) {
bytes = bytes / TSDB_NCHAR_SIZE;
if (pSchema[i].type == TSDB_DATA_TYPE_BINARY || pSchema[i].type == TSDB_DATA_TYPE_NCHAR) {
bytes -= VARSTR_HEADER_SIZE;
if (pSchema[i].type == TSDB_DATA_TYPE_NCHAR) {
bytes = bytes / TSDB_NCHAR_SIZE;
}
}
*(int32_t *)(pRes->data + tscFieldInfoGetOffset(pQueryInfo, 2) * totalNumOfRows + pField->bytes * i) = bytes;
......
src/client/src/tscSecondaryMerge.c → src/client/src/tscLocalMerge.c
浏览文件 @ be9c504e
......@@ -13,7 +13,6 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tscSecondaryMerge.h"
#include "os.h"
#include "tlosertree.h"
#include "tscUtil.h"
......@@ -21,6 +20,7 @@
#include "tsclient.h"
#include "tutil.h"
#include "tscLog.h"
#include "tscLocalMerge.h"
typedef struct SCompareParam {
SLocalDataSource **pLocalData;
......@@ -230,6 +230,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
if (ds == NULL) {
tscError("%p failed to create merge structure", pSql);
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
tfree(pReducer);
return;
}
......@@ -266,6 +267,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
// no data actually, no need to merge result.
if (idx == 0) {
tfree(pReducer);
return;
}
......@@ -282,6 +284,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
pRes->code = tLoserTreeCreate(&pReducer->pLoserTree, pReducer->numOfBuffer, param, treeComparator);
if (pReducer->pLoserTree == NULL || pRes->code != 0) {
tfree(pReducer);
return;
}
......@@ -325,7 +328,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
tfree(pReducer->pResultBuf);
tfree(pReducer->pFinalRes);
tfree(pReducer->prevRowOfInput);
tfree(pReducer);
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
return;
}
......@@ -353,7 +356,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
pRes->numOfGroups = 0;
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);;
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
TSKEY stime = MIN(pQueryInfo->window.skey, pQueryInfo->window.ekey);
int64_t revisedSTime =
......@@ -410,13 +413,13 @@ static int32_t tscFlushTmpBufferImpl(tExtMemBuffer *pMemoryBuf, tOrderDescriptor
}
int32_t tscFlushTmpBuffer(tExtMemBuffer *pMemoryBuf, tOrderDescriptor *pDesc, tFilePage *pPage, int32_t orderType) {
int32_t ret = tscFlushTmpBufferImpl(pMemoryBuf, pDesc, pPage, orderType);
if (ret != 0) {
return -1;
int32_t ret = 0;
if ((ret = tscFlushTmpBufferImpl(pMemoryBuf, pDesc, pPage, orderType)) != 0) {
return ret;
}
if (!tExtMemBufferFlush(pMemoryBuf)) {
return -1;
if ((ret = tExtMemBufferFlush(pMemoryBuf)) != 0) {
return ret;
}
return 0;
......@@ -437,9 +440,9 @@ int32_t saveToBuffer(tExtMemBuffer *pMemoryBuf, tOrderDescriptor *pDesc, tFilePa
// current buffer is full, need to flushed to disk
assert(pPage->num == pModel->capacity);
int32_t ret = tscFlushTmpBuffer(pMemoryBuf, pDesc, pPage, orderType);
if (ret != 0) {
return -1;
int32_t code = tscFlushTmpBuffer(pMemoryBuf, pDesc, pPage, orderType);
if (code != 0) {
return code;
}
int32_t remain = numOfRows - numOfRemainEntries;
......@@ -455,8 +458,8 @@ int32_t saveToBuffer(tExtMemBuffer *pMemoryBuf, tOrderDescriptor *pDesc, tFilePa
tColModelAppend(pModel, pPage, data, numOfRows - remain, numOfWriteElems, numOfRows);
if (pPage->num == pModel->capacity) {
if (tscFlushTmpBuffer(pMemoryBuf, pDesc, pPage, orderType) != TSDB_CODE_SUCCESS) {
return -1;
if ((code = tscFlushTmpBuffer(pMemoryBuf, pDesc, pPage, orderType)) != TSDB_CODE_SUCCESS) {
return code;
}
} else {
pPage->num = numOfWriteElems;
......@@ -685,6 +688,7 @@ int32_t tscLocalReducerEnvCreate(SSqlObj *pSql, tExtMemBuffer ***pMemBuffer, tOr
if (createOrderDescriptor(pOrderDesc, pCmd, pModel) != TSDB_CODE_SUCCESS) {
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
tfree(pSchema);
return pRes->code;
}
......@@ -1092,14 +1096,6 @@ static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx)
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t j = 0; j < size; ++j) {
// SSqlExpr* pExpr = pQueryInfo->fieldsInfo.pSqlExpr[j];
// if (pExpr == NULL) {
// assert(pQueryInfo->fieldsInfo.pExpr[j] != NULL);
//
// maxOutput = 1;
// continue;
// }
/*
* ts, tag, tagprj function can not decide the output number of current query
* the number of output result is decided by main output
......@@ -1109,8 +1105,9 @@ static int64_t getNumOfResultLocal(SQueryInfo *pQueryInfo, SQLFunctionCtx *pCtx)
continue;
}
if (maxOutput < GET_RES_INFO(&pCtx[j])->numOfRes) {
maxOutput = GET_RES_INFO(&pCtx[j])->numOfRes;
SResultInfo* pResInfo = GET_RES_INFO(&pCtx[j]);
if (maxOutput < pResInfo->numOfRes) {
maxOutput = pResInfo->numOfRes;
}
}
......@@ -1260,7 +1257,6 @@ bool doGenerateFinalResults(SSqlObj *pSql, SLocalReducer *pLocalReducer, bool no
#ifdef _DEBUG_VIEW
printf("final result before interpo:\n");
assert(0);
// tColModelDisplay(pLocalReducer->resColModel, pLocalReducer->pBufForInterpo, pResBuf->num, pResBuf->num);
#endif
......
src/client/src/tscParseInsert.c
浏览文件 @ be9c504e
......@@ -252,7 +252,7 @@ int32_t tsParseOneColumnData(SSchema *pSchema, SSQLToken *pToken, char *payload,
numType = tscToInteger(pToken, &iv, &endptr);
if (TK_ILLEGAL == numType) {
return tscInvalidSQLErrMsg(msg, "invalid bigint data", pToken->z);
} else if (errno == ERANGE || iv > INT64_MAX || iv <= INT64_MIN) {
} else if (errno == ERANGE || iv == INT64_MIN) {
return tscInvalidSQLErrMsg(msg, "bigint data overflow", pToken->z);
}
......@@ -594,7 +594,6 @@ int32_t tscAllocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int3
size_t remain = pDataBlock->nAllocSize - pDataBlock->size;
const int factor = 5;
uint32_t nAllocSizeOld = pDataBlock->nAllocSize;
assert(pDataBlock->headerSize >= 0);
// expand the allocated size
if (remain < rowSize * factor) {
......@@ -987,14 +986,16 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
return code;
}
int validateTableName(char *tblName, int len) {
int validateTableName(char *tblName, int len, SSQLToken* psTblToken) {
char buf[TSDB_TABLE_ID_LEN] = {0};
tstrncpy(buf, tblName, sizeof(buf));
SSQLToken token = {.n = len, .type = TK_ID, .z = buf};
tSQLGetToken(buf, &token.type);
psTblToken->n = len;
psTblToken->type = TK_ID;
psTblToken->z = buf;
tSQLGetToken(buf, &psTblToken->type);
return tscValidateName(&token);
return tscValidateName(psTblToken);
}
static int32_t validateDataSource(SSqlCmd *pCmd, int8_t type, const char *sql) {
......@@ -1015,46 +1016,41 @@ static int32_t validateDataSource(SSqlCmd *pCmd, int8_t type, const char *sql) {
* @param pSql
* @return
*/
int doParseInsertSql(SSqlObj *pSql, char *str) {
int tsParseInsertSql(SSqlObj *pSql) {
SSqlCmd *pCmd = &pSql->cmd;
char* str = pCmd->curSql;
int32_t totalNum = 0;
int32_t code = TSDB_CODE_SUCCESS;
STableMetaInfo *pTableMetaInfo = NULL;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
assert(pQueryInfo != NULL);
STableMetaInfo *pTableMetaInfo = NULL;
if (pQueryInfo->numOfTables == 0) {
pTableMetaInfo = tscAddEmptyMetaInfo(pQueryInfo);
} else {
pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
}
// TODO: 2048 is added because TSDB_MAX_TAGS_LEN now is 65536
// but TSDB_PAYLOAD_SIZE is 65380
// TODO: 2048 is added because TSDB_MAX_TAGS_LEN now is 65536, but TSDB_PAYLOAD_SIZE is 65380
if ((code = tscAllocPayload(pCmd, TSDB_PAYLOAD_SIZE + 2048)) != TSDB_CODE_SUCCESS) {
return code;
}
assert(((NULL == pCmd->curSql) && (NULL == pCmd->pTableList))
|| ((NULL != pCmd->curSql) && (NULL != pCmd->pTableList)));
if ((NULL == pCmd->curSql) && (NULL == pCmd->pTableList)) {
if (NULL == pCmd->pTableList) {
pCmd->pTableList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false);
pSql->cmd.pDataBlocks = tscCreateBlockArrayList();
if (NULL == pCmd->pTableList || NULL == pSql->cmd.pDataBlocks) {
code = TSDB_CODE_TSC_OUT_OF_MEMORY;
goto _error_clean;
goto _error;
}
} else {
assert((NULL != pCmd->curSql) && (NULL != pCmd->pTableList));
str = pCmd->curSql;
}
tscTrace("%p create data block list for submit data:%p, curSql:%p, pTableList:%p", pSql, pSql->cmd.pDataBlocks, pCmd->curSql, pCmd->pTableList);
tscTrace("%p create data block list for submit data:%p, pTableList:%p", pSql, pCmd->pDataBlocks, pCmd->pTableList);
while (1) {
int32_t index = 0;
......@@ -1076,48 +1072,41 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
*/
if (totalNum == 0) {
code = TSDB_CODE_TSC_INVALID_SQL;
goto _error_clean;
goto _error;
} else {
break;
}
}
pCmd->curSql = sToken.z;
SSQLToken sTblToken;
// Check if the table name available or not
if (validateTableName(sToken.z, sToken.n) != TSDB_CODE_SUCCESS) {
if (validateTableName(sToken.z, sToken.n, &sTblToken) != TSDB_CODE_SUCCESS) {
code = tscInvalidSQLErrMsg(pCmd->payload, "table name invalid", sToken.z);
goto _error_clean;
goto _error;
}
if ((code = tscSetTableFullName(pTableMetaInfo, &sToken, pSql)) != TSDB_CODE_SUCCESS) {
goto _error_clean;
if ((code = tscSetTableFullName(pTableMetaInfo, &sTblToken, pSql)) != TSDB_CODE_SUCCESS) {
goto _error;
}
ptrdiff_t pos = pCmd->curSql - pSql->sqlstr;
if ((code = tscCheckIfCreateTable(&str, pSql)) != TSDB_CODE_SUCCESS) {
/*
* For async insert, after get the table meta from server, the sql string will not be
* parsed using the new table meta to avoid the overhead cause by get table meta data information.
* And during the getMeterMetaCallback function, the sql string will be parsed from the
* interrupted position.
* After retrieving the table meta from server, the sql string will be parsed from the paused position.
* And during the getTableMetaCallback function, the sql string will be parsed from the paused position.
*/
if (TSDB_CODE_TSC_ACTION_IN_PROGRESS == code) {
tscTrace("%p waiting for get table meta during insert, then resume from offset: %" PRId64 " , %s", pSql,
pos, pCmd->curSql);
return code;
}
// todo add to return
tscError("%p async insert parse error, code:%d, %s", pSql, code, tstrerror(code));
tscError("%p async insert parse error, code:%s", pSql, tstrerror(code));
pCmd->curSql = NULL;
goto _error_clean; // TODO: should _clean or _error_clean to async flow ????
goto _error;
}
if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
code = tscInvalidSQLErrMsg(pCmd->payload, "insert data into super table is not supported", NULL);
goto _error_clean;
goto _error;
}
index = 0;
......@@ -1126,7 +1115,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
if (sToken.n == 0) {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES or FILE required", sToken.z);
goto _error_clean;
goto _error;
}
STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
......@@ -1138,7 +1127,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
tscSetAssignedColumnInfo(&spd, pSchema, tinfo.numOfColumns);
if (validateDataSource(pCmd, DATA_FROM_SQL_STRING, sToken.z) != TSDB_CODE_SUCCESS) {
goto _error_clean;
goto _error;
}
/*
......@@ -1147,11 +1136,11 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
*/
code = doParseInsertStatement(pSql, pCmd->pTableList, &str, &spd, &totalNum);
if (code != TSDB_CODE_SUCCESS) {
goto _error_clean;
goto _error;
}
} else if (sToken.type == TK_FILE) {
if (validateDataSource(pCmd, DATA_FROM_DATA_FILE, sToken.z) != TSDB_CODE_SUCCESS) {
goto _error_clean;
goto _error;
}
index = 0;
......@@ -1159,7 +1148,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
str += index;
if (sToken.n == 0) {
code = tscInvalidSQLErrMsg(pCmd->payload, "file path is required following keyword FILE", sToken.z);
goto _error_clean;
goto _error;
}
char fname[PATH_MAX] = {0};
......@@ -1169,7 +1158,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
wordexp_t full_path;
if (wordexp(fname, &full_path, 0) != 0) {
code = tscInvalidSQLErrMsg(pCmd->payload, "invalid filename", sToken.z);
goto _error_clean;
goto _error;
}
strcpy(fname, full_path.we_wordv[0]);
wordfree(&full_path);
......@@ -1180,7 +1169,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
int32_t ret = tscCreateDataBlock(PATH_MAX, tinfo.rowSize, sizeof(SSubmitBlk), pTableMetaInfo->name,
pTableMeta, &pDataBlock);
if (ret != TSDB_CODE_SUCCESS) {
goto _error_clean;
goto _error;
}
tscAppendDataBlock(pCmd->pDataBlocks, pDataBlock);
......@@ -1191,7 +1180,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
SSchema * pSchema = tscGetTableSchema(pTableMeta);
if (validateDataSource(pCmd, DATA_FROM_SQL_STRING, sToken.z) != TSDB_CODE_SUCCESS) {
goto _error_clean;
goto _error;
}
SParsedDataColInfo spd = {0};
......@@ -1227,7 +1216,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
if (spd.hasVal[t] == true) {
code = tscInvalidSQLErrMsg(pCmd->payload, "duplicated column name", sToken.z);
goto _error_clean;
goto _error;
}
spd.hasVal[t] = true;
......@@ -1238,13 +1227,13 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
if (!findColumnIndex) {
code = tscInvalidSQLErrMsg(pCmd->payload, "invalid column name", sToken.z);
goto _error_clean;
goto _error;
}
}
if (spd.numOfAssignedCols == 0 || spd.numOfAssignedCols > tinfo.numOfColumns) {
code = tscInvalidSQLErrMsg(pCmd->payload, "column name expected", sToken.z);
goto _error_clean;
goto _error;
}
index = 0;
......@@ -1253,16 +1242,16 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
if (sToken.type != TK_VALUES) {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES is expected", sToken.z);
goto _error_clean;
goto _error;
}
code = doParseInsertStatement(pSql, pCmd->pTableList, &str, &spd, &totalNum);
if (code != TSDB_CODE_SUCCESS) {
goto _error_clean;
goto _error;
}
} else {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES or FILE are required", sToken.z);
goto _error_clean;
goto _error;
}
}
......@@ -1273,7 +1262,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
if (pCmd->pDataBlocks->nSize > 0) { // merge according to vgId
if ((code = tscMergeTableDataBlocks(pSql, pCmd->pDataBlocks)) != TSDB_CODE_SUCCESS) {
goto _error_clean;
goto _error;
}
} else {
pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks);
......@@ -1282,7 +1271,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
code = TSDB_CODE_SUCCESS;
goto _clean;
_error_clean:
_error:
pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks);
_clean:
......@@ -1295,7 +1284,7 @@ _clean:
return code;
}
int tsParseInsertSql(SSqlObj *pSql) {
int tsInsertInitialCheck(SSqlObj *pSql) {
if (!pSql->pTscObj->writeAuth) {
return TSDB_CODE_TSC_NO_WRITE_AUTH;
}
......@@ -1313,28 +1302,23 @@ int tsParseInsertSql(SSqlObj *pSql) {
SQueryInfo *pQueryInfo = NULL;
tscGetQueryInfoDetailSafely(pCmd, pCmd->clauseIndex, &pQueryInfo);
TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_INSERT);
TSDB_QUERY_SET_TYPE(pQueryInfo->type, pCmd->insertType);
TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_INSERT | pCmd->insertType);
sToken = tStrGetToken(pSql->sqlstr, &index, false, 0, NULL);
if (sToken.type != TK_INTO) {
return tscInvalidSQLErrMsg(pCmd->payload, "keyword INTO is expected", sToken.z);
}
return doParseInsertSql(pSql, pSql->sqlstr + index);
pCmd->curSql = sToken.z + sToken.n;
return TSDB_CODE_SUCCESS;
}
int tsParseSql(SSqlObj *pSql, bool initialParse) {
int tsParseSql(SSqlObj *pSql, bool initial) {
int32_t ret = TSDB_CODE_SUCCESS;
if (initialParse) {
char* p = pSql->sqlstr;
pSql->sqlstr = NULL;
tscPartiallyFreeSqlObj(pSql);
pSql->sqlstr = p;
} else {
tscTrace("continue parse sql: %s", pSql->cmd.curSql);
SSqlCmd* pCmd = &pSql->cmd;
if ((!pCmd->parseFinished) && (!initial)) {
tscTrace("%p resume to parse sql: %s", pSql, pCmd->curSql);
}
if (tscIsInsertData(pSql->sqlstr)) {
......@@ -1342,18 +1326,22 @@ int tsParseSql(SSqlObj *pSql, bool initialParse) {
* Set the fp before parse the sql string, in case of getTableMeta failed, in which
* the error handle callback function can rightfully restore the user-defined callback function (fp).
*/
if (initialParse && (pSql->cmd.insertType != TSDB_QUERY_TYPE_STMT_INSERT)) {
if (initial && (pSql->cmd.insertType != TSDB_QUERY_TYPE_STMT_INSERT)) {
pSql->fetchFp = pSql->fp;
pSql->fp = (void(*)())tscHandleMultivnodeInsert;
}
if (initial && ((ret = tsInsertInitialCheck(pSql)) != TSDB_CODE_SUCCESS)) {
return ret;
}
ret = tsParseInsertSql(pSql);
} else {
ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
if (TSDB_CODE_SUCCESS != ret) {
return ret;
}
SSqlInfo SQLInfo = qSQLParse(pSql->sqlstr);
ret = tscToSQLCmd(pSql, &SQLInfo);
SQLInfoDestroy(&SQLInfo);
......@@ -1361,11 +1349,9 @@ int tsParseSql(SSqlObj *pSql, bool initialParse) {
/*
* the pRes->code may be modified or released by another thread in tscTableMetaCallBack function,
* so do NOT use pRes->code to determine if the getTableMeta/getMetricMeta function
* invokes new threads to get data from mnode or simply retrieves data from cache.
*
* do NOT assign return code to pRes->code for the same reason since it may be released by another thread
* pRes->code = ret;
* so do NOT use pRes->code to determine if the getTableMeta function
* invokes new threads to get data from mgmt node or simply retrieves data from cache.
* do NOT assign return code to pRes->code for the same reason since it may be released by another thread already.
*/
return ret;
}
......
src/client/src/tscSQLParser.c
浏览文件 @ be9c504e
此差异已折叠。
src/client/src/tscSchemaUtil.c
浏览文件 @ be9c504e
......@@ -32,7 +32,6 @@ int32_t tscGetNumOfTags(const STableMeta* pTableMeta) {
}
if (pTableMeta->tableType == TSDB_SUPER_TABLE || pTableMeta->tableType == TSDB_CHILD_TABLE) {
assert(tinfo.numOfTags >= 0);
return tinfo.numOfTags;
}
......@@ -64,14 +63,6 @@ SSchema* tscGetTableTagSchema(const STableMeta* pTableMeta) {
STableComInfo tscGetTableInfo(const STableMeta* pTableMeta) {
assert(pTableMeta != NULL);
#if 0
if (pTableMeta->tableType == TSDB_CHILD_TABLE) {
assert (pTableMeta->pSTable != NULL);
return pTableMeta->pSTable->tableInfo;
}
#endif
return pTableMeta->tableInfo;
}
......@@ -119,11 +110,24 @@ bool isValidSchema(struct SSchema* pSchema, int32_t numOfCols) {
return (rowLen <= TSDB_MAX_BYTES_PER_ROW);
}
SSchema* tscGetTableColumnSchema(const STableMeta* pTableMeta, int32_t startCol) {
SSchema* tscGetTableColumnSchema(const STableMeta* pTableMeta, int32_t colIndex) {
assert(pTableMeta != NULL);
SSchema* pSchema = (SSchema*) pTableMeta->schema;
return &pSchema[startCol];
return &pSchema[colIndex];
}
// TODO for large number of columns, employ the binary search method
SSchema* tscGetTableColumnSchemaById(STableMeta* pTableMeta, int16_t colId) {
STableComInfo tinfo = tscGetTableInfo(pTableMeta);
for(int32_t i = 0; i < tinfo.numOfColumns + tinfo.numOfTags; ++i) {
if (pTableMeta->schema[i].colId == colId) {
return &pTableMeta->schema[i];
}
}
return NULL;
}
struct SSchema tscGetTbnameColumnSchema() {
......
src/client/src/tscServer.c
浏览文件 @ be9c504e
......@@ -14,20 +14,18 @@
*/
#include "os.h"
#include "qsqltype.h"
#include "tcache.h"
#include "trpc.h"
#include "tscLocalMerge.h"
#include "tscLog.h"
#include "tscProfile.h"
#include "tscSecondaryMerge.h"
#include "tscSubquery.h"
#include "tscUtil.h"
#include "tschemautil.h"
#include "tsclient.h"
#include "tsocket.h"
#include "ttime.h"
#include "ttimer.h"
#include "tutil.h"
#include "tscLog.h"
#include "qsqltype.h"
#define TSC_MGMT_VNODE 999
......@@ -198,30 +196,35 @@ int tscSendMsgToServer(SSqlObj *pSql) {
.handle = pSql,
.code = 0
};
rpcSendRequest(pObj->pDnodeConn, &pSql->ipList, &rpcMsg);
pSql->pRpcCtx = rpcSendRequest(pObj->pDnodeConn, &pSql->ipList, &rpcMsg);
return TSDB_CODE_SUCCESS;
}
void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
SSqlObj *pSql = (SSqlObj *)rpcMsg->handle;
if (pSql == NULL) {
if (pSql == NULL || pSql->signature != pSql) {
tscError("%p sql is already released", pSql->signature);
return;
}
if (pSql->signature != pSql) {
tscError("%p sql is already released, signature:%p", pSql, pSql->signature);
return;
}
STscObj *pObj = pSql->pTscObj;
SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd;
STscObj *pObj = pSql->pTscObj;
if (pObj->signature != pObj || pSql->freed == 1) {
tscTrace("%p sqlObj needs to be released or DB connection is closed, freed:%d pObj:%p signature:%p", pSql, pSql->freed,
if (pObj->signature != pObj) {
tscTrace("%p DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command, pObj, pObj->signature);
tscFreeSqlObj(pSql);
rpcFreeCont(rpcMsg->pCont);
return;
}
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
if (pQueryInfo != NULL && pQueryInfo->type == TSDB_QUERY_TYPE_FREE_RESOURCE) {
tscTrace("%p sqlObj needs to be released or DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command,
pObj, pObj->signature);
tscFreeSqlObj(pSql);
rpcFreeCont(rpcMsg->pCont);
return;
......@@ -274,7 +277,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
if (pRes->code != TSDB_CODE_TSC_QUERY_CANCELLED) {
pRes->code = (rpcMsg->code != TSDB_CODE_SUCCESS) ? rpcMsg->code : TSDB_CODE_RPC_NETWORK_UNAVAIL;
} else {
tscTrace("%p query is cancelled, code:%d", pSql, tstrerror(pRes->code));
tscTrace("%p query is cancelled, code:%s", pSql, tstrerror(pRes->code));
}
if (pRes->code == TSDB_CODE_SUCCESS) {
......@@ -414,7 +417,6 @@ void tscKillSTableQuery(SSqlObj *pSql) {
for (int i = 0; i < pSql->numOfSubs; ++i) {
SSqlObj *pSub = pSql->pSubs[i];
if (pSub == NULL) {
continue;
}
......@@ -424,7 +426,7 @@ void tscKillSTableQuery(SSqlObj *pSql) {
* sub-queries not correctly released and master sql object of super table query reaches an abnormal state.
*/
pSql->pSubs[i]->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
// taosStopRpcConn(pSql->pSubs[i]->);
rpcCancelRequest(pSql->pSubs[i]->pRpcCtx);
}
/*
......@@ -644,14 +646,10 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg->numOfGroupCols = htons(pQueryInfo->groupbyExpr.numOfGroupCols);
pQueryMsg->numOfTags = htonl(numOfTags);
pQueryMsg->tagNameRelType = htons(pQueryInfo->tagCond.relType);
pQueryMsg->queryType = htons(pQueryInfo->type);
pQueryMsg->queryType = htonl(pQueryInfo->type);
size_t numOfOutput = tscSqlExprNumOfExprs(pQueryInfo);
pQueryMsg->numOfOutput = htons(numOfOutput);
if (numOfOutput < 0) {
tscError("%p illegal value of number of output columns in query msg: %d", pSql, numOfOutput);
return -1;
}
// set column list ids
size_t numOfCols = taosArrayGetSize(pQueryInfo->colList);
......@@ -665,7 +663,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
if (pCol->colIndex.columnIndex >= tscGetNumOfColumns(pTableMeta) || pColSchema->type < TSDB_DATA_TYPE_BOOL ||
pColSchema->type > TSDB_DATA_TYPE_NCHAR) {
tscError("%p sid:%d uid:%" PRIu64" id:%s, column index out of range, numOfColumns:%d, index:%d, column name:%s",
pSql, pTableMeta->sid, pTableMeta->uid, pTableMetaInfo->name, tscGetNumOfColumns(pTableMeta), pCol->colIndex,
pSql, pTableMeta->sid, pTableMeta->uid, pTableMetaInfo->name, tscGetNumOfColumns(pTableMeta), pCol->colIndex.columnIndex,
pColSchema->name);
return TSDB_CODE_TSC_INVALID_SQL;
......@@ -723,6 +721,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pMsg += sizeof(SSqlFuncMsg);
for (int32_t j = 0; j < pExpr->numOfParams; ++j) {
// todo add log
pSqlFuncExpr->arg[j].argType = htons((uint16_t)pExpr->param[j].nType);
pSqlFuncExpr->arg[j].argBytes = htons(pExpr->param[j].nLen);
......@@ -784,7 +783,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
(pColSchema->type < TSDB_DATA_TYPE_BOOL || pColSchema->type > TSDB_DATA_TYPE_NCHAR)) {
tscError("%p sid:%d uid:%" PRIu64 " id:%s, tag index out of range, totalCols:%d, numOfTags:%d, index:%d, column name:%s",
pSql, pTableMeta->sid, pTableMeta->uid, pTableMetaInfo->name, total, numOfTagColumns,
pCol->colIndex, pColSchema->name);
pCol->colIndex.columnIndex, pColSchema->name);
return TSDB_CODE_TSC_INVALID_SQL;
}
......@@ -800,6 +799,27 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
}
// serialize tag column query condition
if (pQueryInfo->tagCond.pCond != NULL && taosArrayGetSize(pQueryInfo->tagCond.pCond) > 0) {
STagCond* pTagCond = &pQueryInfo->tagCond;
SCond *pCond = tsGetSTableQueryCond(pTagCond, pTableMeta->uid);
if (pCond != NULL && pCond->cond != NULL) {
pQueryMsg->tagCondLen = htons(pCond->len);
memcpy(pMsg, pCond->cond, pCond->len);
pMsg += pCond->len;
}
}
if (pQueryInfo->tagCond.tbnameCond.cond == NULL) {
*pMsg = 0;
pMsg++;
} else {
strcpy(pMsg, pQueryInfo->tagCond.tbnameCond.cond);
pMsg += strlen(pQueryInfo->tagCond.tbnameCond.cond) + 1;
}
// compressed ts block
pQueryMsg->tsOffset = htonl(pMsg - pStart);
int32_t tsLen = 0;
......@@ -824,27 +844,6 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg->tsOrder = htonl(pQueryInfo->tsBuf->tsOrder);
}
// serialize tag column query condition
if (pQueryInfo->tagCond.pCond != NULL && taosArrayGetSize(pQueryInfo->tagCond.pCond) > 0) {
STagCond* pTagCond = &pQueryInfo->tagCond;
SCond *pCond = tsGetSTableQueryCond(pTagCond, pTableMeta->uid);
if (pCond != NULL && pCond->cond != NULL) {
pQueryMsg->tagCondLen = htons(pCond->len);
memcpy(pMsg, pCond->cond, pCond->len);
pMsg += pCond->len;
}
}
if (pQueryInfo->tagCond.tbnameCond.cond == NULL) {
*pMsg = 0;
pMsg++;
} else {
strcpy(pMsg, pQueryInfo->tagCond.tbnameCond.cond);
pMsg += strlen(pQueryInfo->tagCond.tbnameCond.cond) + 1;
}
int32_t msgLen = pMsg - pStart;
tscTrace("%p msg built success,len:%d bytes", pSql, msgLen);
......@@ -983,7 +982,7 @@ int32_t tscBuildDropDbMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SCMDropDbMsg *pDropDbMsg = (SCMDropDbMsg*)pCmd->payload;
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
strncpy(pDropDbMsg->db, pTableMetaInfo->name, tListLen(pDropDbMsg->db));
tstrncpy(pDropDbMsg->db, pTableMetaInfo->name, sizeof(pDropDbMsg->db));
pDropDbMsg->ignoreNotExists = pInfo->pDCLInfo->existsCheck ? 1 : 0;
pCmd->msgType = TSDB_MSG_TYPE_CM_DROP_DB;
......@@ -1053,7 +1052,7 @@ int32_t tscBuildDropAcctMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SCMDropUserMsg *pDropMsg = (SCMDropUserMsg*)pCmd->payload;
STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
strcpy(pDropMsg->user, pTableMetaInfo->name);
tstrncpy(pDropMsg->user, pTableMetaInfo->name, sizeof(pDropMsg->user));
return TSDB_CODE_SUCCESS;
}
......@@ -1175,7 +1174,7 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
strcpy(pCreateTableMsg->tableId, pTableMetaInfo->name);
// use dbinfo from table id without modifying current db info
tscGetDBInfoFromMeterId(pTableMetaInfo->name, pCreateTableMsg->db);
tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pCreateTableMsg->db);
SCreateTableSQL *pCreateTable = pInfo->pCreateTableInfo;
......@@ -1252,7 +1251,7 @@ int tscBuildAlterTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
SCMAlterTableMsg *pAlterTableMsg = (SCMAlterTableMsg *)pCmd->payload;
tscGetDBInfoFromMeterId(pTableMetaInfo->name, pAlterTableMsg->db);
tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pAlterTableMsg->db);
strcpy(pAlterTableMsg->tableId, pTableMetaInfo->name);
pAlterTableMsg->type = htons(pAlterInfo->type);
......@@ -1473,7 +1472,7 @@ int tscBuildTableMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pMsg += len;
}
pCmd->payloadLen = pMsg - (char*)pInfoMsg;;
pCmd->payloadLen = pMsg - (char*)pInfoMsg;
pCmd->msgType = TSDB_MSG_TYPE_CM_TABLE_META;
tfree(tmpData);
......@@ -1577,7 +1576,7 @@ int tscBuildSTableVgroupMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pMsg = pStart;
SMgmtHead *pMgmt = (SMgmtHead *)pMsg;
tscGetDBInfoFromMeterId(pTableMetaInfo->name, pMgmt->db);
tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pMgmt->db);
pMsg += sizeof(SMgmtHead);
......@@ -1813,6 +1812,7 @@ int tscProcessTableMetaRsp(SSqlObj *pSql) {
// todo handle out of memory case
if (pTableMetaInfo->pTableMeta == NULL) {
free(pTableMeta);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
......@@ -1932,113 +1932,6 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
}
int tscProcessSTableVgroupRsp(SSqlObj *pSql) {
#if 0
void ** metricMetaList = NULL;
int32_t * sizes = NULL;
int32_t num = htons(*(int16_t *)rsp);
rsp += sizeof(int16_t);
metricMetaList = calloc(1, POINTER_BYTES * num);
sizes = calloc(1, sizeof(int32_t) * num);
// return with error code
if (metricMetaList == NULL || sizes == NULL) {
tfree(metricMetaList);
tfree(sizes);
pSql->res.code = TSDB_CODE_TSC_OUT_OF_MEMORY;
return pSql->res.code;
}
for (int32_t k = 0; k < num; ++k) {
pMeta = (SSuperTableMeta *)rsp;
size_t size = (size_t)pSql->res.rspLen - 1;
rsp = rsp + sizeof(SSuperTableMeta);
pMeta->numOfTables = htonl(pMeta->numOfTables);
pMeta->numOfVnodes = htonl(pMeta->numOfVnodes);
pMeta->tagLen = htons(pMeta->tagLen);
size += pMeta->numOfVnodes * sizeof(SVnodeSidList *) + pMeta->numOfTables * sizeof(STableIdInfo *);
char *pBuf = calloc(1, size);
if (pBuf == NULL) {
pSql->res.code = TSDB_CODE_TSC_OUT_OF_MEMORY;
goto _error_clean;
}
SSuperTableMeta *pNewMetricMeta = (SSuperTableMeta *)pBuf;
metricMetaList[k] = pNewMetricMeta;
pNewMetricMeta->numOfTables = pMeta->numOfTables;
pNewMetricMeta->numOfVnodes = pMeta->numOfVnodes;
pNewMetricMeta->tagLen = pMeta->tagLen;
pBuf = pBuf + sizeof(SSuperTableMeta) + pNewMetricMeta->numOfVnodes * sizeof(SVnodeSidList *);
for (int32_t i = 0; i < pMeta->numOfVnodes; ++i) {
SVnodeSidList *pSidLists = (SVnodeSidList *)rsp;
memcpy(pBuf, pSidLists, sizeof(SVnodeSidList));
pNewMetricMeta->list[i] = pBuf - (char *)pNewMetricMeta; // offset value
SVnodeSidList *pLists = (SVnodeSidList *)pBuf;
tscTrace("%p metricmeta:vid:%d,numOfTables:%d", pSql, i, pLists->numOfSids);
pBuf += sizeof(SVnodeSidList) + sizeof(STableIdInfo *) * pSidLists->numOfSids;
rsp += sizeof(SVnodeSidList);
size_t elemSize = sizeof(STableIdInfo) + pNewMetricMeta->tagLen;
for (int32_t j = 0; j < pSidLists->numOfSids; ++j) {
pLists->pSidExtInfoList[j] = pBuf - (char *)pLists;
memcpy(pBuf, rsp, elemSize);
((STableIdInfo *)pBuf)->uid = htobe64(((STableIdInfo *)pBuf)->uid);
((STableIdInfo *)pBuf)->sid = htonl(((STableIdInfo *)pBuf)->sid);
rsp += elemSize;
pBuf += elemSize;
}
}
sizes[k] = pBuf - (char *)pNewMetricMeta;
}
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
for (int32_t i = 0; i < num; ++i) {
char name[TSDB_MAX_TAGS_LEN + 1] = {0};
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, i);
tscGetMetricMetaCacheKey(pQueryInfo, name, pTableMetaInfo->pTableMeta->uid);
#ifdef _DEBUG_VIEW
printf("generate the metric key:%s, index:%d\n", name, i);
#endif
// release the used metricmeta
taosCacheRelease(tscCacheHandle, (void **)&(pTableMetaInfo->pMetricMeta), false);
pTableMetaInfo->pMetricMeta = (SSuperTableMeta *)taosCachePut(tscCacheHandle, name, (char *)metricMetaList[i],
sizes[i], tsMetricMetaKeepTimer);
tfree(metricMetaList[i]);
// failed to put into cache
if (pTableMetaInfo->pMetricMeta == NULL) {
pSql->res.code = TSDB_CODE_TSC_OUT_OF_MEMORY;
goto _error_clean;
}
}
_error_clean:
// free allocated resource
for (int32_t i = 0; i < num; ++i) {
tfree(metricMetaList[i]);
}
free(sizes);
free(metricMetaList);
#endif
SSqlRes* pRes = &pSql->res;
// NOTE: the order of several table must be preserved.
......@@ -2291,7 +2184,7 @@ int tscProcessRetrieveRspFromNode(SSqlObj *pSql) {
}
pRes->row = 0;
tscTrace("%p numOfRows:%d, offset:%d, complete:%d", pSql, pRes->numOfRows, pRes->offset, pRes->completed);
tscTrace("%p numOfRows:%" PRId64 ", offset:%" PRId64 ", complete:%d", pSql, pRes->numOfRows, pRes->offset, pRes->completed);
return 0;
}
......@@ -2432,6 +2325,7 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, int32_t clauseIndex) {
SQueryInfo *pNewQueryInfo = NULL;
if ((code = tscGetQueryInfoDetailSafely(&pNew->cmd, 0, &pNewQueryInfo)) != TSDB_CODE_SUCCESS) {
tscFreeSqlObj(pNew);
return code;
}
......
src/client/src/tscSql.c
浏览文件 @ be9c504e
......@@ -20,14 +20,9 @@
#include "tnote.h"
#include "trpc.h"
#include "tscLog.h"
#include "tscProfile.h"
#include "tscSecondaryMerge.h"
#include "tscSubquery.h"
#include "tscUtil.h"
#include "tsclient.h"
#include "tscompression.h"
#include "tsocket.h"
#include "ttimer.h"
#include "ttokendef.h"
#include "tutil.h"
......@@ -152,6 +147,8 @@ static void syncConnCallback(void *param, TAOS_RES *tres, int code) {
TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port) {
tscTrace("try to create a connection to %s:%u, user:%s db:%s", ip, port, user, db);
if (user == NULL) user = TSDB_DEFAULT_USER;
if (pass == NULL) pass = TSDB_DEFAULT_PASS;
STscObj* pObj = NULL;
SSqlObj *pSql = taosConnectImpl(ip, user, pass, db, port, syncConnCallback, NULL, (void**) &pObj);
......@@ -491,8 +488,7 @@ static bool tscFreeQhandleInVnode(SSqlObj* pSql) {
(pCmd->command == TSDB_SQL_SELECT && pSql->pStream == NULL && pTableMetaInfo->pTableMeta != NULL)) {
pCmd->command = (pCmd->command > TSDB_SQL_MGMT) ? TSDB_SQL_RETRIEVE : TSDB_SQL_FETCH;
tscTrace("%p start to send msg to free qhandle in dnode, command:%s", pSql, sqlCmd[pCmd->command]);
pSql->freed = 1;
tscTrace("%p send msg to dnode to free qhandle ASAP, command:%s", pSql, sqlCmd[pCmd->command]);
tscProcessSql(pSql);
// in case of sync model query, waits for response and then goes on
......@@ -513,18 +509,17 @@ static bool tscFreeQhandleInVnode(SSqlObj* pSql) {
void taos_free_result(TAOS_RES *res) {
SSqlObj *pSql = (SSqlObj *)res;
tscTrace("%p start to free result", res);
if (pSql == NULL || pSql->signature != pSql) {
tscTrace("%p result has been freed", pSql);
tscTrace("%p sqlObj has been freed", pSql);
return;
}
// The semaphore can not be changed while freeing async sub query objects.
SSqlRes *pRes = &pSql->res;
if (pRes == NULL || pRes->qhandle == 0) {
tscTrace("%p SqlObj is freed by app, qhandle is null", pSql);
tscFreeSqlObj(pSql);
tscTrace("%p SqlObj is freed by app, qhandle is null", pSql);
return;
}
......@@ -532,6 +527,7 @@ void taos_free_result(TAOS_RES *res) {
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
if (pQueryInfo == NULL) {
tscFreeSqlObj(pSql);
tscTrace("%p SqlObj is freed by app", pSql);
return;
}
......@@ -589,7 +585,7 @@ char *taos_errstr(TAOS_RES *tres) {
void taos_config(int debug, char *log_path) {
uDebugFlag = debug;
strcpy(tsLogDir, log_path);
tstrncpy(tsLogDir, log_path, TSDB_FILENAME_LEN);
}
char *taos_get_server_info(TAOS *taos) {
......@@ -612,7 +608,9 @@ int* taos_fetch_lengths(TAOS_RES *res) {
char *taos_get_client_info() { return version; }
void taos_stop_query(TAOS_RES *res) {
if (res == NULL) return;
if (res == NULL) {
return;
}
SSqlObj *pSql = (SSqlObj *)res;
SSqlCmd *pCmd = &pSql->cmd;
......@@ -632,7 +630,7 @@ void taos_stop_query(TAOS_RES *res) {
return;
}
//taosStopRpcConn(pSql->thandle);
rpcCancelRequest(pSql->pRpcCtx);
tscTrace("%p query is cancelled", res);
}
......@@ -714,7 +712,6 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd;
pRes->numOfRows = 1;
pRes->numOfTotal = 0;
pRes->numOfClauseTotal = 0;
......@@ -724,6 +721,7 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
if (sqlLen > tsMaxSQLStringLen) {
tscError("%p sql too long", pSql);
pRes->code = TSDB_CODE_TSC_INVALID_SQL;
tfree(pSql);
return pRes->code;
}
......@@ -732,6 +730,7 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
tscError("%p failed to malloc sql string buffer", pSql);
tscTrace("%p Valid SQL result:%d, %s pObj:%p", pSql, pRes->code, taos_errstr(taos), pObj);
tfree(pSql);
return pRes->code;
}
......@@ -856,6 +855,7 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
if (tblListLen > MAX_TABLE_NAME_LENGTH) {
tscError("%p tableNameList too long, length:%d, maximum allowed:%d", pSql, tblListLen, MAX_TABLE_NAME_LENGTH);
pRes->code = TSDB_CODE_TSC_INVALID_SQL;
tfree(pSql);
return pRes->code;
}
......@@ -863,6 +863,7 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
if (str == NULL) {
pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY;
tscError("%p failed to malloc sql string buffer", pSql);
tfree(pSql);
return pRes->code;
}
......@@ -878,6 +879,7 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
free(str);
if (pRes->code != TSDB_CODE_SUCCESS) {
tscFreeSqlObj(pSql);
return pRes->code;
}
......
src/client/src/tscStream.c
浏览文件 @ be9c504e
......@@ -125,7 +125,7 @@ static void tscProcessStreamTimer(void *handle, void *tmrId) {
}
// launch stream computing in a new thread
SSchedMsg schedMsg;
SSchedMsg schedMsg = { 0 };
schedMsg.fp = tscProcessStreamLaunchQuery;
schedMsg.ahandle = pStream;
schedMsg.thandle = (void *)1;
......@@ -239,7 +239,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
/* no resuls in the query range, retry */
// todo set retry dynamic time
int32_t retry = tsProjectExecInterval;
tscError("%p stream:%p, retrieve no data, code:%d, retry in %" PRId64 "ms", pSql, pStream, numOfRows, retry);
tscError("%p stream:%p, retrieve no data, code:%d, retry in %" PRId32 "ms", pSql, pStream, numOfRows, retry);
tscSetRetryTimer(pStream, pStream->pSql, retry);
return;
......@@ -250,7 +250,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
}
}
tscTrace("%p stream:%p, query on:%s, fetch result completed, fetched rows:%d", pSql, pStream, pTableMetaInfo->name,
tscTrace("%p stream:%p, query on:%s, fetch result completed, fetched rows:%" PRId64, pSql, pStream, pTableMetaInfo->name,
pStream->numOfRes);
// release the metric/meter meta information reference, so data in cache can be updated
......@@ -499,12 +499,14 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
if (pSql->sqlstr == NULL) {
tscError("%p failed to malloc sql string buffer", pSql);
tscFreeSqlObj(pSql);
return NULL;;
return NULL;
}
strtolower(pSql->sqlstr, sqlstr);
tscDump("%p SQL: %s", pSql, pSql->sqlstr);
tsem_init(&pSql->rspSem, 0, 0);
int32_t code = doAsyncParseSql(pSql);
int32_t code = tsParseSql(pSql, true);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
sem_wait(&pSql->rspSem);
}
......
src/client/src/tscSub.c
浏览文件 @ be9c504e
......@@ -182,21 +182,25 @@ static SArray* getTableList( SSqlObj* pSql ) {
char* sql = alloca(strlen(p) + 32);
sprintf(sql, "select tbid(tbname)%s", p);
SSqlObj* pSql1 = taos_query(pSql->pTscObj, sql);
if (terrno != TSDB_CODE_SUCCESS) {
tscError("failed to retrieve table id: %s", tstrerror(terrno));
SSqlObj* pNew = taos_query(pSql->pTscObj, sql);
if (pNew == NULL) {
tscError("failed to retrieve table id: cannot create new sql object.");
return NULL;
} else if (taos_errno(pNew) != TSDB_CODE_SUCCESS) {
tscError("failed to retrieve table id: %s", tstrerror(taos_errno(pNew)));
return NULL;
}
TAOS_ROW row;
SArray* result = taosArrayInit( 128, sizeof(STidTags) );
while ((row = taos_fetch_row(pSql1))) {
while ((row = taos_fetch_row(pNew))) {
STidTags tags;
memcpy(&tags, row[0], sizeof(tags));
taosArrayPush(result, &tags);
}
taos_free_result(pSql1);
taos_free_result(pNew);
return result;
}
......@@ -222,6 +226,9 @@ static int tscUpdateSubscription(STscObj* pObj, SSub* pSub) {
}
SArray* tables = getTableList(pSql);
if (tables == NULL) {
return 0;
}
size_t numOfTables = taosArrayGetSize(tables);
SArray* progress = taosArrayInit(numOfTables, sizeof(SSubscriptionProgress));
......@@ -238,10 +245,11 @@ static int tscUpdateSubscription(STscObj* pObj, SSub* pSub) {
if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
taosArraySort( tables, tscCompareTidTags );
tscBuildVgroupTableInfo( pTableMetaInfo, tables );
tscBuildVgroupTableInfo(pSql, pTableMetaInfo, tables);
}
taosArrayDestroy(tables);
TSDB_QUERY_SET_TYPE(tscGetQueryInfoDetail(pCmd, 0)->type, TSDB_QUERY_TYPE_MULTITABLE_QUERY);
return 1;
}
......@@ -291,7 +299,7 @@ static int tscLoadSubscriptionProgress(SSub* pSub) {
fclose(fp);
taosArraySort(progress, tscCompareSubscriptionProgress);
tscTrace("subscription progress loaded, %d tables: %s", taosArrayGetSize(progress), pSub->topic);
tscTrace("subscription progress loaded, %zu tables: %s", taosArrayGetSize(progress), pSub->topic);
return 1;
}
......@@ -350,7 +358,7 @@ TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char* topic, const char
pSub->interval = interval;
if (fp != NULL) {
tscTrace("asynchronize subscription, create new timer", topic);
tscTrace("asynchronize subscription, create new timer: %s", topic);
pSub->fp = fp;
pSub->param = param;
taosTmrReset(tscProcessSubscriptionTimer, interval, pSub, tscTmr, &pSub->pTimer);
......@@ -413,7 +421,7 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
}
if (pRes->code != TSDB_CODE_SUCCESS) {
tscError("failed to query data, error code=%d", pRes->code);
tscError("failed to query data: %s", tstrerror(pRes->code));
tscRemoveFromSqlList(pSql);
return NULL;
}
......@@ -435,7 +443,9 @@ void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress) {
} else {
char path[256];
sprintf(path, "%s/subscribe/%s", tsDataDir, pSub->topic);
remove(path);
if (remove(path) != 0) {
tscError("failed to remove progress file, topic = %s, error = %s", pSub->topic, strerror(errno));
}
}
tscFreeSqlObj(pSub->pSql);
......
src/client/src/tscSubquery.c
浏览文件 @ be9c504e
此差异已折叠。
src/client/src/tscSystem.c
浏览文件 @ be9c504e
......@@ -220,7 +220,7 @@ static int taos_options_imp(TSDB_OPTION option, const char *pStr) {
if (strlen(tsLocale) == 0) { // locale does not set yet
char* defaultLocale = setlocale(LC_CTYPE, "");
strcpy(tsLocale, defaultLocale);
tstrncpy(tsLocale, defaultLocale, sizeof(tsLocale));
}
// set the user specified locale
......@@ -304,7 +304,7 @@ static int taos_options_imp(TSDB_OPTION option, const char *pStr) {
assert(cfg != NULL);
if (cfg->cfgStatus <= TAOS_CFG_CSTATUS_OPTION) {
strcpy(tsTimezone, pStr);
tstrncpy(tsTimezone, pStr, sizeof(tsTimezone));
tsSetTimeZone();
cfg->cfgStatus = TAOS_CFG_CSTATUS_OPTION;
tscTrace("timezone set:%s, input:%s by taos_options", tsTimezone, pStr);
......
src/client/src/tscUtil.c
浏览文件 @ be9c504e
......@@ -22,7 +22,7 @@
#include "tkey.h"
#include "tmd5.h"
#include "tscProfile.h"
#include "tscSecondaryMerge.h"
#include "tscLocalMerge.h"
#include "tscSubquery.h"
#include "tschemautil.h"
#include "tsclient.h"
......@@ -32,7 +32,7 @@
static void freeQueryInfoImpl(SQueryInfo* pQueryInfo);
static void clearAllTableMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool removeFromCache);
SCond* tsGetSTableQueryCond(STagCond* pTagCond, uint64_t uid) {
if (pTagCond->pCond == NULL) {
return NULL;
......@@ -70,13 +70,6 @@ void tsSetSTableQueryCond(STagCond* pTagCond, uint64_t uid, SBufferWriter* bw) {
taosArrayPush(pTagCond->pCond, &cond);
}
bool tscQueryOnSTable(SSqlCmd* pCmd) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
return ((pQueryInfo->type & TSDB_QUERY_TYPE_STABLE_QUERY) == TSDB_QUERY_TYPE_STABLE_QUERY) &&
(pCmd->msgType == TSDB_MSG_TYPE_QUERY);
}
bool tscQueryTags(SQueryInfo* pQueryInfo) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i);
......@@ -95,32 +88,8 @@ bool tscQueryTags(SQueryInfo* pQueryInfo) {
return true;
}
bool tscIsSelectivityWithTagQuery(SSqlCmd* pCmd) {
bool hasTags = false;
int32_t numOfSelectivity = 0;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
int32_t functId = tscSqlExprGet(pQueryInfo, i)->functionId;
if (functId == TSDB_FUNC_TAG_DUMMY) {
hasTags = true;
continue;
}
if ((aAggs[functId].nStatus & TSDB_FUNCSTATE_SELECTIVITY) != 0) {
numOfSelectivity++;
}
}
if (numOfSelectivity > 0 && hasTags) {
return true;
}
return false;
}
void tscGetDBInfoFromMeterId(char* tableId, char* db) {
// todo refactor, extract methods and move the common module
void tscGetDBInfoFromTableFullName(char* tableId, char* db) {
char* st = strstr(tableId, TS_PATH_DELIMITER);
if (st != NULL) {
char* end = strstr(st + 1, TS_PATH_DELIMITER);
......@@ -181,8 +150,14 @@ bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) {
for (int32_t i = 0; i < numOfExprs; ++i) {
int32_t functionId = tscSqlExprGet(pQueryInfo, i)->functionId;
if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TAGPRJ && functionId != TSDB_FUNC_TAG &&
functionId != TSDB_FUNC_TS && functionId != TSDB_FUNC_ARITHM) {
if (functionId != TSDB_FUNC_PRJ &&
functionId != TSDB_FUNC_TAGPRJ &&
functionId != TSDB_FUNC_TAG &&
functionId != TSDB_FUNC_TS &&
functionId != TSDB_FUNC_ARITHM &&
functionId != TSDB_FUNC_TS_COMP &&
functionId != TSDB_FUNC_TID_TAG) {
return false;
}
}
......@@ -209,10 +184,14 @@ bool tscOrderedProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableInde
return pQueryInfo->order.orderColId >= 0;
}
bool tscProjectionQueryOnTable(SQueryInfo* pQueryInfo) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
bool tscIsProjectionQuery(SQueryInfo* pQueryInfo) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < size; ++i) {
int32_t functionId = tscSqlExprGet(pQueryInfo, i)->functionId;
if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TS) {
if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TAGPRJ && functionId != TSDB_FUNC_TAG &&
functionId != TSDB_FUNC_TS && functionId != TSDB_FUNC_ARITHM) {
return false;
}
}
......@@ -225,9 +204,10 @@ bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo) {
for (int32_t i = 0; i < size; ++i) {
SSqlExpr* pExpr = tscSqlExprGet(pQueryInfo, i);
if (pExpr == NULL) {
return false;
}
assert(pExpr != NULL);
// if (pExpr == NULL) {
// return false;
// }
int32_t functionId = pExpr->functionId;
if (functionId == TSDB_FUNC_TAG) {
......@@ -238,6 +218,7 @@ bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo) {
return false;
}
}
return true;
}
......@@ -383,7 +364,6 @@ void tscPartiallyFreeSqlObj(SSqlObj* pSql) {
tscFreeSqlResult(pSql);
tfree(pSql->pSubs);
pSql->freed = 0;
pSql->numOfSubs = 0;
tscResetSqlCmdObj(pCmd);
......@@ -1600,24 +1580,6 @@ void tscClearSubqueryInfo(SSqlCmd* pCmd) {
}
}
void doRemoveTableMetaInfo(SQueryInfo* pQueryInfo, int32_t index, bool removeFromCache) {
if (index < 0 || index >= pQueryInfo->numOfTables) {
return;
}
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index);
tscClearTableMetaInfo(pTableMetaInfo, removeFromCache);
free(pTableMetaInfo);
int32_t after = pQueryInfo->numOfTables - index - 1;
if (after > 0) {
memmove(&pQueryInfo->pTableMetaInfo[index], &pQueryInfo->pTableMetaInfo[index + 1], after * POINTER_BYTES);
}
pQueryInfo->numOfTables -= 1;
}
void clearAllTableMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool removeFromCache) {
tscTrace("%p deref the table meta in cache, numOfTables:%d", address, pQueryInfo->numOfTables);
......@@ -1690,6 +1652,47 @@ void tscResetForNextRetrieve(SSqlRes* pRes) {
pRes->numOfRows = 0;
}
SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cmd) {
SSqlObj* pNew = (SSqlObj*)calloc(1, sizeof(SSqlObj));
if (pNew == NULL) {
tscError("%p new subquery failed, tableIndex:%d", pSql, 0);
return NULL;
}
pNew->pTscObj = pSql->pTscObj;
pNew->signature = pNew;
SSqlCmd* pCmd = &pNew->cmd;
pCmd->command = cmd;
pCmd->parseFinished = 1;
if (tscAddSubqueryInfo(pCmd) != TSDB_CODE_SUCCESS) {
tscFreeSqlObj(pNew);
return NULL;
}
pNew->fp = fp;
pNew->param = param;
pNew->maxRetry = TSDB_MAX_REPLICA_NUM;
pNew->sqlstr = strdup(pSql->sqlstr);
if (pNew->sqlstr == NULL) {
tscError("%p new subquery failed", pSql);
free(pNew);
return NULL;
}
SQueryInfo* pQueryInfo = NULL;
tscGetQueryInfoDetailSafely(pCmd, 0, &pQueryInfo);
assert(pSql->cmd.clauseIndex == 0);
STableMetaInfo* pMasterTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, pSql->cmd.clauseIndex, 0);
tscAddTableMetaInfo(pQueryInfo, pMasterTableMetaInfo->name, NULL, NULL, NULL);
return pNew;
}
SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void* param, int32_t cmd, SSqlObj* pPrevSql) {
SSqlCmd* pCmd = &pSql->cmd;
SSqlObj* pNew = (SSqlObj*)calloc(1, sizeof(SSqlObj));
......@@ -1722,6 +1725,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
pnCmd->numOfClause = 0;
pnCmd->clauseIndex = 0;
pnCmd->pDataBlocks = NULL;
pnCmd->parseFinished = 1;
if (tscAddSubqueryInfo(pnCmd) != TSDB_CODE_SUCCESS) {
tscFreeSqlObj(pNew);
......@@ -1772,7 +1776,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
SQueryInfo* pPrevQueryInfo = tscGetQueryInfoDetail(&pPrevSql->cmd, pPrevSql->cmd.clauseIndex);
pNewQueryInfo->type = pPrevQueryInfo->type;
} else {
pNewQueryInfo->type |= TSDB_QUERY_TYPE_SUBQUERY; // it must be the subquery
TSDB_QUERY_SET_TYPE(pNewQueryInfo->type, TSDB_QUERY_TYPE_SUBQUERY);// it must be the subquery
}
uint64_t uid = pTableMetaInfo->pTableMeta->uid;
......@@ -1797,19 +1801,26 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
}
// make sure the the sqlExpr for each fields is correct
// todo handle the agg arithmetic expression
// todo handle the agg arithmetic expression
numOfExprs = tscSqlExprNumOfExprs(pNewQueryInfo);
for(int32_t f = 0; f < pNewQueryInfo->fieldsInfo.numOfOutput; ++f) {
TAOS_FIELD* field = tscFieldInfoGetField(&pNewQueryInfo->fieldsInfo, f);
numOfExprs = tscSqlExprNumOfExprs(pNewQueryInfo);
bool matched = false;
for(int32_t k1 = 0; k1 < numOfExprs; ++k1) {
SSqlExpr* pExpr1 = tscSqlExprGet(pNewQueryInfo, k1);
if (strcmp(field->name, pExpr1->aliasName) == 0) { // eatablish link according to the result field name
if (strcmp(field->name, pExpr1->aliasName) == 0) { // establish link according to the result field name
SFieldSupInfo* pInfo = tscFieldInfoGetSupp(&pNewQueryInfo->fieldsInfo, f);
pInfo->pSqlExpr = pExpr1;
matched = true;
break;
}
}
assert(matched);
}
tscFieldInfoUpdateOffset(pNewQueryInfo);
......@@ -1825,6 +1836,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
if (pPrevSql == NULL) {
STableMeta* pTableMeta = taosCacheAcquireByData(tscCacheHandle, pTableMetaInfo->pTableMeta); // get by name may failed due to the cache cleanup
assert(pTableMeta != NULL);
pFinalInfo = tscAddTableMetaInfo(pNewQueryInfo, name, pTableMeta, pTableMetaInfo->vgroupList, pTableMetaInfo->tagColList);
} else { // transfer the ownership of pTableMeta to the newly create sql object.
STableMetaInfo* pPrevInfo = tscGetTableMetaInfoFromCmd(&pPrevSql->cmd, pPrevSql->cmd.clauseIndex, 0);
......@@ -1847,12 +1859,12 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
assert(pFinalInfo->vgroupList != NULL);
}
if (cmd == TSDB_SQL_SELECT) {
size_t size = taosArrayGetSize(pNewQueryInfo->colList);
tscTrace(
"%p new subquery:%p, tableIndex:%d, vgroupIndex:%d, type:%d, exprInfo:%d, colList:%d,"
"%p new subquery:%p, tableIndex:%d, vgroupIndex:%d, type:%d, exprInfo:%zu, colList:%zu,"
"fieldInfo:%d, name:%s, qrang:%" PRId64 " - %" PRId64 " order:%d, limit:%" PRId64,
pSql, pNew, tableIndex, pTableMetaInfo->vgroupIndex, pNewQueryInfo->type, tscSqlExprNumOfExprs(pNewQueryInfo),
size, pNewQueryInfo->fieldsInfo.numOfOutput, pFinalInfo->name, pNewQueryInfo->window.skey,
......@@ -1898,16 +1910,21 @@ void tscDoQuery(SSqlObj* pSql) {
}
if (QUERY_IS_JOIN_QUERY(type)) {
if ((pQueryInfo->type & TSDB_QUERY_TYPE_SUBQUERY) == 0) {
if (!TSDB_QUERY_HAS_TYPE(type, TSDB_QUERY_TYPE_SUBQUERY)) {
tscHandleMasterJoinQuery(pSql);
return;
} else {
// for first stage sub query, iterate all vnodes to get all timestamp
if ((pQueryInfo->type & TSDB_QUERY_TYPE_JOIN_SEC_STAGE) != TSDB_QUERY_TYPE_JOIN_SEC_STAGE) {
// doProcessSql(pSql);
assert(0);
} else { // for first stage sub query, iterate all vnodes to get all timestamp
if (!TSDB_QUERY_HAS_TYPE(type, TSDB_QUERY_TYPE_JOIN_SEC_STAGE)) {
tscProcessSql(pSql);
} else { // secondary stage join query.
if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { // super table query
tscHandleMasterSTableQuery(pSql);
} else {
tscProcessSql(pSql);
}
}
}
return;
} else if (tscIsTwoStageSTableQuery(pQueryInfo, 0)) { // super table query
tscHandleMasterSTableQuery(pSql);
return;
......@@ -1917,11 +1934,13 @@ void tscDoQuery(SSqlObj* pSql) {
}
}
int16_t tscGetJoinTagColIndexByUid(STagCond* pTagCond, uint64_t uid) {
int16_t tscGetJoinTagColIdByUid(STagCond* pTagCond, uint64_t uid) {
if (pTagCond->joinInfo.left.uid == uid) {
return pTagCond->joinInfo.left.tagCol;
return pTagCond->joinInfo.left.tagColId;
} else if (pTagCond->joinInfo.right.uid == uid) {
return pTagCond->joinInfo.right.tagColId;
} else {
return pTagCond->joinInfo.right.tagCol;
assert(0);
}
}
......@@ -1965,6 +1984,11 @@ bool tscHasReachLimitation(SQueryInfo* pQueryInfo, SSqlRes* pRes) {
return (pQueryInfo->clauseLimit > 0 && pRes->numOfClauseTotal >= pQueryInfo->clauseLimit);
}
bool tscResultsetFetchCompleted(TAOS_RES *result) {
SSqlRes* pRes = result;
return pRes->completed;
}
char* tscGetErrorMsgPayload(SSqlCmd* pCmd) { return pCmd->payload; }
/**
......@@ -1978,11 +2002,10 @@ bool hasMoreVnodesToTry(SSqlObj* pSql) {
return false;
}
assert(pRes->completed);
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
assert(pRes->completed);
// for normal table, no need to try any more if results are all retrieved from one vnode
if (!UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo) || (pTableMetaInfo->vgroupList == NULL)) {
return false;
......@@ -2004,12 +2027,11 @@ void tscTryQueryNextVnode(SSqlObj* pSql, __async_cb_func_t fp) {
* if case of: multi-vnode super table projection query
*/
assert(pRes->numOfRows == 0 && tscNonOrderedProjectionQueryOnSTable(pQueryInfo, 0) && !tscHasReachLimitation(pQueryInfo, pRes));
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
int32_t totalVgroups = pTableMetaInfo->vgroupList->numOfVgroups;
while (++pTableMetaInfo->vgroupIndex < totalVgroups) {
tscTrace("%p results from vgroup index:%d completed, try next:%d. total vgroups:%d. current numOfRes:%d", pSql,
tscTrace("%p results from vgroup index:%d completed, try next:%d. total vgroups:%d. current numOfRes:%" PRId64, pSql,
pTableMetaInfo->vgroupIndex - 1, pTableMetaInfo->vgroupIndex, totalVgroups, pRes->numOfClauseTotal);
/*
......
src/common/inc/tdataformat.h
浏览文件 @ be9c504e
......@@ -69,8 +69,8 @@ typedef struct {
int version; // version
int numOfCols; // Number of columns appended
int tlen; // maximum length of a SDataRow without the header part
uint16_t flen; // First part length in a SDataRow after the header part
uint16_t vlen; // pure value part length, excluded the overhead
uint16_t flen; // First part length in a SDataRow after the header part
uint16_t vlen; // pure value part length, excluded the overhead
STColumn columns[];
} STSchema;
......@@ -83,8 +83,8 @@ typedef struct {
#define tdFreeSchema(s) tfree((s))
STSchema *tdDupSchema(STSchema *pSchema);
void * tdEncodeSchema(void *dst, STSchema *pSchema);
STSchema *tdDecodeSchema(void **psrc);
int tdEncodeSchema(void **buf, STSchema *pSchema);
void * tdDecodeSchema(void *buf, STSchema **pRSchema);
static FORCE_INLINE int comparColId(const void *key1, const void *key2) {
if (*(int16_t *)key1 > ((STColumn *)key2)->colId) {
......@@ -107,8 +107,8 @@ typedef struct {
int tCols;
int nCols;
int tlen;
uint16_t flen;
uint16_t vlen;
uint16_t flen;
uint16_t vlen;
int version;
STColumn *columns;
} STSchemaBuilder;
......@@ -288,7 +288,7 @@ typedef struct {
SKVRow tdKVRowDup(SKVRow row);
int tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value);
void * tdEncodeKVRow(void *buf, SKVRow row);
int tdEncodeKVRow(void **buf, SKVRow row);
void * tdDecodeKVRow(void *buf, SKVRow *row);
static FORCE_INLINE int comparTagId(const void *key1, const void *key2) {
......
src/common/inc/tglobal.h
浏览文件 @ be9c504e
......@@ -92,10 +92,6 @@ extern int32_t tsNumOfMnodes;
extern int32_t tsMaxShellConns;
extern int32_t tsMaxTables;
extern char tsDefaultDB[];
extern char tsDefaultUser[];
extern char tsDefaultPass[];
extern char tsMqttBrokerAddress[];
extern char tsMqttBrokerClientId[];
......
src/common/src/tdataformat.c
浏览文件 @ be9c504e
......@@ -14,6 +14,7 @@
*/
#include "tdataformat.h"
#include "talgo.h"
#include "tcoding.h"
#include "wchar.h"
/**
......@@ -33,50 +34,50 @@ STSchema *tdDupSchema(STSchema *pSchema) {
/**
* Encode a schema to dst, and return the next pointer
*/
void *tdEncodeSchema(void *dst, STSchema *pSchema) {
int tdEncodeSchema(void **buf, STSchema *pSchema) {
int tlen = 0;
tlen += taosEncodeFixedI32(buf, schemaVersion(pSchema));
tlen += taosEncodeFixedI32(buf, schemaNCols(pSchema));
T_APPEND_MEMBER(dst, pSchema, STSchema, version);
T_APPEND_MEMBER(dst, pSchema, STSchema, numOfCols);
for (int i = 0; i < schemaNCols(pSchema); i++) {
STColumn *pCol = schemaColAt(pSchema, i);
T_APPEND_MEMBER(dst, pCol, STColumn, type);
T_APPEND_MEMBER(dst, pCol, STColumn, colId);
T_APPEND_MEMBER(dst, pCol, STColumn, bytes);
tlen += taosEncodeFixedI8(buf, colType(pCol));
tlen += taosEncodeFixedI16(buf, colColId(pCol));
tlen += taosEncodeFixedI32(buf, colBytes(pCol));
}
return dst;
return tlen;
}
/**
* Decode a schema from a binary.
*/
STSchema *tdDecodeSchema(void **psrc) {
int totalCols = 0;
void *tdDecodeSchema(void *buf, STSchema **pRSchema) {
int version = 0;
STSchemaBuilder schemaBuilder = {0};
int numOfCols = 0;
STSchemaBuilder schemaBuilder;
T_READ_MEMBER(*psrc, int, version);
T_READ_MEMBER(*psrc, int, totalCols);
buf = taosDecodeFixedI32(buf, &version);
buf = taosDecodeFixedI32(buf, &numOfCols);
if (tdInitTSchemaBuilder(&schemaBuilder, version) < 0) return NULL;
for (int i = 0; i < totalCols; i++) {
for (int i = 0; i < numOfCols; i++) {
int8_t type = 0;
int16_t colId = 0;
int32_t bytes = 0;
T_READ_MEMBER(*psrc, int8_t, type);
T_READ_MEMBER(*psrc, int16_t, colId);
T_READ_MEMBER(*psrc, int32_t, bytes);
buf = taosDecodeFixedI8(buf, &type);
buf = taosDecodeFixedI16(buf, &colId);
buf = taosDecodeFixedI32(buf, &bytes);
if (tdAddColToSchema(&schemaBuilder, type, colId, bytes) < 0) {
tdDestroyTSchemaBuilder(&schemaBuilder);
return NULL;
}
}
STSchema *pSchema = tdGetSchemaFromBuilder(&schemaBuilder);
*pRSchema = tdGetSchemaFromBuilder(&schemaBuilder);
tdDestroyTSchemaBuilder(&schemaBuilder);
return pSchema;
return buf;
}
int tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, int32_t version) {
......@@ -105,7 +106,7 @@ void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, int32_t version) {
}
int tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int16_t colId, int32_t bytes) {
if (!isValidDataType(type, 0)) return -1;
if (!isValidDataType(type)) return -1;
if (pBuilder->nCols >= pBuilder->tCols) {
pBuilder->tCols *= 2;
......@@ -605,14 +606,19 @@ int tdSetKVRowDataOfCol(SKVRow *orow, int16_t colId, int8_t type, void *value) {
return 0;
}
void *tdEncodeKVRow(void *buf, SKVRow row) {
int tdEncodeKVRow(void **buf, SKVRow row) {
// May change the encode purpose
kvRowCpy(buf, row);
return POINTER_SHIFT(buf, kvRowLen(row));
if (buf != NULL) {
kvRowCpy(*buf, row);
*buf = POINTER_SHIFT(*buf, kvRowLen(row));
}
return kvRowLen(row);
}
void *tdDecodeKVRow(void *buf, SKVRow *row) {
*row = tdKVRowDup(buf);
if (*row == NULL) return NULL;
return POINTER_SHIFT(buf, kvRowLen(*row));
}
......
src/common/src/tglobal.c
浏览文件 @ be9c504e
......@@ -109,10 +109,6 @@ int32_t tsReplications = TSDB_DEFAULT_REPLICA_NUM;
int16_t tsAffectedRowsMod = 0;
int32_t tsNumOfMnodes = 3;
int32_t tsMaxShellConns = 5000;
char tsDefaultDB[TSDB_DB_NAME_LEN] = {0};
char tsDefaultUser[64] = "root";
char tsDefaultPass[64] = "taosdata";
int32_t tsMaxConnections = 5000;
int32_t tsBalanceInterval = 300; // seconds
......@@ -713,37 +709,6 @@ static void doInitGlobalConfig() {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
// login configs
cfg.option = "defaultDB";
cfg.ptr = tsDefaultDB;
cfg.valType = TAOS_CFG_VTYPE_STRING;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;
cfg.maxValue = 0;
cfg.ptrLength = TSDB_DB_NAME_LEN - 1;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "defaultUser";
cfg.ptr = tsDefaultUser;
cfg.valType = TAOS_CFG_VTYPE_STRING;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;
cfg.maxValue = 0;
cfg.ptrLength = TSDB_USER_LEN - 1;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "defaultPass";
cfg.ptr = tsDefaultPass;
cfg.valType = TAOS_CFG_VTYPE_STRING;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT | TSDB_CFG_CTYPE_B_NOT_PRINT;
cfg.minValue = 0;
cfg.maxValue = 0;
cfg.ptrLength = TSDB_PASSWORD_LEN - 1;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "mqttBrokerAddress";
cfg.ptr = tsMqttBrokerAddress;
cfg.valType = TAOS_CFG_VTYPE_STRING;
......
src/common/src/ttypes.c
浏览文件 @ be9c504e
......@@ -363,16 +363,8 @@ char tTokenTypeSwitcher[13] = {
TSDB_DATA_TYPE_NCHAR, // TK_NCHAR
};
bool isValidDataType(int32_t type, int32_t length) {
if (type < TSDB_DATA_TYPE_NULL || type > TSDB_DATA_TYPE_NCHAR) {
return false;
}
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
// return length >= 0 && length <= TSDB_MAX_BINARY_LEN;
}
return true;
bool isValidDataType(int32_t type) {
return type >= TSDB_DATA_TYPE_NULL && type <= TSDB_DATA_TYPE_NCHAR;
}
bool isNull(const char *val, int32_t type) {
......
src/connector/jdbc/.classpath 0 → 100644
浏览文件 @ be9c504e
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
<classpathentry kind="src" output="target/classes" path="src/main/java">
<attributes>
<attribute name="optional" value="true"/>
<attribute name="maven.pomderived" value="true"/>
</attributes>
</classpathentry>
<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources">
<attributes>
<attribute name="maven.pomderived" value="true"/>
</attributes>
</classpathentry>
<classpathentry kind="src" output="target/test-classes" path="src/test/java">
<attributes>
<attribute name="optional" value="true"/>
<attribute name="maven.pomderived" value="true"/>
<attribute name="test" value="true"/>
</attributes>
</classpathentry>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8">
<attributes>
<attribute name="maven.pomderived" value="true"/>
</attributes>
</classpathentry>
<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
<attributes>
<attribute name="maven.pomderived" value="true"/>
</attributes>
</classpathentry>
<classpathentry kind="src" path="target/generated-sources/annotations">
<attributes>
<attribute name="optional" value="true"/>
<attribute name="maven.pomderived" value="true"/>
<attribute name="ignore_optional_problems" value="true"/>
<attribute name="m2e-apt" value="true"/>
</attributes>
</classpathentry>
<classpathentry kind="src" output="target/test-classes" path="target/generated-test-sources/test-annotations">
<attributes>
<attribute name="optional" value="true"/>
<attribute name="maven.pomderived" value="true"/>
<attribute name="ignore_optional_problems" value="true"/>
<attribute name="m2e-apt" value="true"/>
<attribute name="test" value="true"/>
</attributes>
</classpathentry>
<classpathentry kind="output" path="target/classes"/>
</classpath>
src/connector/jdbc/.project 0 → 100644
浏览文件 @ be9c504e
此差异已折叠。
src/connector/jdbc/.settings/org.eclipse.core.resources.prefs 已删除 100755 → 0
浏览文件 @ 1b6ca91d
eclipse.preferences.version=1
encoding//src/main/java=GBK
encoding//src/main/java/com/taosdata/jdbc/TSDBJNIConnector.java=UTF-8
encoding//src/main/resources=UTF-8
encoding/<project>=UTF-8
src/connector/jdbc/.settings/org.eclipse.m2e.core.prefs 已删除 100755 → 0
浏览文件 @ 1b6ca91d
activeProfiles=
eclipse.preferences.version=1
resolveWorkspaceProjects=true
version=1
src/connector/jdbc/pom.xml
浏览文件 @ be9c504e
此差异已折叠。
src/cq/src/cqMain.c
浏览文件 @ be9c504e
此差异已折叠。
src/cq/test/cqtest.c
浏览文件 @ be9c504e
此差异已折叠。
src/dnode/src/dnodeMWrite.c
浏览文件 @ be9c504e
此差异已折叠。
src/dnode/src/dnodeMgmt.c
浏览文件 @ be9c504e
此差异已折叠。
src/inc/query.h
浏览文件 @ be9c504e
此差异已折叠。
src/inc/taos.h
浏览文件 @ be9c504e
此差异已折叠。
src/inc/taosdef.h
浏览文件 @ be9c504e
此差异已折叠。
src/inc/taoserror.h
浏览文件 @ be9c504e
此差异已折叠。
src/inc/taosmsg.h
浏览文件 @ be9c504e
此差异已折叠。
src/inc/trpc.h
浏览文件 @ be9c504e
此差异已折叠。
src/inc/tsdb.h
浏览文件 @ be9c504e
此差异已折叠。
src/inc/ttokendef.h
浏览文件 @ be9c504e
此差异已折叠。
src/inc/vnode.h
浏览文件 @ be9c504e
此差异已折叠。
src/kit/shell/src/shellEngine.c
浏览文件 @ be9c504e
此差异已折叠。
src/kit/shell/src/shellMain.c
浏览文件 @ be9c504e
此差异已折叠。
src/kit/taosdemo/taosdemo.c
浏览文件 @ be9c504e
此差异已折叠。
src/kit/taosdump/taosdump.c
浏览文件 @ be9c504e
此差异已折叠。
src/kit/taosnetwork/client.c
浏览文件 @ be9c504e
此差异已折叠。
src/kit/taosnetwork/server.c
浏览文件 @ be9c504e
此差异已折叠。
src/kit/taosnetwork/taosnetwork_client.c 已删除 100644 → 0
浏览文件 @ 1b6ca91d
此差异已折叠。
src/kit/taosnetwork/taosnetwork_server.c 已删除 100644 → 0
浏览文件 @ 1b6ca91d
此差异已折叠。
src/kit/taosnetwork/test_client.c 已删除 100644 → 0
浏览文件 @ 1b6ca91d
此差异已折叠。
src/kit/taosnetwork/test_server.c 已删除 100644 → 0
浏览文件 @ 1b6ca91d
此差异已折叠。
src/mnode/inc/mnodeDef.h
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/inc/mnodeDnode.h
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/inc/mnodeInt.h
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/inc/mnodeSdb.h
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/inc/mnodeUser.h
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/inc/mnodeVgroup.h
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeAcct.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeDb.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeDnode.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeMnode.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeProfile.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeSdb.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeShow.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeTable.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeUser.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeVgroup.c
浏览文件 @ be9c504e
此差异已折叠。
src/mnode/src/mnodeWrite.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/gcHandle.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/httpAuth.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/httpContext.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/httpJson.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/httpSession.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/httpSql.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/httpUtil.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/restHandle.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/http/src/tgHandle.c
浏览文件 @ be9c504e
此差异已折叠。
src/plugins/mqtt/src/mqttSystem.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/inc/qextbuffer.h
浏览文件 @ be9c504e
此差异已折叠。
src/query/inc/qtsbuf.h
浏览文件 @ be9c504e
此差异已折叠。
src/query/inc/sql.y
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/qExecutor.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/qast.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/qextbuffer.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/qparserImpl.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/qpercentile.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/qtokenizer.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/qtsbuf.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/sql.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/src/tvariant.c
浏览文件 @ be9c504e
此差异已折叠。
src/query/tests/tsBufTest.cpp
浏览文件 @ be9c504e
此差异已折叠。
src/query/tests/unitTest.cpp
浏览文件 @ be9c504e
此差异已折叠。
src/rpc/inc/rpcTcp.h
浏览文件 @ be9c504e
此差异已折叠。
src/rpc/inc/rpcUdp.h
浏览文件 @ be9c504e
此差异已折叠。
src/rpc/src/rpcMain.c
浏览文件 @ be9c504e
此差异已折叠。
src/rpc/src/rpcTcp.c
浏览文件 @ be9c504e
此差异已折叠。
src/rpc/src/rpcUdp.c
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/inc/tsdbMain.h
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/src/tsdbBuffer.c 0 → 100644
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/src/tsdbCache.c 已删除 100644 → 0
浏览文件 @ 1b6ca91d
此差异已折叠。
src/tsdb/src/tsdbFile.c
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/src/tsdbMain.c
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/src/tsdbMemTable.c 0 → 100644
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/src/tsdbMeta.c
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/src/tsdbMetaFile.c 已删除 100644 → 0
浏览文件 @ 1b6ca91d
此差异已折叠。
src/tsdb/src/tsdbRWHelper.c
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/src/tsdbRead.c
浏览文件 @ be9c504e
此差异已折叠。
src/tsdb/tests/tsdbTests.cpp
浏览文件 @ be9c504e
此差异已折叠。
src/util/inc/tcoding.h
浏览文件 @ be9c504e
此差异已折叠。
src/util/inc/tkvstore.h
浏览文件 @ be9c504e
此差异已折叠。
src/util/inc/tlist.h
浏览文件 @ be9c504e
此差异已折叠。
src/util/inc/tlog.h
浏览文件 @ be9c504e
此差异已折叠。
src/util/inc/tref.h
浏览文件 @ be9c504e
此差异已折叠。
src/util/inc/tstoken.h
浏览文件 @ be9c504e
此差异已折叠。
src/util/inc/tutil.h
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/hash.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tarray.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tcache.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tcompare.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tcompression.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tconfig.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tcrc32c.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tdes.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/terror.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tkvstore.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tlist.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tlog.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tmem.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tnote.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tskiplist.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tsocket.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/ttime.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/ttimer.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/src/tutil.c
浏览文件 @ be9c504e
此差异已折叠。
src/util/tests/codingTests.cpp
浏览文件 @ be9c504e
此差异已折叠。
src/vnode/inc/vnodeInt.h
浏览文件 @ be9c504e
此差异已折叠。
src/vnode/src/vnodeMain.c
浏览文件 @ be9c504e
此差异已折叠。
src/vnode/src/vnodeRead.c
浏览文件 @ be9c504e
此差异已折叠。
src/vnode/src/vnodeWrite.c
浏览文件 @ be9c504e
此差异已折叠。
tests/examples/c/demo.c
浏览文件 @ be9c504e
此差异已折叠。
tests/examples/c/stream.c 100755 → 100644
浏览文件 @ be9c504e
文件模式从 100755 更改为 100644
tests/pytest/alter/alter_table_crash.py 0 → 100644
浏览文件 @ be9c504e
此差异已折叠。
tests/pytest/fulltest.sh
浏览文件 @ be9c504e
此差异已折叠。
tests/pytest/query/queryJoin.py 0 → 100644
浏览文件 @ be9c504e
此差异已折叠。
tests/pytest/query/querytest.py
浏览文件 @ be9c504e
此差异已折叠。
tests/pytest/regressiontest.sh
浏览文件 @ be9c504e
此差异已折叠。
tests/pytest/stable/query_after_reset.py 0 → 100644
浏览文件 @ be9c504e
此差异已折叠。
tests/pytest/util/sql.py
浏览文件 @ be9c504e
此差异已折叠。
tests/script/fullGeneralSuite.sim
浏览文件 @ be9c504e
此差异已折叠。
tests/script/general/tag/add.sim
浏览文件 @ be9c504e
此差异已折叠。
tests/script/general/tag/change.sim
浏览文件 @ be9c504e
此差异已折叠。
tests/script/general/tag/commit.sim
浏览文件 @ be9c504e
此差异已折叠。
tests/script/general/tag/delete.sim
浏览文件 @ be9c504e
此差异已折叠。
tests/script/general/tag/set.sim
浏览文件 @ be9c504e
此差异已折叠。
tests/script/jenkins/basic.txt
浏览文件 @ be9c504e
此差异已折叠。
tests/script/jenkins/general.txt 已删除 100644 → 0
浏览文件 @ 1b6ca91d
此差异已折叠。
tests/script/jenkins/simple.txt
浏览文件 @ be9c504e
此差异已折叠。
tests/script/jenkins/unique.txt
浏览文件 @ be9c504e
此差异已折叠。
tests/script/loop.sh 0 → 100755
浏览文件 @ be9c504e
此差异已折叠。
tests/script/regressionSuite.sim
浏览文件 @ be9c504e
此差异已折叠。
tests/script/sh/exec.sh
浏览文件 @ be9c504e
此差异已折叠。
tests/script/tmp/prepare.sim
浏览文件 @ be9c504e
此差异已折叠。
tests/test-all.sh
浏览文件 @ be9c504e
此差异已折叠。
tests/test/c/importOneRow.c
浏览文件 @ be9c504e
此差异已折叠。
tests/test/c/importPerTable.c
浏览文件 @ be9c504e
此差异已折叠。
tests/test/c/insertPerRow.c
浏览文件 @ be9c504e
此差异已折叠。
tests/test/c/insertPerTable.c
浏览文件 @ be9c504e
此差异已折叠。
tests/tsim/src/simExe.c
浏览文件 @ be9c504e
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或