Skip to content

T
TDengine

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

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
提交 942c8814 编写于 作者: H Hongze Cheng
浏览文件
操作

Merge branch 'develop' into feature/TD-1925_new

上级 0948fdca 9242cef8
隐藏空白更改
内联 并排
Showing with 591 addition and 301 deletion
documentation20/webdocs/markdowndocs/TAOS SQL-ch.md
浏览文件 @ 942c8814
...@@ -213,16 +213,19 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic ...@@ -213,16 +213,19 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic
如果表是通过[超级表](../super-table/)创建，更改表结构的操作只能对超级表进行。同时针对超级表的结构更改对所有通过该结构创建的表生效。对于不是通过超级表创建的表，可以直接修改表结构 如果表是通过[超级表](../super-table/)创建，更改表结构的操作只能对超级表进行。同时针对超级表的结构更改对所有通过该结构创建的表生效。对于不是通过超级表创建的表，可以直接修改表结构
## 超级表STable管理 ## 超级表STable管理
注意：在 2.0.15 以前的版本中，并不支持 STABLE 保留字，而是写作 TABLE。也即，在本节后文的指令说明中，CREATE、DROP、ALTER 三个指令在老版本中保留字需写作 TABLE 而不是 STABLE。
- **创建超级表** - **创建超级表**
```mysql ```mysql
CREATE TABLE [IF NOT EXISTS] stb_name (timestamp_field_name TIMESTAMP, field1_name data_type1 [, field2_name data_type2 ...]) TAGS (tag1_name tag_type1, tag2_name tag_type2 [, tag3_name tag_type3]); CREATE STABLE [IF NOT EXISTS] stb_name (timestamp_field_name TIMESTAMP, field1_name data_type1 [, field2_name data_type2 ...]) TAGS (tag1_name tag_type1, tag2_name tag_type2 [, tag3_name tag_type3]);
``` ```
创建STable, 与创建表的SQL语法相似，但需指定TAGS字段的名称和类型 创建 STable，与创建表的 SQL 语法相似，但需指定 TAGS 字段的名称和类型
说明： 说明：
1) TAGS 列的数据类型不能是timestamp类型； 1) TAGS 列的数据类型不能是 timestamp 类型；
2) TAGS 列名不能与其他列名相同； 2) TAGS 列名不能与其他列名相同；
...@@ -233,16 +236,16 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic ...@@ -233,16 +236,16 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic
- **删除超级表** - **删除超级表**
```mysql ```mysql
DROP TABLE [IF EXISTS] stb_name; DROP STABLE [IF EXISTS] stb_name;
``` ```
删除STable会自动删除通过STable创建的子表。 删除 STable 会自动删除通过 STable 创建的子表。
- **显示当前数据库下的所有超级表信息** - **显示当前数据库下的所有超级表信息**
```mysql ```mysql
SHOW STABLES [LIKE tb_name_wildcar]; SHOW STABLES [LIKE tb_name_wildcar];
``` ```
查看数据库内全部STable，及其相关信息，包括STable的名称、创建时间、列数量、标签（TAG）数量、通过该STable建表的数量。 查看数据库内全部 STable，及其相关信息，包括 STable 的名称、创建时间、列数量、标签（TAG）数量、通过该 STable 建表的数量。
- **获取超级表的结构信息** - **获取超级表的结构信息**
...@@ -253,43 +256,43 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic ...@@ -253,43 +256,43 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic
- **超级表增加列** - **超级表增加列**
```mysql ```mysql
ALTER TABLE stb_name ADD COLUMN field_name data_type; ALTER STABLE stb_name ADD COLUMN field_name data_type;
``` ```
- **超级表删除列** - **超级表删除列**
```mysql ```mysql
ALTER TABLE stb_name DROP COLUMN field_name; ALTER STABLE stb_name DROP COLUMN field_name;
``` ```
## 超级表 STable 中 TAG 管理 ## 超级表 STable 中 TAG 管理
- **添加标签** - **添加标签**
```mysql ```mysql
ALTER TABLE stb_name ADD TAG new_tag_name tag_type; ALTER STABLE stb_name ADD TAG new_tag_name tag_type;
``` ```
为STable增加一个新的标签，并指定新标签的类型。标签总数不能超过128个，总长度不超过16k个字符。 为 STable 增加一个新的标签，并指定新标签的类型。标签总数不能超过 128 个，总长度不超过 16k 个字符。
- **删除标签** - **删除标签**
```mysql ```mysql
ALTER TABLE stb_name DROP TAG tag_name; ALTER STABLE stb_name DROP TAG tag_name;
``` ```
删除超级表的一个标签，从超级表删除某个标签后，该超级表下的所有子表也会自动删除该标签。 删除超级表的一个标签，从超级表删除某个标签后，该超级表下的所有子表也会自动删除该标签。
- **修改标签名** - **修改标签名**
```mysql ```mysql
ALTER TABLE stb_name CHANGE TAG old_tag_name new_tag_name; ALTER STABLE stb_name CHANGE TAG old_tag_name new_tag_name;
``` ```
修改超级表的标签名，从超级表修改某个标签名后，该超级表下的所有子表也会自动更新该标签名。 修改超级表的标签名，从超级表修改某个标签名后，该超级表下的所有子表也会自动更新该标签名。
- **修改子表标签值** - **修改子表标签值**
```mysql ```mysql
ALTER TABLE tb_name SET TAG tag_name=new_tag_value; ALTER STABLE tb_name SET TAG tag_name=new_tag_value;
``` ```
说明：除了更新标签的值的操作是针对子表进行，其他所有的标签操作（添加标签、删除标签等）均只能作用于STable，不能对单个子表操作。对STable添加标签以后，依托于该STable建立的所有表将自动增加了一个标签，所有新增标签的默认值都是NULL。 说明：除了更新标签的值的操作是针对子表进行，其他所有的标签操作（添加标签、删除标签等）均只能作用于 STable，不能对单个子表操作。对 STable 添加标签以后，依托于该 STable 建立的所有表将自动增加了一个标签，所有新增标签的默认值都是 NULL。
## 数据写入 ## 数据写入
...@@ -467,6 +470,17 @@ Query OK, 2 row(s) in set (0.003112s) ...@@ -467,6 +470,17 @@ Query OK, 2 row(s) in set (0.003112s)
注意：普通表的通配符 * 中并不包含 _标签列_。 注意：普通表的通配符 * 中并不包含 _标签列_。
##### 获取标签列的去重取值
从 2.0.15 版本开始，支持在超级表查询标签列时，指定 distinct 关键字，这样将返回指定标签列的所有不重复取值。
```mysql
SELECT DISTINCT tag_name FROM stb_name;
```
注意：目前 distinct 关键字只支持对超级表的标签列进行去重，而不能用于普通列。
#### 结果集列名 #### 结果集列名
```SELECT```子句中，如果不指定返回结果集合的列名，结果集列名称默认使用```SELECT```子句中的表达式名称作为列名称。此外，用户可使用```AS```来重命名返回结果集合中列的名称。例如： ```SELECT```子句中，如果不指定返回结果集合的列名，结果集列名称默认使用```SELECT```子句中的表达式名称作为列名称。此外，用户可使用```AS```来重命名返回结果集合中列的名称。例如：
......
documentation20/webdocs/markdowndocs/administrator-ch.md
浏览文件 @ 942c8814
...@@ -12,7 +12,7 @@ ...@@ -12,7 +12,7 @@
Memory Size = maxVgroupsPerDb * (blocks * cache + 10Mb) + numOfTables * (tagSizePerTable + 0.5Kb) Memory Size = maxVgroupsPerDb * (blocks * cache + 10Mb) + numOfTables * (tagSizePerTable + 0.5Kb)
``` ```
示例：假设是4核机器，cache是缺省大小16M, blocks是缺省值6，假设有10万张表，标签总长度是256字节，则总的内存需求为：4\*(16\*6+10) + 100000*(0.25+0.5)/1000 = 499M。 示例：假设是4核机器，cache是缺省大小16M, blocks是缺省值6，假设有10万张表，标签总长度是256字节，则总的内存需求为：4\*(16\*6+10) + 100000\*(0.25+0.5)/1000 = 499M。
实际运行的系统往往会根据数据特点的不同，将数据存放在不同的DB里。因此做规划时，也需要考虑。 实际运行的系统往往会根据数据特点的不同，将数据存放在不同的DB里。因此做规划时，也需要考虑。
...@@ -35,7 +35,7 @@ TDengine相对于通用数据库，有超高的压缩比，在绝大多数场景 ...@@ -35,7 +35,7 @@ TDengine相对于通用数据库，有超高的压缩比，在绝大多数场景
Raw DataSize = numOfTables * rowSizePerTable * rowsPerTable Raw DataSize = numOfTables * rowSizePerTable * rowsPerTable
``` ```
示例：1000万台智能电表，每台电表每15分钟采集一次数据，每次采集的数据128字节，那么一年的原始数据量是：10000000\*128\*24\*60/15*365 = 44.8512T。TDengine大概需要消耗44.851/5=8.97024T空间。 示例：1000万台智能电表，每台电表每15分钟采集一次数据，每次采集的数据128字节，那么一年的原始数据量是：10000000\*128\*24\*60/15\*365 = 44.8512T。TDengine大概需要消耗44.851/5=8.97024T空间。
用户可以通过参数keep，设置数据在磁盘中的最大保存时长。为进一步减少存储成本，TDengine还提供多级存储，最冷的数据可以存放在最廉价的存储介质上，应用的访问不用做任何调整，只是读取速度降低了。 用户可以通过参数keep，设置数据在磁盘中的最大保存时长。为进一步减少存储成本，TDengine还提供多级存储，最冷的数据可以存放在最廉价的存储介质上，应用的访问不用做任何调整，只是读取速度降低了。
...@@ -181,7 +181,7 @@ taos -C 或 taos --dump-config ...@@ -181,7 +181,7 @@ taos -C 或 taos --dump-config
客户端的输入的字符均采用操作系统当前默认的编码格式，在Linux系统上多为UTF-8，部分中文系统编码则可能是GB18030或GBK等。在docker环境中默认的编码是POSIX。在中文版Windows系统中，编码则是CP936。客户端需要确保正确设置自己所使用的字符集，即客户端运行的操作系统当前编码字符集，才能保证nchar中的数据正确转换为UCS4-LE编码格式。 客户端的输入的字符均采用操作系统当前默认的编码格式，在Linux系统上多为UTF-8，部分中文系统编码则可能是GB18030或GBK等。在docker环境中默认的编码是POSIX。在中文版Windows系统中，编码则是CP936。客户端需要确保正确设置自己所使用的字符集，即客户端运行的操作系统当前编码字符集，才能保证nchar中的数据正确转换为UCS4-LE编码格式。
在 Linux 中 locale 的命名规则为: <语言>_<地区>.<字符集编码> 如：zh_CN.UTF-8，zh代表中文，CN代表大陆地区，UTF-8表示字符集。字符集编码为客户端正确解析本地字符串提供编码转换的说明。Linux系统与 Mac OSX 系统可以通过设置locale来确定系统的字符编码，由于Windows使用的locale中不是POSIX标准的locale格式，因此在Windows下需要采用另一个配置参数charset来指定字符编码。在Linux 系统中也可以使用charset来指定字符编码。 在 Linux 中 locale 的命名规则为: <语言>\_<地区>.<字符集编码> 如：zh_CN.UTF-8，zh代表中文，CN代表大陆地区，UTF-8表示字符集。字符集编码为客户端正确解析本地字符串提供编码转换的说明。Linux系统与 Mac OSX 系统可以通过设置locale来确定系统的字符编码，由于Windows使用的locale中不是POSIX标准的locale格式，因此在Windows下需要采用另一个配置参数charset来指定字符编码。在Linux 系统中也可以使用charset来指定字符编码。
- charset - charset
...@@ -452,39 +452,39 @@ TDengine的所有可执行文件默认存放在 _/usr/local/taos/bin_ 目录下 ...@@ -452,39 +452,39 @@ TDengine的所有可执行文件默认存放在 _/usr/local/taos/bin_ 目录下
| 关键字列表 | | | | | | 关键字列表 | | | | |
| ---------- | ----------- | ------------ | ---------- | --------- | | ---------- | ----------- | ------------ | ---------- | --------- |
| ABLOCKS | CONNECTION | GT | MINUS | SHOW | | ABLOCKS | CONNECTION | GROUP | MINUS | SLASH |
| ABORT | CONNECTIONS | ID | MNODES | SLASH | | ABORT | CONNECTIONS | GT | MNODES | SLIDING |
| ACCOUNT | COPY | IF | MODULES | SLIDING | | ACCOUNT | COPY | ID | MODULES | SMALLINT |
| ACCOUNTS | COUNT | IGNORE | NCHAR | SMALLINT | | ACCOUNTS | COUNT | IF | NCHAR | SPREAD |
| ADD | CREATE | IMMEDIATE | NE | SPREAD | | ADD | CREATE | IGNORE | NE | STABLE |
| AFTER | CTIME | IMPORT | NONE | STAR | | AFTER | CTIME | IMMEDIATE | NONE | STABLES |
| ALL | DATABASE | IN | NOT | STATEMENT | | ALL | DATABASE | IMPORT | NOT | STAR |
| ALTER | DATABASES | INITIALLY | NOTNULL | STDDEV | | ALTER | DATABASES | IN | NOTNULL | STATEMENT |
| AND | DAYS | INSERT | NOW | STREAM | | AND | DAYS | INITIALLY | NOW | STDDEV |
| AS | DEFERRED | INSTEAD | OF | STREAMS | | AS | DEFERRED | INSERT | OF | STREAM |
| ASC | DELIMITERS | INTEGER | OFFSET | STRING | | ASC | DELIMITERS | INSTEAD | OFFSET | STREAMS |
| ATTACH | DESC | INTERVAL | OR | SUM | | ATTACH | DESC | INTEGER | OR | STRING |
| AVG | DESCRIBE | INTO | ORDER | TABLE | | AVG | DESCRIBE | INTERVAL | ORDER | SUM |
| BEFORE | DETACH | IP | PASS | TABLES | | BEFORE | DETACH | INTO | PASS | TABLE |
| BEGIN | DIFF | IS | PERCENTILE | TAG | | BEGIN | DIFF | IP | PERCENTILE | TABLES |
| BETWEEN | DIVIDE | ISNULL | PLUS | TAGS | | BETWEEN | DISTINCT | IS | PLUS | TAG |
| BIGINT | DNODE | JOIN | PRAGMA | TBLOCKS | | BIGINT | DIVIDE | ISNULL | PRAGMA | TAGS |
| BINARY | DNODES | KEEP | PREV | TBNAME | | BINARY | DNODE | JOIN | PREV | TBLOCKS |
| BITAND | DOT | KEY | PRIVILEGE | TIMES | | BITAND | DNODES | KEEP | PRIVILEGE | TBNAME |
| BITNOT | DOUBLE | KILL | QUERIES | TIMESTAMP | | BITNOT | DOT | KEY | QUERIES | TIMES |
| BITOR | DROP | LAST | QUERY | TINYINT | | BITOR | DOUBLE | KILL | QUERY | TIMESTAMP |
| BOOL | EACH | LE | RAISE | TOP | | BOOL | DROP | LAST | RAISE | TINYINT |
| BOTTOM | END | LEASTSQUARES | REM | TRIGGER | | BOTTOM | EACH | LE | REM | TOP |
| BY | EQ | LIKE | REPLACE | UMINUS | | BY | END | LEASTSQUARES | REPLACE | TRIGGER |
| CACHE | EXISTS | LIMIT | REPLICA | UPLUS | | CACHE | EQ | LIKE | REPLICA | UMINUS |
| CASCADE | EXPLAIN | LINEAR | RESET | USE | | CASCADE | EXISTS | LIMIT | RESET | UPLUS |
| CHANGE | FAIL | LOCAL | RESTRICT | USER | | CHANGE | EXPLAIN | LINEAR | RESTRICT | USE |
| CLOG | FILL | LP | ROW | USERS | | CLOG | FAIL | LOCAL | ROW | USER |
| CLUSTER | FIRST | LSHIFT | ROWS | USING | | CLUSTER | FILL | LP | ROWS | USERS |
| COLON | FLOAT | LT | RP | VALUES | | COLON | FIRST | LSHIFT | RP | USING |
| COLUMN | FOR | MATCH | RSHIFT | VARIABLE | | COLUMN | FLOAT | LT | RSHIFT | VALUES |
| COMMA | FROM | MAX | SCORES | VGROUPS | | COMMA | FOR | MATCH | SCORES | VARIABLE |
| COMP | GE | METRIC | SELECT | VIEW | | COMP | FROM | MAX | SELECT | VGROUPS |
| CONCAT | GLOB | METRICS | SEMI | WAVG | | CONCAT | GE | METRIC | SEMI | VIEW |
| CONFIGS | GRANTS | MIN | SET | WHERE | | CONFIGS | GLOB | METRICS | SET | WAVG |
| CONFLICT | GROUP | | | | | CONFLICT | GRANTS | MIN | SHOW | WHERE |
\ No newline at end of file
documentation20/webdocs/markdowndocs/architecture-ch.md
浏览文件 @ 942c8814
...@@ -344,7 +344,6 @@ dataDir /mnt/disk6/taos 2 ...@@ -344,7 +344,6 @@ dataDir /mnt/disk6/taos 2
挂载的盘也可以是非本地的网络盘，只要系统能访问即可。 挂载的盘也可以是非本地的网络盘，只要系统能访问即可。
注：多级存储功能仅企业版支持 注：多级存储功能仅企业版支持
**提示：该功能暂未提供**
## 数据查询 ## 数据查询
TDengine提供了多种多样针对表和超级表的查询处理功能，除了常规的聚合查询之外，还提供针对时序数据的窗口查询、统计聚合等功能。TDengine的查询处理需要客户端、vnode, mnode节点协同完成。 TDengine提供了多种多样针对表和超级表的查询处理功能，除了常规的聚合查询之外，还提供针对时序数据的窗口查询、统计聚合等功能。TDengine的查询处理需要客户端、vnode, mnode节点协同完成。
......
packaging/cfg/taos.cfg
浏览文件 @ 942c8814
...@@ -270,3 +270,9 @@ ...@@ -270,3 +270,9 @@
# in retrieve blocking model, only in 50% query threads will be used in query processing in dnode # in retrieve blocking model, only in 50% query threads will be used in query processing in dnode
# retrieveBlockingModel 0 # retrieveBlockingModel 0
# the maximum allowed query buffer size in MB during query processing for each data node
# -1 no limit (default)
# 0 no query allowed, queries are disabled
# queryBufferSize -1
packaging/deb/makedeb.sh
浏览文件 @ 942c8814
...@@ -43,11 +43,13 @@ mkdir -p ${pkg_dir}${install_home_path}/include ...@@ -43,11 +43,13 @@ mkdir -p ${pkg_dir}${install_home_path}/include
mkdir -p ${pkg_dir}${install_home_path}/init.d mkdir -p ${pkg_dir}${install_home_path}/init.d
mkdir -p ${pkg_dir}${install_home_path}/script mkdir -p ${pkg_dir}${install_home_path}/script
echo "" > ${pkg_dir}${install_home_path}/email
cp ${compile_dir}/../packaging/cfg/taos.cfg ${pkg_dir}${install_home_path}/cfg cp ${compile_dir}/../packaging/cfg/taos.cfg ${pkg_dir}${install_home_path}/cfg
cp ${compile_dir}/../packaging/deb/taosd ${pkg_dir}${install_home_path}/init.d cp ${compile_dir}/../packaging/deb/taosd ${pkg_dir}${install_home_path}/init.d
cp ${compile_dir}/../packaging/tools/post.sh ${pkg_dir}${install_home_path}/script cp ${compile_dir}/../packaging/tools/post.sh ${pkg_dir}${install_home_path}/script
cp ${compile_dir}/../packaging/tools/preun.sh ${pkg_dir}${install_home_path}/script cp ${compile_dir}/../packaging/tools/preun.sh ${pkg_dir}${install_home_path}/script
cp ${compile_dir}/../packaging/tools/startPre.sh ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/../packaging/tools/set_core.sh ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/../packaging/tools/taosd-dump-cfg.gdb ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/build/bin/taosdemo ${pkg_dir}${install_home_path}/bin cp ${compile_dir}/build/bin/taosdemo ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/build/bin/taosdemox ${pkg_dir}${install_home_path}/bin cp ${compile_dir}/build/bin/taosdemox ${pkg_dir}${install_home_path}/bin
cp ${compile_dir}/build/bin/taosdump ${pkg_dir}${install_home_path}/bin cp ${compile_dir}/build/bin/taosdump ${pkg_dir}${install_home_path}/bin
......
packaging/rpm/tdengine.spec
浏览文件 @ 942c8814
...@@ -51,11 +51,13 @@ mkdir -p %{buildroot}%{homepath}/include ...@@ -51,11 +51,13 @@ mkdir -p %{buildroot}%{homepath}/include
mkdir -p %{buildroot}%{homepath}/init.d mkdir -p %{buildroot}%{homepath}/init.d
mkdir -p %{buildroot}%{homepath}/script mkdir -p %{buildroot}%{homepath}/script
echo "" > %{buildroot}%{homepath}/email
cp %{_compiledir}/../packaging/cfg/taos.cfg %{buildroot}%{homepath}/cfg cp %{_compiledir}/../packaging/cfg/taos.cfg %{buildroot}%{homepath}/cfg
cp %{_compiledir}/../packaging/rpm/taosd %{buildroot}%{homepath}/init.d cp %{_compiledir}/../packaging/rpm/taosd %{buildroot}%{homepath}/init.d
cp %{_compiledir}/../packaging/tools/post.sh %{buildroot}%{homepath}/script cp %{_compiledir}/../packaging/tools/post.sh %{buildroot}%{homepath}/script
cp %{_compiledir}/../packaging/tools/preun.sh %{buildroot}%{homepath}/script cp %{_compiledir}/../packaging/tools/preun.sh %{buildroot}%{homepath}/script
cp %{_compiledir}/../packaging/tools/startPre.sh %{buildroot}%{homepath}/bin
cp %{_compiledir}/../packaging/tools/set_core.sh %{buildroot}%{homepath}/bin
cp %{_compiledir}/../packaging/tools/taosd-dump-cfg.gdb %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/taos %{buildroot}%{homepath}/bin cp %{_compiledir}/build/bin/taos %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/taosd %{buildroot}%{homepath}/bin cp %{_compiledir}/build/bin/taosd %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/bin/taosdemo %{buildroot}%{homepath}/bin cp %{_compiledir}/build/bin/taosdemo %{buildroot}%{homepath}/bin
......
packaging/tools/install.sh
浏览文件 @ 942c8814
...@@ -604,9 +604,7 @@ function install_service_on_systemd() { ...@@ -604,9 +604,7 @@ function install_service_on_systemd() {
${csudo} bash -c "echo '[Service]' >> ${taosd_service_config}" ${csudo} bash -c "echo '[Service]' >> ${taosd_service_config}"
${csudo} bash -c "echo 'Type=simple' >> ${taosd_service_config}" ${csudo} bash -c "echo 'Type=simple' >> ${taosd_service_config}"
${csudo} bash -c "echo 'ExecStart=/usr/bin/taosd' >> ${taosd_service_config}" ${csudo} bash -c "echo 'ExecStart=/usr/bin/taosd' >> ${taosd_service_config}"
#${csudo} bash -c "echo 'ExecStartPre=/usr/local/taos/bin/setDelay.sh' >> ${taosd_service_config}" ${csudo} bash -c "echo 'ExecStartPre=/usr/local/taos/bin/startPre.sh' >> ${taosd_service_config}"
#${csudo} bash -c "echo 'ExecStartPost=/usr/local/taos/bin/resetDelay.sh' >> ${taosd_service_config}"
#${csudo} bash -c "echo 'ExecStopPost=/usr/local/taos/bin/resetDelay.sh' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitNOFILE=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitNOFILE=infinity' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitNPROC=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitNPROC=infinity' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitCORE=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitCORE=infinity' >> ${taosd_service_config}"
......
packaging/tools/install_power.sh
浏览文件 @ 942c8814
...@@ -578,6 +578,7 @@ function install_service_on_systemd() { ...@@ -578,6 +578,7 @@ function install_service_on_systemd() {
${csudo} bash -c "echo '[Service]' >> ${powerd_service_config}" ${csudo} bash -c "echo '[Service]' >> ${powerd_service_config}"
${csudo} bash -c "echo 'Type=simple' >> ${powerd_service_config}" ${csudo} bash -c "echo 'Type=simple' >> ${powerd_service_config}"
${csudo} bash -c "echo 'ExecStart=/usr/bin/powerd' >> ${powerd_service_config}" ${csudo} bash -c "echo 'ExecStart=/usr/bin/powerd' >> ${powerd_service_config}"
${csudo} bash -c "echo 'ExecStartPre=/usr/local/power/bin/startPre.sh' >> ${powerd_service_config}"
${csudo} bash -c "echo 'LimitNOFILE=infinity' >> ${powerd_service_config}" ${csudo} bash -c "echo 'LimitNOFILE=infinity' >> ${powerd_service_config}"
${csudo} bash -c "echo 'LimitNPROC=infinity' >> ${powerd_service_config}" ${csudo} bash -c "echo 'LimitNPROC=infinity' >> ${powerd_service_config}"
${csudo} bash -c "echo 'LimitCORE=infinity' >> ${powerd_service_config}" ${csudo} bash -c "echo 'LimitCORE=infinity' >> ${powerd_service_config}"
......
packaging/tools/make_install.sh
浏览文件 @ 942c8814
...@@ -149,10 +149,12 @@ function install_bin() { ...@@ -149,10 +149,12 @@ function install_bin() {
${csudo} rm -f ${bin_link_dir}/rmtaos || : ${csudo} rm -f ${bin_link_dir}/rmtaos || :
${csudo} cp -r ${binary_dir}/build/bin/* ${install_main_dir}/bin ${csudo} cp -r ${binary_dir}/build/bin/* ${install_main_dir}/bin
${csudo} cp -r ${script_dir}/taosd-dump-cfg.gdb ${install_main_dir}/bin
if [ "$osType" != "Darwin" ]; then if [ "$osType" != "Darwin" ]; then
${csudo} cp -r ${script_dir}/remove.sh ${install_main_dir}/bin ${csudo} cp -r ${script_dir}/remove.sh ${install_main_dir}/bin
${csudo} cp -r ${script_dir}/set_core.sh ${install_main_dir}/bin ${csudo} cp -r ${script_dir}/set_core.sh ${install_main_dir}/bin
${csudo} cp -r ${script_dir}/startPre.sh ${install_main_dir}/bin
else else
${csudo} cp -r ${script_dir}/remove_client.sh ${install_main_dir}/bin ${csudo} cp -r ${script_dir}/remove_client.sh ${install_main_dir}/bin
fi fi
...@@ -330,6 +332,7 @@ function install_service_on_systemd() { ...@@ -330,6 +332,7 @@ function install_service_on_systemd() {
${csudo} bash -c "echo '[Service]' >> ${taosd_service_config}" ${csudo} bash -c "echo '[Service]' >> ${taosd_service_config}"
${csudo} bash -c "echo 'Type=simple' >> ${taosd_service_config}" ${csudo} bash -c "echo 'Type=simple' >> ${taosd_service_config}"
${csudo} bash -c "echo 'ExecStart=/usr/bin/taosd' >> ${taosd_service_config}" ${csudo} bash -c "echo 'ExecStart=/usr/bin/taosd' >> ${taosd_service_config}"
${csudo} bash -c "echo 'ExecStartPre=/usr/local/taos/bin/startPre.sh' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitNOFILE=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitNOFILE=infinity' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitNPROC=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitNPROC=infinity' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitCORE=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitCORE=infinity' >> ${taosd_service_config}"
......
packaging/tools/makeclient.sh
浏览文件 @ 942c8814
...@@ -45,7 +45,8 @@ if [ "$osType" != "Darwin" ]; then ...@@ -45,7 +45,8 @@ if [ "$osType" != "Darwin" ]; then
strip ${build_dir}/bin/taos strip ${build_dir}/bin/taos
bin_files="${build_dir}/bin/taos ${script_dir}/remove_client.sh" bin_files="${build_dir}/bin/taos ${script_dir}/remove_client.sh"
else else
bin_files="${build_dir}/bin/taos ${build_dir}/bin/taosdump ${build_dir}/bin/taosdemo ${build_dir}/bin/taosdemox ${script_dir}/remove_client.sh ${script_dir}/set_core.sh ${script_dir}/get_client.sh" bin_files="${build_dir}/bin/taos ${build_dir}/bin/taosdump ${build_dir}/bin/taosdemo ${build_dir}/bin/taosdemox\
${script_dir}/remove_client.sh ${script_dir}/set_core.sh ${script_dir}/get_client.sh ${script_dir}/taosd-dump-cfg.gdb"
fi fi
lib_files="${build_dir}/lib/libtaos.so.${version}" lib_files="${build_dir}/lib/libtaos.so.${version}"
else else
......
packaging/tools/makeclient_power.sh
浏览文件 @ 942c8814
...@@ -81,6 +81,7 @@ if [ "$osType" != "Darwin" ]; then ...@@ -81,6 +81,7 @@ if [ "$osType" != "Darwin" ]; then
cp ${build_dir}/bin/taosdump ${install_dir}/bin/powerdump cp ${build_dir}/bin/taosdump ${install_dir}/bin/powerdump
cp ${script_dir}/set_core.sh ${install_dir}/bin cp ${script_dir}/set_core.sh ${install_dir}/bin
cp ${script_dir}/get_client.sh ${install_dir}/bin cp ${script_dir}/get_client.sh ${install_dir}/bin
cp ${script_dir}/taosd-dump-cfg.gdb ${install_dir}/bin
fi fi
else else
cp ${bin_files} ${install_dir}/bin cp ${bin_files} ${install_dir}/bin
......
packaging/tools/makepkg.sh
浏览文件 @ 942c8814
...@@ -36,7 +36,8 @@ if [ "$pagMode" == "lite" ]; then ...@@ -36,7 +36,8 @@ if [ "$pagMode" == "lite" ]; then
strip ${build_dir}/bin/taos strip ${build_dir}/bin/taos
bin_files="${build_dir}/bin/taosd ${build_dir}/bin/taos ${script_dir}/remove.sh" bin_files="${build_dir}/bin/taosd ${build_dir}/bin/taos ${script_dir}/remove.sh"
else else
bin_files="${build_dir}/bin/taosd ${build_dir}/bin/taos ${build_dir}/bin/taosdump ${build_dir}/bin/taosdemo ${build_dir}/bin/taosdemox ${build_dir}/bin/tarbitrator ${script_dir}/remove.sh ${script_dir}/set_core.sh ${script_dir}/get_client.sh" bin_files="${build_dir}/bin/taosd ${build_dir}/bin/taos ${build_dir}/bin/taosdump ${build_dir}/bin/taosdemo ${build_dir}/bin/taosdemox ${build_dir}/bin/tarbitrator\
${script_dir}/remove.sh ${script_dir}/set_core.sh ${script_dir}/startPre.sh ${script_dir}/taosd-dump-cfg.gdb"
fi fi
lib_files="${build_dir}/lib/libtaos.so.${version}" lib_files="${build_dir}/lib/libtaos.so.${version}"
......
packaging/tools/makepkg_power.sh
浏览文件 @ 942c8814
...@@ -36,7 +36,8 @@ fi ...@@ -36,7 +36,8 @@ fi
# strip ${build_dir}/bin/taos # strip ${build_dir}/bin/taos
# bin_files="${build_dir}/bin/powerd ${build_dir}/bin/power ${script_dir}/remove_power.sh" # bin_files="${build_dir}/bin/powerd ${build_dir}/bin/power ${script_dir}/remove_power.sh"
#else #else
# bin_files="${build_dir}/bin/powerd ${build_dir}/bin/power ${build_dir}/bin/powerdemo ${build_dir}/bin/tarbitrator ${script_dir}/remove_power.sh ${script_dir}/set_core.sh" # bin_files="${build_dir}/bin/powerd ${build_dir}/bin/power ${build_dir}/bin/powerdemo ${build_dir}/bin/tarbitrator ${script_dir}/remove_power.sh\
# ${script_dir}/set_core.sh ${script_dir}/startPre.sh ${script_dir}/taosd-dump-cfg.gdb"
#fi #fi
lib_files="${build_dir}/lib/libtaos.so.${version}" lib_files="${build_dir}/lib/libtaos.so.${version}"
...@@ -82,6 +83,8 @@ else ...@@ -82,6 +83,8 @@ else
cp ${build_dir}/bin/tarbitrator ${install_dir}/bin cp ${build_dir}/bin/tarbitrator ${install_dir}/bin
cp ${script_dir}/set_core.sh ${install_dir}/bin cp ${script_dir}/set_core.sh ${install_dir}/bin
cp ${script_dir}/get_client.sh ${install_dir}/bin cp ${script_dir}/get_client.sh ${install_dir}/bin
cp ${script_dir}/startPre.sh ${install_dir}/bin
cp ${script_dir}/taosd-dump-cfg.gdb ${install_dir}/bin
fi fi
chmod a+x ${install_dir}/bin/* || : chmod a+x ${install_dir}/bin/* || :
......
packaging/tools/post.sh
浏览文件 @ 942c8814
...@@ -406,6 +406,7 @@ function install_service_on_systemd() { ...@@ -406,6 +406,7 @@ function install_service_on_systemd() {
${csudo} bash -c "echo '[Service]' >> ${taosd_service_config}" ${csudo} bash -c "echo '[Service]' >> ${taosd_service_config}"
${csudo} bash -c "echo 'Type=simple' >> ${taosd_service_config}" ${csudo} bash -c "echo 'Type=simple' >> ${taosd_service_config}"
${csudo} bash -c "echo 'ExecStart=/usr/bin/taosd' >> ${taosd_service_config}" ${csudo} bash -c "echo 'ExecStart=/usr/bin/taosd' >> ${taosd_service_config}"
${csudo} bash -c "echo 'ExecStartPre=/usr/local/taos/bin/startPre.sh' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitNOFILE=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitNOFILE=infinity' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitNPROC=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitNPROC=infinity' >> ${taosd_service_config}"
${csudo} bash -c "echo 'LimitCORE=infinity' >> ${taosd_service_config}" ${csudo} bash -c "echo 'LimitCORE=infinity' >> ${taosd_service_config}"
......
packaging/tools/startPre.sh 0 → 100644
浏览文件 @ 942c8814
#!/bin/bash
#
# if enable core dump, set start count to 3, disable core dump, set start count to 20.
# set -e
# set -x
taosd=/etc/systemd/system/taosd.service
line=`grep StartLimitBurst ${taosd}`
num=${line##*=}
#echo "burst num: ${num}"
startSeqFile=/usr/local/taos/.startSeq
recordFile=/usr/local/taos/.startRecord
startSeq=0
if [[ ! -e ${startSeqFile} ]]; then
startSeq=0
else
startSeq=$(cat ${startSeqFile})
fi
nextSeq=`expr $startSeq + 1`
echo "${nextSeq}" > ${startSeqFile}
curTime=$(date "+%Y-%m-%d %H:%M:%S")
echo "startSeq:${startSeq} startPre.sh exec ${curTime}, burstCnt:${num}" >> ${recordFile}
coreFlag=`ulimit -c`
echo "coreFlag: ${coreFlag}" >> ${recordFile}
if [ ${coreFlag} = "0" ];then
#echo "core is 0"
if [ ${num} != "20" ];then
sed -i "s/^.*StartLimitBurst.*$/StartLimitBurst=20/" ${taosd}
systemctl daemon-reload
echo "modify burst count from ${num} to 20" >> ${recordFile}
fi
fi
if [ ${coreFlag} = "unlimited" ];then
#echo "core is unlimited"
if [ ${num} != "3" ];then
sed -i "s/^.*StartLimitBurst.*$/StartLimitBurst=3/" ${taosd}
systemctl daemon-reload
echo "modify burst count from ${num} to 3" >> ${recordFile}
fi
fi
src/client/inc/tsclient.h
浏览文件 @ 942c8814
...@@ -297,6 +297,11 @@ typedef struct { ...@@ -297,6 +297,11 @@ typedef struct {
struct SLocalMerger *pLocalMerger; struct SLocalMerger *pLocalMerger;
} SSqlRes; } SSqlRes;
typedef struct {
char key[512];
void *pDnodeConn;
} SRpcObj;
typedef struct STscObj { typedef struct STscObj {
void * signature; void * signature;
void * pTimer; void * pTimer;
...@@ -312,8 +317,8 @@ typedef struct STscObj { ...@@ -312,8 +317,8 @@ typedef struct STscObj {
int64_t hbrid; int64_t hbrid;
struct SSqlObj * sqlList; struct SSqlObj * sqlList;
struct SSqlStream *streamList; struct SSqlStream *streamList;
SRpcCorEpSet *tscCorMgmtEpSet; SRpcObj *pRpcObj;
void* pDnodeConn; SRpcCorEpSet *tscCorMgmtEpSet;
pthread_mutex_t mutex; pthread_mutex_t mutex;
int32_t numOfObj; // number of sqlObj from this tscObj int32_t numOfObj; // number of sqlObj from this tscObj
} STscObj; } STscObj;
...@@ -390,8 +395,10 @@ typedef struct SSqlStream { ...@@ -390,8 +395,10 @@ typedef struct SSqlStream {
void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable); void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable);
int32_t tscInitRpc(const char *user, const char *secret, void** pDnodeConn);
void tscInitMsgsFp(); int tscAcquireRpc(const char *key, const char *user, const char *secret,void **pRpcObj);
void tscReleaseRpc(void *param);
void tscInitMsgsFp();
int tsParseSql(SSqlObj *pSql, bool initial); int tsParseSql(SSqlObj *pSql, bool initial);
......
src/client/src/tscSQLParser.c
浏览文件 @ 942c8814
...@@ -5427,6 +5427,7 @@ int32_t validateColumnName(char* name) { ...@@ -5427,6 +5427,7 @@ int32_t validateColumnName(char* name) {
if (token.type == TK_STRING) { if (token.type == TK_STRING) {
strdequote(token.z); strdequote(token.z);
strntolower(token.z, token.z, token.n);
token.n = (uint32_t)strtrim(token.z); token.n = (uint32_t)strtrim(token.z);
int32_t k = tSQLGetToken(token.z, &token.type); int32_t k = tSQLGetToken(token.z, &token.type);
......
src/client/src/tscServer.c
浏览文件 @ 942c8814
...@@ -157,13 +157,16 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) { ...@@ -157,13 +157,16 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
SRpcEpSet *epSet = &pRsp->epSet; SRpcEpSet *epSet = &pRsp->epSet;
if (epSet->numOfEps > 0) { if (epSet->numOfEps > 0) {
tscEpSetHtons(epSet); tscEpSetHtons(epSet);
if (!tscEpSetIsEqual(&pSql->pTscObj->tscCorMgmtEpSet->epSet, epSet)) {
tscTrace("%p updating epset: numOfEps: %d, inUse: %d", pSql, epSet->numOfEps, epSet->inUse); //SRpcCorEpSet *pCorEpSet = pSql->pTscObj->tscCorMgmtEpSet;
for (int8_t i = 0; i < epSet->numOfEps; i++) { //if (!tscEpSetIsEqual(&pCorEpSet->epSet, epSet)) {
tscTrace("endpoint %d: fqdn=%s, port=%d", i, epSet->fqdn[i], epSet->port[i]); // tscTrace("%p updating epset: numOfEps: %d, inUse: %d", pSql, epSet->numOfEps, epSet->inUse);
} // for (int8_t i = 0; i < epSet->numOfEps; i++) {
tscUpdateMgmtEpSet(pSql, epSet); // tscTrace("endpoint %d: fqdn=%s, port=%d", i, epSet->fqdn[i], epSet->port[i]);
} // }
//}
//concurrency problem, update mgmt epset anyway
tscUpdateMgmtEpSet(pSql, epSet);
} }
pSql->pTscObj->connId = htonl(pRsp->connId); pSql->pTscObj->connId = htonl(pRsp->connId);
...@@ -270,7 +273,8 @@ int tscSendMsgToServer(SSqlObj *pSql) { ...@@ -270,7 +273,8 @@ int tscSendMsgToServer(SSqlObj *pSql) {
.code = 0 .code = 0
}; };
rpcSendRequest(pObj->pDnodeConn, &pSql->epSet, &rpcMsg, &pSql->rpcRid);
rpcSendRequest(pObj->pRpcObj->pDnodeConn, &pSql->epSet, &rpcMsg, &pSql->rpcRid);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -292,8 +296,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { ...@@ -292,8 +296,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
if (pObj->signature != pObj) { if (pObj->signature != pObj) {
tscDebug("%p DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command, pObj, pObj->signature); tscDebug("%p DB connection is closed, cmd:%d pObj:%p signature:%p", pSql, pCmd->command, pObj, pObj->signature);
taosRemoveRef(tscObjRef, pSql->self); taosRemoveRef(tscObjRef, handle);
taosReleaseRef(tscObjRef, pSql->self); taosReleaseRef(tscObjRef, handle);
rpcFreeCont(rpcMsg->pCont); rpcFreeCont(rpcMsg->pCont);
return; return;
} }
...@@ -303,8 +307,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { ...@@ -303,8 +307,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
tscDebug("%p sqlObj needs to be released or DB connection is closed, cmd:%d type:%d, pObj:%p signature:%p", tscDebug("%p sqlObj needs to be released or DB connection is closed, cmd:%d type:%d, pObj:%p signature:%p",
pSql, pCmd->command, pQueryInfo->type, pObj, pObj->signature); pSql, pCmd->command, pQueryInfo->type, pObj, pObj->signature);
taosRemoveRef(tscObjRef, pSql->self); taosRemoveRef(tscObjRef, handle);
taosReleaseRef(tscObjRef, pSql->self); taosReleaseRef(tscObjRef, handle);
rpcFreeCont(rpcMsg->pCont); rpcFreeCont(rpcMsg->pCont);
return; return;
} }
...@@ -350,7 +354,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { ...@@ -350,7 +354,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
// if there is an error occurring, proceed to the following error handling procedure. // if there is an error occurring, proceed to the following error handling procedure.
if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
taosReleaseRef(tscObjRef, pSql->self); taosReleaseRef(tscObjRef, handle);
rpcFreeCont(rpcMsg->pCont); rpcFreeCont(rpcMsg->pCont);
return; return;
} }
...@@ -418,13 +422,15 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { ...@@ -418,13 +422,15 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
(*pSql->fp)(pSql->param, pSql, rpcMsg->code); (*pSql->fp)(pSql->param, pSql, rpcMsg->code);
} }
taosReleaseRef(tscObjRef, pSql->self);
if (shouldFree) { // in case of table-meta/vgrouplist query, automatically free it if (shouldFree) { // in case of table-meta/vgrouplist query, automatically free it
taosRemoveRef(tscObjRef, pSql->self); taosRemoveRef(tscObjRef, handle);
tscDebug("%p sqlObj is automatically freed", pSql); tscDebug("%p sqlObj is automatically freed", pSql);
} }
taosReleaseRef(tscObjRef, handle);
rpcFreeCont(rpcMsg->pCont); rpcFreeCont(rpcMsg->pCont);
} }
......
src/client/src/tscSql.c
浏览文件 @ 942c8814
...@@ -90,9 +90,11 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa ...@@ -90,9 +90,11 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
} else { } else {
if (tscSetMgmtEpSetFromCfg(tsFirst, tsSecond, &corMgmtEpSet) < 0) return NULL; if (tscSetMgmtEpSetFromCfg(tsFirst, tsSecond, &corMgmtEpSet) < 0) return NULL;
} }
char rpcKey[512] = {0};
snprintf(rpcKey, sizeof(rpcKey), "%s:%s:%s:%d", user, pass, ip, port);
void *pDnodeConn = NULL; void *pRpcObj = NULL;
if (tscInitRpc(user, secretEncrypt, &pDnodeConn) != 0) { if (tscAcquireRpc(rpcKey, user, secretEncrypt, &pRpcObj) != 0) {
terrno = TSDB_CODE_RPC_NETWORK_UNAVAIL; terrno = TSDB_CODE_RPC_NETWORK_UNAVAIL;
return NULL; return NULL;
} }
...@@ -100,23 +102,21 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa ...@@ -100,23 +102,21 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
STscObj *pObj = (STscObj *)calloc(1, sizeof(STscObj)); STscObj *pObj = (STscObj *)calloc(1, sizeof(STscObj));
if (NULL == pObj) { if (NULL == pObj) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
rpcClose(pDnodeConn); tscReleaseRpc(pRpcObj);
return NULL; return NULL;
} }
// set up tscObj's mgmtEpSet
pObj->tscCorMgmtEpSet = (SRpcCorEpSet *)malloc(sizeof(SRpcCorEpSet)); pObj->tscCorMgmtEpSet = malloc(sizeof(SRpcCorEpSet));
if (NULL == pObj->tscCorMgmtEpSet) { if (pObj->tscCorMgmtEpSet == NULL) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
rpcClose(pDnodeConn); tscReleaseRpc(pRpcObj);
free(pObj->tscCorMgmtEpSet);
free(pObj); free(pObj);
return NULL; return NULL;
} }
memcpy(pObj->tscCorMgmtEpSet, &corMgmtEpSet, sizeof(SRpcCorEpSet)); memcpy(pObj->tscCorMgmtEpSet, &corMgmtEpSet, sizeof(corMgmtEpSet));
pObj->signature = pObj;
pObj->pDnodeConn = pDnodeConn;
pObj->signature = pObj;
pObj->pRpcObj = (SRpcObj *)pRpcObj;
tstrncpy(pObj->user, user, sizeof(pObj->user)); tstrncpy(pObj->user, user, sizeof(pObj->user));
secretEncryptLen = MIN(secretEncryptLen, sizeof(pObj->pass)); secretEncryptLen = MIN(secretEncryptLen, sizeof(pObj->pass));
memcpy(pObj->pass, secretEncrypt, secretEncryptLen); memcpy(pObj->pass, secretEncrypt, secretEncryptLen);
...@@ -126,8 +126,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa ...@@ -126,8 +126,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
/* db name is too long */ /* db name is too long */
if (len >= TSDB_DB_NAME_LEN) { if (len >= TSDB_DB_NAME_LEN) {
terrno = TSDB_CODE_TSC_INVALID_DB_LENGTH; terrno = TSDB_CODE_TSC_INVALID_DB_LENGTH;
rpcClose(pDnodeConn); tscReleaseRpc(pRpcObj);
free(pObj->tscCorMgmtEpSet);
free(pObj); free(pObj);
return NULL; return NULL;
} }
...@@ -144,8 +143,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa ...@@ -144,8 +143,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj)); SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj));
if (NULL == pSql) { if (NULL == pSql) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
rpcClose(pDnodeConn); tscReleaseRpc(pRpcObj);
free(pObj->tscCorMgmtEpSet);
free(pObj); free(pObj);
return NULL; return NULL;
} }
...@@ -161,9 +159,8 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa ...@@ -161,9 +159,8 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
if (TSDB_CODE_SUCCESS != tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE)) { if (TSDB_CODE_SUCCESS != tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE)) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
rpcClose(pDnodeConn); tscReleaseRpc(pRpcObj);
free(pSql); free(pSql);
free(pObj->tscCorMgmtEpSet);
free(pObj); free(pObj);
return NULL; return NULL;
} }
...@@ -202,7 +199,7 @@ TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass, ...@@ -202,7 +199,7 @@ TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass,
return NULL; return NULL;
} }
tscDebug("%p DB connection is opening, dnodeConn:%p", pObj, pObj->pDnodeConn); tscDebug("%p DB connection is opening, rpcObj: %p, dnodeConn:%p", pObj, pObj->pRpcObj, pObj->pRpcObj->pDnodeConn);
taos_free_result(pSql); taos_free_result(pSql);
// version compare only requires the first 3 segments of the version string // version compare only requires the first 3 segments of the version string
...@@ -279,7 +276,7 @@ void taos_close(TAOS *taos) { ...@@ -279,7 +276,7 @@ void taos_close(TAOS *taos) {
return; return;
} }
tscDebug("%p try to free tscObj and close dnodeConn:%p", pObj, pObj->pDnodeConn); tscDebug("%p try to free tscObj", pObj);
if (pObj->signature != pObj) { if (pObj->signature != pObj) {
tscDebug("%p already closed or invalid tscObj", pObj); tscDebug("%p already closed or invalid tscObj", pObj);
return; return;
...@@ -303,7 +300,7 @@ void taos_close(TAOS *taos) { ...@@ -303,7 +300,7 @@ void taos_close(TAOS *taos) {
} }
} }
tscDebug("%p all sqlObj are freed, free tscObj and close dnodeConn:%p", pObj, pObj->pDnodeConn); tscDebug("%p all sqlObj are freed, free tscObj", pObj);
taosRemoveRef(tscRefId, pObj->rid); taosRemoveRef(tscRefId, pObj->rid);
} }
......
src/client/src/tscSystem.c
浏览文件 @ 942c8814
...@@ -42,41 +42,74 @@ void *tscTmr; ...@@ -42,41 +42,74 @@ void *tscTmr;
void *tscQhandle; void *tscQhandle;
int32_t tscRefId = -1; int32_t tscRefId = -1;
int32_t tscNumOfObj = 0; // number of sqlObj in current process. int32_t tscNumOfObj = 0; // number of sqlObj in current process.
static void *tscCheckDiskUsageTmr; static void *tscCheckDiskUsageTmr;
void *tscRpcCache; // cache to keep rpc obj
int32_t tscNumOfThreads = 1; // num of rpc threads
static pthread_mutex_t rpcObjMutex; // mutex to protect open the rpc obj concurrently
static pthread_once_t tscinit = PTHREAD_ONCE_INIT; static pthread_once_t tscinit = PTHREAD_ONCE_INIT;
void tscCheckDiskUsage(void *UNUSED_PARAM(para), void *UNUSED_PARAM(param)) { void tscCheckDiskUsage(void *UNUSED_PARAM(para), void *UNUSED_PARAM(param)) {
taosGetDisk(); taosGetDisk();
taosTmrReset(tscCheckDiskUsage, 1000, NULL, tscTmr, &tscCheckDiskUsageTmr); taosTmrReset(tscCheckDiskUsage, 1000, NULL, tscTmr, &tscCheckDiskUsageTmr);
} }
void tscFreeRpcObj(void *param) {
assert(param);
SRpcObj *pRpcObj = (SRpcObj *)(param);
tscDebug("free rpcObj:%p and free pDnodeConn: %p", pRpcObj, pRpcObj->pDnodeConn);
rpcClose(pRpcObj->pDnodeConn);
}
int32_t tscInitRpc(const char *user, const char *secretEncrypt, void **pDnodeConn) { void tscReleaseRpc(void *param) {
SRpcInit rpcInit; if (param == NULL) {
return;
if (*pDnodeConn == NULL) {
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localPort = 0;
rpcInit.label = "TSC";
rpcInit.numOfThreads = 1; // every DB connection has only one thread
rpcInit.cfp = tscProcessMsgFromServer;
rpcInit.sessions = tsMaxConnections;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.user = (char *)user;
rpcInit.idleTime = 2000;
rpcInit.ckey = "key";
rpcInit.spi = 1;
rpcInit.secret = (char *)secretEncrypt;
*pDnodeConn = rpcOpen(&rpcInit);
if (*pDnodeConn == NULL) {
tscError("failed to init connection to TDengine");
return -1;
} else {
tscDebug("dnodeConn:%p is created, user:%s", *pDnodeConn, user);
}
} }
pthread_mutex_lock(&rpcObjMutex);
taosCacheRelease(tscRpcCache, (void *)&param, false);
pthread_mutex_unlock(&rpcObjMutex);
}
int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncrypt, void **ppRpcObj) {
pthread_mutex_lock(&rpcObjMutex);
SRpcObj *pRpcObj = (SRpcObj *)taosCacheAcquireByKey(tscRpcCache, key, strlen(key));
if (pRpcObj != NULL) {
*ppRpcObj = pRpcObj;
pthread_mutex_unlock(&rpcObjMutex);
return 0;
}
SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localPort = 0;
rpcInit.label = "TSC";
rpcInit.numOfThreads = tscNumOfThreads * 2;
rpcInit.cfp = tscProcessMsgFromServer;
rpcInit.sessions = tsMaxConnections;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.user = (char *)user;
rpcInit.idleTime = 2000;
rpcInit.ckey = "key";
rpcInit.spi = 1;
rpcInit.secret = (char *)secretEncrypt;
SRpcObj rpcObj;
memset(&rpcObj, 0, sizeof(rpcObj));
strncpy(rpcObj.key, key, strlen(key));
rpcObj.pDnodeConn = rpcOpen(&rpcInit);
if (rpcObj.pDnodeConn == NULL) {
pthread_mutex_unlock(&rpcObjMutex);
tscError("failed to init connection to TDengine");
return -1;
}
pRpcObj = taosCachePut(tscRpcCache, rpcObj.key, strlen(rpcObj.key), &rpcObj, sizeof(rpcObj), 1000*10);
if (pRpcObj == NULL) {
rpcClose(rpcObj.pDnodeConn);
pthread_mutex_unlock(&rpcObjMutex);
return -1;
}
*ppRpcObj = pRpcObj;
pthread_mutex_unlock(&rpcObjMutex);
return 0; return 0;
} }
...@@ -117,7 +150,7 @@ void taos_init_imp(void) { ...@@ -117,7 +150,7 @@ void taos_init_imp(void) {
int queueSize = tsMaxConnections*2; int queueSize = tsMaxConnections*2;
double factor = (tscEmbedded == 0)? 2.0:4.0; double factor = (tscEmbedded == 0)? 2.0:4.0;
int32_t tscNumOfThreads = (int)(tsNumOfCores * tsNumOfThreadsPerCore / factor); tscNumOfThreads = (int)(tsNumOfCores * tsNumOfThreadsPerCore / factor);
if (tscNumOfThreads < 2) { if (tscNumOfThreads < 2) {
tscNumOfThreads = 2; tscNumOfThreads = 2;
} }
...@@ -139,6 +172,10 @@ void taos_init_imp(void) { ...@@ -139,6 +172,10 @@ void taos_init_imp(void) {
tscTableMetaInfo = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); tscTableMetaInfo = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
tscDebug("TableMeta:%p", tscTableMetaInfo); tscDebug("TableMeta:%p", tscTableMetaInfo);
} }
int refreshTime = 5;
tscRpcCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, refreshTime, true, tscFreeRpcObj, "rpcObj");
pthread_mutex_init(&rpcObjMutex, NULL);
tscRefId = taosOpenRef(200, tscCloseTscObj); tscRefId = taosOpenRef(200, tscCloseTscObj);
...@@ -180,10 +217,16 @@ void taos_cleanup(void) { ...@@ -180,10 +217,16 @@ void taos_cleanup(void) {
taosCleanupKeywordsTable(); taosCleanupKeywordsTable();
taosCloseLog(); taosCloseLog();
if (tscEmbedded == 0) { p = tscRpcCache;
rpcCleanup(); tscRpcCache = NULL;
if (p != NULL) {
taosCacheCleanup(p);
pthread_mutex_destroy(&rpcObjMutex);
} }
if (tscEmbedded == 0) rpcCleanup();
p = tscTmr; p = tscTmr;
tscTmr = NULL; tscTmr = NULL;
taosTmrCleanUp(p); taosTmrCleanUp(p);
......
src/client/src/tscUtil.c
浏览文件 @ 942c8814
...@@ -32,6 +32,14 @@ ...@@ -32,6 +32,14 @@
static void freeQueryInfoImpl(SQueryInfo* pQueryInfo); static void freeQueryInfoImpl(SQueryInfo* pQueryInfo);
static void clearAllTableMetaInfo(SQueryInfo* pQueryInfo); static void clearAllTableMetaInfo(SQueryInfo* pQueryInfo);
static void tscStrToLower(char *str, int32_t n) {
if (str == NULL || n <= 0) { return;}
for (int32_t i = 0; i < n; i++) {
if (str[i] >= 'A' && str[i] <= 'Z') {
str[i] -= ('A' - 'a');
}
}
}
SCond* tsGetSTableQueryCond(STagCond* pTagCond, uint64_t uid) { SCond* tsGetSTableQueryCond(STagCond* pTagCond, uint64_t uid) {
if (pTagCond->pCond == NULL) { if (pTagCond->pCond == NULL) {
return NULL; return NULL;
...@@ -447,7 +455,6 @@ void tscFreeRegisteredSqlObj(void *pSql) { ...@@ -447,7 +455,6 @@ void tscFreeRegisteredSqlObj(void *pSql) {
SSqlObj* p = *(SSqlObj**)pSql; SSqlObj* p = *(SSqlObj**)pSql;
STscObj* pTscObj = p->pTscObj; STscObj* pTscObj = p->pTscObj;
assert(RID_VALID(p->self)); assert(RID_VALID(p->self));
int32_t num = atomic_sub_fetch_32(&pTscObj->numOfObj, 1); int32_t num = atomic_sub_fetch_32(&pTscObj->numOfObj, 1);
...@@ -898,16 +905,10 @@ void tscCloseTscObj(void *param) { ...@@ -898,16 +905,10 @@ void tscCloseTscObj(void *param) {
pObj->signature = NULL; pObj->signature = NULL;
taosTmrStopA(&(pObj->pTimer)); taosTmrStopA(&(pObj->pTimer));
void* p = pObj->pDnodeConn;
if (pObj->pDnodeConn != NULL) {
rpcClose(pObj->pDnodeConn);
pObj->pDnodeConn = NULL;
}
tfree(pObj->tscCorMgmtEpSet); tfree(pObj->tscCorMgmtEpSet);
tscReleaseRpc(pObj->pRpcObj);
pthread_mutex_destroy(&pObj->mutex); pthread_mutex_destroy(&pObj->mutex);
tscDebug("%p DB connection is closed, dnodeConn:%p", pObj, p);
tfree(pObj); tfree(pObj);
} }
...@@ -1420,9 +1421,11 @@ int32_t tscValidateName(SStrToken* pToken) { ...@@ -1420,9 +1421,11 @@ int32_t tscValidateName(SStrToken* pToken) {
char* sep = strnchr(pToken->z, TS_PATH_DELIMITER[0], pToken->n, true); char* sep = strnchr(pToken->z, TS_PATH_DELIMITER[0], pToken->n, true);
if (sep == NULL) { // single part if (sep == NULL) { // single part
if (pToken->type == TK_STRING) { if (pToken->type == TK_STRING) {
strdequote(pToken->z); strdequote(pToken->z);
tscStrToLower(pToken->z, pToken->n);
pToken->n = (uint32_t)strtrim(pToken->z); pToken->n = (uint32_t)strtrim(pToken->z);
int len = tSQLGetToken(pToken->z, &pToken->type); int len = tSQLGetToken(pToken->z, &pToken->type);
// single token, validate it // single token, validate it
...@@ -1474,7 +1477,7 @@ int32_t tscValidateName(SStrToken* pToken) { ...@@ -1474,7 +1477,7 @@ int32_t tscValidateName(SStrToken* pToken) {
if (pToken->type == TK_STRING && validateQuoteToken(pToken) != TSDB_CODE_SUCCESS) { if (pToken->type == TK_STRING && validateQuoteToken(pToken) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_TSC_INVALID_SQL; return TSDB_CODE_TSC_INVALID_SQL;
} }
// re-build the whole name string // re-build the whole name string
if (pStr[firstPartLen] == TS_PATH_DELIMITER[0]) { if (pStr[firstPartLen] == TS_PATH_DELIMITER[0]) {
// first part do not have quote do nothing // first part do not have quote do nothing
...@@ -1486,6 +1489,8 @@ int32_t tscValidateName(SStrToken* pToken) { ...@@ -1486,6 +1489,8 @@ int32_t tscValidateName(SStrToken* pToken) {
} }
pToken->n += (firstPartLen + sizeof(TS_PATH_DELIMITER[0])); pToken->n += (firstPartLen + sizeof(TS_PATH_DELIMITER[0]));
pToken->z = pStr; pToken->z = pStr;
tscStrToLower(pToken->z,pToken->n);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
......
src/common/inc/tglobal.h
浏览文件 @ 942c8814
...@@ -59,7 +59,8 @@ extern int32_t tsCompressMsgSize; ...@@ -59,7 +59,8 @@ extern int32_t tsCompressMsgSize;
extern char tsTempDir[]; extern char tsTempDir[];
//query buffer management //query buffer management
extern int32_t tsQueryBufferSize; // maximum allowed usage buffer for each data node during query processing extern int32_t tsQueryBufferSize; // maximum allowed usage buffer size in MB for each data node during query processing
extern int64_t tsQueryBufferSizeBytes; // maximum allowed usage buffer size in byte for each data node during query processing
extern int32_t tsRetrieveBlockingModel;// retrieve threads will be blocked extern int32_t tsRetrieveBlockingModel;// retrieve threads will be blocked
extern int8_t tsKeepOriginalColumnName; extern int8_t tsKeepOriginalColumnName;
......
src/common/src/tglobal.c
浏览文件 @ 942c8814
...@@ -105,6 +105,7 @@ int64_t tsMaxRetentWindow = 24 * 3600L; // maximum time window tolerance ...@@ -105,6 +105,7 @@ int64_t tsMaxRetentWindow = 24 * 3600L; // maximum time window tolerance
// 0 no query allowed, queries are disabled // 0 no query allowed, queries are disabled
// positive value (in MB) // positive value (in MB)
int32_t tsQueryBufferSize = -1; int32_t tsQueryBufferSize = -1;
int64_t tsQueryBufferSizeBytes = -1;
// in retrieve blocking model, the retrieve threads will wait for the completion of the query processing. // in retrieve blocking model, the retrieve threads will wait for the completion of the query processing.
int32_t tsRetrieveBlockingModel = 0; int32_t tsRetrieveBlockingModel = 0;
...@@ -291,7 +292,7 @@ bool taosCfgDynamicOptions(char *msg) { ...@@ -291,7 +292,7 @@ bool taosCfgDynamicOptions(char *msg) {
int32_t cfgLen = (int32_t)strlen(cfg->option); int32_t cfgLen = (int32_t)strlen(cfg->option);
if (cfgLen != olen) continue; if (cfgLen != olen) continue;
if (strncasecmp(option, cfg->option, olen) != 0) continue; if (strncasecmp(option, cfg->option, olen) != 0) continue;
if (cfg->valType != TAOS_CFG_VTYPE_INT32) { if (cfg->valType == TAOS_CFG_VTYPE_INT32) {
*((int32_t *)cfg->ptr) = vint; *((int32_t *)cfg->ptr) = vint;
} else { } else {
*((int8_t *)cfg->ptr) = (int8_t)vint; *((int8_t *)cfg->ptr) = (int8_t)vint;
...@@ -1530,6 +1531,10 @@ int32_t taosCheckGlobalCfg() { ...@@ -1530,6 +1531,10 @@ int32_t taosCheckGlobalCfg() {
tsSyncPort = tsServerPort + TSDB_PORT_SYNC; tsSyncPort = tsServerPort + TSDB_PORT_SYNC;
tsHttpPort = tsServerPort + TSDB_PORT_HTTP; tsHttpPort = tsServerPort + TSDB_PORT_HTTP;
if (tsQueryBufferSize >= 0) {
tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;
}
taosPrintGlobalCfg(); taosPrintGlobalCfg();
return 0; return 0;
......
src/common/src/ttypes.c
浏览文件 @ 942c8814
...@@ -632,7 +632,7 @@ int32_t tStrToInteger(const char* z, int16_t type, int32_t n, int64_t* value, bo ...@@ -632,7 +632,7 @@ int32_t tStrToInteger(const char* z, int16_t type, int32_t n, int64_t* value, bo
} }
// the string may be overflow according to errno // the string may be overflow according to errno
*value = issigned? strtoll(z, &endPtr, radix):strtoul(z, &endPtr, radix); *value = issigned? strtoll(z, &endPtr, radix):strtoull(z, &endPtr, radix);
// not a valid integer number, return error // not a valid integer number, return error
if (endPtr - z != n || errno == ERANGE) { if (endPtr - z != n || errno == ERANGE) {
......
src/cq/src/cqMain.c
浏览文件 @ 942c8814
...@@ -50,10 +50,13 @@ typedef struct { ...@@ -50,10 +50,13 @@ typedef struct {
void *dbConn; void *dbConn;
void *tmrCtrl; void *tmrCtrl;
pthread_mutex_t mutex; pthread_mutex_t mutex;
int32_t delete;
int32_t cqObjNum;
} SCqContext; } SCqContext;
typedef struct SCqObj { typedef struct SCqObj {
tmr_h tmrId; tmr_h tmrId;
int64_t rid;
uint64_t uid; uint64_t uid;
int32_t tid; // table ID int32_t tid; // table ID
int32_t rowSize; // bytes of a row int32_t rowSize; // bytes of a row
...@@ -69,6 +72,84 @@ typedef struct SCqObj { ...@@ -69,6 +72,84 @@ typedef struct SCqObj {
static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row); static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row);
static void cqCreateStream(SCqContext *pContext, SCqObj *pObj); static void cqCreateStream(SCqContext *pContext, SCqObj *pObj);
int32_t cqObjRef = -1;
void cqRmFromList(SCqObj *pObj) {
//LOCK in caller
SCqContext *pContext = pObj->pContext;
if (pObj->prev) {
pObj->prev->next = pObj->next;
} else {
pContext->pHead = pObj->next;
}
if (pObj->next) {
pObj->next->prev = pObj->prev;
}
}
void cqFree(void *handle) {
if (tsEnableStream == 0) {
return;
}
SCqObj *pObj = handle;
SCqContext *pContext = pObj->pContext;
int32_t delete = 0;
pthread_mutex_lock(&pContext->mutex);
// free the resources associated
if (pObj->pStream) {
taos_close_stream(pObj->pStream);
pObj->pStream = NULL;
} else {
taosTmrStop(pObj->tmrId);
pObj->tmrId = 0;
}
cInfo("vgId:%d, id:%d CQ:%s is dropped", pContext->vgId, pObj->tid, pObj->sqlStr);
tdFreeSchema(pObj->pSchema);
free(pObj->dstTable);
free(pObj->sqlStr);
free(pObj);
pContext->cqObjNum--;
if (pContext->cqObjNum <= 0 && pContext->delete) {
delete = 1;
}
pthread_mutex_unlock(&pContext->mutex);
if (delete) {
pthread_mutex_unlock(&pContext->mutex);
pthread_mutex_destroy(&pContext->mutex);
taosTmrCleanUp(pContext->tmrCtrl);
pContext->tmrCtrl = NULL;
cDebug("vgId:%d, CQ is closed", pContext->vgId);
free(pContext);
}
}
void cqCreateRef() {
int32_t ref = atomic_load_32(&cqObjRef);
if (ref == -1) {
ref = taosOpenRef(4096, cqFree);
if (atomic_val_compare_exchange_32(&cqObjRef, -1, ref) != -1) {
taosCloseRef(ref);
}
}
}
void *cqOpen(void *ahandle, const SCqCfg *pCfg) { void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
if (tsEnableStream == 0) { if (tsEnableStream == 0) {
return NULL; return NULL;
...@@ -79,6 +160,8 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) { ...@@ -79,6 +160,8 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
return NULL; return NULL;
} }
cqCreateRef();
pContext->tmrCtrl = taosTmrInit(0, 0, 0, "CQ"); pContext->tmrCtrl = taosTmrInit(0, 0, 0, "CQ");
tstrncpy(pContext->user, pCfg->user, sizeof(pContext->user)); tstrncpy(pContext->user, pCfg->user, sizeof(pContext->user));
...@@ -97,6 +180,7 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) { ...@@ -97,6 +180,7 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
pthread_mutex_init(&pContext->mutex, NULL); pthread_mutex_init(&pContext->mutex, NULL);
cDebug("vgId:%d, CQ is opened", pContext->vgId); cDebug("vgId:%d, CQ is opened", pContext->vgId);
return pContext; return pContext;
...@@ -109,30 +193,30 @@ void cqClose(void *handle) { ...@@ -109,30 +193,30 @@ void cqClose(void *handle) {
SCqContext *pContext = handle; SCqContext *pContext = handle;
if (handle == NULL) return; if (handle == NULL) return;
pContext->delete = 1;
// stop all CQs // stop all CQs
cqStop(pContext); cqStop(pContext);
// free all resources int64_t rid = 0;
pthread_mutex_lock(&pContext->mutex);
SCqObj *pObj = pContext->pHead;
while (pObj) {
SCqObj *pTemp = pObj;
pObj = pObj->next;
tdFreeSchema(pTemp->pSchema);
tfree(pTemp->sqlStr);
free(pTemp);
}
pthread_mutex_unlock(&pContext->mutex);
pthread_mutex_destroy(&pContext->mutex); while (1) {
pthread_mutex_lock(&pContext->mutex);
taosTmrCleanUp(pContext->tmrCtrl); SCqObj *pObj = pContext->pHead;
pContext->tmrCtrl = NULL; if (pObj) {
cqRmFromList(pObj);
cDebug("vgId:%d, CQ is closed", pContext->vgId); rid = pObj->rid;
free(pContext); } else {
pthread_mutex_unlock(&pContext->mutex);
break;
}
pthread_mutex_unlock(&pContext->mutex);
taosRemoveRef(cqObjRef, rid);
}
} }
void cqStart(void *handle) { void cqStart(void *handle) {
...@@ -191,7 +275,8 @@ void *cqCreate(void *handle, uint64_t uid, int32_t sid, const char* dstTable, ch ...@@ -191,7 +275,8 @@ void *cqCreate(void *handle, uint64_t uid, int32_t sid, const char* dstTable, ch
return NULL; return NULL;
} }
SCqContext *pContext = handle; SCqContext *pContext = handle;
int64_t rid = 0;
SCqObj *pObj = calloc(sizeof(SCqObj), 1); SCqObj *pObj = calloc(sizeof(SCqObj), 1);
if (pObj == NULL) return NULL; if (pObj == NULL) return NULL;
...@@ -213,32 +298,36 @@ void *cqCreate(void *handle, uint64_t uid, int32_t sid, const char* dstTable, ch ...@@ -213,32 +298,36 @@ void *cqCreate(void *handle, uint64_t uid, int32_t sid, const char* dstTable, ch
if (pContext->pHead) pContext->pHead->prev = pObj; if (pContext->pHead) pContext->pHead->prev = pObj;
pContext->pHead = pObj; pContext->pHead = pObj;
pContext->cqObjNum++;
pObj->rid = taosAddRef(cqObjRef, pObj);
cqCreateStream(pContext, pObj); cqCreateStream(pContext, pObj);
rid = pObj->rid;
pthread_mutex_unlock(&pContext->mutex); pthread_mutex_unlock(&pContext->mutex);
return pObj;
return (void *)rid;
} }
void cqDrop(void *handle) { void cqDrop(void *handle) {
if (tsEnableStream == 0) { if (tsEnableStream == 0) {
return; return;
} }
SCqObj *pObj = handle;
SCqContext *pContext = pObj->pContext;
pthread_mutex_lock(&pContext->mutex);
if (pObj->prev) { SCqObj* pObj = (SCqObj*)taosAcquireRef(cqObjRef, (int64_t)handle);
pObj->prev->next = pObj->next; if (pObj == NULL) {
} else { return;
pContext->pHead = pObj->next;
}
if (pObj->next) {
pObj->next->prev = pObj->prev;
} }
SCqContext *pContext = pObj->pContext;
pthread_mutex_lock(&pContext->mutex);
cqRmFromList(pObj);
// free the resources associated // free the resources associated
if (pObj->pStream) { if (pObj->pStream) {
taos_close_stream(pObj->pStream); taos_close_stream(pObj->pStream);
...@@ -248,17 +337,18 @@ void cqDrop(void *handle) { ...@@ -248,17 +337,18 @@ void cqDrop(void *handle) {
pObj->tmrId = 0; pObj->tmrId = 0;
} }
cInfo("vgId:%d, id:%d CQ:%s is dropped", pContext->vgId, pObj->tid, pObj->sqlStr);
tdFreeSchema(pObj->pSchema);
free(pObj->dstTable);
free(pObj->sqlStr);
free(pObj);
pthread_mutex_unlock(&pContext->mutex); pthread_mutex_unlock(&pContext->mutex);
taosRemoveRef(cqObjRef, (int64_t)handle);
taosReleaseRef(cqObjRef, (int64_t)handle);
} }
static void doCreateStream(void *param, TAOS_RES *result, int32_t code) { static void doCreateStream(void *param, TAOS_RES *result, int32_t code) {
SCqObj* pObj = (SCqObj*)param; SCqObj* pObj = (SCqObj*)taosAcquireRef(cqObjRef, (int64_t)param);
if (pObj == NULL) {
return;
}
SCqContext* pContext = pObj->pContext; SCqContext* pContext = pObj->pContext;
SSqlObj* pSql = (SSqlObj*)result; SSqlObj* pSql = (SSqlObj*)result;
if (atomic_val_compare_exchange_ptr(&(pContext->dbConn), NULL, pSql->pTscObj) != NULL) { if (atomic_val_compare_exchange_ptr(&(pContext->dbConn), NULL, pSql->pTscObj) != NULL) {
...@@ -267,10 +357,16 @@ static void doCreateStream(void *param, TAOS_RES *result, int32_t code) { ...@@ -267,10 +357,16 @@ static void doCreateStream(void *param, TAOS_RES *result, int32_t code) {
pthread_mutex_lock(&pContext->mutex); pthread_mutex_lock(&pContext->mutex);
cqCreateStream(pContext, pObj); cqCreateStream(pContext, pObj);
pthread_mutex_unlock(&pContext->mutex); pthread_mutex_unlock(&pContext->mutex);
taosReleaseRef(cqObjRef, (int64_t)param);
} }
static void cqProcessCreateTimer(void *param, void *tmrId) { static void cqProcessCreateTimer(void *param, void *tmrId) {
SCqObj* pObj = (SCqObj*)param; SCqObj* pObj = (SCqObj*)taosAcquireRef(cqObjRef, (int64_t)param);
if (pObj == NULL) {
return;
}
SCqContext* pContext = pObj->pContext; SCqContext* pContext = pObj->pContext;
if (pContext->dbConn == NULL) { if (pContext->dbConn == NULL) {
...@@ -281,6 +377,8 @@ static void cqProcessCreateTimer(void *param, void *tmrId) { ...@@ -281,6 +377,8 @@ static void cqProcessCreateTimer(void *param, void *tmrId) {
cqCreateStream(pContext, pObj); cqCreateStream(pContext, pObj);
pthread_mutex_unlock(&pContext->mutex); pthread_mutex_unlock(&pContext->mutex);
} }
taosReleaseRef(cqObjRef, (int64_t)param);
} }
static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) { static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
...@@ -288,13 +386,13 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) { ...@@ -288,13 +386,13 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
if (pContext->dbConn == NULL) { if (pContext->dbConn == NULL) {
cDebug("vgId:%d, create dbConn after 1000 ms", pContext->vgId); cDebug("vgId:%d, create dbConn after 1000 ms", pContext->vgId);
pObj->tmrId = taosTmrStart(cqProcessCreateTimer, 1000, pObj, pContext->tmrCtrl); pObj->tmrId = taosTmrStart(cqProcessCreateTimer, 1000, (void *)pObj->rid, pContext->tmrCtrl);
return; return;
} }
pObj->tmrId = 0; pObj->tmrId = 0;
if (pObj->pStream == NULL) { if (pObj->pStream == NULL) {
pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, 0, pObj, NULL); pObj->pStream = taos_open_stream(pContext->dbConn, pObj->sqlStr, cqProcessStreamRes, 0, (void *)pObj->rid, NULL);
// TODO the pObj->pStream may be released if error happens // TODO the pObj->pStream may be released if error happens
if (pObj->pStream) { if (pObj->pStream) {
...@@ -308,18 +406,28 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) { ...@@ -308,18 +406,28 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
} }
static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) { static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
SCqObj *pObj = (SCqObj *)param; SCqObj* pObj = (SCqObj*)taosAcquireRef(cqObjRef, (int64_t)param);
if (pObj == NULL) {
return;
}
if (tres == NULL && row == NULL) { if (tres == NULL && row == NULL) {
taos_close_stream(pObj->pStream); taos_close_stream(pObj->pStream);
pObj->pStream = NULL; pObj->pStream = NULL;
taosReleaseRef(cqObjRef, (int64_t)param);
return; return;
} }
SCqContext *pContext = pObj->pContext; SCqContext *pContext = pObj->pContext;
STSchema *pSchema = pObj->pSchema; STSchema *pSchema = pObj->pSchema;
if (pObj->pStream == NULL) return; if (pObj->pStream == NULL) {
taosReleaseRef(cqObjRef, (int64_t)param);
return;
}
cDebug("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr); cDebug("vgId:%d, id:%d CQ:%s stream result is ready", pContext->vgId, pObj->tid, pObj->sqlStr);
int32_t size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + TD_DATA_ROW_HEAD_SIZE + pObj->rowSize; int32_t size = sizeof(SWalHead) + sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + TD_DATA_ROW_HEAD_SIZE + pObj->rowSize;
...@@ -370,5 +478,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) { ...@@ -370,5 +478,7 @@ static void cqProcessStreamRes(void *param, TAOS_RES *tres, TAOS_ROW row) {
// write into vnode write queue // write into vnode write queue
pContext->cqWrite(pContext->vgId, pHead, TAOS_QTYPE_CQ, NULL); pContext->cqWrite(pContext->vgId, pHead, TAOS_QTYPE_CQ, NULL);
free(buffer); free(buffer);
taosReleaseRef(cqObjRef, (int64_t)param);
} }
src/dnode/src/dnodeCfg.c
浏览文件 @ 942c8814
...@@ -156,7 +156,7 @@ static int32_t dnodeWriteCfg() { ...@@ -156,7 +156,7 @@ static int32_t dnodeWriteCfg() {
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");
fwrite(content, 1, len, fp); fwrite(content, 1, len, fp);
fflush(fp); fsync(fileno(fp));
fclose(fp); fclose(fp);
free(content); free(content);
terrno = 0; terrno = 0;
......
src/dnode/src/dnodeEps.c
浏览文件 @ 942c8814
...@@ -277,7 +277,7 @@ static int32_t dnodeWriteEps() { ...@@ -277,7 +277,7 @@ static int32_t dnodeWriteEps() {
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");
fwrite(content, 1, len, fp); fwrite(content, 1, len, fp);
fflush(fp); fsync(fileno(fp));
fclose(fp); fclose(fp);
free(content); free(content);
terrno = 0; terrno = 0;
......
src/dnode/src/dnodeMInfos.c
浏览文件 @ 942c8814
...@@ -286,7 +286,7 @@ static int32_t dnodeWriteMInfos() { ...@@ -286,7 +286,7 @@ static int32_t dnodeWriteMInfos() {
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");
fwrite(content, 1, len, fp); fwrite(content, 1, len, fp);
fflush(fp); fsync(fileno(fp));
fclose(fp); fclose(fp);
free(content); free(content);
terrno = 0; terrno = 0;
......
src/mnode/src/mnodeUser.c
浏览文件 @ 942c8814
...@@ -123,7 +123,7 @@ static void mnodePrintUserAuth() { ...@@ -123,7 +123,7 @@ static void mnodePrintUserAuth() {
mnodeDecUserRef(pUser); mnodeDecUserRef(pUser);
} }
fflush(fp); fsync(fileno(fp));
fclose(fp); fclose(fp);
} }
......
src/query/src/qExecutor.c
浏览文件 @ 942c8814
...@@ -7730,15 +7730,15 @@ static int64_t getQuerySupportBufSize(size_t numOfTables) { ...@@ -7730,15 +7730,15 @@ static int64_t getQuerySupportBufSize(size_t numOfTables) {
int32_t checkForQueryBuf(size_t numOfTables) { int32_t checkForQueryBuf(size_t numOfTables) {
int64_t t = getQuerySupportBufSize(numOfTables); int64_t t = getQuerySupportBufSize(numOfTables);
if (tsQueryBufferSize < 0) { if (tsQueryBufferSizeBytes < 0) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (tsQueryBufferSize > 0) { } else if (tsQueryBufferSizeBytes > 0) {
while(1) { while(1) {
int64_t s = tsQueryBufferSize; int64_t s = tsQueryBufferSizeBytes;
int64_t remain = s - t; int64_t remain = s - t;
if (remain >= 0) { if (remain >= 0) {
if (atomic_val_compare_exchange_64(&tsQueryBufferSize, s, remain) == s) { if (atomic_val_compare_exchange_64(&tsQueryBufferSizeBytes, s, remain) == s) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
} else { } else {
...@@ -7752,14 +7752,14 @@ int32_t checkForQueryBuf(size_t numOfTables) { ...@@ -7752,14 +7752,14 @@ int32_t checkForQueryBuf(size_t numOfTables) {
} }
void releaseQueryBuf(size_t numOfTables) { void releaseQueryBuf(size_t numOfTables) {
if (tsQueryBufferSize <= 0) { if (tsQueryBufferSizeBytes < 0) {
return; return;
} }
int64_t t = getQuerySupportBufSize(numOfTables); int64_t t = getQuerySupportBufSize(numOfTables);
// restore value is not enough buffer available // restore value is not enough buffer available
atomic_add_fetch_64(&tsQueryBufferSize, t); atomic_add_fetch_64(&tsQueryBufferSizeBytes, t);
} }
void* qGetResultRetrieveMsg(qinfo_t qinfo) { void* qGetResultRetrieveMsg(qinfo_t qinfo) {
......
src/rpc/src/rpcTcp.c
浏览文件 @ 942c8814
...@@ -48,6 +48,13 @@ typedef struct SThreadObj { ...@@ -48,6 +48,13 @@ typedef struct SThreadObj {
void *(*processData)(SRecvInfo *pPacket); void *(*processData)(SRecvInfo *pPacket);
} SThreadObj; } SThreadObj;
typedef struct {
char label[TSDB_LABEL_LEN];
int32_t index;
int numOfThreads;
SThreadObj **pThreadObj;
} SClientObj;
typedef struct { typedef struct {
SOCKET fd; SOCKET fd;
uint32_t ip; uint32_t ip;
...@@ -116,6 +123,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread ...@@ -116,6 +123,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
pThreadObj->processData = fp; pThreadObj->processData = fp;
tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label)); tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
pThreadObj->shandle = shandle; pThreadObj->shandle = shandle;
pThreadObj->stop = false;
} }
// initialize mutex, thread, fd which may fail // initialize mutex, thread, fd which may fail
...@@ -166,6 +174,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread ...@@ -166,6 +174,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread
} }
static void taosStopTcpThread(SThreadObj* pThreadObj) { static void taosStopTcpThread(SThreadObj* pThreadObj) {
if (pThreadObj == NULL) { return;}
// save thread into local variable and signal thread to stop // save thread into local variable and signal thread to stop
pthread_t thread = pThreadObj->thread; pthread_t thread = pThreadObj->thread;
if (!taosCheckPthreadValid(thread)) { if (!taosCheckPthreadValid(thread)) {
...@@ -282,49 +291,76 @@ static void *taosAcceptTcpConnection(void *arg) { ...@@ -282,49 +291,76 @@ static void *taosAcceptTcpConnection(void *arg) {
return NULL; return NULL;
} }
void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *fp, void *shandle) { void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle) {
SThreadObj *pThreadObj; SClientObj *pClientObj = (SClientObj *)calloc(1, sizeof(SClientObj));
pthread_attr_t thattr; if (pClientObj == NULL) {
tError("TCP:%s no enough memory", label);
pThreadObj = (SThreadObj *)malloc(sizeof(SThreadObj));
memset(pThreadObj, 0, sizeof(SThreadObj));
tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
pThreadObj->ip = ip;
pThreadObj->shandle = shandle;
if (pthread_mutex_init(&(pThreadObj->mutex), NULL) < 0) {
tError("%s failed to init TCP client mutex(%s)", label, strerror(errno));
free(pThreadObj);
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
return NULL; return NULL;
} }
pThreadObj->pollFd = (EpollFd)epoll_create(10); // size does not matter tstrncpy(pClientObj->label, label, sizeof(pClientObj->label));
if (pThreadObj->pollFd < 0) { pClientObj->numOfThreads = numOfThreads;
tError("%s failed to create TCP client epoll", label); pClientObj->pThreadObj = (SThreadObj **)calloc(numOfThreads, sizeof(SThreadObj*));
free(pThreadObj); if (pClientObj->pThreadObj == NULL) {
tError("TCP:%s no enough memory", label);
tfree(pClientObj);
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
return NULL;
} }
pThreadObj->processData = fp; int code = 0;
pthread_attr_t thattr;
pthread_attr_init(&thattr); pthread_attr_init(&thattr);
pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE); pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
int code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
pthread_attr_destroy(&thattr); for (int i = 0; i < numOfThreads; ++i) {
if (code != 0) { SThreadObj *pThreadObj = (SThreadObj *)calloc(1, sizeof(SThreadObj));
EpollClose(pThreadObj->pollFd); if (pThreadObj == NULL) {
pThreadObj->pollFd = -1; tError("TCP:%s no enough memory", label);
free(pThreadObj); terrno = TAOS_SYSTEM_ERROR(errno);
terrno = TAOS_SYSTEM_ERROR(errno); for (int j=0; j<i; ++j) free(pClientObj->pThreadObj[j]);
tError("%s failed to create TCP read data thread(%s)", label, strerror(errno)); free(pClientObj);
return NULL; pthread_attr_destroy(&thattr);
return NULL;
}
pClientObj->pThreadObj[i] = pThreadObj;
taosResetPthread(&pThreadObj->thread);
pThreadObj->ip = ip;
pThreadObj->stop = false;
tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label));
pThreadObj->shandle = shandle;
pThreadObj->processData = fp;
} }
tDebug("%s TCP client is initialized, ip:%u:%hu", label, ip, port); // initialize mutex, thread, fd which may fail
for (int i = 0; i < numOfThreads; ++i) {
SThreadObj *pThreadObj = pClientObj->pThreadObj[i];
code = pthread_mutex_init(&(pThreadObj->mutex), NULL);
if (code < 0) {
tError("%s failed to init TCP process data mutex(%s)", label, strerror(errno));
break;
}
return pThreadObj; pThreadObj->pollFd = (int64_t)epoll_create(10); // size does not matter
if (pThreadObj->pollFd < 0) {
tError("%s failed to create TCP epoll", label);
code = -1;
break;
}
code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj));
if (code != 0) {
tError("%s failed to create TCP process data thread(%s)", label, strerror(errno));
break;
}
pThreadObj->threadId = i;
}
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosCleanUpTcpClient(pClientObj);
pClientObj = NULL;
}
return pClientObj;
} }
void taosStopTcpClient(void *chandle) { void taosStopTcpClient(void *chandle) {
...@@ -335,15 +371,23 @@ void taosStopTcpClient(void *chandle) { ...@@ -335,15 +371,23 @@ void taosStopTcpClient(void *chandle) {
} }
void taosCleanUpTcpClient(void *chandle) { void taosCleanUpTcpClient(void *chandle) {
SThreadObj *pThreadObj = chandle; SClientObj *pClientObj = chandle;
if (pThreadObj == NULL) return; if (pClientObj == NULL) return;
for (int i = 0; i < pClientObj->numOfThreads; ++i) {
tDebug ("%s TCP client will be cleaned up", pThreadObj->label); SThreadObj *pThreadObj= pClientObj->pThreadObj[i];
taosStopTcpThread(pThreadObj); taosStopTcpThread(pThreadObj);
}
tDebug("%s TCP client is cleaned up", pClientObj->label);
tfree(pClientObj->pThreadObj);
tfree(pClientObj);
} }
void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) { void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) {
SThreadObj * pThreadObj = shandle; SClientObj * pClientObj = shandle;
int32_t index = atomic_load_32(&pClientObj->index) % pClientObj->numOfThreads;
atomic_store_32(&pClientObj->index, index + 1);
SThreadObj *pThreadObj = pClientObj->pThreadObj[index];
SOCKET fd = taosOpenTcpClientSocket(ip, port, pThreadObj->ip); SOCKET fd = taosOpenTcpClientSocket(ip, port, pThreadObj->ip);
if (fd <= 0) return NULL; if (fd <= 0) return NULL;
......
src/vnode/src/vnodeCfg.c
浏览文件 @ 942c8814
...@@ -341,7 +341,7 @@ int32_t vnodeWriteCfg(SCreateVnodeMsg *pMsg) { ...@@ -341,7 +341,7 @@ int32_t vnodeWriteCfg(SCreateVnodeMsg *pMsg) {
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");
fwrite(content, 1, len, fp); fwrite(content, 1, len, fp);
fflush(fp); fsync(fileno(fp));
fclose(fp); fclose(fp);
free(content); free(content);
terrno = 0; terrno = 0;
......
src/vnode/src/vnodeVersion.c
浏览文件 @ 942c8814
...@@ -90,7 +90,7 @@ int32_t vnodeSaveVersion(SVnodeObj *pVnode) { ...@@ -90,7 +90,7 @@ int32_t vnodeSaveVersion(SVnodeObj *pVnode) {
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");
fwrite(content, 1, len, fp); fwrite(content, 1, len, fp);
fflush(fp); fsync(fileno(fp));
fclose(fp); fclose(fp);
free(content); free(content);
terrno = 0; terrno = 0;
......
tests/examples/JDBC/JDBCDemo/pom.xml
浏览文件 @ 942c8814
...@@ -50,6 +50,7 @@ ...@@ -50,6 +50,7 @@
<configuration> <configuration>
<source>8</source> <source>8</source>
<target>8</target> <target>8</target>
<encoding>UTF-8</encoding>
</configuration> </configuration>
</plugin> </plugin>
......
tests/examples/JDBC/SpringJdbcTemplate/pom.xml
浏览文件 @ 942c8814
...@@ -47,7 +47,7 @@ ...@@ -47,7 +47,7 @@
<dependency> <dependency>
<groupId>com.taosdata.jdbc</groupId> <groupId>com.taosdata.jdbc</groupId>
<artifactId>taos-jdbcdriver</artifactId> <artifactId>taos-jdbcdriver</artifactId>
<version>2.0.4</version> <version>2.0.18</version>
</dependency> </dependency>
</dependencies> </dependencies>
...@@ -69,7 +69,7 @@ ...@@ -69,7 +69,7 @@
<configuration> <configuration>
<archive> <archive>
<manifest> <manifest>
<mainClass>com.taosdata.jdbc.example.jdbcTemplate.App</mainClass> <mainClass>com.taosdata.example.jdbcTemplate.App</mainClass>
</manifest> </manifest>
</archive> </archive>
<descriptorRefs> <descriptorRefs>
......
tests/examples/JDBC/SpringJdbcTemplate/readme.md
浏览文件 @ 942c8814
...@@ -8,18 +8,16 @@ ...@@ -8,18 +8,16 @@
修改 `src/main/resources/applicationContext.xml` 文件中 TDengine 的配置信息： 修改 `src/main/resources/applicationContext.xml` 文件中 TDengine 的配置信息：
```xml ```xml
<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource"> <property name="driverClassName" value="com.taosdata.jdbc.TSDBDriver"></property>
<property name="driverClassName" value="com.taosdata.jdbc.TSDBDriver"></property> <property name="url" value="jdbc:TAOS://127.0.0.1:6030/log"></property>
<property name="url" value="jdbc:TAOS://127.0.0.1:6030/log"></property> <property name="username" value="root"></property>
<property name="username" value="root"></property> <property name="password" value="taosdata"></property>
<property name="password" value="taosdata"></property> </bean>
</bean>
<bean id = "jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate" >
<property name="dataSource" ref = "dataSource" ></property>
<bean id = "jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate" > </bean>
<property name="dataSource" ref = "dataSource" ></property>
</bean>
``` ```
### 打包运行 ### 打包运行
......
tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/jdbc/example/jdbcTemplate/App.java → tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/example/jdbcTemplate/App.java
浏览文件 @ 942c8814
package com.taosdata.jdbc.example.jdbcTemplate; package com.taosdata.example.jdbcTemplate;
import com.taosdata.jdbc.example.jdbcTemplate.dao.ExecuteAsStatement; import com.taosdata.example.jdbcTemplate.dao.ExecuteAsStatement;
import com.taosdata.jdbc.example.jdbcTemplate.dao.WeatherDao; import com.taosdata.example.jdbcTemplate.dao.WeatherDao;
import com.taosdata.jdbc.example.jdbcTemplate.domain.Weather; import com.taosdata.example.jdbcTemplate.domain.Weather;
import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext;
......
tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/jdbc/example/jdbcTemplate/dao/ExecuteAsStatement.java → tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/example/jdbcTemplate/dao/ExecuteAsStatement.java
浏览文件 @ 942c8814
package com.taosdata.jdbc.example.jdbcTemplate.dao; package com.taosdata.example.jdbcTemplate.dao;
public interface ExecuteAsStatement{ public interface ExecuteAsStatement{
......
tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/jdbc/example/jdbcTemplate/dao/impl/ExecuteAsStatementImpl.java → tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/example/jdbcTemplate/dao/ExecuteAsStatementImpl.java
浏览文件 @ 942c8814
package com.taosdata.jdbc.example.jdbcTemplate.dao.impl; package com.taosdata.example.jdbcTemplate.dao;
import com.taosdata.jdbc.example.jdbcTemplate.dao.ExecuteAsStatement;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository; import org.springframework.stereotype.Repository;
......
tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/jdbc/example/jdbcTemplate/dao/WeatherDao.java → tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/example/jdbcTemplate/dao/WeatherDao.java
浏览文件 @ 942c8814
package com.taosdata.jdbc.example.jdbcTemplate.dao; package com.taosdata.example.jdbcTemplate.dao;
import com.taosdata.jdbc.example.jdbcTemplate.domain.Weather; import com.taosdata.example.jdbcTemplate.domain.Weather;
import java.util.List; import java.util.List;
......
tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/jdbc/example/jdbcTemplate/dao/impl/WeatherDaoImpl.java → tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/example/jdbcTemplate/dao/WeatherDaoImpl.java
浏览文件 @ 942c8814
package com.taosdata.jdbc.example.jdbcTemplate.dao.impl; package com.taosdata.example.jdbcTemplate.dao;
import com.taosdata.jdbc.example.jdbcTemplate.dao.WeatherDao; import com.taosdata.example.jdbcTemplate.domain.Weather;
import com.taosdata.jdbc.example.jdbcTemplate.domain.Weather; import com.taosdata.example.jdbcTemplate.dao.WeatherDao;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.BatchPreparedStatementSetter; import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.SqlParameterSourceUtils;
import org.springframework.jdbc.core.simple.SimpleJdbcInsert;
import org.springframework.stereotype.Repository; import org.springframework.stereotype.Repository;
import java.sql.PreparedStatement; import java.sql.PreparedStatement;
import java.sql.SQLException; import java.sql.SQLException;
import java.sql.Timestamp; import java.sql.Timestamp;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map;
@Repository @Repository
public class WeatherDaoImpl implements WeatherDao { public class WeatherDaoImpl implements WeatherDao {
......
tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/jdbc/example/jdbcTemplate/domain/Weather.java → tests/examples/JDBC/SpringJdbcTemplate/src/main/java/com/taosdata/example/jdbcTemplate/domain/Weather.java
浏览文件 @ 942c8814
package com.taosdata.jdbc.example.jdbcTemplate.domain; package com.taosdata.example.jdbcTemplate.domain;
import java.sql.Timestamp; import java.sql.Timestamp;
......
tests/examples/JDBC/SpringJdbcTemplate/src/main/resources/applicationContext.xml
浏览文件 @ 942c8814
...@@ -10,7 +10,7 @@ ...@@ -10,7 +10,7 @@
<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource"> <bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
<property name="driverClassName" value="com.taosdata.jdbc.TSDBDriver"></property> <property name="driverClassName" value="com.taosdata.jdbc.TSDBDriver"></property>
<property name="url" value="jdbc:TAOS://192.168.236.137:6030/"></property> <property name="url" value="jdbc:TAOS://127.0.0.1:6030/"></property>
<property name="username" value="root"></property> <property name="username" value="root"></property>
<property name="password" value="taosdata"></property> <property name="password" value="taosdata"></property>
</bean> </bean>
...@@ -20,6 +20,6 @@ ...@@ -20,6 +20,6 @@
<property name="dataSource" ref="dataSource"></property> <property name="dataSource" ref="dataSource"></property>
</bean> </bean>
<context:component-scan base-package="com.taosdata.jdbc.example.jdbcTemplate"/> <context:component-scan base-package="com.taosdata.example.jdbcTemplate"/>
</beans> </beans>
tests/examples/JDBC/SpringJdbcTemplate/src/test/java/com/taosdata/jdbc/example/jdbcTemplate/BatcherInsertTest.java → tests/examples/JDBC/SpringJdbcTemplate/src/test/java/com/taosdata/example/jdbcTemplate/BatcherInsertTest.java
浏览文件 @ 942c8814
package com.taosdata.jdbc.example.jdbcTemplate; package com.taosdata.example.jdbcTemplate;
import com.taosdata.jdbc.example.jdbcTemplate.dao.ExecuteAsStatement; import com.taosdata.example.jdbcTemplate.dao.ExecuteAsStatement;
import com.taosdata.jdbc.example.jdbcTemplate.dao.WeatherDao; import com.taosdata.example.jdbcTemplate.dao.WeatherDao;
import com.taosdata.jdbc.example.jdbcTemplate.domain.Weather; import com.taosdata.example.jdbcTemplate.domain.Weather;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith; import org.junit.runner.RunWith;
......
tests/examples/JDBC/SpringJdbcTemplate/src/test/java/com/taosdata/jdbc/AppTest.java 已删除 100644 → 0
浏览文件 @ 0948fdca
package com.taosdata.jdbc;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
/**
* Unit test for simple App.
*/
public class AppTest {
/**
* Rigorous Test :-)
*/
@Test
public void shouldAnswerWithTrue() {
assertTrue(true);
}
}
tests/examples/JDBC/readme.md 0 → 100644
浏览文件 @ 942c8814
# TDengine examples
| No. | Name | Describe |
| :--: | :----------------: | ------------------------------------------------------------ |
| 1 | JDBCDemo | Example codes for JDBC-JNI, JDBC-RESTful, Subscribe |
| 2 | connectionPools | Example codes for HikariCP, Druid, dbcp, c3p0 connection pools |
| 3 | SpringJdbcTemplate | Example codes for spring jdbcTemplate |
| 4 | mybatisplus-demo | Example codes for mybatis |
| 5 | springbootdemo | Example codes for springboot |
| 6 | taosdemo | This is an internal tool for testing Our JDBC-JNI, JDBC-RESTful, RESTful interfaces |
more detail: https://www.taosdata.com/cn//documentation20/connector-java/
\ No newline at end of file
tests/script/general/parser/dbtbnameValidate.sim
浏览文件 @ 942c8814
...@@ -26,11 +26,11 @@ sql use ' XYZ ' ...@@ -26,11 +26,11 @@ sql use ' XYZ '
sql drop database 'abc123' sql drop database 'abc123'
sql drop database '_ab1234' sql drop database '_ab1234'
sql drop database 'ABC123' sql_error drop database 'ABC123'
sql drop database '_ABC123' sql drop database '_ABC123'
sql drop database 'aABb123' sql drop database 'aABb123'
sql drop database ' xyz ' sql drop database ' xyz '
sql drop database ' XYZ ' sql_error drop database ' XYZ '
sql use abc sql use abc
...@@ -67,9 +67,9 @@ sql describe mt ...@@ -67,9 +67,9 @@ sql describe mt
sql describe sub_001 sql describe sub_001
sql describe sub_dy_tbl sql describe sub_dy_tbl
sql_error describe Dd sql describe Dd
sql_error describe FF sql describe FF
sql_error describe gG sql describe gG
sql drop table abc.cc sql drop table abc.cc
sql drop table 'abc.Dd' sql drop table 'abc.Dd'
...@@ -119,4 +119,4 @@ if $rows != 4 then ...@@ -119,4 +119,4 @@ if $rows != 4 then
return -1 return -1
endi endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file
tests/test-all.sh
浏览文件 @ 942c8814
...@@ -155,7 +155,7 @@ if [ "$2" != "python" ]; then ...@@ -155,7 +155,7 @@ if [ "$2" != "python" ]; then
elif [ "$1" == "b1" ]; then elif [ "$1" == "b1" ]; then
echo "### run TSIM b1 test ###" echo "### run TSIM b1 test ###"
runSimCaseOneByOne jenkins/basic_1.txt runSimCaseOneByOne jenkins/basic_1.txt
# runSimCaseOneByOne jenkins/basic_4.txt runSimCaseOneByOne jenkins/basic_4.txt
elif [ "$1" == "b2" ]; then elif [ "$1" == "b2" ]; then
echo "### run TSIM b2 test ###" echo "### run TSIM b2 test ###"
runSimCaseOneByOne jenkins/basic_2.txt runSimCaseOneByOne jenkins/basic_2.txt
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或