Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1193
Star 22018
Fork 4786
T
TDengine
提交 51259460 编写于 作者: wmmhello's avatar wmmhello
浏览文件
操作

merge from master

上级 c0a989fe 7fd66c21
展开全部 隐藏空白更改
内联 并排
Showing with 4548 addition and 3462 deletion
.gitignore
浏览文件 @ 51259460
......@@ -81,6 +81,10 @@ tests/comparisonTest/opentsdb/opentsdbtest/.settings/
tests/examples/JDBC/JDBCDemo/.classpath
tests/examples/JDBC/JDBCDemo/.project
tests/examples/JDBC/JDBCDemo/.settings/
tests/script/api/batchprepare
tests/script/api/stmt
tests/script/api/stmtBatchTest
tests/script/api/stmtTest
# Emacs
# -*- mode: gitignore; -*-
......
Jenkinsfile
浏览文件 @ 51259460
......@@ -111,6 +111,77 @@ def pre_test(){
'''
return 1
}
def pre_test_noinstall(){
sh'hostname'
sh'''
cd ${WKC}
git reset --hard HEAD~10 >/dev/null
'''
script {
if (env.CHANGE_TARGET == 'master') {
sh '''
cd ${WKC}
git checkout master
'''
}
else if(env.CHANGE_TARGET == '2.0'){
sh '''
cd ${WKC}
git checkout 2.0
'''
}
else{
sh '''
cd ${WKC}
git checkout develop
'''
}
}
sh'''
cd ${WKC}
git pull >/dev/null
git fetch origin +refs/pull/${CHANGE_ID}/merge
git checkout -qf FETCH_HEAD
git clean -dfx
git submodule update --init --recursive
cd ${WK}
git reset --hard HEAD~10
'''
script {
if (env.CHANGE_TARGET == 'master') {
sh '''
cd ${WK}
git checkout master
'''
}
else if(env.CHANGE_TARGET == '2.0'){
sh '''
cd ${WK}
git checkout 2.0
'''
}
else{
sh '''
cd ${WK}
git checkout develop
'''
}
}
sh '''
cd ${WK}
git pull >/dev/null
export TZ=Asia/Harbin
date
git clean -dfx
mkdir debug
cd debug
cmake .. > /dev/null
make
'''
return 1
}
def pre_test_win(){
bat '''
taskkill /f /t /im python.exe
......@@ -179,9 +250,9 @@ def pre_test_win(){
git clean -dfx
mkdir debug
cd debug
call "C:\\Program Files (x86)\\Microsoft Visual Studio 14.0\\VC\\vcvarsall.bat" amd64
call "C:\\Program Files (x86)\\Microsoft Visual Studio\\2017\\Community\\VC\\Auxiliary\\Build\\vcvarsall.bat" amd64
cmake ../ -G "NMake Makefiles"
nmake || exit 8
set CL=/MP nmake nmake || exit 8
nmake install || exit 8
xcopy /e/y/i/f C:\\workspace\\TDinternal\\debug\\build\\lib\\taos.dll C:\\Windows\\System32 || exit 8
cd C:\\workspace\\TDinternal\\community\\src\\connector\\python
......@@ -290,7 +361,7 @@ pipeline {
stage('python_3_s6') {
agent{label " slave6 || slave16 "}
steps {
timeout(time: 55, unit: 'MINUTES'){
timeout(time: 65, unit: 'MINUTES'){
pre_test()
sh '''
date
......@@ -396,7 +467,7 @@ pipeline {
stage('test_b4_s7') {
agent{label " slave7 || slave17 "}
steps {
timeout(time: 55, unit: 'MINUTES'){
timeout(time: 105, unit: 'MINUTES'){
pre_test()
sh '''
date
......@@ -428,7 +499,7 @@ pipeline {
stage('test_b6_s9') {
agent{label " slave9 || slave19 "}
steps {
timeout(time: 55, unit: 'MINUTES'){
timeout(time: 105, unit: 'MINUTES'){
pre_test()
sh '''
date
......@@ -451,36 +522,90 @@ pipeline {
}
}
}
stage('arm64centos7') {
agent{label " arm64centos7 "}
steps {
pre_test_noinstall()
}
}
stage('arm64centos8') {
agent{label " arm64centos8 "}
steps {
pre_test_noinstall()
}
}
stage('arm32bionic') {
agent{label " arm32bionic "}
steps {
pre_test_noinstall()
}
}
stage('arm64bionic') {
agent{label " arm64bionic "}
steps {
pre_test_noinstall()
}
}
stage('arm64focal') {
agent{label " arm64focal "}
steps {
pre_test_noinstall()
}
}
stage('centos7') {
agent{label " centos7 "}
steps {
pre_test_noinstall()
}
}
stage('ubuntu:trusty') {
agent{label " trusty "}
steps {
pre_test_noinstall()
}
}
stage('ubuntu:xenial') {
agent{label " xenial "}
steps {
pre_test_noinstall()
}
}
stage('ubuntu:bionic') {
agent{label " bionic "}
steps {
pre_test_noinstall()
}
}
// stage('build'){
// agent{label " wintest "}
// steps {
// pre_test()
// script{
// while(win_stop == 0){
// sleep(1)
// }
// }
// }
// }
// stage('test'){
// agent{label "win"}
// steps{
stage('build'){
agent{label " wintest "}
steps {
pre_test()
script{
while(win_stop == 0){
sleep(1)
}
}
}
}
stage('test'){
agent{label "win"}
steps{
// catchError(buildResult: 'FAILURE', stageResult: 'FAILURE') {
// pre_test_win()
// timeout(time: 20, unit: 'MINUTES'){
// bat'''
// cd C:\\workspace\\TDinternal\\community\\tests\\pytest
// .\\test-all.bat Wintest
// '''
// }
// }
// script{
// win_stop=1
// }
// }
// }
catchError(buildResult: 'FAILURE', stageResult: 'FAILURE') {
pre_test_win()
timeout(time: 20, unit: 'MINUTES'){
bat'''
cd C:\\workspace\\TDinternal\\community\\tests\\pytest
.\\test-all.bat wintest
'''
}
}
script{
win_stop=1
}
}
}
}
......
cmake/define.inc
浏览文件 @ 51259460
......@@ -117,7 +117,7 @@ IF (TD_MIPS_32)
SET(COMMON_FLAGS "-Wall -Werror -fPIC -D_FILE_OFFSET_BITS=64 -D_LARGE_FILE")
ENDIF ()
IF (TD_APLHINE)
IF (TD_ALPINE)
SET(COMMON_FLAGS "${COMMON_FLAGS} -largp")
link_libraries(/usr/lib/libargp.a)
ADD_DEFINITIONS(-D_ALPINE)
......@@ -136,7 +136,12 @@ IF (TD_LINUX)
ENDIF ()
IF (TD_MEMORY_SANITIZER)
SET(DEBUG_FLAGS "-fsanitize=address -fsanitize=undefined -fno-sanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -static-libasan -O0 -g3 -DDEBUG")
IF (TD_ARCHLINUX)
SET(DEBUG_FLAGS "-fsanitize=address -fsanitize=undefined -fno-sanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -O0 -g3 -DDEBUG")
ELSE ()
SET(DEBUG_FLAGS "-fsanitize=address -fsanitize=undefined -fno-sanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -static-libasan -O0 -g3 -DDEBUG")
ENDIF ()
MESSAGE(STATUS "${BoldRed}Will compile with memory sanitizer! ${ColourReset}")
ELSE ()
SET(DEBUG_FLAGS "-O0 -g3 -DDEBUG")
ENDIF ()
......
cmake/install.inc
浏览文件 @ 51259460
IF (TD_LINUX)
SET(TD_MAKE_INSTALL_SH "${TD_COMMUNITY_DIR}/packaging/tools/make_install.sh")
INSTALL(CODE "MESSAGE(\"make install script: ${TD_MAKE_INSTALL_SH}\")")
INSTALL(CODE "execute_process(COMMAND chmod 777 ${TD_MAKE_INSTALL_SH})")
INSTALL(CODE "execute_process(COMMAND ${TD_MAKE_INSTALL_SH} ${TD_COMMUNITY_DIR} ${PROJECT_BINARY_DIR} Linux ${TD_VER_NUMBER})")
INSTALL(CODE "execute_process(COMMAND bash ${TD_MAKE_INSTALL_SH} ${TD_COMMUNITY_DIR} ${PROJECT_BINARY_DIR} Linux ${TD_VER_NUMBER})")
ELSEIF (TD_WINDOWS)
IF (TD_POWER)
SET(CMAKE_INSTALL_PREFIX C:/PowerDB)
......@@ -41,6 +40,5 @@ ELSEIF (TD_WINDOWS)
ELSEIF (TD_DARWIN)
SET(TD_MAKE_INSTALL_SH "${TD_COMMUNITY_DIR}/packaging/tools/make_install.sh")
INSTALL(CODE "MESSAGE(\"make install script: ${TD_MAKE_INSTALL_SH}\")")
INSTALL(CODE "execute_process(COMMAND chmod 777 ${TD_MAKE_INSTALL_SH})")
INSTALL(CODE "execute_process(COMMAND ${TD_MAKE_INSTALL_SH} ${TD_COMMUNITY_DIR} ${PROJECT_BINARY_DIR} Darwin ${TD_VER_NUMBER})")
INSTALL(CODE "execute_process(COMMAND bash ${TD_MAKE_INSTALL_SH} ${TD_COMMUNITY_DIR} ${PROJECT_BINARY_DIR} Darwin ${TD_VER_NUMBER})")
ENDIF ()
cmake/platform.inc
浏览文件 @ 51259460
......@@ -21,7 +21,7 @@ SET(TD_LINUX FALSE)
SET(TD_ARM_32 FALSE)
SET(TD_MIPS_64 FALSE)
SET(TD_MIPS_32 FALSE)
SET(TD_APLHINE FALSE)
SET(TD_ALPINE FALSE)
SET(TD_NINGSI FALSE)
SET(TD_NINGSI_60 FALSE)
SET(TD_NINGSI_80 FALSE)
......@@ -36,7 +36,7 @@ IF (${CMAKE_SYSTEM_NAME} MATCHES "Linux")
# Get OS information and store in variable TD_OS_INFO.
#
execute_process(COMMAND chmod 777 ${TD_COMMUNITY_DIR}/packaging/tools/get_os.sh)
execute_process(COMMAND ${TD_COMMUNITY_DIR}/packaging/tools/get_os.sh "" OUTPUT_VARIABLE TD_OS_INFO)
execute_process(COMMAND sh ${TD_COMMUNITY_DIR}/packaging/tools/get_os.sh "" OUTPUT_VARIABLE TD_OS_INFO)
MESSAGE(STATUS "The current os is " ${TD_OS_INFO})
SET(TD_LINUX TRUE)
......@@ -52,8 +52,13 @@ IF (${CMAKE_SYSTEM_NAME} MATCHES "Linux")
ENDIF ()
IF (${TD_OS_INFO} MATCHES "Alpine")
SET(TD_APLHINE TRUE)
MESSAGE(STATUS "The current OS is Alpine, append extra flags")
SET(TD_ALPINE TRUE)
MESSAGE(STATUS "The current OS is Alpine Linux, append extra flags")
ELSEIF (${TD_OS_INFO} MATCHES "Arch")
SET(TD_ARCHLINUX TRUE)
MESSAGE(STATUS "The current OS is Arch Linux")
ELSE ()
MESSAGE(STATUS "Ths distro is " ${TD_OS_INFO})
ENDIF()
ELSEIF (${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
SET(TD_DARWIN TRUE)
......@@ -155,7 +160,7 @@ ELSEIF (${OSTYPE} MATCHES "Linux")
MESSAGE(STATUS "input osType: Linux")
ELSEIF (${OSTYPE} MATCHES "Alpine")
MESSAGE(STATUS "input osType: Alpine")
SET(TD_APLHINE TRUE)
SET(TD_ALPINE TRUE)
ELSE ()
MESSAGE(STATUS "The user specified osType is unknown: " ${OSTYPE})
ENDIF ()
cmake/version.inc
浏览文件 @ 51259460
......@@ -4,7 +4,7 @@ PROJECT(TDengine)
IF (DEFINED VERNUMBER)
SET(TD_VER_NUMBER ${VERNUMBER})
ELSE ()
SET(TD_VER_NUMBER "2.2.1.1")
SET(TD_VER_NUMBER "2.2.1.3")
ENDIF ()
IF (DEFINED VERCOMPATIBLE)
......
documentation20/cn/09.connections/docs.md
浏览文件 @ 51259460
......@@ -3,7 +3,7 @@
## <a class="anchor" id="grafana"></a>Grafana
TDengine能够与开源数据可视化系统[Grafana](https://www.grafana.com/)快速集成搭建数据监测报警系统，整个过程无需任何代码开发，TDengine中数据表中内容可以在仪表盘(DashBoard)上进行可视化展现。
TDengine 能够与开源数据可视化系统 [Grafana](https://www.grafana.com/)快速集成搭建数据监测报警系统，整个过程无需任何代码开发，TDengine 中数据表中内容可以在仪表盘(DashBoard)上进行可视化展现。关于TDengine插件的使用您可以在[GitHub](https://github.com/taosdata/grafanaplugin/blob/master/README.md)中了解更多。
### 安装Grafana
......@@ -11,12 +11,24 @@ TDengine能够与开源数据可视化系统[Grafana](https://www.grafana.com/)
### 配置Grafana
TDengine的Grafana插件在安装包的/usr/local/taos/connector/grafanaplugin目录下。
TDengine 的 Grafana 插件请从 <https://github.com/taosdata/grafanaplugin/releases/latest> 下载。
以CentOS 7.2操作系统为例，将grafanaplugin目录拷贝到/var/lib/grafana/plugins目录下，重新启动grafana即可。
```bash
GF_VERSION=3.1.1
wget https://github.com/taosdata/grafanaplugin/releases/download/v$GF_VERSION/tdengine-datasource-$GF_VERSION.zip
```
以 CentOS 7.2 操作系统为例，将插件包解压到 /var/lib/grafana/plugins 目录下，重新启动 grafana 即可。
```bash
sudo cp -rf /usr/local/taos/connector/grafanaplugin /var/lib/grafana/plugins/tdengine
sudo unzip tdengine-datasource-$GF_VERSION.zip -d /var/lib/grafana/plugins/
```
Grafana 7.3+ / 8.x 版本会对插件进行签名检查，因此还需要在 grafana.ini 文件中增加如下行，才能正确使用插件：
```ini
[plugins]
allow_loading_unsigned_plugins = tdengine-datasource
```
### 使用 Grafana
......@@ -55,7 +67,6 @@ sudo cp -rf /usr/local/taos/connector/grafanaplugin /var/lib/grafana/plugins/tde
* ALIAS BY：可设置当前查询别名。
* GENERATE SQL： 点击该按钮会自动替换相应变量，并生成最终执行的语句。
按照默认提示查询当前 TDengine 部署所在服务器指定间隔系统内存平均使用量如下：
![img](page://images/connections/create_dashboard2.jpg)
......@@ -64,16 +75,15 @@ sudo cp -rf /usr/local/taos/connector/grafanaplugin /var/lib/grafana/plugins/tde
#### 导入 Dashboard
在 Grafana 插件目录 /usr/local/taos/connector/grafana/tdengine/dashboard/ 下提供了一个 `tdengine-grafana.json` 可导入的 dashboard。
我们提供一个 TDengine Dashboard 可以作为 TDengine 集群的监控可视化工具使用，见 [Grafana Dashboard 15146](https://grafana.com/grafana/dashboards/15146)。
点击左侧 `Import` 按钮，并上传 `tdengine-grafana.json` 文件：
点击左侧 `Import` 按钮，选择 **Grafana.com Dashboard**,j将id `15146` 填入并加载：
![img](page://images/connections/import_dashboard1.jpg)
导入完成之后可看到如下效果：
![img](page://images/connections/import_dashboard2.jpg)
![img](../images/connections/dashboard-15146.png)
## <a class="anchor" id="matlab"></a>MATLAB
......
documentation20/cn/14.devops/01.telegraf/docs.md 0 → 100644
浏览文件 @ 51259460
# 使用 TDengine + Telegraf + Grafana 快速搭建 IT 运维展示系统
## 背景介绍
TDengine是涛思数据专为物联网、车联网、工业互联网、IT运维等设计和优化的大数据平台。自从 2019年 7 月开源以来，凭借创新的数据建模设计、快捷的安装方式、易用的编程接口和强大的数据写入查询性能博得了大量时序数据开发者的青睐。
IT 运维监测数据通常都是对时间特性比较敏感的数据，例如：
- 系统资源指标：CPU、内存、IO、带宽等。
- 软件系统指标：存活状态、连接数目、请求数目、超时数目、错误数目、响应时间、服务类型及其他与业务有关的指标。
当前主流的 IT 运维系统通常包含一个数据采集模块，一个数据存储模块，和一个可视化显示模块。Telegraf 和 Grafana 分别是当前最流行的数据采集模块和可视化显示模块之一。而数据存储模块可供选择的软件比较多，其中 OpenTSDB 或 InfluxDB 比较流行。而 TDengine 作为新兴的时序大数据平台，具备极强的高性能、高可靠、易管理、易维护的优势。
本文介绍不需要写一行代码，通过简单修改几行配置文件，就可以快速搭建一个基于 TDengine + Telegraf + Grafana 的 IT 运维系统。架构如下图：
![IT-DevOps-Solutions-Telegraf.png](../../images/IT-DevOps-Solutions-Telegraf.png)
## 安装步骤
### 安装 Telegraf，Grafana 和 TDengine
安装 Telegraf、Grafana 和 TDengine 请参考相关官方文档。
### Telegraf
请参考[官方文档](https://portal.influxdata.com/downloads/)。
### Grafana
请参考[官方文档](https://grafana.com/grafana/download)。
### 安装 TDengine
从涛思数据官网[下载](http://taosdata.com/cn/all-downloads/）页面下载最新 TDengine-server 2.3.0.0 或以上版本安装。
## 数据链路设置
### 下载 TDengine 插件到 grafana 插件目录
```bash
1. wget -c https://github.com/taosdata/grafanaplugin/releases/download/v3.1.1/tdengine-datasource-3.1.1.zip
2. sudo unzip tdengine-datasource-3.1.1.zip -d /var/lib/grafana/plugins/
3. sudo chown grafana:grafana -R /var/lib/grafana/plugins/tdengine
4. echo -e "[plugins]\nallow_loading_unsigned_plugins = tdengine-datasource\n" | sudo tee -a /etc/grafana/grafana.ini
5. sudo systemctl restart grafana-server.service
```
### 修改 /etc/telegraf/telegraf.conf
配置方法，在 /etc/telegraf/telegraf.conf 增加如下文字，其中 database name 请填写希望在 TDengine 保存 Telegraf 数据的数据库名，TDengine server/cluster host、username和 password 填写 TDengine 实际值：
```
[[outputs.http]]
url = "http://<TDengine server/cluster host>:6041/influxdb/v1/write?db=<database name>"
method = "POST"
timeout = "5s"
username = "<TDengine's username>"
password = "<TDengine's password>"
data_format = "influx"
influx_max_line_bytes = 250
```
然后重启 telegraf：
```
sudo systemctl start telegraf
```
### 导入 Dashboard
使用 Web 浏览器访问 IP:3000 登录 Grafana 界面，系统初始用户名密码为 admin/admin。
点击左侧齿轮图标并选择 Plugins，应该可以找到 TDengine data source 插件图标。
点击左侧加号图标并选择 Import，从 https://github.com/taosdata/grafanaplugin/blob/master/examples/telegraf/grafana/dashboards/telegraf-dashboard-v0.1.0.json 下载 dashboard JSON 文件后导入。之后可以看到如下界面的仪表盘：
![IT-DevOps-Solutions-telegraf-dashboard.png](../../images/IT-DevOps-Solutions-telegraf-dashboard.png)
## 总结
以上演示如何快速搭建一个完整的 IT 运维展示系统。得力于 TDengine 2.3.0.0 版本中新增的 schemaless 协议解析功能，以及强大的生态软件适配能力，用户可以短短数分钟就可以搭建一个高效易用的 IT 运维系统。TDengine 强大的数据写入查询性能和其他丰富功能请参考官方文档和产品落地案例。
documentation20/cn/14.devops/02.collectd/docs.md 0 → 100644
浏览文件 @ 51259460
# 使用 TDengine + collectd/StatsD + Grafana 快速搭建 IT 运维监控系统
## 背景介绍
TDengine是涛思数据专为物联网、车联网、工业互联网、IT运维等设计和优化的大数据平台。自从 2019年 7 月开源以来，凭借创新的数据建模设计、快捷的安装方式、易用的编程接口和强大的数据写入查询性能博得了大量时序数据开发者的青睐。
IT 运维监测数据通常都是对时间特性比较敏感的数据，例如：
- 系统资源指标：CPU、内存、IO、带宽等。
- 软件系统指标：存活状态、连接数目、请求数目、超时数目、错误数目、响应时间、服务类型及其他与业务有关的指标。
当前主流的 IT 运维系统通常包含一个数据采集模块，一个数据存储模块，和一个可视化显示模块。collectd / statsD 作为老牌开源数据采集工具，具有广泛的用户群。但是 collectd / StatsD 自身功能有限，往往需要配合 Telegraf、Grafana 以及时序数据库组合搭建成为完整的监控系统。而 TDengine 新版本支持多种数据协议接入，可以直接接受 collectd 和 statsD 的数据写入，并提供 Grafana dashboard 进行图形化展示。
本文介绍不需要写一行代码，通过简单修改几行配置文件，就可以快速搭建一个基于 TDengine + collectd / statsD + Grafana 的 IT 运维系统。架构如下图：
![IT-DevOps-Solutions-Collectd-StatsD.png](../../images/IT-DevOps-Solutions-Collectd-StatsD.png)
## 安装步骤
安装 collectd， StatsD， Grafana 和 TDengine 请参考相关官方文档。
### 安装 collectd
请参考[官方文档](https://collectd.org/documentation.shtml)。
### 安装 StatsD
请参考[官方文档](https://github.com/statsd/statsd)。
### 安装 Grafana
请参考[官方文档](https://grafana.com/grafana/download)。
### 安装 TDengine
从涛思数据官网[下载](http://taosdata.com/cn/all-downloads/）页面下载最新 TDengine-server 2.3.0.0 或以上版本安装。
## 数据链路设置
### 复制 TDengine 插件到 grafana 插件目录
```bash
1. wget -c https://github.com/taosdata/grafanaplugin/releases/download/v3.1.1/tdengine-datasource-3.1.1.zip
2. sudo unzip tdengine-datasource-3.1.1.zip -d /var/lib/grafana/plugins/
3. sudo chown grafana:grafana -R /var/lib/grafana/plugins/tdengine
4. echo -e "[plugins]\nallow_loading_unsigned_plugins = tdengine-datasource\n" | sudo tee -a /etc/grafana/grafana.ini
5. sudo systemctl restart grafana-server.service
```
### 配置 collectd
在 /etc/collectd/collectd.conf 文件中增加如下内容，其中 host 和 port 请填写 TDengine 和 BLM3 配置的实际值：
```
LoadPlugin network
<Plugin network>
Server "<TDengine cluster/server host>" "<port for collectd>"
</Plugin>
sudo systemctl start collectd
```
### 配置 StatsD
在 config.js 文件中增加如下内容后启动 StatsD，其中 host 和 port 请填写 TDengine 和 BLM3 配置的实际值：
```
backends 部分添加 "./backends/repeater"
repeater 部分添加 { host:'<TDengine server/cluster host>', port: <port for StatsD>}
```
### 导入 Dashboard
使用 Web 浏览器访问运行 Grafana 的服务器的3000端口 host:3000 登录 Grafana 界面，系统初始用户名密码为 admin/admin。
点击左侧齿轮图标并选择 Plugins，应该可以找到 TDengine data source 插件图标。
#### 导入 collectd 仪表盘
从 https://github.com/taosdata/grafanaplugin/blob/master/examples/collectd/grafana/dashboards/collect-metrics-with-tdengine-v0.1.0.json 下载 dashboard json 文件，点击左侧加号图标并选择 Import，按照界面提示选择 JSON 文件导入。之后可以看到如下界面的仪表盘：
![IT-DevOps-Solutions-collectd-dashboard.png](../../images/IT-DevOps-Solutions-collectd-dashboard.png)
#### 导入 StatsD 仪表盘
从 https://github.com/taosdata/grafanaplugin/blob/master/examples/statsd/dashboards/statsd-with-tdengine-v0.1.0.json 下载 dashboard json 文件，点击左侧加号图标并选择 Import，按照界面提示导入JSON文件。之后可以看到如下界面的仪表盘：
![IT-DevOps-Solutions-statsd-dashboard.png](../../images/IT-DevOps-Solutions-statsd-dashboard.png)
## 总结
TDengine 作为新兴的时序大数据平台，具备极强的高性能、高可靠、易管理、易维护的优势。得力于 TDengine 2.3.0.0 版本中新增的 schemaless 协议解析功能，以及强大的生态软件适配能力，用户可以短短数分钟就可以搭建一个高效易用的 IT 运维系统或者适配一个已存在的系统。
TDengine 强大的数据写入查询性能和其他丰富功能请参考官方文档和产品成功落地案例。
documentation20/en/09.connections/docs.md
浏览文件 @ 51259460
......@@ -12,12 +12,17 @@ https://grafana.com/grafana/download.
### Configure Grafana
TDengine Grafana plugin is in the /usr/local/taos/connector/grafanaplugin directory.
Download grafana plugin from <https://github.com/taosdata/grafanaplugin/releases/latest> .
```bash
GF_VERSION=3.1.1
wget https://github.com/taosdata/grafanaplugin/releases/download/v$GF_VERSION/tdengine-datasource-$GF_VERSION.zip
```
Taking Centos 7.2 as an example, just copy grafanaplugin directory to /var/lib/grafana/plugins directory and restart Grafana.
```bash
sudo cp -rf /usr/local/taos/connector/grafanaplugin /var/lib/grafana/plugins/tdengine
sudo unzip tdengine-datasource-$GF_VERSION.zip /var/lib/grafana/plugins/
```
### Use Grafana
......@@ -64,15 +69,15 @@ According to the default prompt, query the average system memory usage at the sp
#### Import Dashboard
A `tdengine-grafana.json` importable dashboard is provided under the Grafana plug-in directory/usr/local/taos/connector/grafana/tdengine/dashboard/.
We provide an example dashboard [Grafana Dashboard 15146](https://grafana.com/grafana/dashboards/15146)。
Click the `Import` button on the left panel and upload the `tdengine-grafana.json` file:
Click the `Import` button on the left panel and load the grafana id:
![img](page://images/connections/import_dashboard1.jpg)
You can see as follows after Dashboard imported.
![img](page://images/connections/import_dashboard2.jpg)
![img](../images/connections/dashboard-15146.png)
## <a class="anchor" id="matlab"></a> MATLAB
......
packaging/deb/makedeb.sh
浏览文件 @ 51259460
......@@ -58,12 +58,6 @@ cp ${compile_dir}/build/lib/${libfile} ${pkg_dir}${install_home_pat
cp ${compile_dir}/../src/inc/taos.h ${pkg_dir}${install_home_path}/include
cp ${compile_dir}/../src/inc/taoserror.h ${pkg_dir}${install_home_path}/include
cp -r ${top_dir}/tests/examples/* ${pkg_dir}${install_home_path}/examples
if [ -d "${top_dir}/src/connector/grafanaplugin/dist" ]; then
cp -r ${top_dir}/src/connector/grafanaplugin/dist ${pkg_dir}${install_home_path}/connector/grafanaplugin
else
echo "grafanaplugin bundled directory not found!"
exit 1
fi
cp -r ${top_dir}/src/connector/python ${pkg_dir}${install_home_path}/connector
cp -r ${top_dir}/src/connector/go ${pkg_dir}${install_home_path}/connector
cp -r ${top_dir}/src/connector/nodejs ${pkg_dir}${install_home_path}/connector
......
packaging/release.sh
浏览文件 @ 51259460
......@@ -194,6 +194,7 @@ fi
if [[ "$dbName" == "pro" ]]; then
sed -i "s/taos config/prodb config/g" ${top_dir}/src/util/src/tconfig.c
sed -i "s/TDengine/ProDB/g" ${top_dir}/src/dnode/src/dnodeSystem.c
fi
......
packaging/rpm/tdengine.spec
浏览文件 @ 51259460
......@@ -67,12 +67,6 @@ cp %{_compiledir}/build/bin/taosdump %{buildroot}%{homepath}/bin
cp %{_compiledir}/build/lib/${libfile} %{buildroot}%{homepath}/driver
cp %{_compiledir}/../src/inc/taos.h %{buildroot}%{homepath}/include
cp %{_compiledir}/../src/inc/taoserror.h %{buildroot}%{homepath}/include
if [ -d %{_compiledir}/../src/connector/grafanaplugin/dist ]; then
cp -r %{_compiledir}/../src/connector/grafanaplugin/dist %{buildroot}%{homepath}/connector/grafanaplugin
else
echo grafanaplugin bundled directory not found!
exit 1
fi
cp -r %{_compiledir}/../src/connector/python %{buildroot}%{homepath}/connector
cp -r %{_compiledir}/../src/connector/go %{buildroot}%{homepath}/connector
cp -r %{_compiledir}/../src/connector/nodejs %{buildroot}%{homepath}/connector
......
packaging/tools/make_install.sh
浏览文件 @ 51259460
......@@ -316,11 +316,6 @@ function install_data() {
}
function install_connector() {
if [ -d "${source_dir}/src/connector/grafanaplugin/dist" ]; then
${csudo} cp -rf ${source_dir}/src/connector/grafanaplugin/dist ${install_main_dir}/connector/grafanaplugin
else
echo "WARNING: grafanaplugin bundled dir not found, please check if want to use it!"
fi
if find ${source_dir}/src/connector/go -mindepth 1 -maxdepth 1 | read; then
${csudo} cp -r ${source_dir}/src/connector/go ${install_main_dir}/connector
else
......
packaging/tools/makeclient.sh
浏览文件 @ 51259460
......@@ -150,11 +150,6 @@ if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
if [ "$osType" != "Darwin" ]; then
cp ${build_dir}/lib/*.jar ${install_dir}/connector ||:
fi
if [ -d "${connector_dir}/grafanaplugin/dist" ]; then
cp -r ${connector_dir}/grafanaplugin/dist ${install_dir}/connector/grafanaplugin
else
echo "WARNING: grafanaplugin bundled dir not found, please check if want to use it!"
fi
if find ${connector_dir}/go -mindepth 1 -maxdepth 1 | read; then
cp -r ${connector_dir}/go ${install_dir}/connector
else
......
packaging/tools/makeclient_power.sh
浏览文件 @ 51259460
......@@ -210,11 +210,6 @@ if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
if [ "$osType" != "Darwin" ]; then
cp ${build_dir}/lib/*.jar ${install_dir}/connector ||:
fi
if [ -d "${connector_dir}/grafanaplugin/dist" ]; then
cp -r ${connector_dir}/grafanaplugin/dist ${install_dir}/connector/grafanaplugin
else
echo "WARNING: grafanaplugin bunlded dir not found, please check if want to use it!"
fi
if find ${connector_dir}/go -mindepth 1 -maxdepth 1 | read; then
cp -r ${connector_dir}/go ${install_dir}/connector
else
......
packaging/tools/makeclient_pro.sh
浏览文件 @ 51259460
......@@ -172,11 +172,6 @@ if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
if [ "$osType" != "Darwin" ]; then
cp ${build_dir}/lib/*.jar ${install_dir}/connector ||:
fi
if [ -d "${connector_dir}/grafanaplugin/dist" ]; then
cp -r ${connector_dir}/grafanaplugin/dist ${install_dir}/connector/grafanaplugin
else
echo "WARNING: grafanaplugin bunlded dir not found, please check if want to use it!"
fi
if find ${connector_dir}/go -mindepth 1 -maxdepth 1 | read; then
cp -r ${connector_dir}/go ${install_dir}/connector
else
......
packaging/tools/makeclient_tq.sh
浏览文件 @ 51259460
......@@ -177,11 +177,6 @@ if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
if [ "$osType" != "Darwin" ]; then
cp ${build_dir}/lib/*.jar ${install_dir}/connector ||:
fi
if [ -d "${connector_dir}/grafanaplugin/dist" ]; then
cp -r ${connector_dir}/grafanaplugin/dist ${install_dir}/connector/grafanaplugin
else
echo "WARNING: grafanaplugin bunlded dir not found, please check if want to use it!"
fi
if find ${connector_dir}/go -mindepth 1 -maxdepth 1 | read; then
cp -r ${connector_dir}/go ${install_dir}/connector
else
......
packaging/tools/makepkg.sh
浏览文件 @ 51259460
......@@ -183,11 +183,6 @@ connector_dir="${code_dir}/connector"
mkdir -p ${install_dir}/connector
if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
cp ${build_dir}/lib/*.jar ${install_dir}/connector ||:
if [ -d "${connector_dir}/grafanaplugin/dist" ]; then
cp -r ${connector_dir}/grafanaplugin/dist ${install_dir}/connector/grafanaplugin
else
echo "WARNING: grafanaplugin bundled dir not found, please check if you want to use it!"
fi
if find ${connector_dir}/go -mindepth 1 -maxdepth 1 | read; then
cp -r ${connector_dir}/go ${install_dir}/connector
else
......
packaging/tools/makepkg_power.sh
浏览文件 @ 51259460
......@@ -167,11 +167,6 @@ mkdir -p ${install_dir}/connector
if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
cp ${build_dir}/lib/*.jar ${install_dir}/connector ||:
if [ -d "${connector_dir}/grafanaplugin/dist" ]; then
cp -r ${connector_dir}/grafanaplugin/dist ${install_dir}/connector/grafanaplugin
else
echo "WARNING: grafanaplugin bundled dir not found, please check if want to use it!"
fi
if find ${connector_dir}/go -mindepth 1 -maxdepth 1 | read; then
cp -r ${connector_dir}/go ${install_dir}/connector
else
......
packaging/tools/makepkg_pro.sh
浏览文件 @ 51259460
......@@ -153,11 +153,6 @@ mkdir -p ${install_dir}/driver && cp ${lib_files} ${install_dir}/driver && echo
#if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
# cp ${build_dir}/lib/*.jar ${install_dir}/connector ||:
# if [ -d "${connector_dir}/grafanaplugin/dist" ]; then
# cp -r ${connector_dir}/grafanaplugin/dist ${install_dir}/connector/grafanaplugin
# else
# echo "WARNING: grafanaplugin bundled dir not found, please check if want to use it!"
# fi
# if find ${connector_dir}/go -mindepth 1 -maxdepth 1 | read; then
# cp -r ${connector_dir}/go ${install_dir}/connector
# else
......
packaging/tools/makepkg_tq.sh
浏览文件 @ 51259460
......@@ -167,11 +167,6 @@ mkdir -p ${install_dir}/connector
if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
cp ${build_dir}/lib/*.jar ${install_dir}/connector ||:
if [ -d "${connector_dir}/grafanaplugin/dist" ]; then
cp -r ${connector_dir}/grafanaplugin/dist ${install_dir}/connector/grafanaplugin
else
echo "WARNING: grafanaplugin bundled dir not found, please check if want to use it!"
fi
if find ${connector_dir}/go -mindepth 1 -maxdepth 1 | read; then
cp -r ${connector_dir}/go ${install_dir}/connector
else
......
snap/snapcraft.yaml
浏览文件 @ 51259460
name: tdengine
base: core18
version: '2.2.1.1'
version: '2.2.1.3'
icon: snap/gui/t-dengine.svg
summary: an open-source big data platform designed and optimized for IoT.
description: |
......@@ -72,7 +72,7 @@ parts:
- usr/bin/taosd
- usr/bin/taos
- usr/bin/taosdemo
- usr/lib/libtaos.so.2.2.1.1
- usr/lib/libtaos.so.2.2.1.3
- usr/lib/libtaos.so.1
- usr/lib/libtaos.so
......
src/client/inc/tscUtil.h
浏览文件 @ 51259460
......@@ -108,6 +108,7 @@ typedef struct SBlockKeyInfo {
int32_t converToStr(char *str, int type, void *buf, int32_t bufSize, int32_t *len);
int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, SName* name, STableMeta* pTableMeta, STableDataBlocks** dataBlocks);
int32_t tscCreateDataBlockData(STableDataBlocks* dataBuf, size_t defaultSize, int32_t rowSize, int32_t startOffset);
void tscDestroyDataBlock(STableDataBlocks* pDataBlock, bool removeMeta);
void tscSortRemoveDataBlockDupRowsRaw(STableDataBlocks* dataBuf);
int tscSortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKeyInfo);
......
src/client/src/tscLocal.c
浏览文件 @ 51259460
......@@ -358,9 +358,13 @@ static int32_t tscGetTableTagValue(SCreateBuilder *builder, char *result) {
int num_fields = taos_num_fields(pSql);
TAOS_FIELD *fields = taos_fetch_fields(pSql);
char buf[TSDB_COL_NAME_LEN + 16];
for (int i = 0; i < num_fields; i++) {
memset(buf, 0, sizeof(buf));
char *buf = calloc(1, lengths[i] + 1);
if (buf == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
memset(buf, 0, lengths[i] + 1);
int32_t ret = tscGetNthFieldResult(row, fields, lengths, i, buf);
if (i == 0) {
......@@ -373,10 +377,13 @@ static int32_t tscGetTableTagValue(SCreateBuilder *builder, char *result) {
} else {
snprintf(result + strlen(result), TSDB_MAX_BINARY_LEN - strlen(result), "%s,", buf);
}
free(buf);
if (i == num_fields - 1) {
sprintf(result + strlen(result) - 1, "%s", ")");
}
}
}
if (0 == strlen(result)) {
return TSDB_CODE_TSC_INVALID_TABLE_NAME;
......
src/client/src/tscPrepare.c
浏览文件 @ 51259460
......@@ -48,12 +48,14 @@ typedef struct SMultiTbStmt {
bool nameSet;
bool tagSet;
bool subSet;
bool tagColSet;
uint64_t currentUid;
char *sqlstr;
uint32_t tbNum;
SStrToken tbname;
SStrToken stbname;
SStrToken values;
SStrToken tagCols;
SArray *tags;
STableDataBlocks *lastBlock;
SHashObj *pTableHash;
......@@ -1246,6 +1248,12 @@ static void insertBatchClean(STscStmt* pStmt) {
pCmd->insertParam.pDataBlocks = tscDestroyBlockArrayList(pCmd->insertParam.pDataBlocks);
pCmd->insertParam.numOfTables = 0;
STableDataBlocks** p = taosHashIterate(pCmd->insertParam.pTableBlockHashList, NULL);
while(p) {
tfree((*p)->pData);
p = taosHashIterate(pCmd->insertParam.pTableBlockHashList, p);
}
taosHashClear(pCmd->insertParam.pTableBlockHashList);
tscFreeSqlResult(pSql);
tscFreeSubobj(pSql);
......@@ -1337,9 +1345,40 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
pStmt->mtb.stbname = sToken;
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n <= 0 || sToken.type != TK_TAGS) {
tscError("keyword TAGS expected, sql:%s", pCmd->insertParam.sql);
return tscSQLSyntaxErrMsg(pCmd->payload, "keyword TAGS expected", sToken.z ? sToken.z : pCmd->insertParam.sql);
if (sToken.n <= 0 || ((sToken.type != TK_TAGS) && (sToken.type != TK_LP))) {
tscError("invalid token, sql:%s", pCmd->insertParam.sql);
return tscSQLSyntaxErrMsg(pCmd->payload, "invalid token", sToken.z ? sToken.z : pCmd->insertParam.sql);
}
// ... (tag_col_list) TAGS(tag_val_list) ...
int32_t tagColsCnt = 0;
if (sToken.type == TK_LP) {
pStmt->mtb.tagColSet = true;
pStmt->mtb.tagCols = sToken;
int32_t tagColsStart = index;
while (1) {
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.type == TK_ILLEGAL) {
return tscSQLSyntaxErrMsg(pCmd->payload, "unrecognized token", sToken.z);
}
if (sToken.type == TK_ID) {
++tagColsCnt;
}
if (sToken.type == TK_RP) {
break;
}
}
if (tagColsCnt == 0) {
tscError("tag column list expected, sql:%s", pCmd->insertParam.sql);
return tscSQLSyntaxErrMsg(pCmd->payload, "tag column list expected", pCmd->insertParam.sql);
}
pStmt->mtb.tagCols.n = index - tagColsStart + 1;
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n <= 0 || sToken.type != TK_TAGS) {
tscError("keyword TAGS expected, sql:%s", pCmd->insertParam.sql);
return tscSQLSyntaxErrMsg(pCmd->payload, "keyword TAGS expected", sToken.z ? sToken.z : pCmd->insertParam.sql);
}
}
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
......@@ -1379,6 +1418,11 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
return tscSQLSyntaxErrMsg(pCmd->payload, "no tags", pCmd->insertParam.sql);
}
if (tagColsCnt > 0 && taosArrayGetSize(pStmt->mtb.tags) != tagColsCnt) {
tscError("not match tags, sql:%s", pCmd->insertParam.sql);
return tscSQLSyntaxErrMsg(pCmd->payload, "not match tags", pCmd->insertParam.sql);
}
sToken = tStrGetToken(pCmd->insertParam.sql, &index, false);
if (sToken.n <= 0 || (sToken.type != TK_VALUES && sToken.type != TK_LP)) {
tscError("sql error, sql:%s", pCmd->insertParam.sql);
......@@ -1401,7 +1445,13 @@ int stmtGenInsertStatement(SSqlObj* pSql, STscStmt* pStmt, const char* name, TAO
int32_t j = 0;
while (1) {
len = (size_t)snprintf(str, size - 1, "insert into %s using %.*s tags(", name, pStmt->mtb.stbname.n, pStmt->mtb.stbname.z);
if (pStmt->mtb.tagColSet) {
len = (size_t)snprintf(str, size - 1, "insert into %s using %.*s %.*s tags(",
name, pStmt->mtb.stbname.n, pStmt->mtb.stbname.z, pStmt->mtb.tagCols.n, pStmt->mtb.tagCols.z);
} else {
len = (size_t)snprintf(str, size - 1, "insert into %s using %.*s tags(", name, pStmt->mtb.stbname.n, pStmt->mtb.stbname.z);
}
if (len >= (size -1)) {
size *= 2;
free(str);
......@@ -1637,6 +1687,13 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
STMT_RET(TSDB_CODE_TSC_APP_ERROR);
}
if ((*t1)->pData == NULL) {
code = tscCreateDataBlockData(*t1, TSDB_PAYLOAD_SIZE, (*t1)->pTableMeta->tableInfo.rowSize, sizeof(SSubmitBlk));
if (code != TSDB_CODE_SUCCESS) {
STMT_RET(code);
}
}
SSubmitBlk* pBlk = (SSubmitBlk*) (*t1)->pData;
pCmd->batchSize = pBlk->numOfRows;
if (pBlk->numOfRows == 0) {
......@@ -1695,7 +1752,6 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
STMT_RET(TSDB_CODE_SUCCESS);
}
if (pStmt->mtb.tagSet) {
pStmt->mtb.tbname = tscReplaceStrToken(&pSql->sqlstr, &pStmt->mtb.tbname, name);
} else {
......@@ -1762,7 +1818,6 @@ int taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags
STMT_RET(code);
}
int taos_stmt_set_sub_tbname(TAOS_STMT* stmt, const char* name) {
STscStmt* pStmt = (STscStmt*)stmt;
STMT_CHECK
......@@ -1770,8 +1825,6 @@ int taos_stmt_set_sub_tbname(TAOS_STMT* stmt, const char* name) {
return taos_stmt_set_tbname_tags(stmt, name, NULL);
}
int taos_stmt_set_tbname(TAOS_STMT* stmt, const char* name) {
STscStmt* pStmt = (STscStmt*)stmt;
STMT_CHECK
......@@ -1779,7 +1832,6 @@ int taos_stmt_set_tbname(TAOS_STMT* stmt, const char* name) {
return taos_stmt_set_tbname_tags(stmt, name, NULL);
}
int taos_stmt_close(TAOS_STMT* stmt) {
STscStmt* pStmt = (STscStmt*)stmt;
if (pStmt == NULL || pStmt->taos == NULL) {
......@@ -1846,7 +1898,6 @@ int taos_stmt_bind_param(TAOS_STMT* stmt, TAOS_BIND* bind) {
}
}
int taos_stmt_bind_param_batch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind) {
STscStmt* pStmt = (STscStmt*)stmt;
......@@ -1910,8 +1961,6 @@ int taos_stmt_bind_single_param_batch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, in
STMT_RET(insertStmtBindParamBatch(pStmt, bind, colIdx));
}
int taos_stmt_add_batch(TAOS_STMT* stmt) {
STscStmt* pStmt = (STscStmt*)stmt;
STMT_CHECK
......@@ -2052,7 +2101,6 @@ int taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes) {
}
}
char *taos_stmt_errstr(TAOS_STMT *stmt) {
STscStmt* pStmt = (STscStmt*)stmt;
......@@ -2063,8 +2111,6 @@ char *taos_stmt_errstr(TAOS_STMT *stmt) {
return taos_errstr(pStmt->pSql);
}
const char *taos_data_type(int type) {
switch (type) {
case TSDB_DATA_TYPE_NULL: return "TSDB_DATA_TYPE_NULL";
......@@ -2081,4 +2127,3 @@ const char *taos_data_type(int type) {
default: return "UNKNOWN";
}
}
src/client/src/tscSQLParser.c
浏览文件 @ 51259460
......@@ -5610,7 +5610,7 @@ int32_t validateOrderbyNode(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SSqlNode* pSq
bool udf = false;
if (pQueryInfo->pUdfInfo && taosArrayGetSize(pQueryInfo->pUdfInfo) > 0) {
int32_t usize = taosArrayGetSize(pQueryInfo->pUdfInfo);
int32_t usize = (int32_t)taosArrayGetSize(pQueryInfo->pUdfInfo);
for (int32_t i = 0; i < usize; ++i) {
SUdfInfo* pUdfInfo = taosArrayGet(pQueryInfo->pUdfInfo, i);
......@@ -8406,7 +8406,7 @@ int32_t loadAllTableMeta(SSqlObj* pSql, struct SSqlInfo* pInfo) {
for (int32_t j = 0; j < usize; ++j) {
SUdfInfo* pUdfInfo = taosArrayGet(pQueryInfo->pUdfInfo, j);
int32_t len = strlen(pUdfInfo->name);
int32_t len = (int32_t)strlen(pUdfInfo->name);
if (len == t->n && strncasecmp(info.name, pUdfInfo->name, t->n) == 0) {
exist = 1;
break;
......
src/client/src/tscUtil.c
浏览文件 @ 51259460
......@@ -919,23 +919,31 @@ SSDataBlock* doGetDataBlock(void* param, bool* newgroup) {
pBlock->info.rows = pRes->numOfRows;
if (pRes->numOfRows != 0) {
doSetupSDataBlock(pRes, pBlock, pInput->pFilterInfo, pInput->numOfFilterCols);
*newgroup = false;
return pBlock;
if (pBlock->info.rows > 0) {
*newgroup = false;
return pBlock;
}
}
// No data block exists. So retrieve and transfer it into to SSDataBlock
TAOS_ROW pRow = NULL;
taos_fetch_block(pSql, &pRow);
SSDataBlock* result = NULL;
do {
// No data block exists. So retrieve and transfer it into to SSDataBlock
TAOS_ROW pRow = NULL;
taos_fetch_block(pSql, &pRow);
if (pRes->numOfRows == 0) {
pOperator->status = OP_EXEC_DONE;
return NULL;
}
if (pRes->numOfRows == 0) {
pOperator->status = OP_EXEC_DONE;
result = NULL;
break;
}
pBlock->info.rows = pRes->numOfRows;
doSetupSDataBlock(pRes, pBlock, pInput->pFilterInfo, pInput->numOfFilterCols);
*newgroup = false;
return pBlock;
pBlock->info.rows = pRes->numOfRows;
doSetupSDataBlock(pRes, pBlock, pInput->pFilterInfo, pInput->numOfFilterCols);
*newgroup = false;
result = pBlock;
} while (result->info.rows == 0);
return result;
}
static void fetchNextBlockIfCompleted(SOperatorInfo* pOperator, bool* newgroup) {
......@@ -1797,6 +1805,32 @@ int32_t tscCreateDataBlock(size_t defaultSize, int32_t rowSize, int32_t startOff
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
int32_t code = tscCreateDataBlockData(dataBuf, defaultSize, rowSize, startOffset);
if (code != TSDB_CODE_SUCCESS) {
tfree(dataBuf);
return code;
}
//Here we keep the tableMeta to avoid it to be remove by other threads.
dataBuf->pTableMeta = tscTableMetaDup(pTableMeta);
SParsedDataColInfo* pColInfo = &dataBuf->boundColumnInfo;
SSchema* pSchema = tscGetTableSchema(dataBuf->pTableMeta);
tscSetBoundColumnInfo(pColInfo, pSchema, dataBuf->pTableMeta->tableInfo.numOfColumns);
dataBuf->vgId = dataBuf->pTableMeta->vgId;
tNameAssign(&dataBuf->tableName, name);
assert(defaultSize > 0 && pTableMeta != NULL && dataBuf->pTableMeta != NULL);
*dataBlocks = dataBuf;
return TSDB_CODE_SUCCESS;
}
int32_t tscCreateDataBlockData(STableDataBlocks* dataBuf, size_t defaultSize, int32_t rowSize, int32_t startOffset) {
assert(dataBuf != NULL);
dataBuf->nAllocSize = (uint32_t)defaultSize;
dataBuf->headerSize = startOffset;
......@@ -1809,30 +1843,16 @@ int32_t tscCreateDataBlock(size_t defaultSize, int32_t rowSize, int32_t startOff
dataBuf->pData = malloc(dataBuf->nAllocSize);
if (dataBuf->pData == NULL) {
tscError("failed to allocated memory, reason:%s", strerror(errno));
tfree(dataBuf);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
memset(dataBuf->pData, 0, sizeof(SSubmitBlk));
//Here we keep the tableMeta to avoid it to be remove by other threads.
dataBuf->pTableMeta = tscTableMetaDup(pTableMeta);
SParsedDataColInfo* pColInfo = &dataBuf->boundColumnInfo;
SSchema* pSchema = tscGetTableSchema(dataBuf->pTableMeta);
tscSetBoundColumnInfo(pColInfo, pSchema, dataBuf->pTableMeta->tableInfo.numOfColumns);
dataBuf->ordered = true;
dataBuf->prevTS = INT64_MIN;
dataBuf->rowSize = rowSize;
dataBuf->size = startOffset;
dataBuf->tsSource = -1;
dataBuf->vgId = dataBuf->pTableMeta->vgId;
tNameAssign(&dataBuf->tableName, name);
assert(defaultSize > 0 && pTableMeta != NULL && dataBuf->pTableMeta != NULL);
*dataBlocks = dataBuf;
return TSDB_CODE_SUCCESS;
}
......
src/common/inc/tglobal.h
浏览文件 @ 51259460
......@@ -45,7 +45,7 @@ extern int32_t tsDnodeId;
// common
extern int tsRpcTimer;
extern int tsRpcMaxTime;
extern int tsRpcForceTcp; // all commands go to tcp protocol if this is enabled
extern int tsRpcForceTcp; // all commands go to tcp protocol if this is enabled
extern int32_t tsMaxConnections;
extern int32_t tsMaxShellConns;
extern int32_t tsShellActivityTimer;
......@@ -56,18 +56,19 @@ extern float tsRatioOfQueryCores;
extern int8_t tsDaylight;
extern char tsTimezone[];
extern char tsLocale[];
extern char tsCharset[]; // default encode string
extern char tsCharset[]; // default encode string
extern int8_t tsEnableCoreFile;
extern int32_t tsCompressMsgSize;
extern int32_t tsMaxNumOfDistinctResults;
extern char tsTempDir[];
//query buffer management
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
// query buffer management
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 int8_t tsKeepOriginalColumnName;
extern int8_t tsKeepOriginalColumnName;
// client
extern int32_t tsMaxSQLStringLen;
......@@ -126,6 +127,7 @@ extern int8_t tsHttpEnableCompress;
extern int8_t tsHttpEnableRecordSql;
extern int8_t tsTelegrafUseFieldNum;
extern int8_t tsHttpDbNameMandatory;
extern int32_t tsHttpKeepAlive;
// mqtt
extern int8_t tsEnableMqttModule;
......@@ -162,22 +164,22 @@ extern int64_t tsTickPerDay[3];
extern int32_t tsTopicBianryLen;
// system info
extern char tsOsName[];
extern int64_t tsPageSize;
extern int64_t tsOpenMax;
extern int64_t tsStreamMax;
extern int32_t tsNumOfCores;
extern float tsTotalLogDirGB;
extern float tsTotalTmpDirGB;
extern float tsTotalDataDirGB;
extern float tsAvailLogDirGB;
extern float tsAvailTmpDirectorySpace;
extern float tsAvailDataDirGB;
extern float tsUsedDataDirGB;
extern float tsMinimalLogDirGB;
extern float tsReservedTmpDirectorySpace;
extern float tsMinimalDataDirGB;
extern int32_t tsTotalMemoryMB;
extern char tsOsName[];
extern int64_t tsPageSize;
extern int64_t tsOpenMax;
extern int64_t tsStreamMax;
extern int32_t tsNumOfCores;
extern float tsTotalLogDirGB;
extern float tsTotalTmpDirGB;
extern float tsTotalDataDirGB;
extern float tsAvailLogDirGB;
extern float tsAvailTmpDirectorySpace;
extern float tsAvailDataDirGB;
extern float tsUsedDataDirGB;
extern float tsMinimalLogDirGB;
extern float tsReservedTmpDirectorySpace;
extern float tsMinimalDataDirGB;
extern int32_t tsTotalMemoryMB;
extern uint32_t tsVersion;
// build info
......@@ -188,26 +190,26 @@ extern char gitinfoOfInternal[];
extern char buildinfo[];
// log
extern int8_t tsAsyncLog;
extern int32_t tsNumOfLogLines;
extern int32_t tsLogKeepDays;
extern int32_t dDebugFlag;
extern int32_t vDebugFlag;
extern int32_t mDebugFlag;
extern int8_t tsAsyncLog;
extern int32_t tsNumOfLogLines;
extern int32_t tsLogKeepDays;
extern int32_t dDebugFlag;
extern int32_t vDebugFlag;
extern int32_t mDebugFlag;
extern uint32_t cDebugFlag;
extern int32_t jniDebugFlag;
extern int32_t tmrDebugFlag;
extern int32_t sdbDebugFlag;
extern int32_t httpDebugFlag;
extern int32_t mqttDebugFlag;
extern int32_t monDebugFlag;
extern int32_t uDebugFlag;
extern int32_t rpcDebugFlag;
extern int32_t odbcDebugFlag;
extern int32_t jniDebugFlag;
extern int32_t tmrDebugFlag;
extern int32_t sdbDebugFlag;
extern int32_t httpDebugFlag;
extern int32_t mqttDebugFlag;
extern int32_t monDebugFlag;
extern int32_t uDebugFlag;
extern int32_t rpcDebugFlag;
extern int32_t odbcDebugFlag;
extern uint32_t qDebugFlag;
extern int32_t wDebugFlag;
extern int32_t cqDebugFlag;
extern int32_t debugFlag;
extern int32_t wDebugFlag;
extern int32_t cqDebugFlag;
extern int32_t debugFlag;
extern int8_t tsClientMerge;
......@@ -218,7 +220,7 @@ extern double fPrecision;
extern double dPrecision;
extern uint32_t maxRange;
extern uint32_t curRange;
extern char Compressor[];
extern char Compressor[];
#endif
......
src/common/src/tdataformat.c
浏览文件 @ 51259460
......@@ -253,9 +253,10 @@ int dataColAppendVal(SDataCol *pCol, const void *value, int numOfRows, int maxPo
}
if(tdAllocMemForCol(pCol, maxPoints) < 0) return -1;
if (numOfRows > 0) {
if (((rowOffset == 0) && (numOfRows > 0)) || ((rowOffset == -1) && (numOfRows >= 0))) {
// Find the first not null value, fill all previouse values as NULL
dataColSetNEleNull(pCol, numOfRows);
dataColSetNEleNull(pCol, numOfRows - rowOffset);
}
}
......@@ -463,9 +464,7 @@ static void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols
int rcol = 0;
int dcol = 0;
while (dcol < pCols->numOfCols) {
bool setCol = 0;
SDataCol *pDataCol = &(pCols->cols[dcol]);
if (rcol >= schemaNCols(pSchema)) {
dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints, rowOffset);
......@@ -476,14 +475,22 @@ static void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols
STColumn *pRowCol = schemaColAt(pSchema, rcol);
if (pRowCol->colId == pDataCol->colId) {
void *value = tdGetRowDataOfCol(row, pRowCol->type, pRowCol->offset + TD_DATA_ROW_HEAD_SIZE);
if(!isNull(value, pDataCol->type)) setCol = 1;
dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints, rowOffset);
if (rowOffset == 0) {
dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints, rowOffset);
} else if (rowOffset == -1) {
// for update 2
if (!isNull(value, pDataCol->type)) {
dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints, rowOffset);
}
} else {
ASSERT(0);
}
dcol++;
rcol++;
} else if (pRowCol->colId < pDataCol->colId) {
rcol++;
} else {
if(forceSetNull || setCol) {
if(forceSetNull) {
dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints, rowOffset);
}
dcol++;
......@@ -501,7 +508,6 @@ static void tdAppendKvRowToDataCol(SKVRow row, STSchema *pSchema, SDataCols *pCo
int nRowCols = kvRowNCols(row);
while (dcol < pCols->numOfCols) {
bool setCol = 0;
SDataCol *pDataCol = &(pCols->cols[dcol]);
if (rcol >= nRowCols || rcol >= schemaNCols(pSchema)) {
dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints, rowOffset);
......@@ -513,14 +519,22 @@ static void tdAppendKvRowToDataCol(SKVRow row, STSchema *pSchema, SDataCols *pCo
if (colIdx->colId == pDataCol->colId) {
void *value = tdGetKvRowDataOfCol(row, colIdx->offset);
if(!isNull(value, pDataCol->type)) setCol = 1;
dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints, rowOffset);
if (rowOffset == 0) {
dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints, rowOffset);
} else if (rowOffset == -1) {
// for update 2
if (!isNull(value, pDataCol->type)) {
dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints, rowOffset);
}
} else {
ASSERT(0);
}
++dcol;
++rcol;
} else if (colIdx->colId < pDataCol->colId) {
++rcol;
} else {
if (forceSetNull || setCol) {
if (forceSetNull) {
dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints, rowOffset);
}
++dcol;
......
src/common/src/tglobal.c
浏览文件 @ 51259460
......@@ -14,18 +14,18 @@
*/
#define _DEFAULT_SOURCE
#include "tglobal.h"
#include "monitor.h"
#include "os.h"
#include "taosdef.h"
#include "taoserror.h"
#include "tulog.h"
#include "tcompare.h"
#include "tconfig.h"
#include "tglobal.h"
#include "monitor.h"
#include "tsocket.h"
#include "tutil.h"
#include "tlocale.h"
#include "tsocket.h"
#include "ttimezone.h"
#include "tcompare.h"
#include "tulog.h"
#include "tutil.h"
// TSDB
bool tsdbForceKeepFile = false;
......@@ -51,16 +51,16 @@ char tsEmail[TSDB_FQDN_LEN] = {0};
int32_t tsDnodeId = 0;
// common
int32_t tsRpcTimer = 300;
int32_t tsRpcMaxTime = 600; // seconds;
int32_t tsRpcForceTcp = 0; //disable this, means query, show command use udp protocol as default
int32_t tsMaxShellConns = 50000;
int32_t tsRpcTimer = 300;
int32_t tsRpcMaxTime = 600; // seconds;
int32_t tsRpcForceTcp = 0; // disable this, means query, show command use udp protocol as default
int32_t tsMaxShellConns = 50000;
int32_t tsMaxConnections = 5000;
int32_t tsShellActivityTimer = 3; // second
int32_t tsShellActivityTimer = 3; // second
float tsNumOfThreadsPerCore = 1.0f;
int32_t tsNumOfCommitThreads = 4;
float tsRatioOfQueryCores = 1.0f;
int8_t tsDaylight = 0;
int8_t tsDaylight = 0;
char tsTimezone[TSDB_TIMEZONE_LEN] = {0};
char tsLocale[TSDB_LOCALE_LEN] = {0};
char tsCharset[TSDB_LOCALE_LEN] = {0}; // default encode string
......@@ -90,7 +90,7 @@ int32_t tsMaxNumOfOrderedResults = 1000000;
int32_t tsMinSlidingTime = 10;
// the maxinum number of distict query result
int32_t tsMaxNumOfDistinctResults = 1000 * 10000;
int32_t tsMaxNumOfDistinctResults = 1000 * 10000;
// 1 us for interval time range, changed accordingly
int32_t tsMinIntervalTime = 1;
......@@ -102,7 +102,7 @@ int32_t tsMaxStreamComputDelay = 20000;
int32_t tsStreamCompStartDelay = 10000;
// the stream computing delay time after executing failed, change accordingly
int32_t tsRetryStreamCompDelay = 10*1000;
int32_t tsRetryStreamCompDelay = 10 * 1000;
// The delayed computing ration. 10% of the whole computing time window by default.
float tsStreamComputDelayRatio = 0.1f;
......@@ -121,26 +121,26 @@ int64_t tsQueryBufferSizeBytes = -1;
int32_t tsRetrieveBlockingModel = 0;
// last_row(*), first(*), last_row(ts, col1, col2) query, the result fields will be the original column name
int8_t tsKeepOriginalColumnName = 0;
int8_t tsKeepOriginalColumnName = 0;
// db parameters
int32_t tsCacheBlockSize = TSDB_DEFAULT_CACHE_BLOCK_SIZE;
int32_t tsBlocksPerVnode = TSDB_DEFAULT_TOTAL_BLOCKS;
int16_t tsDaysPerFile = TSDB_DEFAULT_DAYS_PER_FILE;
int32_t tsDaysToKeep = TSDB_DEFAULT_KEEP;
int16_t tsDaysPerFile = TSDB_DEFAULT_DAYS_PER_FILE;
int32_t tsDaysToKeep = TSDB_DEFAULT_KEEP;
int32_t tsMinRowsInFileBlock = TSDB_DEFAULT_MIN_ROW_FBLOCK;
int32_t tsMaxRowsInFileBlock = TSDB_DEFAULT_MAX_ROW_FBLOCK;
int16_t tsCommitTime = TSDB_DEFAULT_COMMIT_TIME; // seconds
int16_t tsCommitTime = TSDB_DEFAULT_COMMIT_TIME; // seconds
int32_t tsTimePrecision = TSDB_DEFAULT_PRECISION;
int8_t tsCompression = TSDB_DEFAULT_COMP_LEVEL;
int8_t tsWAL = TSDB_DEFAULT_WAL_LEVEL;
int32_t tsFsyncPeriod = TSDB_DEFAULT_FSYNC_PERIOD;
int32_t tsReplications = TSDB_DEFAULT_DB_REPLICA_OPTION;
int32_t tsQuorum = TSDB_DEFAULT_DB_QUORUM_OPTION;
int16_t tsPartitons = TSDB_DEFAULT_DB_PARTITON_OPTION;
int8_t tsUpdate = TSDB_DEFAULT_DB_UPDATE_OPTION;
int8_t tsCacheLastRow = TSDB_DEFAULT_CACHE_LAST_ROW;
int32_t tsMaxVgroupsPerDb = 0;
int8_t tsCompression = TSDB_DEFAULT_COMP_LEVEL;
int8_t tsWAL = TSDB_DEFAULT_WAL_LEVEL;
int32_t tsFsyncPeriod = TSDB_DEFAULT_FSYNC_PERIOD;
int32_t tsReplications = TSDB_DEFAULT_DB_REPLICA_OPTION;
int32_t tsQuorum = TSDB_DEFAULT_DB_QUORUM_OPTION;
int16_t tsPartitons = TSDB_DEFAULT_DB_PARTITON_OPTION;
int8_t tsUpdate = TSDB_DEFAULT_DB_UPDATE_OPTION;
int8_t tsCacheLastRow = TSDB_DEFAULT_CACHE_LAST_ROW;
int32_t tsMaxVgroupsPerDb = 0;
int32_t tsMinTablePerVnode = TSDB_TABLES_STEP;
int32_t tsMaxTablePerVnode = TSDB_DEFAULT_TABLES;
int32_t tsTableIncStepPerVnode = TSDB_TABLES_STEP;
......@@ -148,7 +148,7 @@ int32_t tsTableIncStepPerVnode = TSDB_TABLES_STEP;
// balance
int8_t tsEnableBalance = 1;
int8_t tsAlternativeRole = 0;
int32_t tsBalanceInterval = 300; // seconds
int32_t tsBalanceInterval = 300; // seconds
int32_t tsOfflineThreshold = 86400 * 10; // seconds of 10 days
int32_t tsMnodeEqualVnodeNum = 4;
int8_t tsEnableFlowCtrl = 1;
......@@ -166,15 +166,16 @@ int8_t tsHttpEnableCompress = 1;
int8_t tsHttpEnableRecordSql = 0;
int8_t tsTelegrafUseFieldNum = 0;
int8_t tsHttpDbNameMandatory = 0;
int32_t tsHttpKeepAlive = 30000;
// mqtt
int8_t tsEnableMqttModule = 0; // not finished yet, not started it by default
char tsMqttHostName[TSDB_MQTT_HOSTNAME_LEN] = "test.mosquitto.org";
char tsMqttPort[TSDB_MQTT_PORT_LEN] = "1883";
char tsMqttUser[TSDB_MQTT_USER_LEN] = {0};
char tsMqttPass[TSDB_MQTT_PASS_LEN] = {0};
char tsMqttClientId[TSDB_MQTT_CLIENT_ID_LEN] = "TDengineMqttSubscriber";
char tsMqttTopic[TSDB_MQTT_TOPIC_LEN] = "/test"; // #
char tsMqttHostName[TSDB_MQTT_HOSTNAME_LEN] = "test.mosquitto.org";
char tsMqttPort[TSDB_MQTT_PORT_LEN] = "1883";
char tsMqttUser[TSDB_MQTT_USER_LEN] = {0};
char tsMqttPass[TSDB_MQTT_PASS_LEN] = {0};
char tsMqttClientId[TSDB_MQTT_CLIENT_ID_LEN] = "TDengineMqttSubscriber";
char tsMqttTopic[TSDB_MQTT_TOPIC_LEN] = "/test"; // #
// monitor
int8_t tsEnableMonitorModule = 1;
......@@ -183,7 +184,7 @@ char tsInternalPass[] = "secretkey";
int32_t tsMonitorInterval = 30; // seconds
// stream
int8_t tsEnableStream = 1;
int8_t tsEnableStream = 1;
// internal
int8_t tsCompactMnodeWal = 0;
......@@ -199,7 +200,7 @@ char tsDataDir[PATH_MAX] = {0};
char tsScriptDir[PATH_MAX] = {0};
char tsTempDir[PATH_MAX] = "/tmp/";
int32_t tsDiskCfgNum = 0;
int32_t tsDiskCfgNum = 0;
int32_t tsTopicBianryLen = 16000;
#ifndef _STORAGE
......@@ -217,42 +218,42 @@ SDiskCfg tsDiskCfg[TSDB_MAX_DISKS];
int64_t tsTickPerDay[] = {86400000L, 86400000000L, 86400000000000L};
// system info
char tsOsName[10] = "Linux";
int64_t tsPageSize;
int64_t tsOpenMax;
int64_t tsStreamMax;
int32_t tsNumOfCores = 1;
float tsTotalTmpDirGB = 0;
float tsTotalDataDirGB = 0;
float tsAvailTmpDirectorySpace = 0;
float tsAvailDataDirGB = 0;
float tsUsedDataDirGB = 0;
float tsReservedTmpDirectorySpace = 1.0f;
float tsMinimalDataDirGB = 2.0f;
int32_t tsTotalMemoryMB = 0;
char tsOsName[10] = "Linux";
int64_t tsPageSize;
int64_t tsOpenMax;
int64_t tsStreamMax;
int32_t tsNumOfCores = 1;
float tsTotalTmpDirGB = 0;
float tsTotalDataDirGB = 0;
float tsAvailTmpDirectorySpace = 0;
float tsAvailDataDirGB = 0;
float tsUsedDataDirGB = 0;
float tsReservedTmpDirectorySpace = 1.0f;
float tsMinimalDataDirGB = 2.0f;
int32_t tsTotalMemoryMB = 0;
uint32_t tsVersion = 0;
// log
int32_t tsNumOfLogLines = 10000000;
int32_t mDebugFlag = 131;
int32_t sdbDebugFlag = 131;
int32_t dDebugFlag = 135;
int32_t vDebugFlag = 135;
int32_t tsNumOfLogLines = 10000000;
int32_t mDebugFlag = 131;
int32_t sdbDebugFlag = 131;
int32_t dDebugFlag = 135;
int32_t vDebugFlag = 135;
uint32_t cDebugFlag = 131;
int32_t jniDebugFlag = 131;
int32_t odbcDebugFlag = 131;
int32_t httpDebugFlag = 131;
int32_t mqttDebugFlag = 131;
int32_t monDebugFlag = 131;
int32_t jniDebugFlag = 131;
int32_t odbcDebugFlag = 131;
int32_t httpDebugFlag = 131;
int32_t mqttDebugFlag = 131;
int32_t monDebugFlag = 131;
uint32_t qDebugFlag = 131;
int32_t rpcDebugFlag = 131;
int32_t uDebugFlag = 131;
int32_t debugFlag = 0;
int32_t sDebugFlag = 135;
int32_t wDebugFlag = 135;
int32_t tsdbDebugFlag = 131;
int32_t cqDebugFlag = 131;
int32_t fsDebugFlag = 135;
int32_t rpcDebugFlag = 131;
int32_t uDebugFlag = 131;
int32_t debugFlag = 0;
int32_t sDebugFlag = 135;
int32_t wDebugFlag = 135;
int32_t tsdbDebugFlag = 131;
int32_t cqDebugFlag = 131;
int32_t fsDebugFlag = 135;
int8_t tsClientMerge = 0;
......@@ -260,13 +261,14 @@ int8_t tsClientMerge = 0;
//
// lossy compress 6
//
char lossyColumns[32] = ""; // "float|double" means all float and double columns can be lossy compressed. set empty can close lossy compress.
// below option can take effect when tsLossyColumns not empty
double fPrecision = 1E-8; // float column precision
double dPrecision = 1E-16; // double column precision
uint32_t maxRange = 500; // max range
uint32_t curRange = 100; // range
char Compressor[32] = "ZSTD_COMPRESSOR"; // ZSTD_COMPRESSOR or GZIP_COMPRESSOR
char lossyColumns[32] = ""; // "float|double" means all float and double columns can be lossy compressed. set empty
// can close lossy compress.
// below option can take effect when tsLossyColumns not empty
double fPrecision = 1E-8; // float column precision
double dPrecision = 1E-16; // double column precision
uint32_t maxRange = 500; // max range
uint32_t curRange = 100; // range
char Compressor[32] = "ZSTD_COMPRESSOR"; // ZSTD_COMPRESSOR or GZIP_COMPRESSOR
#endif
int32_t (*monStartSystemFp)() = NULL;
......@@ -278,7 +280,7 @@ char *qtypeStr[] = {"rpc", "fwd", "wal", "cq", "query"};
static pthread_once_t tsInitGlobalCfgOnce = PTHREAD_ONCE_INIT;
void taosSetAllDebugFlag() {
if (debugFlag != 0) {
if (debugFlag != 0) {
mDebugFlag = debugFlag;
sdbDebugFlag = debugFlag;
dDebugFlag = debugFlag;
......@@ -289,7 +291,7 @@ void taosSetAllDebugFlag() {
httpDebugFlag = debugFlag;
mqttDebugFlag = debugFlag;
monDebugFlag = debugFlag;
qDebugFlag = debugFlag;
qDebugFlag = debugFlag;
rpcDebugFlag = debugFlag;
uDebugFlag = debugFlag;
sDebugFlag = debugFlag;
......@@ -301,12 +303,13 @@ void taosSetAllDebugFlag() {
}
bool taosCfgDynamicOptions(char *msg) {
char *option, *value;
int32_t olen, vlen;
int32_t vint = 0;
char *option, *value;
int32_t olen, vlen;
int32_t vint = 0;
paGetToken(msg, &option, &olen);
if (olen == 0) return false;;
if (olen == 0) return false;
;
paGetToken(option + olen + 1, &value, &vlen);
if (vlen == 0)
......@@ -319,9 +322,9 @@ bool taosCfgDynamicOptions(char *msg) {
for (int32_t i = 0; i < tsGlobalConfigNum; ++i) {
SGlobalCfg *cfg = tsGlobalConfig + i;
//if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_LOG)) continue;
// if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_LOG)) continue;
if (cfg->valType != TAOS_CFG_VTYPE_INT32 && cfg->valType != TAOS_CFG_VTYPE_INT8) continue;
int32_t cfgLen = (int32_t)strlen(cfg->option);
if (cfgLen != olen) continue;
if (strncasecmp(option, cfg->option, olen) != 0) continue;
......@@ -350,7 +353,7 @@ bool taosCfgDynamicOptions(char *msg) {
return true;
}
if (strncasecmp(cfg->option, "debugFlag", olen) == 0) {
taosSetAllDebugFlag();
taosSetAllDebugFlag();
}
return true;
}
......@@ -407,7 +410,7 @@ static void taosCheckDataDirCfg() {
}
static int32_t taosCheckTmpDir(void) {
if (strlen(tsTempDir) <= 0){
if (strlen(tsTempDir) <= 0) {
uError("tempDir is not set");
return -1;
}
......@@ -428,7 +431,7 @@ static void doInitGlobalConfig(void) {
srand(taosSafeRand());
SGlobalCfg cfg = {0};
// ip address
cfg.option = "firstEp";
cfg.ptr = tsFirst;
......@@ -557,12 +560,12 @@ static void doInitGlobalConfig(void) {
cfg.ptr = &tsMaxNumOfDistinctResults;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 10*10000;
cfg.maxValue = 10000*10000;
cfg.minValue = 10 * 10000;
cfg.maxValue = 10000 * 10000;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "numOfMnodes";
cfg.ptr = &tsNumOfMnodes;
cfg.valType = TAOS_CFG_VTYPE_INT32;
......@@ -1149,7 +1152,7 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
// module configs
// module configs
cfg.option = "flowctrl";
cfg.ptr = &tsEnableFlowCtrl;
cfg.valType = TAOS_CFG_VTYPE_INT8;
......@@ -1280,6 +1283,17 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
// pContext in cache
cfg.option = "httpKeepAlive";
cfg.ptr = &tsHttpKeepAlive;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG;
cfg.minValue = 3000;
cfg.maxValue = 3600000;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
// debug flag
cfg.option = "numOfLogLines";
cfg.ptr = &tsNumOfLogLines;
......@@ -1361,7 +1375,6 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "sdbDebugFlag";
cfg.ptr = &sdbDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32;
......@@ -1648,21 +1661,18 @@ static void doInitGlobalConfig(void) {
#else
assert(tsGlobalConfigNum == TSDB_CFG_MAX_NUM - 5);
#endif
}
void taosInitGlobalCfg() {
pthread_once(&tsInitGlobalCfgOnce, doInitGlobalConfig);
}
void taosInitGlobalCfg() { pthread_once(&tsInitGlobalCfgOnce, doInitGlobalConfig); }
int32_t taosCheckGlobalCfg() {
char fqdn[TSDB_FQDN_LEN];
char fqdn[TSDB_FQDN_LEN];
uint16_t port;
if (debugFlag & DEBUG_TRACE || debugFlag & DEBUG_DEBUG || debugFlag & DEBUG_DUMP) {
taosSetAllDebugFlag();
}
if (tsLocalFqdn[0] == 0) {
taosGetFqdn(tsLocalFqdn);
}
......@@ -1689,7 +1699,7 @@ int32_t taosCheckGlobalCfg() {
if (taosCheckTmpDir()) {
return -1;
}
taosGetSystemInfo();
tsSetLocale();
......@@ -1711,8 +1721,8 @@ int32_t taosCheckGlobalCfg() {
}
if (tsMaxTablePerVnode < tsMinTablePerVnode) {
uError("maxTablesPerVnode(%d) < minTablesPerVnode(%d), reset to minTablesPerVnode(%d)",
tsMaxTablePerVnode, tsMinTablePerVnode, tsMinTablePerVnode);
uError("maxTablesPerVnode(%d) < minTablesPerVnode(%d), reset to minTablesPerVnode(%d)", tsMaxTablePerVnode,
tsMinTablePerVnode, tsMinTablePerVnode);
tsMaxTablePerVnode = tsMinTablePerVnode;
}
......@@ -1734,7 +1744,7 @@ int32_t taosCheckGlobalCfg() {
}
tsDnodeShellPort = tsServerPort + TSDB_PORT_DNODESHELL; // udp[6035-6039] tcp[6035]
tsDnodeDnodePort = tsServerPort + TSDB_PORT_DNODEDNODE; // udp/tcp
tsDnodeDnodePort = tsServerPort + TSDB_PORT_DNODEDNODE; // udp/tcp
tsSyncPort = tsServerPort + TSDB_PORT_SYNC;
tsHttpPort = tsServerPort + TSDB_PORT_HTTP;
......@@ -1754,17 +1764,17 @@ int taosGetFqdnPortFromEp(const char *ep, char *fqdn, uint16_t *port) {
strcpy(fqdn, ep);
char *temp = strchr(fqdn, ':');
if (temp) {
if (temp) {
*temp = 0;
*port = atoi(temp+1);
}
*port = atoi(temp + 1);
}
if (*port == 0) {
*port = tsServerPort;
return -1;
}
return 0;
return 0;
}
/*
......
src/connector/jdbc/pom.xml
浏览文件 @ 51259460
......@@ -112,6 +112,7 @@
<include>**/*Test.java</include>
</includes>
<excludes>
<exclude>**/HttpClientPoolUtilTest.java</exclude>
<exclude>**/AppMemoryLeakTest.java</exclude>
<exclude>**/ConnectMultiTaosdByRestfulWithDifferentTokenTest.java</exclude>
<exclude>**/DatetimeBefore1970Test.java</exclude>
......
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBErrorNumbers.java
浏览文件 @ 51259460
......@@ -31,6 +31,8 @@ public class TSDBErrorNumbers {
public static final int ERROR_RESTFul_Client_IOException = 0x2318;
public static final int ERROR_USER_IS_REQUIRED = 0x2319; // user is required
public static final int ERROR_PASSWORD_IS_REQUIRED = 0x231a; // password is required
public static final int ERROR_INVALID_JSON_FORMAT = 0x231b;
public static final int ERROR_HTTP_ENTITY_IS_NULL = 0x231c; //http entity is null
public static final int ERROR_UNKNOWN = 0x2350; //unknown error
......@@ -72,6 +74,8 @@ public class TSDBErrorNumbers {
errorNumbers.add(ERROR_RESTFul_Client_IOException);
errorNumbers.add(ERROR_USER_IS_REQUIRED);
errorNumbers.add(ERROR_PASSWORD_IS_REQUIRED);
errorNumbers.add(ERROR_INVALID_JSON_FORMAT);
errorNumbers.add(ERROR_HTTP_ENTITY_IS_NULL);
errorNumbers.add(ERROR_RESTFul_Client_Protocol_Exception);
......
src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/HttpClientPoolUtil.java
浏览文件 @ 51259460
......@@ -5,6 +5,7 @@ import com.taosdata.jdbc.TSDBErrorNumbers;
import org.apache.http.HeaderElement;
import org.apache.http.HeaderElementIterator;
import org.apache.http.HttpEntity;
import org.apache.http.NoHttpResponseException;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.HttpRequestRetryHandler;
import org.apache.http.client.config.RequestConfig;
......@@ -30,12 +31,12 @@ import java.sql.SQLException;
public class HttpClientPoolUtil {
private static final String DEFAULT_CONTENT_TYPE = "application/json";
private static final int DEFAULT_MAX_TOTAL = 200;
private static final int DEFAULT_MAX_PER_ROUTE = 20;
private static final int DEFAULT_TIME_OUT = 15000;
private static final int DEFAULT_HTTP_KEEP_TIME = 15000;
private static final int DEFAULT_MAX_RETRY_COUNT = 5;
private static final int DEFAULT_MAX_TOTAL = 50;
private static final int DEFAULT_MAX_PER_ROUTE = 5;
private static final int DEFAULT_HTTP_KEEP_TIME = -1;
private static final ConnectionKeepAliveStrategy DEFAULT_KEEP_ALIVE_STRATEGY = (response, context) -> {
HeaderElementIterator it = new BasicHeaderElementIterator(response.headerIterator(HTTP.CONN_KEEP_ALIVE));
while (it.hasNext()) {
......@@ -52,29 +53,19 @@ public class HttpClientPoolUtil {
return DEFAULT_HTTP_KEEP_TIME * 1000;
};
private static final HttpRequestRetryHandler retryHandler = (exception, executionCount, httpContext) -> {
if (executionCount >= DEFAULT_MAX_RETRY_COUNT)
// do not retry if over max retry count
return false;
if (exception instanceof InterruptedIOException)
// timeout
return false;
if (exception instanceof UnknownHostException)
// unknown host
return false;
if (exception instanceof SSLException)
// SSL handshake exception
return false;
return true;
};
private static CloseableHttpClient httpClient;
static {
PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
connectionManager.setMaxTotal(DEFAULT_MAX_TOTAL);
connectionManager.setDefaultMaxPerRoute(DEFAULT_MAX_PER_ROUTE);
httpClient = HttpClients.custom().setKeepAliveStrategy(DEFAULT_KEEP_ALIVE_STRATEGY).setConnectionManager(connectionManager).setRetryHandler(retryHandler).build();
httpClient = HttpClients.custom()
.setKeepAliveStrategy(DEFAULT_KEEP_ALIVE_STRATEGY)
.setConnectionManager(connectionManager)
.setRetryHandler((exception, executionCount, httpContext) -> executionCount < DEFAULT_MAX_RETRY_COUNT)
.build();
}
/*** execute GET request ***/
......@@ -118,9 +109,10 @@ public class HttpClientPoolUtil {
HttpContext context = HttpClientContext.create();
CloseableHttpResponse httpResponse = httpClient.execute(method, context);
httpEntity = httpResponse.getEntity();
if (httpEntity != null) {
responseBody = EntityUtils.toString(httpEntity, StandardCharsets.UTF_8);
if (httpEntity == null) {
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_HTTP_ENTITY_IS_NULL, "httpEntity is null, sql: " + data);
}
responseBody = EntityUtils.toString(httpEntity, StandardCharsets.UTF_8);
} catch (ClientProtocolException e) {
e.printStackTrace();
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_RESTFul_Client_Protocol_Exception, e.getMessage());
......@@ -139,9 +131,6 @@ public class HttpClientPoolUtil {
private static HttpRequestBase getRequest(String uri, String methodName) {
HttpRequestBase method;
RequestConfig requestConfig = RequestConfig.custom()
.setSocketTimeout(DEFAULT_TIME_OUT * 1000)
.setConnectTimeout(DEFAULT_TIME_OUT * 1000)
.setConnectionRequestTimeout(DEFAULT_TIME_OUT * 1000)
.setExpectContinueEnabled(false)
.build();
if (HttpPut.METHOD_NAME.equalsIgnoreCase(methodName)) {
......
src/connector/jdbc/src/test/java/com/taosdata/jdbc/utils/HttpClientPoolUtilTest.java 0 → 100644
浏览文件 @ 51259460
package com.taosdata.jdbc.utils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.taosdata.jdbc.TSDBDriver;
import com.taosdata.jdbc.TSDBError;
import org.junit.Test;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class HttpClientPoolUtilTest {
String user = "root";
String password = "taosdata";
String host = "127.0.0.1";
String dbname = "log";
@Test
public void test() {
// given
List<Thread> threads = IntStream.range(0, 4000).mapToObj(i -> new Thread(() -> {
useDB();
// try {
// TimeUnit.SECONDS.sleep(10);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
})).collect(Collectors.toList());
threads.forEach(Thread::start);
for (Thread thread : threads) {
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private void useDB() {
try {
user = URLEncoder.encode(user, StandardCharsets.UTF_8.displayName());
password = URLEncoder.encode(password, StandardCharsets.UTF_8.displayName());
String loginUrl = "http://" + host + ":" + 6041 + "/rest/login/" + user + "/" + password + "";
String result = HttpClientPoolUtil.execute(loginUrl);
JSONObject jsonResult = JSON.parseObject(result);
String status = jsonResult.getString("status");
String token = jsonResult.getString("desc");
if (!status.equals("succ")) {
throw new SQLException(jsonResult.getString("desc"));
}
String url = "http://" + host + ":6041/rest/sql";
String sql = "use " + dbname;
result = HttpClientPoolUtil.execute(url, sql, token);
JSONObject resultJson = JSON.parseObject(result);
if (resultJson.getString("status").equals("error")) {
throw TSDBError.createSQLException(resultJson.getInteger("code"), resultJson.getString("desc"));
}
} catch (UnsupportedEncodingException | SQLException e) {
e.printStackTrace();
}
}
}
\ No newline at end of file
src/inc/taosmsg.h
浏览文件 @ 51259460
......@@ -400,7 +400,7 @@ typedef struct SColIndex {
int16_t colId; // column id
int16_t colIndex; // column index in colList if it is a normal column or index in tagColList if a tag
uint16_t flag; // denote if it is a tag or a normal column
char name[TSDB_COL_NAME_LEN + TSDB_DB_NAME_LEN + 1];
char name[TSDB_COL_NAME_LEN + TSDB_TABLE_NAME_LEN + 1];
} SColIndex;
typedef struct SColumnFilterInfo {
......
src/inc/ttokendef.h
浏览文件 @ 51259460
......@@ -211,6 +211,7 @@
#define TK_INSERT 193
#define TK_INTO 194
#define TK_VALUES 195
#define TK_FILE 196
#define TK_SPACE 300
......@@ -219,7 +220,6 @@
#define TK_HEX 303 // hex number 0x123
#define TK_OCT 304 // oct number
#define TK_BIN 305 // bin format data 0b111
#define TK_FILE 306
#define TK_QUESTION 307 // denoting the placeholder of "?",when invoking statement bind query
#endif
......
src/kit/shell/src/shellImport.c
浏览文件 @ 51259460
......@@ -25,7 +25,7 @@
static char **shellSQLFiles = NULL;
static int32_t shellSQLFileNum = 0;
static char shellTablesSQLFile[TSDB_FILENAME_LEN] = {0};
static char shellTablesSQLFile[4096] = {0};
typedef struct {
pthread_t threadID;
......
src/kit/taosdemo/taosdemo.c
浏览文件 @ 51259460
此差异已折叠。
src/kit/taosdump/taosdump.c
浏览文件 @ 51259460
......@@ -1972,7 +1972,7 @@ static int getTableDes(
while ((row = taos_fetch_row(res)) != NULL) {
tstrncpy(tableDes->cols[colCount].field,
(char *)row[TSDB_DESCRIBE_METRIC_FIELD_INDEX],
min(TSDB_COL_NAME_LEN + 1,
min(TSDB_COL_NAME_LEN,
fields[TSDB_DESCRIBE_METRIC_FIELD_INDEX].bytes + 1));
tstrncpy(tableDes->cols[colCount].type,
(char *)row[TSDB_DESCRIBE_METRIC_TYPE_INDEX],
......
src/mnode/src/mnodeDb.c
浏览文件 @ 51259460
......@@ -1136,7 +1136,7 @@ static int32_t mnodeAlterDbFp(SMnodeMsg *pMsg) {
return sdbUpdateRow(&row);
}
//bnNotify();
bnNotify();
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}
......
src/mnode/src/mnodeVgroup.c
浏览文件 @ 51259460
......@@ -742,19 +742,6 @@ static int32_t mnodeGetVgroupMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *p
return 0;
}
static bool mnodeFilterVgroups(SVgObj *pVgroup, STableObj *pTable) {
if (NULL == pTable || pTable->type == TSDB_SUPER_TABLE) {
return true;
}
SCTableObj *pCTable = (SCTableObj *)pTable;
if (pVgroup->vgId == pCTable->vgId) {
return true;
} else {
return false;
}
}
static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
int32_t numOfRows = 0;
SVgObj *pVgroup = NULL;
......@@ -770,11 +757,6 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v
return 0;
}
STableObj *pTable = NULL;
if (pShow->payloadLen > 0 ) {
pTable = mnodeGetTable(pShow->payload);
}
while (numOfRows < rows) {
pShow->pIter = mnodeGetNextVgroup(pShow->pIter, &pVgroup);
if (pVgroup == NULL) break;
......@@ -784,11 +766,6 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v
continue;
}
if (!mnodeFilterVgroups(pVgroup, pTable)) {
mnodeDecVgroupRef(pVgroup);
continue;
}
cols = 0;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
......@@ -842,7 +819,6 @@ static int32_t mnodeRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, v
mnodeVacuumResult(data, pShow->numOfColumns, numOfRows, rows, pShow);
pShow->numOfReads += numOfRows;
mnodeDecTableRef(pTable);
mnodeDecDbRef(pDb);
return numOfRows;
......
src/os/src/darwin/dwSysInfo.c
浏览文件 @ 51259460
......@@ -65,6 +65,8 @@ static void taosGetSystemTimezone() {
struct tm tm1;
localtime_r(&tx1, &tm1);
tsDaylight = daylight;
/*
* format example:
*
......
src/os/src/detail/osSysinfo.c
浏览文件 @ 51259460
......@@ -213,6 +213,8 @@ static void taosGetSystemTimezone() {
int32_t tz = (-timezone * MILLISECOND_PER_SECOND) / MILLISECOND_PER_HOUR;
tz += daylight;
tsDaylight = daylight;
/*
* format example:
*
......
src/os/src/windows/wSysinfo.c
浏览文件 @ 51259460
......@@ -91,6 +91,17 @@ static void taosGetSystemTimezone() {
strcpy(tsTimezone, tz);
}
cfg_timezone->cfgStatus = TAOS_CFG_CSTATUS_DEFAULT;
#ifdef _MSC_VER
#if _MSC_VER >= 1900
int64_t timezone = _timezone;
int32_t daylight = _daylight;
char **tzname = _tzname;
#endif
#endif
tsDaylight = daylight;
uInfo("timezone not configured, use default");
}
}
......
src/plugins/http/inc/httpInt.h
浏览文件 @ 51259460
......@@ -140,28 +140,29 @@ typedef enum {
} EHTTP_CONTEXT_FAILED_CAUSE;
typedef struct HttpContext {
int32_t refCount;
SOCKET fd;
uint32_t accessTimes;
uint32_t lastAccessTime;
int32_t state;
uint8_t reqType;
uint8_t parsed;
uint8_t error;
char ipstr[22];
char user[TSDB_USER_LEN]; // parsed from auth token or login message
char pass[HTTP_PASSWORD_LEN];
char db[/*TSDB_ACCT_ID_LEN + */TSDB_DB_NAME_LEN];
TAOS * taos;
void * ppContext;
HttpSession *session;
z_stream gzipStream;
HttpParser *parser;
HttpSqlCmd singleCmd;
HttpSqlCmds *multiCmds;
JsonBuf * jsonBuf;
HttpEncodeMethod *encodeMethod;
HttpDecodeMethod *decodeMethod;
int32_t refCount;
SOCKET fd;
uint32_t accessTimes;
uint32_t lastAccessTime;
int32_t state;
uint8_t reqType;
uint8_t parsed;
uint8_t error;
char ipstr[22];
char user[TSDB_USER_LEN]; // parsed from auth token or login message
char pass[HTTP_PASSWORD_LEN];
char db[/*TSDB_ACCT_ID_LEN + */TSDB_DB_NAME_LEN];
TAOS * taos;
void * ppContext;
pthread_mutex_t ctxMutex;
HttpSession *session;
z_stream gzipStream;
HttpParser *parser;
HttpSqlCmd singleCmd;
HttpSqlCmds *multiCmds;
JsonBuf *jsonBuf;
HttpEncodeMethod *encodeMethod;
HttpDecodeMethod *decodeMethod;
struct HttpThread *pThread;
} HttpContext;
......
src/plugins/http/src/httpContext.c
浏览文件 @ 51259460
......@@ -67,6 +67,8 @@ static void httpDestroyContext(void *data) {
pContext->parser = NULL;
}
pthread_mutex_destroy(&pContext->ctxMutex);
tfree(pContext);
}
......@@ -121,13 +123,15 @@ HttpContext *httpCreateContext(SOCKET fd) {
TSDB_CACHE_PTR_TYPE handleVal = (TSDB_CACHE_PTR_TYPE)pContext;
HttpContext **ppContext = taosCachePut(tsHttpServer.contextCache, &handleVal, sizeof(TSDB_CACHE_PTR_TYPE), &pContext,
sizeof(TSDB_CACHE_PTR_TYPE), 3000);
sizeof(TSDB_CACHE_PTR_TYPE), tsHttpKeepAlive);
pContext->ppContext = ppContext;
httpDebug("context:%p, fd:%d, is created, data:%p", pContext, fd, ppContext);
// set the ref to 0
taosCacheRelease(tsHttpServer.contextCache, (void **)&ppContext, false);
pthread_mutex_init(&pContext->ctxMutex, NULL);
return pContext;
}
......
src/plugins/http/src/httpHandle.c
浏览文件 @ 51259460
......@@ -45,15 +45,14 @@ bool httpProcessData(HttpContext* pContext) {
httpTrace("context:%p, fd:%d, process options request", pContext, pContext->fd);
httpSendOptionResp(pContext, "process options request success");
} else {
if (!httpDecodeRequest(pContext)) {
/*
* httpCloseContextByApp has been called when parsing the error
*/
// httpCloseContextByApp(pContext);
} else {
pthread_mutex_lock(&pContext->ctxMutex);
if (httpDecodeRequest(pContext)) {
httpClearParser(pContext->parser);
httpProcessRequest(pContext);
}
pthread_mutex_unlock(&pContext->ctxMutex);
}
return true;
......
src/plugins/http/src/httpSql.c
浏览文件 @ 51259460
......@@ -406,7 +406,14 @@ void httpProcessRequestCb(void *param, TAOS_RES *result, int32_t code) {
if (pContext->session == NULL) {
httpSendErrorResp(pContext, TSDB_CODE_HTTP_SESSION_FULL);
} else {
// httpProcessRequestCb called by another thread and a subsequent thread calls this
// function again, if this function called by httpProcessRequestCb executes memset
// just before the subsequent thread executes *Cmd function, nativSql will be NULL
pthread_mutex_lock(&pContext->ctxMutex);
httpExecCmd(pContext);
pthread_mutex_unlock(&pContext->ctxMutex);
}
}
......
src/query/inc/qExecutor.h
浏览文件 @ 51259460
......@@ -43,6 +43,8 @@ typedef int32_t (*__block_search_fn_t)(char* data, int32_t num, int64_t key, int
#define GET_NUM_OF_RESULTS(_r) (((_r)->outputBuf) == NULL? 0:((_r)->outputBuf)->info.rows)
#define RESET_NUM_OF_RESULTS(_r) (((_r)->outputBuf) == NULL? 0:(((_r)->outputBuf)->info.rows = 0))
enum {
// when query starts to execute, this status will set
QUERY_NOT_COMPLETED = 0x1u,
......@@ -389,6 +391,7 @@ typedef struct SQInfo {
int32_t dataReady; // denote if query result is ready or not
void* rspContext; // response context
int64_t startExecTs; // start to exec timestamp
int64_t lastRetrieveTs; // last retrieve timestamp
char* sql; // query sql string
SQueryCostInfo summary;
} SQInfo;
......
src/query/inc/sql.y
浏览文件 @ 51259460
......@@ -128,12 +128,6 @@ cmd ::= SHOW dbPrefix(X) VGROUPS. {
setShowOptions(pInfo, TSDB_MGMT_TABLE_VGROUP, &token, 0);
}
cmd ::= SHOW dbPrefix(X) VGROUPS ids(Y). {
SStrToken token;
tSetDbName(&token, &X);
setShowOptions(pInfo, TSDB_MGMT_TABLE_VGROUP, &token, &Y);
}
//drop configure for tables
cmd ::= DROP TABLE ifexists(Y) ids(X) cpxName(Z). {
X.n += Z.n;
......@@ -919,4 +913,4 @@ cmd ::= KILL QUERY INTEGER(X) COLON(Z) INTEGER(Y). {X.n += (Z.n + Y.n); s
%fallback ID ABORT AFTER ASC ATTACH BEFORE BEGIN CASCADE CLUSTER CONFLICT COPY DATABASE DEFERRED
DELIMITERS DESC DETACH EACH END EXPLAIN FAIL FOR GLOB IGNORE IMMEDIATE INITIALLY INSTEAD
LIKE MATCH KEY OF OFFSET RAISE REPLACE RESTRICT ROW STATEMENT TRIGGER VIEW ALL
NOW IPTOKEN SEMI NONE PREV LINEAR IMPORT TBNAME JOIN STABLE NULL INSERT INTO VALUES.
NOW IPTOKEN SEMI NONE PREV LINEAR IMPORT TBNAME JOIN STABLE NULL INSERT INTO VALUES FILE.
src/query/src/qAggMain.c
浏览文件 @ 51259460
......@@ -758,7 +758,7 @@ static int32_t lastDistFuncRequired(SQLFunctionCtx *pCtx, STimeWindow* w, int32_
if (pInfo->hasResult != DATA_SET_FLAG) {
return BLK_DATA_ALL_NEEDED;
} else {
return (pInfo->ts > w->ekey) ? BLK_DATA_NO_NEEDED : BLK_DATA_ALL_NEEDED;
return (pInfo->ts >= w->ekey) ? BLK_DATA_NO_NEEDED : BLK_DATA_ALL_NEEDED;
}
}
......
src/query/src/qExecutor.c
浏览文件 @ 51259460
......@@ -2430,11 +2430,11 @@ bool isQueryKilled(SQInfo *pQInfo) {
// query has been executed more than tsShellActivityTimer, and the retrieve has not arrived
// abort current query execution.
if (pQInfo->owner != 0 && ((taosGetTimestampSec() - pQInfo->startExecTs) > getMaximumIdleDurationSec()) &&
if (pQInfo->owner != 0 && ((taosGetTimestampSec() - pQInfo->lastRetrieveTs/1000) > getMaximumIdleDurationSec()) &&
(!needBuildResAfterQueryComplete(pQInfo))) {
assert(pQInfo->startExecTs != 0);
qDebug("QInfo:%" PRIu64 " retrieve not arrive beyond %d sec, abort current query execution, start:%" PRId64
qDebug("QInfo:%" PRIu64 " retrieve not arrive beyond %d ms, abort current query execution, start:%" PRId64
", current:%d", pQInfo->qId, 1, pQInfo->startExecTs, taosGetTimestampSec());
return true;
}
......
src/query/src/queryMain.c
浏览文件 @ 51259460
......@@ -213,7 +213,10 @@ bool qTableQuery(qinfo_t qinfo, uint64_t *qId) {
}
*qId = pQInfo->qId;
pQInfo->startExecTs = taosGetTimestampSec();
if(pQInfo->startExecTs == 0) {
pQInfo->startExecTs = taosGetTimestampMs();
pQInfo->lastRetrieveTs = pQInfo->startExecTs;
}
if (isQueryKilled(pQInfo)) {
qDebug("QInfo:0x%"PRIx64" it is already killed, abort", pQInfo->qId);
......@@ -343,6 +346,9 @@ int32_t qDumpRetrieveResult(qinfo_t qinfo, SRetrieveTableRsp **pRsp, int32_t *co
setQueryStatus(pRuntimeEnv, QUERY_OVER);
}
RESET_NUM_OF_RESULTS(&(pQInfo->runtimeEnv));
pQInfo->lastRetrieveTs = taosGetTimestampMs();
pQInfo->rspContext = NULL;
pQInfo->dataReady = QUERY_RESULT_NOT_READY;
......
src/query/src/sql.c
浏览文件 @ 51259460
此差异已折叠。
src/sync/src/syncRetrieve.c
浏览文件 @ 51259460
......@@ -233,7 +233,7 @@ static int64_t syncProcessLastWal(SSyncPeer *pPeer, char *wname, int64_t index)
int32_t once = 0; // last WAL has once ever been processed
int64_t offset = 0;
uint64_t fversion = 0;
char fname[TSDB_FILENAME_LEN * 2] = {0}; // full path to wal file
char fname[TSDB_FILENAME_LEN * 3] = {0}; // full path to wal file
// get full path to wal file
snprintf(fname, sizeof(fname), "%s/%s", pNode->path, wname);
......
src/tsdb/inc/tsdbMeta.h
浏览文件 @ 51259460
......@@ -41,6 +41,7 @@ typedef struct STable {
int16_t restoreColumnNum;
bool hasRestoreLastColumn;
int lastColSVersion;
int16_t cacheLastConfigVersion;
T_REF_DECLARE()
} STable;
......
src/tsdb/inc/tsdbint.h
浏览文件 @ 51259460
......@@ -79,8 +79,8 @@ struct STsdbRepo {
STsdbCfg save_config; // save apply config
bool config_changed; // config changed flag
pthread_mutex_t save_mutex; // protect save config
uint8_t hasCachedLastColumn;
int16_t cacheLastConfigVersion;
STsdbAppH appH;
STsdbStat stat;
......@@ -110,7 +110,8 @@ int tsdbUnlockRepo(STsdbRepo* pRepo);
STsdbMeta* tsdbGetMeta(STsdbRepo* pRepo);
int tsdbCheckCommit(STsdbRepo* pRepo);
int tsdbRestoreInfo(STsdbRepo* pRepo);
int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg);
UNUSED_FUNC int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg);
int32_t tsdbLoadLastCache(STsdbRepo *pRepo, STable* pTable);
void tsdbGetRootDir(int repoid, char dirName[]);
void tsdbGetDataDir(int repoid, char dirName[]);
......
src/tsdb/src/tsdbCommit.c
浏览文件 @ 51259460
......@@ -223,7 +223,7 @@ int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf) {
SBlockIdx *pBlkIdx;
size_t nidx = taosArrayGetSize(pIdxA);
int tlen = 0, size;
int64_t offset;
int64_t offset = 0;
if (nidx <= 0) {
// All data are deleted
......
src/tsdb/src/tsdbCommitQueue.c
浏览文件 @ 51259460
......@@ -146,7 +146,9 @@ static void tsdbApplyRepoConfig(STsdbRepo *pRepo) {
if (oldCfg.cacheLastRow != pRepo->config.cacheLastRow) {
if (tsdbLockRepo(pRepo) < 0) return;
tsdbCacheLastData(pRepo, &oldCfg);
// tsdbCacheLastData(pRepo, &oldCfg);
// lazy load last cache when query or update
++pRepo->cacheLastConfigVersion;
tsdbUnlockRepo(pRepo);
}
......
src/tsdb/src/tsdbCompact.c
浏览文件 @ 51259460
......@@ -64,7 +64,9 @@ int tsdbCompact(STsdbRepo *pRepo) { return tsdbAsyncCompact(pRepo); }
void *tsdbCompactImpl(STsdbRepo *pRepo) {
// Check if there are files in TSDB FS to compact
if (REPO_FS(pRepo)->cstatus->pmf == NULL) {
tsdbInfo("vgId:%d no file to compact in FS", REPO_ID(pRepo));
pRepo->compactState = TSDB_NO_COMPACT;
tsem_post(&(pRepo->readyToCommit));
tsdbInfo("vgId:%d compact over, no file to compact in FS", REPO_ID(pRepo));
return NULL;
}
......
src/tsdb/src/tsdbMain.c
浏览文件 @ 51259460
......@@ -562,7 +562,7 @@ static STsdbRepo *tsdbNewRepo(STsdbCfg *pCfg, STsdbAppH *pAppH) {
return NULL;
}
pRepo->config_changed = false;
atomic_store_8(&pRepo->hasCachedLastColumn, 0);
pRepo->cacheLastConfigVersion = 0;
code = tsem_init(&(pRepo->readyToCommit), 0, 1);
if (code != 0) {
......@@ -711,7 +711,7 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
tsdbGetBlockStatis(pReadh, pBlockStatis, (int)numColumns);
loadStatisData = true;
}
TSDB_WLOCK_TABLE(pTable); // lock when update pTable->lastCols[]
for (int16_t i = 0; i < numColumns && numColumns > pTable->restoreColumnNum; ++i) {
STColumn *pCol = schemaColAt(pSchema, i);
// ignore loaded columns
......@@ -760,6 +760,7 @@ static int tsdbRestoreLastColumns(STsdbRepo *pRepo, STable *pTable, SReadH* pRea
break;
}
}
TSDB_WUNLOCK_TABLE(pTable);
}
out:
......@@ -774,7 +775,6 @@ out:
}
static int tsdbRestoreLastRow(STsdbRepo *pRepo, STable *pTable, SReadH* pReadh, SBlockIdx *pIdx) {
ASSERT(pTable->lastRow == NULL);
if (tsdbLoadBlockInfo(pReadh, NULL) < 0) {
return -1;
}
......@@ -788,21 +788,32 @@ static int tsdbRestoreLastRow(STsdbRepo *pRepo, STable *pTable, SReadH* pReadh,
// Get the data in row
STSchema *pSchema = tsdbGetTableSchema(pTable);
pTable->lastRow = taosTMalloc(memRowMaxBytesFromSchema(pSchema));
if (pTable->lastRow == NULL) {
SMemRow lastRow = taosTMalloc(memRowMaxBytesFromSchema(pSchema));
if (lastRow == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
memRowSetType(pTable->lastRow, SMEM_ROW_DATA);
tdInitDataRow(memRowDataBody(pTable->lastRow), pSchema);
memRowSetType(lastRow, SMEM_ROW_DATA);
tdInitDataRow(memRowDataBody(lastRow), pSchema);
for (int icol = 0; icol < schemaNCols(pSchema); icol++) {
STColumn *pCol = schemaColAt(pSchema, icol);
SDataCol *pDataCol = pReadh->pDCols[0]->cols + icol;
tdAppendColVal(memRowDataBody(pTable->lastRow), tdGetColDataOfRow(pDataCol, pBlock->numOfRows - 1), pCol->type,
tdAppendColVal(memRowDataBody(lastRow), tdGetColDataOfRow(pDataCol, pBlock->numOfRows - 1), pCol->type,
pCol->offset);
}
pTable->lastKey = memRowKey(pTable->lastRow);
TSKEY lastKey = memRowKey(lastRow);
// during the load data in file, new data would be inserted and last row has been updated
TSDB_WLOCK_TABLE(pTable);
if (pTable->lastRow == NULL) {
pTable->lastKey = lastKey;
pTable->lastRow = lastRow;
TSDB_WUNLOCK_TABLE(pTable);
} else {
TSDB_WUNLOCK_TABLE(pTable);
taosTZfree(lastRow);
}
return 0;
}
......@@ -874,14 +885,105 @@ int tsdbRestoreInfo(STsdbRepo *pRepo) {
tsdbDestroyReadH(&readh);
if (CACHE_LAST_NULL_COLUMN(pCfg)) {
atomic_store_8(&pRepo->hasCachedLastColumn, 1);
// if (CACHE_LAST_NULL_COLUMN(pCfg)) {
// atomic_store_8(&pRepo->hasCachedLastColumn, 1);
// }
return 0;
}
int32_t tsdbLoadLastCache(STsdbRepo *pRepo, STable *pTable) {
SFSIter fsiter;
SReadH readh;
SDFileSet *pSet;
int cacheLastRowTableNum = 0;
int cacheLastColTableNum = 0;
bool cacheLastRow = CACHE_LAST_ROW(&(pRepo->config));
bool cacheLastCol = CACHE_LAST_NULL_COLUMN(&(pRepo->config));
tsdbDebug("tsdbLoadLastCache for %s, cacheLastRow:%d, cacheLastCol:%d", pTable->name->data, cacheLastRow, cacheLastCol);
pTable->cacheLastConfigVersion = pRepo->cacheLastConfigVersion;
if (!cacheLastRow && pTable->lastRow != NULL) {
taosTZfree(pTable->lastRow);
pTable->lastRow = NULL;
}
if (!cacheLastCol && pTable->lastCols != NULL) {
tsdbFreeLastColumns(pTable);
}
if (!cacheLastRow && !cacheLastCol) {
return 0;
}
cacheLastRowTableNum = (cacheLastRow && pTable->lastRow == NULL) ? 1 : 0;
cacheLastColTableNum = (cacheLastCol && pTable->lastCols == NULL) ? 1 : 0;
if (cacheLastRowTableNum == 0 && cacheLastColTableNum == 0) {
return 0;
}
if (tsdbInitReadH(&readh, pRepo) < 0) {
return -1;
}
tsdbRLockFS(REPO_FS(pRepo));
tsdbFSIterInit(&fsiter, REPO_FS(pRepo), TSDB_FS_ITER_BACKWARD);
while ((cacheLastRowTableNum > 0 || cacheLastColTableNum > 0) && (pSet = tsdbFSIterNext(&fsiter)) != NULL) {
if (tsdbSetAndOpenReadFSet(&readh, pSet) < 0) {
tsdbUnLockFS(REPO_FS(pRepo));
tsdbDestroyReadH(&readh);
return -1;
}
if (tsdbLoadBlockIdx(&readh) < 0) {
tsdbUnLockFS(REPO_FS(pRepo));
tsdbDestroyReadH(&readh);
return -1;
}
// tsdbDebug("tsdbRestoreInfo restore vgId:%d,table:%s", REPO_ID(pRepo), pTable->name->data);
if (tsdbSetReadTable(&readh, pTable) < 0) {
tsdbUnLockFS(REPO_FS(pRepo));
tsdbDestroyReadH(&readh);
return -1;
}
SBlockIdx *pIdx = readh.pBlkIdx;
if (pIdx && (cacheLastRowTableNum > 0) && (pTable->lastRow == NULL)) {
if (tsdbRestoreLastRow(pRepo, pTable, &readh, pIdx) != 0) {
tsdbUnLockFS(REPO_FS(pRepo));
tsdbDestroyReadH(&readh);
return -1;
}
cacheLastRowTableNum -= 1;
}
// restore NULL columns
if (pIdx && (cacheLastColTableNum > 0) && !pTable->hasRestoreLastColumn) {
if (tsdbRestoreLastColumns(pRepo, pTable, &readh) != 0) {
tsdbUnLockFS(REPO_FS(pRepo));
tsdbDestroyReadH(&readh);
return -1;
}
if (pTable->hasRestoreLastColumn) {
cacheLastColTableNum -= 1;
}
}
}
tsdbUnLockFS(REPO_FS(pRepo));
tsdbDestroyReadH(&readh);
return 0;
}
int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
UNUSED_FUNC int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
bool cacheLastRow = false, cacheLastCol = false;
SFSIter fsiter;
SReadH readh;
......@@ -915,9 +1017,9 @@ int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
// if close last option,need to free data
if (need_free_last_row || need_free_last_col) {
if (need_free_last_col) {
atomic_store_8(&pRepo->hasCachedLastColumn, 0);
}
// if (need_free_last_col) {
// atomic_store_8(&pRepo->hasCachedLastColumn, 0);
// }
tsdbInfo("free cache last data since cacheLast option changed");
for (int i = 1; i <= maxTableIdx; i++) {
STable *pTable = pMeta->tables[i];
......@@ -995,9 +1097,9 @@ int tsdbCacheLastData(STsdbRepo *pRepo, STsdbCfg* oldCfg) {
tsdbDestroyReadH(&readh);
if (cacheLastCol) {
atomic_store_8(&pRepo->hasCachedLastColumn, 1);
}
// if (cacheLastCol) {
// atomic_store_8(&pRepo->hasCachedLastColumn, 1);
// }
return 0;
}
src/tsdb/src/tsdbMemTable.c
浏览文件 @ 51259460
......@@ -996,7 +996,8 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SMemRow ro
if ((value == NULL) || isNull(value, pTCol->type)) {
continue;
}
// lock
TSDB_WLOCK_TABLE(pTable);
SDataCol *pDataCol = &(pLatestCols[idx]);
if (pDataCol->pData == NULL) {
pDataCol->pData = malloc(pTCol->bytes);
......@@ -1012,6 +1013,8 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SMemRow ro
memcpy(pDataCol->pData, value, bytes);
//tsdbInfo("updateTableLatestColumn vgId:%d cache column %d for %d,%s", REPO_ID(pRepo), j, pDataCol->bytes, (char*)pDataCol->pData);
pDataCol->ts = memRowKey(row);
// unlock
TSDB_WUNLOCK_TABLE(pTable);
}
}
......@@ -1058,5 +1061,8 @@ static int tsdbUpdateTableLatestInfo(STsdbRepo *pRepo, STable *pTable, SMemRow r
updateTableLatestColumn(pRepo, pTable, row);
}
}
pTable->cacheLastConfigVersion = pRepo->cacheLastConfigVersion;
return 0;
}
src/tsdb/src/tsdbMeta.c
浏览文件 @ 51259460
......@@ -346,7 +346,7 @@ int tsdbUpdateTableTagValue(STsdbRepo *repo, SUpdateTableTagValMsg *pMsg) {
tsdbError(
"vgId:%d failed to update tag value of table %s since version out of date, client tag version %d server tag "
"version %d",
REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), pMsg->tversion, schemaVersion(pTable->tagSchema));
REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), pMsg->tversion, schemaVersion(pTable->pSuper->tagSchema));
terrno = TSDB_CODE_TDB_TAG_VER_OUT_OF_DATE;
return -1;
}
......@@ -627,27 +627,30 @@ int16_t tsdbGetLastColumnsIndexByColId(STable* pTable, int16_t colId) {
}
int tsdbInitColIdCacheWithSchema(STable* pTable, STSchema* pSchema) {
ASSERT(pTable->lastCols == NULL);
TSDB_WLOCK_TABLE(pTable);
if (pTable->lastCols == NULL) {
int16_t numOfColumn = pSchema->numOfCols;
int16_t numOfColumn = pSchema->numOfCols;
pTable->lastCols = (SDataCol *)malloc(numOfColumn * sizeof(SDataCol));
if (pTable->lastCols == NULL) {
TSDB_WUNLOCK_TABLE(pTable);
return -1;
}
pTable->lastCols = (SDataCol*)malloc(numOfColumn * sizeof(SDataCol));
if (pTable->lastCols == NULL) {
return -1;
}
for (int16_t i = 0; i < numOfColumn; ++i) {
STColumn *pCol = schemaColAt(pSchema, i);
SDataCol *pDataCol = &(pTable->lastCols[i]);
pDataCol->bytes = 0;
pDataCol->pData = NULL;
pDataCol->colId = pCol->colId;
}
for (int16_t i = 0; i < numOfColumn; ++i) {
STColumn *pCol = schemaColAt(pSchema, i);
SDataCol* pDataCol = &(pTable->lastCols[i]);
pDataCol->bytes = 0;
pDataCol->pData = NULL;
pDataCol->colId = pCol->colId;
pTable->lastColSVersion = schemaVersion(pSchema);
pTable->maxColNum = numOfColumn;
pTable->restoreColumnNum = 0;
pTable->hasRestoreLastColumn = false;
}
pTable->lastColSVersion = schemaVersion(pSchema);
pTable->maxColNum = numOfColumn;
pTable->restoreColumnNum = 0;
pTable->hasRestoreLastColumn = false;
TSDB_WUNLOCK_TABLE(pTable);
return 0;
}
......@@ -797,6 +800,7 @@ static STable *tsdbNewTable() {
pTable->lastCols = NULL;
pTable->restoreColumnNum = 0;
pTable->cacheLastConfigVersion = 0;
pTable->maxColNum = 0;
pTable->hasRestoreLastColumn = false;
pTable->lastColSVersion = -1;
......
src/tsdb/src/tsdbRead.c
浏览文件 @ 51259460
......@@ -156,6 +156,7 @@ typedef struct STableGroupSupporter {
static STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList);
static int32_t checkForCachedLastRow(STsdbQueryHandle* pQueryHandle, STableGroupInfo *groupList);
static int32_t checkForCachedLast(STsdbQueryHandle* pQueryHandle);
static int32_t lazyLoadCacheLast(STsdbQueryHandle* pQueryHandle);
static int32_t tsdbGetCachedLastRow(STable* pTable, SMemRow* pRes, TSKEY* lastKey);
static void changeQueryHandleForInterpQuery(TsdbQueryHandleT pHandle);
......@@ -589,6 +590,28 @@ void tsdbResetQueryHandleForNewTable(TsdbQueryHandleT queryHandle, STsdbQueryCon
pQueryHandle->next = doFreeColumnInfoData(pQueryHandle->next);
}
static int32_t lazyLoadCacheLast(STsdbQueryHandle* pQueryHandle) {
STsdbRepo* pRepo = pQueryHandle->pTsdb;
size_t numOfTables = taosArrayGetSize(pQueryHandle->pTableCheckInfo);
int32_t code = 0;
for (size_t i = 0; i < numOfTables; ++i) {
STableCheckInfo* pCheckInfo = taosArrayGet(pQueryHandle->pTableCheckInfo, i);
STable* pTable = pCheckInfo->pTableObj;
if (pTable->cacheLastConfigVersion == pRepo->cacheLastConfigVersion) {
continue;
}
code = tsdbLoadLastCache(pRepo, pTable);
if (code != 0) {
tsdbError("%p uid:%" PRId64 ", tid:%d, failed to load last cache since %s", pQueryHandle, pTable->tableId.uid,
pTable->tableId.tid, tstrerror(terrno));
break;
}
}
return code;
}
TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, SMemRef* pMemRef) {
pCond->twindow = updateLastrowForEachGroup(groupList);
......@@ -602,6 +625,8 @@ TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STable
return NULL;
}
lazyLoadCacheLast(pQueryHandle);
int32_t code = checkForCachedLastRow(pQueryHandle, groupList);
if (code != TSDB_CODE_SUCCESS) { // set the numOfTables to be 0
terrno = code;
......@@ -616,13 +641,14 @@ TsdbQueryHandleT tsdbQueryLastRow(STsdbRepo *tsdb, STsdbQueryCond *pCond, STable
return pQueryHandle;
}
TsdbQueryHandleT tsdbQueryCacheLast(STsdbRepo *tsdb, STsdbQueryCond *pCond, STableGroupInfo *groupList, uint64_t qId, SMemRef* pMemRef) {
STsdbQueryHandle *pQueryHandle = (STsdbQueryHandle*) tsdbQueryTables(tsdb, pCond, groupList, qId, pMemRef);
if (pQueryHandle == NULL) {
return NULL;
}
lazyLoadCacheLast(pQueryHandle);
int32_t code = checkForCachedLast(pQueryHandle);
if (code != TSDB_CODE_SUCCESS) { // set the numOfTables to be 0
terrno = code;
......@@ -1526,7 +1552,7 @@ static void mergeTwoRowFromMem(STsdbQueryHandle* pQueryHandle, int32_t capacity,
int16_t offset;
bool isRow1DataRow = isDataRow(row1);
bool isRow2DataRow;
bool isRow2DataRow = false;
bool isChosenRowDataRow;
int32_t chosen_itr;
void *value;
......@@ -2781,6 +2807,9 @@ static bool loadCachedLast(STsdbQueryHandle* pQueryHandle) {
}
int32_t i = 0, j = 0;
// lock pTable->lastCols[i] as it would be released when schema update(tsdbUpdateLastColSchema)
TSDB_RLOCK_TABLE(pTable);
while(i < tgNumOfCols && j < numOfCols) {
pColInfo = taosArrayGet(pQueryHandle->pColumns, i);
if (pTable->lastCols[j].colId < pColInfo->info.colId) {
......@@ -2867,6 +2896,7 @@ static bool loadCachedLast(STsdbQueryHandle* pQueryHandle) {
i++;
j++;
}
TSDB_RUNLOCK_TABLE(pTable);
// leave the real ts column as the last row, because last function only (not stable) use the last row as res
if (priKey != TSKEY_INITIAL_VAL) {
......@@ -3198,7 +3228,9 @@ int32_t checkForCachedLast(STsdbQueryHandle* pQueryHandle) {
int32_t code = 0;
if (pQueryHandle->pTsdb && atomic_load_8(&pQueryHandle->pTsdb->hasCachedLastColumn)){
STsdbRepo* pRepo = pQueryHandle->pTsdb;
if (pRepo && CACHE_LAST_NULL_COLUMN(&(pRepo->config))) {
pQueryHandle->cachelastrow = TSDB_CACHED_TYPE_LAST;
}
......@@ -3621,6 +3653,9 @@ static bool tableFilterFp(const void* pNode, void* param) {
return (val != NULL) && (!isNull(val, pInfo->sch.type));
}
} else if (pInfo->optr == TSDB_RELATION_IN) {
if(val == NULL) {
return false;
}
int type = pInfo->sch.type;
if (type == TSDB_DATA_TYPE_BOOL || IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_TIMESTAMP) {
int64_t v;
......
src/util/inc/tconfig.h
浏览文件 @ 51259460
......@@ -20,7 +20,7 @@
extern "C" {
#endif
#define TSDB_CFG_MAX_NUM 120
#define TSDB_CFG_MAX_NUM 121
#define TSDB_CFG_PRINT_LEN 23
#define TSDB_CFG_OPTION_LEN 24
#define TSDB_CFG_VALUE_LEN 41
......
src/util/src/talgo.c
浏览文件 @ 51259460
......@@ -230,7 +230,7 @@ void taosheapadjust(void *base, int32_t size, int32_t start, int32_t end, const
{
int32_t parent;
int32_t child;
char *buf;
char *buf = NULL;
if (base && size > 0 && compar) {
parent = start;
......
src/util/src/tcompare.c
浏览文件 @ 51259460
......@@ -357,14 +357,20 @@ int32_t compareFindItemInSet(const void *pLeft, const void* pRight) {
int32_t compareWStrPatternComp(const void* pLeft, const void* pRight) {
SPatternCompareInfo pInfo = {'%', '_'};
size_t size = varDataLen(pLeft)/TSDB_NCHAR_SIZE;
assert(varDataLen(pRight) <= TSDB_MAX_FIELD_LEN * TSDB_NCHAR_SIZE);
wchar_t *pattern = calloc(varDataLen(pRight) + 1, sizeof(wchar_t));
wchar_t *str = calloc(size + 1, sizeof(wchar_t));
memcpy(pattern, varDataVal(pRight), varDataLen(pRight));
memcpy(str, varDataVal(pLeft), size * sizeof(wchar_t));
int32_t ret = WCSPatternMatch(pattern, str, size, &pInfo);
int32_t ret = WCSPatternMatch(pattern, varDataVal(pLeft), varDataLen(pLeft)/TSDB_NCHAR_SIZE, &pInfo);
free(pattern);
free(str);
return (ret == TSDB_PATTERN_MATCH) ? 0 : 1;
}
......
src/util/src/tlog.c
浏览文件 @ 51259460
......@@ -566,7 +566,7 @@ static int32_t taosPushLogBuffer(SLogBuff *tLogBuff, char *msg, int32_t msgLen)
int32_t end = 0;
int32_t remainSize = 0;
static int64_t lostLine = 0;
char tmpBuf[40] = {0};
char tmpBuf[60] = {0};
int32_t tmpBufLen = 0;
if (tLogBuff == NULL || tLogBuff->stop) return -1;
......
tests/pytest/fulltest.sh
浏览文件 @ 51259460
......@@ -269,6 +269,7 @@ python3 ./test.py -f query/queryCnameDisplay.py
python3 ./test.py -f query/operator_cost.py
python3 test.py -f query/nestedQuery/queryWithSpread.py
python3 ./test.py -f query/bug6586.py
python3 ./test.py -f query/bug5903.py
#stream
python3 ./test.py -f stream/metric_1.py
......
tests/pytest/query/bug5903.py 0 → 100644
浏览文件 @ 51259460
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
from util.log import *
from util.cases import *
from util.sql import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
# TD-5903 show db.vgroups xxx. xxx is invalid content, but still returns results.
tdSql.execute("create database if not exists test_5903")
tdSql.execute("show test_5903.vgroups")
tdSql.error("show test_5903.vgroups xxx")
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())
tests/pytest/query/query.py
浏览文件 @ 51259460
......@@ -148,6 +148,22 @@ class TDTestCase:
tdLog.info("case for bug_6387")
self.bug_6387()
#JIRA TS-583
tdLog.info("case for JIRA TS-583")
tdSql.execute("create database test2")
tdSql.execute("use test2")
tdSql.execute("create table stb(ts timestamp, c1 int) tags(t1 binary(120))")
tdSql.execute("create table t0 using stb tags('abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz')")
tdSql.query("show create table t0")
tdSql.checkRows(1)
tdSql.execute("create table stb2(ts timestamp, c1 int) tags(t1 nchar(120))")
tdSql.execute("create table t1 using stb2 tags('abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz')")
tdSql.query("show create table t1")
tdSql.checkRows(1)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
......
tests/pytest/query/udf.py
浏览文件 @ 51259460
此差异已折叠。
tests/pytest/table/create.py
浏览文件 @ 51259460
......@@ -61,6 +61,9 @@ class TDTestCase:
tdSql.execute("use db2")
tdSql.execute("create table stb(ts timestamp, c int) tags(t int)")
tdSql.error("insert into db2.tb6 using db2.stb tags(1) values(now 1) tb2 using db2. tags( )values(now 2)")
# case for TD-10691
tdSql.error("create table ttb1(ts timestamp, file int )")
def stop(self):
tdSql.close()
......
tests/pytest/tools/taosdemoTestTblAlt.py
浏览文件 @ 51259460
......@@ -54,29 +54,36 @@ class TDTestCase:
binPath = buildPath + "/build/bin/"
if(threadID == 0):
print("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT -m t" %
print("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT" %
(binPath, self.numberOfTables, self.numberOfRecords))
os.system("%staosdemo -y -t %d -n %d -b INT,INT,INT,INT" %
(binPath, self.numberOfTables, self.numberOfRecords))
if(threadID == 1):
time.sleep(2)
print("use test")
while True:
max_try = 100
count = 0
while (count < max_try):
try:
tdSql.execute("use test")
break
except Exception as e:
tdLog.info("use database test failed")
time.sleep(1)
time.sleep(2)
count += 1
print("try %d times" % count)
continue
# check if all the tables have heen created
while True:
count = 0
while (count < max_try):
try:
tdSql.query("show tables")
except Exception as e:
tdLog.info("show tables test failed")
time.sleep(1)
time.sleep(2)
count += 1
print("try %d times" % count)
continue
rows = tdSql.queryRows
......@@ -85,13 +92,17 @@ class TDTestCase:
break
time.sleep(1)
# check if there are any records in the last created table
while True:
count = 0
while (count < max_try):
print("query started")
print("try %d times" % count)
try:
tdSql.query("select * from test.t7")
tdSql.query("select * from test.d7")
except Exception as e:
tdLog.info("select * test failed")
time.sleep(2)
count += 1
print("try %d times" % count)
continue
rows = tdSql.queryRows
......@@ -102,8 +113,8 @@ class TDTestCase:
print("alter table test.meters add column c10 int")
tdSql.execute("alter table test.meters add column c10 int")
print("insert into test.t7 values (now, 1, 2, 3, 4, 0)")
tdSql.execute("insert into test.t7 values (now, 1, 2, 3, 4, 0)")
print("insert into test.d7 values (now, 1, 2, 3, 4, 0)")
tdSql.execute("insert into test.d7 values (now, 1, 2, 3, 4, 0)")
def run(self):
tdSql.prepare()
......
tests/script/api/batchprepare.c
浏览文件 @ 51259460
......@@ -1761,7 +1761,7 @@ int stmt_funcb_autoctb_e1(TAOS_STMT *stmt) {
int code = taos_stmt_prepare(stmt, sql, 0);
if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
return -1;
exit(1);
}
int id = 0;
......@@ -1797,9 +1797,44 @@ int stmt_funcb_autoctb_e1(TAOS_STMT *stmt) {
return 0;
}
int stmt_multi_insert_check(TAOS_STMT *stmt) {
char *sql;
// The number of tag column list is not equal to the number of tag value list
sql = "insert into ? using stb1 (id1) tags(1,?) values(?,?,?,?,?,?,?,?,?,?)";
if (0 == taos_stmt_prepare(stmt, sql, 0)) {
printf("failed to check taos_stmt_prepare. sql:%s\n", sql);
exit(1);
}
// The number of column list is not equal to the number of value list
sql = "insert into ? using stb1 tags(1,?,2,?,4,?,6.0,?,'b') "
"(ts, b, v1, v2, v4, v8, f4, f8, bin) values(?,?,?,?,?,?,?,?,?,?)";
if (0 == taos_stmt_prepare(stmt, sql, 0)) {
printf("failed to check taos_stmt_prepare. sql:%s\n", sql);
exit(1);
}
sql = "insert into ? using stb1 () tags(1,?) values(?,?,?,?,?,?,?,?,?,?)";
if (0 == taos_stmt_prepare(stmt, sql, 0)) {
printf("failed to check taos_stmt_prepare. sql:%s\n", sql);
exit(1);
}
sql = "insert into ? using stb1 ( tags(1,?) values(?,?,?,?,?,?,?,?,?,?)";
if (0 == taos_stmt_prepare(stmt, sql, 0)) {
printf("failed to check taos_stmt_prepare. sql:%s\n", sql);
exit(1);
}
sql = "insert into ? using stb1 ) tags(1,?) values(?,?,?,?,?,?,?,?,?,?)";
if (0 == taos_stmt_prepare(stmt, sql, 0)) {
printf("failed to check taos_stmt_prepare. sql:%s\n", sql);
exit(1);
}
return 0;
}
//1 tables 10 records
int stmt_funcb_autoctb_e2(TAOS_STMT *stmt) {
......@@ -4505,7 +4540,6 @@ void* runcase(void *par) {
(void)idx;
#if 1
prepare(taos, 1, 1);
......@@ -4819,6 +4853,16 @@ void* runcase(void *par) {
#endif
#if 1
prepare(taos, 1, 0);
stmt = taos_stmt_init(taos);
printf("stmt_multi_insert_check start\n");
stmt_multi_insert_check(stmt);
printf("stmt_multi_insert_check end\n");
taos_stmt_close(stmt);
#endif
#if 1
prepare(taos, 1, 1);
......@@ -5007,7 +5051,6 @@ void* runcase(void *par) {
printf("check result end\n");
#endif
#if 1
preparem(taos, 0, idx);
......
tests/script/general/parser/udf_dll.sim
浏览文件 @ 51259460
......@@ -10,7 +10,7 @@ sql connect
print ======================== dnode1 start
sql create function add_one as '/tmp/add_one.so' outputtype int;
sql create aggregate function sum_double as '/tmp/sum_double.so' outputtype int;
sql create aggregate function sum_double as '/tmp/sum_double.so' outputtype bigint;
sql show functions;
if $rows != 2 then
return -1
......
tests/script/general/parser/udf_dll_stable.sim
浏览文件 @ 51259460
......@@ -11,7 +11,7 @@ print ======================== dnode1 start
sql create function add_one as '/tmp/add_one.so' outputtype int;
sql create function add_one_64232 as '/tmp/add_one_64232.so' outputtype int;
sql create aggregate function sum_double as '/tmp/sum_double.so' outputtype int;
sql create aggregate function sum_double as '/tmp/sum_double.so' outputtype bigint;
sql show functions;
if $rows != 3 then
return -1
......
tests/script/sh/abs_max.c
浏览文件 @ 51259460
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <inttypes.h>
typedef struct SUdfInit{
int maybe_null; /* 1 if function can return NULL */
int decimals; /* for real functions */
long long length; /* For string functions */
int64_t length; /* For string functions */
char *ptr; /* free pointer for function data */
int const_item; /* 0 if result is independent of arguments */
} SUdfInit;
......@@ -14,31 +15,36 @@ typedef struct SUdfInit{
#define TSDB_DATA_INT_NULL 0x80000000L
#define TSDB_DATA_BIGINT_NULL 0x8000000000000000L
void abs_max(char* data, short itype, short ibytes, int numOfRows, long long* ts, char* dataOutput, char* interBuf, char* tsOutput,
void abs_max(char* data, short itype, short ibytes, int numOfRows, int64_t* ts, char* dataOutput, char* interBuf, char* tsOutput,
int* numOfOutput, short otype, short obytes, SUdfInit* buf) {
int i;
int r = 0;
printf("abs_max input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
int64_t r = 0;
// printf("abs_max input data:%p, type:%d, rows:%d, ts:%p, %" PRId64 ", dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
if (itype == 5) {
r=*(long *)dataOutput;
r=*(int64_t *)dataOutput;
*numOfOutput=0;
for(i=0;i<numOfRows;++i) {
if (*((long *)data + i) == TSDB_DATA_BIGINT_NULL) {
if (*((int64_t *)data + i) == TSDB_DATA_BIGINT_NULL) {
continue;
}
*numOfOutput=1;
long v = abs(*((long *)data + i));
//int64_t v = abs(*((int64_t *)data + i));
int64_t v = *((int64_t *)data + i);
if (v < 0) {
v = 0 - v;
}
if (v > r) {
r = v;
}
}
*(long *)dataOutput=r;
*(int64_t *)dataOutput=r;
printf("abs_max out, dataoutput:%ld, numOfOutput:%d\n", *(long *)dataOutput, *numOfOutput);
} else {
// printf("abs_max out, dataoutput:%" PRId64", numOfOutput:%d\n", *(int64_t *)dataOutput, *numOfOutput);
}else {
*numOfOutput=0;
}
}
......@@ -47,44 +53,43 @@ void abs_max(char* data, short itype, short ibytes, int numOfRows, long long* ts
void abs_max_finalize(char* dataOutput, char* interBuf, int* numOfOutput, SUdfInit* buf) {
int i;
int r = 0;
printf("abs_max_finalize dataoutput:%p:%d, numOfOutput:%d, buf:%p\n", dataOutput, *dataOutput, *numOfOutput, buf);
printf("abs_max finalize, dataoutput:%ld, numOfOutput:%d\n", *(long *)dataOutput, *numOfOutput);
//int64_t r = 0;
// printf("abs_max_finalize dataoutput:%p:%d, numOfOutput:%d, buf:%p\n", dataOutput, *dataOutput, *numOfOutput, buf);
// *numOfOutput=1;
// printf("abs_max finalize, dataoutput:%" PRId64", numOfOutput:%d\n", *(int64_t *)dataOutput, *numOfOutput);
}
void abs_max_merge(char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput, SUdfInit* buf) {
int r = 0;
int64_t r = 0;
if (numOfRows > 0) {
r = *((long *)data);
r = *((int64_t *)data);
}
printf("abs_max_merge numOfRows:%d, dataoutput:%p, buf:%p\n", numOfRows, dataOutput, buf);
// printf("abs_max_merge numOfRows:%d, dataoutput:%p, buf:%p\n", numOfRows, dataOutput, buf);
for (int i = 1; i < numOfRows; ++i) {
printf("abs_max_merge %d - %ld\n", i, *((long *)data + i));
if (*((long*)data + i) > r) {
r= *((long*)data + i);
// printf("abs_max_merge %d - %" PRId64"\n", i, *((int64_t *)data + i));
if (*((int64_t*)data + i) > r) {
r= *((int64_t*)data + i);
}
}
*(long*)dataOutput=r;
*(int64_t*)dataOutput=r;
if (numOfRows > 0) {
*numOfOutput=1;
} else {
*numOfOutput=0;
}
printf("abs_max_merge, dataoutput:%ld, numOfOutput:%d\n", *(long *)dataOutput, *numOfOutput);
// printf("abs_max_merge, dataoutput:%" PRId64", numOfOutput:%d\n", *(int64_t *)dataOutput, *numOfOutput);
}
int abs_max_init(SUdfInit* buf) {
printf("abs_max init\n");
// printf("abs_max init\n");
return 0;
}
void abs_max_destroy(SUdfInit* buf) {
printf("abs_max destroy\n");
}
// printf("abs_max destroy\n");
}
\ No newline at end of file
tests/script/sh/add_one.c
浏览文件 @ 51259460
......@@ -14,20 +14,18 @@ void add_one(char* data, short itype, short ibytes, int numOfRows, long long* ts
int* numOfOutput, short otype, short obytes, SUdfInit* buf) {
int i;
int r = 0;
printf("add_one input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
// printf("add_one input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
if (itype == 4) {
for(i=0;i<numOfRows;++i) {
printf("input %d - %d", i, *((int *)data + i));
// printf("input %d - %d", i, *((int *)data + i));
*((int *)dataOutput+i)=*((int *)data + i) + 1;
printf(", output %d\n", *((int *)dataOutput+i));
// printf(", output %d\n", *((int *)dataOutput+i));
if (tsOutput) {
*(long long*)tsOutput=1000000;
}
}
*numOfOutput=numOfRows;
printf("add_one out, numOfOutput:%d\n", *numOfOutput);
// printf("add_one out, numOfOutput:%d\n", *numOfOutput);
}
}
}
\ No newline at end of file
tests/script/sh/add_one_64232.c
浏览文件 @ 51259460
......@@ -17,7 +17,7 @@ void add_one_64232(char* data, short itype, short ibytes, int numOfRows, long lo
printf("add_one_64232 input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
if (itype == 5) {
for(i=0;i<numOfRows;++i) {
printf("input %d - %d", i, *((long *)data + i));
printf("input %d - %ld", i, *((long *)data + i));
*((int *)dataOutput+i)=(int)*((long *)data + i) + 1;
printf(", output %d\n", *((int *)dataOutput+i));
if (tsOutput) {
......@@ -28,6 +28,4 @@ void add_one_64232(char* data, short itype, short ibytes, int numOfRows, long lo
printf("add_one_64232 out, numOfOutput:%d\n", *numOfOutput);
}
}
}
\ No newline at end of file
tests/script/sh/sum_double.c
浏览文件 @ 51259460
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <inttypes.h>
typedef struct SUdfInit{
int maybe_null; /* 1 if function can return NULL */
int decimals; /* for real functions */
long long length; /* For string functions */
int64_t length; /* For string functions */
char *ptr; /* free pointer for function data */
int const_item; /* 0 if result is independent of arguments */
} SUdfInit;
......@@ -13,13 +14,13 @@ typedef struct SUdfInit{
#define TSDB_DATA_INT_NULL 0x80000000L
void sum_double(char* data, short itype, short ibytes, int numOfRows, long long* ts, char* dataOutput, char* interBuf, char* tsOutput,
void sum_double(char* data, short itype, short ibytes, int numOfRows, int64_t* ts, char* dataOutput, char* interBuf, char* tsOutput,
int* numOfOutput, short otype, short obytes, SUdfInit* buf) {
int i;
int r = 0;
printf("sum_double input data:%p, type:%d, rows:%d, ts:%p,%lld, dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
int64_t r = 0;
printf("sum_double input data:%p, type:%d, rows:%d, ts:%p,%"PRId64", dataoutput:%p, tsOutput:%p, numOfOutput:%p, buf:%p\n", data, itype, numOfRows, ts, *ts, dataOutput, tsOutput, numOfOutput, buf);
if (itype == 4) {
r=*(int *)dataOutput;
r=*(int64_t *)dataOutput;
*numOfOutput=0;
for(i=0;i<numOfRows;++i) {
......@@ -29,10 +30,10 @@ void sum_double(char* data, short itype, short ibytes, int numOfRows, long long*
*numOfOutput=1;
r+=*((int *)data + i);
*(int *)dataOutput=r;
}
*(int64_t *)dataOutput=r;
}
printf("sum_double out, dataoutput:%d, numOfOutput:%d\n", *(int *)dataOutput, *numOfOutput);
// printf("sum_double out, dataoutput:%"PRId64", numOfOutput:%d\n", *(int64_t *)dataOutput, *numOfOutput);
}
}
......@@ -40,45 +41,44 @@ void sum_double(char* data, short itype, short ibytes, int numOfRows, long long*
void sum_double_finalize(char* dataOutput, char* interBuf, int* numOfOutput, SUdfInit* buf) {
int i;
int r = 0;
printf("sum_double_finalize dataoutput:%p:%d, numOfOutput:%d, buf:%p\n", dataOutput, *dataOutput, *numOfOutput, buf);
*numOfOutput=1;
*(int*)(buf->ptr)=*(int*)dataOutput*2;
*(int*)dataOutput=*(int*)(buf->ptr);
printf("sum_double finalize, dataoutput:%d, numOfOutput:%d\n", *(int *)dataOutput, *numOfOutput);
int64_t r = 0;
// printf("sum_double_finalize dataoutput:%p:%"PRId64", numOfOutput:%d, buf:%p\n", dataOutput, *(int64_t*)dataOutput, *numOfOutput, buf);
// *numOfOutput=1;
*(int64_t*)(buf->ptr)=*(int64_t*)dataOutput*2;
*(int64_t*)dataOutput=*(int64_t*)(buf->ptr);
// printf("sum_double finalize, dataoutput:%"PRId64", numOfOutput:%d\n", *(int64_t *)dataOutput, *numOfOutput);
}
void sum_double_merge(char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput, SUdfInit* buf) {
void sum_double_merge(char* data, int32_t numOfRows, char* dataOutput, int* numOfOutput, SUdfInit* buf) {
int r = 0;
int sum = 0;
int64_t sum = 0;
printf("sum_double_merge numOfRows:%d, dataoutput:%p, buf:%p\n", numOfRows, dataOutput, buf);
// printf("sum_double_merge numOfRows:%d, dataoutput:%p, buf:%p\n", numOfRows, dataOutput, buf);
for (int i = 0; i < numOfRows; ++i) {
printf("sum_double_merge %d - %d\n", i, *((int*)data + i));
sum +=*((int*)data + i);
// printf("sum_double_merge %d - %"PRId64"\n", i, *((int64_t*)data + i));
sum +=*((int64_t*)data + i);
}
*(int*)dataOutput+=sum;
*(int64_t*)dataOutput+=sum;
if (numOfRows > 0) {
*numOfOutput=1;
} else {
*numOfOutput=0;
}
printf("sum_double_merge, dataoutput:%d, numOfOutput:%d\n", *(int *)dataOutput, *numOfOutput);
// printf("sum_double_merge, dataoutput:%"PRId64", numOfOutput:%d\n", *(int64_t *)dataOutput, *numOfOutput);
}
int sum_double_init(SUdfInit* buf) {
buf->maybe_null=1;
buf->ptr = malloc(sizeof(int));
printf("sum_double init\n");
buf->ptr = malloc(sizeof(int64_t));
// printf("sum_double init\n");
return 0;
}
void sum_double_destroy(SUdfInit* buf) {
free(buf->ptr);
printf("sum_double destroy\n");
}
// printf("sum_double destroy\n");
}
\ No newline at end of file
tests/tsim/inc/sim.h
浏览文件 @ 51259460
......@@ -135,7 +135,7 @@ typedef struct _script_t {
int32_t numOfLines; // number of lines in the script
int32_t bgScriptLen;
char fileName[MAX_FILE_NAME_LEN]; // script file name
char error[MAX_ERROR_LEN];
char error[TSDB_MAX_BINARY_LEN + 100];
char * optionBuffer;
SCmdLine *lines; // command list
SVariable variables[MAX_VAR_LEN];
......@@ -178,4 +178,4 @@ bool simExecuteLineInsertCmd(SScript *script, char *option);
bool simExecuteLineInsertErrorCmd(SScript *script, char *option);
void simVisuallizeOption(SScript *script, char *src, char *dst);
#endif
\ No newline at end of file
#endif
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或