提交 be07c1da 编写于 作者: P Ping Xiao

Merge branch 'develop' into xiaoping/add_test_case

......@@ -82,6 +82,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; -*-
......
......@@ -135,6 +135,8 @@ IF ("${BUILD_HTTP}" STREQUAL "")
ELSE ()
SET(BUILD_HTTP "false")
ENDIF ()
ELSEIF (TD_DARWIN)
SET(BUILD_HTTP "false")
ELSE ()
SET(BUILD_HTTP "true")
ENDIF ()
......@@ -150,6 +152,17 @@ IF (TD_BUILD_HTTP)
ADD_DEFINITIONS(-DHTTP_EMBEDDED)
ENDIF ()
IF (${BUILD_LUA} MATCHES "false")
SET(TD_BUILD_LUA FALSE)
ENDIF ()
IF (TD_BUILD_LUA)
MESSAGE("Enable lua")
ADD_DEFINITIONS(-DLUA_EMBEDDED)
ELSE ()
MESSAGE("Disable lua")
ENDIF ()
IF ("${AVRO_SUPPORT}" MATCHES "true")
SET(TD_AVRO_SUPPORT TRUE)
ELSEIF ("${AVRO_SUPPORT}" MATCHES "false")
......
......@@ -92,6 +92,8 @@ ENDIF ()
SET(TD_BUILD_HTTP FALSE)
SET(TD_BUILD_LUA TRUE)
SET(TD_AVRO_SUPPORT FALSE)
SET(TD_MEMORY_SANITIZER FALSE)
......
......@@ -36,7 +36,13 @@ 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 readlink /bin/sh OUTPUT_VARIABLE SHELL_LINK)
MESSAGE(STATUS "The shell is: " ${SHELL_LINK})
IF (${SHELL_LINK} MATCHES "dash")
execute_process(COMMAND ${TD_COMMUNITY_DIR}/packaging/tools/get_os.sh "" OUTPUT_VARIABLE TD_OS_INFO)
ELSE ()
execute_process(COMMAND sh ${TD_COMMUNITY_DIR}/packaging/tools/get_os.sh "" OUTPUT_VARIABLE TD_OS_INFO)
ENDIF()
MESSAGE(STATUS "The current os is " ${TD_OS_INFO})
SET(TD_LINUX TRUE)
......
......@@ -4,7 +4,7 @@ PROJECT(TDengine)
IF (DEFINED VERNUMBER)
SET(TD_VER_NUMBER ${VERNUMBER})
ELSE ()
SET(TD_VER_NUMBER "2.3.1.0")
SET(TD_VER_NUMBER "2.3.2.0")
ENDIF ()
IF (DEFINED VERCOMPATIBLE)
......
......@@ -15,7 +15,10 @@ ADD_SUBDIRECTORY(cJson)
ADD_SUBDIRECTORY(wepoll)
ADD_SUBDIRECTORY(MsvcLibX)
ADD_SUBDIRECTORY(rmonotonic)
ADD_SUBDIRECTORY(lua)
IF (TD_BUILD_LUA)
ADD_SUBDIRECTORY(lua)
ENDIF ()
IF (TD_LINUX AND TD_MQTT)
ADD_SUBDIRECTORY(MQTT-C)
......@@ -55,6 +58,7 @@ IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
MESSAGE("")
MESSAGE("setup deps/jemalloc, current source dir:" ${CMAKE_CURRENT_SOURCE_DIR})
MESSAGE("binary dir:" ${CMAKE_BINARY_DIR})
include(ExternalProject)
ExternalProject_Add(jemalloc
PREFIX "jemalloc"
SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/jemalloc
......@@ -62,6 +66,7 @@ IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
CONFIGURE_COMMAND ./autogen.sh COMMAND ./configure --prefix=${CMAKE_BINARY_DIR}/build/
BUILD_COMMAND ${MAKE}
)
INCLUDE_DIRECTORIES(${CMAKE_BINARY_DIR}/build/include)
ENDIF ()
IF (${TSZ_ENABLED} MATCHES "true")
......
......@@ -4,7 +4,7 @@
自从 TDengine 2019年 7 月开源以来,凭借创新的数据建模设计、快捷的安装方式、易用的编程接口和强大的数据写入查询性能博得了大量时序数据开发者的青睐。其中写入和查询性能往往令刚接触 TDengine 的用户称叹不已。为了便于用户在最短时间内就可以体验到 TDengine 的高性能特点,我们专门开发了一个应用程序 taosdemo 用于对 TDengine 进行写入和查询的性能测试,用户可以通过 taosdemo 轻松模拟大量设备产生海量数据的场景,并且可以通过 taosdemo 参数灵活控制表的列数、数据类型、乱序比例以及并发线程数量。
运行 taosdemo 很简单,通过下载 TDengine 安装包(https://www.taosdata.com/cn/all-downloads/)或者自行下载 TDengine 代码(https://github.com/taosdata/TDengine)编译都可以在安装目录或者编译结果目录中找到并运行。
运行 taosdemo 很简单,通过下载 TDengine 安装包( https://www.taosdata.com/cn/all-downloads/ )或者自行下载 TDengine 代码( https://github.com/taosdata/TDengine )编译都可以在安装目录或者编译结果目录中找到并运行。
接下来本文为大家讲解 taosdemo 的使用介绍及注意事项。
......
......@@ -178,7 +178,7 @@ taos> select avg(current), max(voltage), min(phase) from test.d10 interval(10s);
## <a class="anchor" id="taosdemo"></a> taosdemo 详细功能列表
taosdemo 命令本身带有很多选项,配置表的数目、记录条数等等,请执行 `taosdemo --help` 详细列出。您可以设置不同参数进行体验。
taosdemo 详细使用方法请参照 [如何使用taosdemo对TDengine进行性能测试](https://www.taosdata.com/cn/documentation/getting-started/taosdemo )
taosdemo 详细使用方法请参照 [如何使用taosdemo对TDengine进行性能测试](https://www.taosdata.com/cn/documentation/getting-started/taosdemo )
## 客户端和报警模块
......@@ -190,7 +190,7 @@ taosdemo 详细使用方法请参照 [如何使用taosdemo对TDengine进行性
### TDengine 服务器支持的平台列表
| | **CentOS 6/7/8** | **Ubuntu 16/18/20** | **Other Linux** | **统信 UOS** | **银河/中标麒麟** | **凝思 V60/V80** | **华为 EulerOS** |
| | **CentOS 7/8** | **Ubuntu 16/18/20** | **Other Linux** | **统信 UOS** | **银河/中标麒麟** | **凝思 V60/V80** | **华为 EulerOS** |
| -------------- | --------------------- | ------------------------ | --------------- | --------------- | ------------------------- | --------------------- | --------------------- |
| X64 | ● | ● | | ○ | ● | ● | ● |
| 龙芯 MIPS64 | | | ● | | | | |
......
......@@ -328,7 +328,7 @@ TDengine的异步API均采用非阻塞调用模式。应用程序可以用多线
除 C/C++ 语言外,TDengine 的 Java 语言 JNI Connector 也提供参数绑定接口支持,具体请另外参见:[参数绑定接口的 Java 用法](https://www.taosdata.com/cn/documentation/connector/java#stmt-java)
接口相关的具体函数如下(也可以参考 [apitest.c](https://github.com/taosdata/TDengine/blob/develop/tests/examples/c/apitest.c) 文件中使用对应函数的方式):
接口相关的具体函数如下(也可以参考 [prepare.c](https://github.com/taosdata/TDengine/blob/develop/tests/examples/c/prepare.c) 文件中使用对应函数的方式):
- `TAOS_STMT* taos_stmt_init(TAOS *taos)`
......
......@@ -719,7 +719,7 @@ Query OK, 1 row(s) in set (0.001091s)
* 暂不支持含列名的四则运算表达式用于条件过滤算子(例如,不支持 `where a*2>6;`,但可以写 `where a>6/2;`)。
* 暂不支持含列名的四则运算表达式作为 SQL 函数的应用对象(例如,不支持 `select min(2*a) from t;`,但可以写 `select 2*min(a) from t;`)。
- WHERE 语句可以使用各种逻辑判断来过滤数字值,或使用通配符来过滤字符串。
- 输出结果缺省按首列时间戳升序排序,但可以指定按降序排序( _c0 指首列时间戳)。使用 ORDER BY 对其他字段进行排序为非法操作
- 输出结果缺省按首列时间戳升序排序,但可以指定按降序排序( _c0 指首列时间戳)。使用 ORDER BY 对其他字段进行排序,排序结果顺序不确定
- 参数 LIMIT 控制输出条数,OFFSET 指定从第几条开始输出。LIMIT/OFFSET 对结果集的执行顺序在 ORDER BY 之后。且 `LIMIT 5 OFFSET 2` 可以简写为 `LIMIT 2, 5`。
* 在有 GROUP BY 子句的情况下,LIMIT 参数控制的是每个分组中至多允许输出的条数。
- 参数 SLIMIT 控制由 GROUP BY 指令划分的分组中,至多允许输出几个分组的数据。且 `SLIMIT 5 SOFFSET 2` 可以简写为 `SLIMIT 2, 5`。
......@@ -743,7 +743,7 @@ Query OK, 1 row(s) in set (0.001091s)
1. <> 算子也可以写为 != ,请注意,这个算子不能用于数据表第一列的 timestamp 字段。
2. like 算子使用通配符字符串进行匹配检查。
* 在通配符字符串中:'%'(百分号)匹配 0 到任意个字符;'\_'(下划线)匹配单个任意字符。
* 在通配符字符串中:'%'(百分号)匹配 0 到任意个字符;'\_'(下划线)匹配单个任意ASCII字符。
* 如果希望匹配字符串中原本就带有的 \_(下划线)字符,那么可以在通配符字符串中写作 `\_`,也即加一个反斜线来进行转义。(从 2.2.0.0 版本开始支持)
* 通配符字符串最长不能超过 20 字节。(从 2.1.6.1 版本开始,通配符字符串的长度放宽到了 100 字节,并可以通过 taos.cfg 中的 maxWildCardsLength 参数来配置这一长度限制。但不建议使用太长的通配符字符串,将有可能严重影响 LIKE 操作的执行性能。)
3. 同时进行多个字段的范围过滤,需要使用关键词 AND 来连接不同的查询条件,暂不支持 OR 连接的不同列之间的查询过滤条件。
......
......@@ -628,7 +628,7 @@ func insertData(threadIndex, start, end int, wg *sync.WaitGroup, successRows []i
buffer.WriteString(",")
for _, field := range subTableInfo.config.Fields {
buffer.WriteString(getFieldValue(currentRow[strings.ToLower(field.Name)]))
buffer.WriteString(getFieldValue(currentRow[strings.ToLower(field.Name)],field.Type))
buffer.WriteString(",")
}
......@@ -708,7 +708,10 @@ func executeBatchInsert(insertSql string, connection *sql.DB) int64 {
return affected
}
func getFieldValue(fieldValue interface{}) string {
func getFieldValue(fieldValue interface{},fieldtype interface{}) string {
if fieldtype == "timestamp" || fieldtype == "bigint" {
return fmt.Sprintf("%v", fieldValue)
}
return fmt.Sprintf("'%v'", fieldValue)
}
......
......@@ -18,6 +18,8 @@ tags = [
fields = [
# 字段列表,name 为字段名称,type 为字段类型
# 除主键外,其他field如果也要设置为timestamp,可以是type ="timestamp" 类型,此时value可同时支持'2006-01-02 15:04:05.000'和millisecond格式
# 也可以是type = "bigint",此时value只支持millisecond格式
{ name = "ts", type = "timestamp" },
{ name = "temperature", type = "int" },
{ name = "humidity", type = "float" },
......
......@@ -299,8 +299,8 @@ keepColumnName 1
# percent of redundant data in tsdb meta will compact meta data,0 means donot compact
# tsdbMetaCompactRatio 0
# default string type used for storing JSON String, options can be binary/nchar, default is binary
# defaultJSONStrType binary
# default string type used for storing JSON String, options can be binary/nchar, default is nchar
# defaultJSONStrType nchar
# force TCP transmission
# rpcForceTcp 0
......
......@@ -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
echo "build ${pagMode} package ..."
......@@ -213,7 +214,12 @@ else
exit 1
fi
make -j8 && ${csudo} make install
if [[ "$allocator" == "jemalloc" ]]; then
# jemalloc need compile first, so disable parallel build
make V=1 && ${csudo} make install
else
make -j8 && ${csudo} make install
fi
cd ${curr_dir}
......
......@@ -62,12 +62,14 @@ NC='\033[0m'
csudo=""
service_mod=2
os_type=0
if [ "$osType" != "Darwin" ]; then
if command -v sudo > /dev/null; then
csudo="sudo"
fi
initd_mod=0
service_mod=2
if pidof systemd &> /dev/null; then
service_mod=0
elif $(which service &> /dev/null); then
......@@ -91,7 +93,6 @@ if [ "$osType" != "Darwin" ]; then
#osinfo=$(awk -F= '/^NAME/{print $2}' /etc/os-release)
osinfo=$(cat /etc/os-release | grep "NAME" | cut -d '"' -f2)
#echo "osinfo: ${osinfo}"
os_type=0
if echo $osinfo | grep -qwi "ubuntu" ; then
echo "this is ubuntu system"
os_type=1
......@@ -122,7 +123,8 @@ function kill_taosadapter() {
}
function kill_taosd() {
pid=$(ps -ef | grep "taosd" | grep -v "grep" | awk '{print $2}')
ps -ef | grep "taosd"
pid=$(ps -ef | grep -w "taosd" | grep -v "grep" | awk '{print $2}')
if [ -n "$pid" ]; then
${csudo} kill -9 $pid || :
fi
......@@ -202,31 +204,31 @@ function install_jemalloc() {
/usr/bin/install -c -d /usr/local/bin
if [ -f "${binary_dir}/build/bin/jemalloc-config" ]; then
/usr/bin/install -c -m 755 ${binary_dir}/build/bin/jemalloc-config /usr/local/bin
${csudo} /usr/bin/install -c -m 755 ${binary_dir}/build/bin/jemalloc-config /usr/local/bin
fi
if [ -f "${binary_dir}/build/bin/jemalloc.sh" ]; then
/usr/bin/install -c -m 755 ${binary_dir}/build/bin/jemalloc.sh /usr/local/bin
${csudo} /usr/bin/install -c -m 755 ${binary_dir}/build/bin/jemalloc.sh /usr/local/bin
fi
if [ -f "${binary_dir}/build/bin/jeprof" ]; then
/usr/bin/install -c -m 755 ${binary_dir}/build/bin/jeprof /usr/local/bin
${csudo} /usr/bin/install -c -m 755 ${binary_dir}/build/bin/jeprof /usr/local/bin
fi
if [ -f "${binary_dir}/build/include/jemalloc/jemalloc.h" ]; then
/usr/bin/install -c -d /usr/local/include/jemalloc
/usr/bin/install -c -m 644 ${binary_dir}/build/include/jemalloc/jemalloc.h\
${csudo} /usr/bin/install -c -d /usr/local/include/jemalloc
${csudo} /usr/bin/install -c -m 644 ${binary_dir}/build/include/jemalloc/jemalloc.h\
/usr/local/include/jemalloc
fi
if [ -f "${binary_dir}/build/lib/libjemalloc.so.2" ]; then
/usr/bin/install -c -d /usr/local/lib
/usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc.so.2 /usr/local/lib
ln -sf libjemalloc.so.2 /usr/local/lib/libjemalloc.so
/usr/bin/install -c -d /usr/local/lib
${csudo} /usr/bin/install -c -d /usr/local/lib
${csudo} /usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc.so.2 /usr/local/lib
${csudo} ln -sf libjemalloc.so.2 /usr/local/lib/libjemalloc.so
${csudo} /usr/bin/install -c -d /usr/local/lib
[ -f ${binary_dir}/build/lib/libjemalloc.a ] &&
/usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc.a /usr/local/lib
${csudo} /usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc.a /usr/local/lib
[ -f ${binary_dir}/build/lib/libjemalloc_pic.a ] &&
/usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc_pic.a /usr/local/lib
${csudo} /usr/bin/install -c -m 755 ${binary_dir}/build/lib/libjemalloc_pic.a /usr/local/lib
if [ -f "${binary_dir}/build/lib/pkgconfig/jemalloc.pc" ]; then
/usr/bin/install -c -d /usr/local/lib/pkgconfig
/usr/bin/install -c -m 644 ${binary_dir}/build/lib/pkgconfig/jemalloc.pc\
${csudo} /usr/bin/install -c -d /usr/local/lib/pkgconfig
${csudo} /usr/bin/install -c -m 644 ${binary_dir}/build/lib/pkgconfig/jemalloc.pc\
/usr/local/lib/pkgconfig
fi
if [ -d /etc/ld.so.conf.d ]; then
......@@ -237,29 +239,28 @@ function install_jemalloc() {
fi
fi
if [ -f "${binary_dir}/build/share/doc/jemalloc/jemalloc.html" ]; then
/usr/bin/install -c -d /usr/local/share/doc/jemalloc
/usr/bin/install -c -m 644 ${binary_dir}/build/share/doc/jemalloc/jemalloc.html\
${csudo} /usr/bin/install -c -d /usr/local/share/doc/jemalloc
${csudo} /usr/bin/install -c -m 644 ${binary_dir}/build/share/doc/jemalloc/jemalloc.html\
/usr/local/share/doc/jemalloc
fi
if [ -f "${binary_dir}/build/share/man/man3/jemalloc.3" ]; then
/usr/bin/install -c -d /usr/local/share/man/man3
/usr/bin/install -c -m 644 ${binary_dir}/build/share/man/man3/jemalloc.3\
${csudo} /usr/bin/install -c -d /usr/local/share/man/man3
${csudo} /usr/bin/install -c -m 644 ${binary_dir}/build/share/man/man3/jemalloc.3\
/usr/local/share/man/man3
fi
fi
}
function install_avro() {
if [ "$osType" != "Darwin" ]; then
if [ -f "${binary_dir}/build/$1/libavro.so.23.0.0" ]; then
/usr/bin/install -c -d /usr/local/$1
/usr/bin/install -c -m 755 ${binary_dir}/build/$1/libavro.so.23.0.0 /usr/local/$1
ln -sf libavro.so.23.0.0 /usr/local/$1/libavro.so.23
ln -sf libavro.so.23 /usr/local/$1/libavro.so
/usr/bin/install -c -d /usr/local/$1
${csudo} /usr/bin/install -c -d /usr/local/$1
${csudo} /usr/bin/install -c -m 755 ${binary_dir}/build/$1/libavro.so.23.0.0 /usr/local/$1
${csudo} ln -sf libavro.so.23.0.0 /usr/local/$1/libavro.so.23
${csudo} ln -sf libavro.so.23 /usr/local/$1/libavro.so
${csudo} /usr/bin/install -c -d /usr/local/$1
[ -f ${binary_dir}/build/$1/libavro.a ] &&
/usr/bin/install -c -m 755 ${binary_dir}/build/$1/libavro.a /usr/local/$1
${csudo} /usr/bin/install -c -m 755 ${binary_dir}/build/$1/libavro.a /usr/local/$1
if [ -d /etc/ld.so.conf.d ]; then
echo "/usr/local/$1" | ${csudo} tee /etc/ld.so.conf.d/libavro.conf
......@@ -543,7 +544,6 @@ function update_TDengine() {
echo -e "${GREEN}Start to update TDengine...${NC}"
# Stop the service if running
if [ "$osType" != "Darwin" ]; then
if pidof taosd &> /dev/null; then
if ((${service_mod}==0)); then
${csudo} systemctl stop taosd || :
......@@ -555,7 +555,6 @@ function update_TDengine() {
fi
sleep 1
fi
fi
install_main_path
......@@ -566,50 +565,35 @@ function update_TDengine() {
install_examples
install_bin
if [ "$osType" != "Darwin" ]; then
install_service
install_taosadapter_service
fi
install_config
install_taosadapter_config
if [ "$osType" != "Darwin" ]; then
echo
echo -e "\033[44;32;1mTDengine is updated successfully!${NC}"
echo
echo -e "${GREEN_DARK}To configure TDengine ${NC}: edit /etc/taos/taos.cfg"
echo -e "${GREEN_DARK}To configure taosadapter (if has) ${NC}: edit /etc/taos/taosadapter.toml"
echo -e "${GREEN_DARK}To configure Taos Adapter (if has) ${NC}: edit /etc/taos/taosadapter.toml"
if ((${service_mod}==0)); then
echo -e "${GREEN_DARK}To start TDengine ${NC}: ${csudo} systemctl start taosd${NC}"
elif ((${service_mod}==1)); then
echo -e "${GREEN_DARK}To start TDengine ${NC}: ${csudo} service taosd start${NC}"
else
echo -e "${GREEN_DARK}To start TDengine ${NC}: ./taosd${NC}"
echo -e "${GREEN_DARK}To start Taos Adapter (if has)${NC}: taosadapter &${NC}"
echo -e "${GREEN_DARK}To start TDengine ${NC}: taosd${NC}"
fi
echo -e "${GREEN_DARK}To access TDengine ${NC}: use ${GREEN_UNDERLINE}taos${NC} in shell${NC}"
echo
echo -e "\033[44;32;1mTDengine is updated successfully!${NC}"
else
echo
echo -e "\033[44;32;1mTDengine Client is updated successfully!${NC}"
echo
echo -e "${GREEN_DARK}To access TDengine Client ${NC}: use ${GREEN_UNDERLINE}taos${NC} in shell${NC}"
echo
echo -e "\033[44;32;1mTDengine Client is updated successfully!${NC}"
fi
}
function install_TDengine() {
# Start to install
if [ "$osType" != "Darwin" ]; then
echo -e "${GREEN}Start to install TDEngine...${NC}"
else
echo -e "${GREEN}Start to install TDEngine Client ...${NC}"
fi
echo -e "${GREEN}Start to install TDengine...${NC}"
install_main_path
......@@ -621,15 +605,12 @@ function install_TDengine() {
install_examples
install_bin
if [ "$osType" != "Darwin" ]; then
install_service
install_taosadapter_service
fi
install_config
install_taosadapter_config
if [ "$osType" != "Darwin" ]; then
# Ask if to start the service
echo
echo -e "\033[44;32;1mTDengine is installed successfully!${NC}"
......@@ -647,11 +628,6 @@ function install_TDengine() {
echo -e "${GREEN_DARK}To access TDengine ${NC}: use ${GREEN_UNDERLINE}taos${NC} in shell${NC}"
echo
echo -e "\033[44;32;1mTDengine is installed successfully!${NC}"
else
echo -e "${GREEN_DARK}To access TDengine ${NC}: use ${GREEN_UNDERLINE}taos${NC} in shell${NC}"
echo
echo -e "\033[44;32;1mTDengine Client is installed successfully!${NC}"
fi
}
## ==============================Main program starts from here============================
......
......@@ -468,8 +468,8 @@ function install_service_on_systemd() {
function install_taosadapter_service() {
if ((${service_mod}==0)); then
[ -f ${script_dir}/cfg/taosadapter.service ] &&\
${csudo} cp ${script_dir}/cfg/taosadapter.service \
[ -f ${script_dir}/../cfg/taosadapter.service ] &&\
${csudo} cp ${script_dir}/../cfg/taosadapter.service \
${service_config_dir}/ || :
${csudo} systemctl daemon-reload
fi
......
name: tdengine
base: core20
version: '2.3.1.0'
version: '2.3.2.0'
icon: snap/gui/t-dengine.svg
summary: an open-source big data platform designed and optimized for IoT.
description: |
......
......@@ -62,7 +62,7 @@ typedef struct SRetrieveSupport {
uint32_t numOfRetry; // record the number of retry times
} SRetrieveSupport;
int32_t tscCreateGlobalMergerEnv(SQueryInfo* pQueryInfo, tExtMemBuffer ***pMemBuffer, int32_t numOfSub, tOrderDescriptor **pDesc, uint32_t nBufferSize, int64_t id);
int32_t tscCreateGlobalMergerEnv(SQueryInfo* pQueryInfo, tExtMemBuffer ***pMemBuffer, int32_t numOfSub, tOrderDescriptor **pDesc, uint32_t* nBufferSize, int64_t id);
void tscDestroyGlobalMergerEnv(tExtMemBuffer **pMemBuffer, tOrderDescriptor *pDesc, int32_t numOfVnodes);
......
......@@ -83,6 +83,11 @@ typedef struct SJoinSupporter {
int32_t totalLen;
int32_t num;
SArray* pVgroupTables;
int16_t fillType; // final result fill type
int64_t * fillVal; // default value for fill
int32_t numOfFillVal; // fill value size
} SJoinSupporter;
......@@ -119,6 +124,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(SSqlObj *pSql, STableDataBlocks* pDataBlock, bool removeMeta);
void tscSortRemoveDataBlockDupRowsRaw(STableDataBlocks* dataBuf);
int tscSortRemoveDataBlockDupRows(STableDataBlocks* dataBuf, SBlockKeyInfo* pBlkKeyInfo);
......@@ -147,6 +153,7 @@ int32_t tscGetDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, i
* @return
*/
bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo);
bool tscGetPointInterpQuery(SQueryInfo* pQueryInfo);
bool tscIsTWAQuery(SQueryInfo* pQueryInfo);
bool tscIsIrateQuery(SQueryInfo* pQueryInfo);
bool tscQueryContainsFunction(SQueryInfo* pQueryInfo, int16_t functionId);
......
......@@ -237,7 +237,7 @@ void taos_fetch_rows_a(TAOS_RES *tres, __async_cb_func_t fp, void *param) {
return;
}
if (pRes->qId == 0) {
if (pRes->qId == 0 && pSql->cmd.command != TSDB_SQL_RETRIEVE_EMPTY_RESULT) {
tscError("qhandle is invalid");
pRes->code = TSDB_CODE_TSC_INVALID_QHANDLE;
tscAsyncResultOnError(pSql);
......
......@@ -233,7 +233,7 @@ static int32_t tscFlushTmpBufferImpl(tExtMemBuffer *pMemoryBuf, tOrderDescriptor
// sort before flush to disk, the data must be consecutively put on tFilePage.
if (pDesc->orderInfo.numOfCols > 0) {
tColDataQSort(pDesc, (int32_t)pPage->num, 0, (int32_t)pPage->num - 1, pPage->data, orderType);
tColDataMergeSort(pDesc, (int32_t)pPage->num, 0, (int32_t)pPage->num - 1, pPage->data, orderType);
}
#ifdef _DEBUG_VIEW
......@@ -364,7 +364,7 @@ static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SQueryInfo*
SExprInfo* pExprInfo = tscExprGet(pQueryInfo, j);
int32_t functionId = pExprInfo->base.functionId;
if (pColIndex->colId == pExprInfo->base.colInfo.colId && (functionId == TSDB_FUNC_PRJ || functionId == TSDB_FUNC_TAG)) {
if (pColIndex->colId == pExprInfo->base.colInfo.colId && (functionId == TSDB_FUNC_PRJ || functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TAGPRJ)) {
orderColIndexList[i] = j;
break;
}
......@@ -407,8 +407,8 @@ static int32_t createOrderDescriptor(tOrderDescriptor **pOrderDesc, SQueryInfo*
}
int32_t tscCreateGlobalMergerEnv(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBuffer, int32_t numOfSub,
tOrderDescriptor **pOrderDesc, uint32_t nBufferSizes, int64_t id) {
SSchema *pSchema = NULL;
tOrderDescriptor **pOrderDesc, uint32_t* nBufferSizes, int64_t id) {
SSchema1 *pSchema = NULL;
SColumnModel *pModel = NULL;
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
......@@ -421,7 +421,7 @@ int32_t tscCreateGlobalMergerEnv(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBu
size_t size = tscNumOfExprs(pQueryInfo);
pSchema = (SSchema *)calloc(1, sizeof(SSchema) * size);
pSchema = (SSchema1 *)calloc(1, sizeof(SSchema1) * size);
if (pSchema == NULL) {
tscError("0x%"PRIx64" failed to allocate memory", id);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
......@@ -440,7 +440,10 @@ int32_t tscCreateGlobalMergerEnv(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBu
int32_t capacity = 0;
if (rlen != 0) {
capacity = nBufferSizes / rlen;
if ((*nBufferSizes) < rlen) {
(*nBufferSizes) = rlen * 2;
}
capacity = (*nBufferSizes) / rlen;
}
pModel = createColumnModel(pSchema, (int32_t)size, capacity);
......@@ -457,7 +460,7 @@ int32_t tscCreateGlobalMergerEnv(SQueryInfo *pQueryInfo, tExtMemBuffer ***pMemBu
assert(numOfSub <= pTableMetaInfo->vgroupList->numOfVgroups);
for (int32_t i = 0; i < numOfSub; ++i) {
(*pMemBuffer)[i] = createExtMemBuffer(nBufferSizes, rlen, pg, pModel);
(*pMemBuffer)[i] = createExtMemBuffer(*nBufferSizes, rlen, pg, pModel);
(*pMemBuffer)[i]->flushModel = MULTIPLE_APPEND_MODEL;
}
......@@ -967,7 +970,6 @@ SSDataBlock* doGlobalAggregate(void* param, bool* newgroup) {
if (pOperator->pRuntimeEnv->pQueryAttr->order.order == TSDB_ORDER_DESC) {
SWAP(w->skey, w->ekey, TSKEY);
assert(w->skey <= w->ekey);
}
}
}
......
......@@ -554,9 +554,9 @@ static int32_t tscGetTableTagColumnName(SSqlObj *pSql, char **result) {
int32_t numOfTags = tscGetNumOfTags(pMeta);
for (int32_t i = 0; i < numOfTags; i++) {
if (i != numOfTags - 1) {
snprintf(buf + strlen(buf), TSDB_MAX_BINARY_LEN - strlen(buf), "%s,", pTagsSchema[i].name);
snprintf(buf + strlen(buf), TSDB_MAX_BINARY_LEN - strlen(buf), "`%s`,", pTagsSchema[i].name);
} else {
snprintf(buf + strlen(buf), TSDB_MAX_BINARY_LEN - strlen(buf), "%s", pTagsSchema[i].name);
snprintf(buf + strlen(buf), TSDB_MAX_BINARY_LEN - strlen(buf), "`%s`", pTagsSchema[i].name);
}
}
......
......@@ -156,13 +156,15 @@ static int32_t getSmlMd5ChildTableName(TAOS_SML_DATA_POINT* point, char* tableNa
SStringBuilder sb; memset(&sb, 0, sizeof(sb));
char sTableName[TSDB_TABLE_NAME_LEN + TS_ESCAPE_CHAR_SIZE] = {0};
strtolower(sTableName, point->stableName);
strncpy(sTableName, point->stableName, strlen(point->stableName));
//strtolower(sTableName, point->stableName);
taosStringBuilderAppendString(&sb, sTableName);
for (int j = 0; j < point->tagNum; ++j) {
taosStringBuilderAppendChar(&sb, ',');
TAOS_SML_KV* tagKv = point->tags + j;
char tagName[TSDB_COL_NAME_LEN + TS_ESCAPE_CHAR_SIZE] = {0};
strtolower(tagName, tagKv->key);
strncpy(tagName, tagKv->key, strlen(tagKv->key));
//strtolower(tagName, tagKv->key);
taosStringBuilderAppendString(&sb, tagName);
taosStringBuilderAppendChar(&sb, '=');
taosStringBuilderAppend(&sb, tagKv->value, tagKv->length);
......@@ -261,10 +263,10 @@ static int32_t buildDataPointSchemas(TAOS_SML_DATA_POINT* points, int numPoint,
static int32_t generateSchemaAction(SSchema* pointColField, SHashObj* dbAttrHash, SArray* dbAttrArray, bool isTag, char sTableName[],
SSchemaAction* action, bool* actionNeeded, SSmlLinesInfo* info) {
char fieldNameLowerCase[TSDB_COL_NAME_LEN + TS_ESCAPE_CHAR_SIZE] = {0};
strtolower(fieldNameLowerCase, pointColField->name);
char fieldName[TSDB_COL_NAME_LEN + TS_ESCAPE_CHAR_SIZE] = {0};
strcpy(fieldName, pointColField->name);
size_t* pDbIndex = taosHashGet(dbAttrHash, fieldNameLowerCase, strlen(fieldNameLowerCase));
size_t* pDbIndex = taosHashGet(dbAttrHash, fieldName, strlen(fieldName));
if (pDbIndex) {
SSchema* dbAttr = taosArrayGet(dbAttrArray, *pDbIndex);
assert(strcasecmp(dbAttr->name, pointColField->name) == 0);
......@@ -297,7 +299,7 @@ static int32_t generateSchemaAction(SSchema* pointColField, SHashObj* dbAttrHash
*actionNeeded = true;
}
if (*actionNeeded) {
tscDebug("SML:0x%" PRIx64 " generate schema action. column name: %s, action: %d", info->id, fieldNameLowerCase,
tscDebug("SML:0x%" PRIx64 " generate schema action. column name: %s, action: %d", info->id, fieldName,
action->action);
}
return 0;
......@@ -536,11 +538,8 @@ static int32_t retrieveTableMeta(TAOS* taos, char* tableName, STableMeta** pTabl
tscDebug("SML:0x%" PRIx64 " retrieve table meta. super table name: %s", info->id, tableName);
char tableNameLowerCase[TSDB_TABLE_NAME_LEN + TS_ESCAPE_CHAR_SIZE];
strtolower(tableNameLowerCase, tableName);
char sql[256];
snprintf(sql, 256, "describe %s", tableNameLowerCase);
snprintf(sql, 256, "describe %s", tableName);
TAOS_RES* res = taos_query(taos, sql);
code = taos_errno(res);
if (code != 0) {
......@@ -561,8 +560,10 @@ static int32_t retrieveTableMeta(TAOS* taos, char* tableName, STableMeta** pTabl
pSql->fp = NULL;
registerSqlObj(pSql);
SStrToken tableToken = {.z = tableNameLowerCase, .n = (uint32_t)strlen(tableNameLowerCase), .type = TK_ID};
tGetToken(tableNameLowerCase, &tableToken.type);
char tableNameBuf[TSDB_TABLE_NAME_LEN + TS_ESCAPE_CHAR_SIZE] = {0};
memcpy(tableNameBuf, tableName, strlen(tableName));
SStrToken tableToken = {.z = tableNameBuf, .n = (uint32_t)strlen(tableName), .type = TK_ID};
tGetToken(tableNameBuf, &tableToken.type);
bool dbIncluded = false;
// Check if the table name available or not
if (tscValidateName(&tableToken, true, &dbIncluded) != TSDB_CODE_SUCCESS) {
......@@ -1839,7 +1840,7 @@ static int32_t parseSmlTimeStamp(TAOS_SML_KV **pTS, const char **index, SSmlLine
const char *start, *cur;
int32_t ret = TSDB_CODE_SUCCESS;
int len = 0;
char key[] = "_ts";
char key[] = "ts";
char *value = NULL;
start = cur = *index;
......@@ -1870,24 +1871,14 @@ static int32_t parseSmlTimeStamp(TAOS_SML_KV **pTS, const char **index, SSmlLine
bool checkDuplicateKey(char *key, SHashObj *pHash, SSmlLinesInfo* info) {
char *val = NULL;
char *cur = key;
char keyLower[TSDB_COL_NAME_LEN];
size_t keyLen = 0;
while(*cur != '\0') {
keyLower[keyLen] = tolower(*cur);
keyLen++;
cur++;
}
keyLower[keyLen] = '\0';
val = taosHashGet(pHash, keyLower, keyLen);
val = taosHashGet(pHash, key, strlen(key));
if (val) {
tscError("SML:0x%"PRIx64" Duplicate key detected:%s", info->id, keyLower);
tscError("SML:0x%"PRIx64" Duplicate key detected:%s", info->id, key);
return true;
}
uint8_t dummy_val = 0;
taosHashPut(pHash, keyLower, strlen(key), &dummy_val, sizeof(uint8_t));
taosHashPut(pHash, key, strlen(key), &dummy_val, sizeof(uint8_t));
return false;
}
......@@ -1925,7 +1916,6 @@ static int32_t parseSmlKey(TAOS_SML_KV *pKV, const char **index, SHashObj *pHash
pKV->key = calloc(len + TS_ESCAPE_CHAR_SIZE + 1, 1);
memcpy(pKV->key, key, len + 1);
strntolower_s(pKV->key, pKV->key, (int32_t)len);
addEscapeCharToString(pKV->key, len);
tscDebug("SML:0x%"PRIx64" Key:%s|len:%d", info->id, pKV->key, len);
*index = cur + 1;
......@@ -2053,7 +2043,7 @@ static int32_t parseSmlMeasurement(TAOS_SML_DATA_POINT *pSml, const char **index
if (*cur == '\\') {
escapeSpecialCharacter(1, &cur);
}
pSml->stableName[len] = tolower(*cur);
pSml->stableName[len] = *cur;
cur++;
len++;
}
......@@ -2129,7 +2119,6 @@ static int32_t parseSmlKvPairs(TAOS_SML_KV **pKVs, int *num_kvs,
if (!isField && childTableNameLen != 0 && strcasecmp(pkv->key, childTableName) == 0) {
smlData->childTableName = malloc(pkv->length + TS_ESCAPE_CHAR_SIZE + 1);
memcpy(smlData->childTableName, pkv->value, pkv->length);
strntolower_s(smlData->childTableName, smlData->childTableName, (int32_t)pkv->length);
addEscapeCharToString(smlData->childTableName, (int32_t)pkv->length);
free(pkv->key);
free(pkv->value);
......
......@@ -65,7 +65,7 @@ static int32_t parseTelnetMetric(TAOS_SML_DATA_POINT *pSml, const char **index,
}
}
pSml->stableName[len] = tolower(*cur);
pSml->stableName[len] = *cur;
cur++;
len++;
......@@ -195,8 +195,9 @@ static int32_t parseTelnetMetricValue(TAOS_SML_KV **pKVs, int *num_kvs, const ch
}
tfree(value);
pVal->key = tcalloc(sizeof(key), 1);
pVal->key = tcalloc(sizeof(key) + TS_ESCAPE_CHAR_SIZE, 1);
memcpy(pVal->key, key, sizeof(key));
addEscapeCharToString(pVal->key, (int32_t)strlen(pVal->key));
*num_kvs += 1;
*index = cur + 1;
......@@ -240,7 +241,6 @@ static int32_t parseTelnetTagKey(TAOS_SML_KV *pKV, const char **index, SHashObj
pKV->key = tcalloc(len + TS_ESCAPE_CHAR_SIZE + 1, 1);
memcpy(pKV->key, key, len + 1);
strntolower_s(pKV->key, pKV->key, (int32_t)len);
addEscapeCharToString(pKV->key, len);
//tscDebug("OTD:0x%"PRIx64" Key:%s|len:%d", info->id, pKV->key, len);
*index = cur + 1;
......@@ -326,7 +326,6 @@ static int32_t parseTelnetTagKvs(TAOS_SML_KV **pKVs, int *num_kvs,
*childTableName = tcalloc(pkv->length + TS_ESCAPE_CHAR_SIZE + 1, 1);
memcpy(*childTableName, pkv->value, pkv->length);
(*childTableName)[pkv->length] = '\0';
strntolower_s(*childTableName, *childTableName, (int32_t)pkv->length);
addEscapeCharToString(*childTableName, pkv->length);
tfree(pkv->key);
tfree(pkv->value);
......@@ -514,7 +513,6 @@ static int32_t parseMetricFromJSON(cJSON *root, TAOS_SML_DATA_POINT* pSml, SSmlL
*/
tstrncpy(pSml->stableName, metric->valuestring, stableLen + 1);
strntolower_s(pSml->stableName, pSml->stableName, (int32_t)stableLen);
addEscapeCharToString(pSml->stableName, (int32_t)stableLen);
return TSDB_CODE_SUCCESS;
......@@ -545,7 +543,6 @@ static int32_t parseTimestampFromJSONObj(cJSON *root, int64_t *tsVal, SSmlLinesI
}
size_t typeLen = strlen(type->valuestring);
strntolower_s(type->valuestring, type->valuestring, (int32_t)typeLen);
if (typeLen == 1 && type->valuestring[0] == 's') {
//seconds
*tsVal = (int64_t)(*tsVal * 1e9);
......@@ -881,8 +878,9 @@ static int32_t parseMetricValueFromJSON(cJSON *root, TAOS_SML_KV **pKVs, int *nu
return ret;
}
pVal->key = tcalloc(sizeof(key), 1);
pVal->key = tcalloc(sizeof(key) + TS_ESCAPE_CHAR_SIZE, 1);
memcpy(pVal->key, key, sizeof(key));
addEscapeCharToString(pVal->key, (int32_t)strlen(pVal->key));
*num_kvs += 1;
return TSDB_CODE_SUCCESS;
......@@ -913,7 +911,6 @@ static int32_t parseTagsFromJSON(cJSON *root, TAOS_SML_KV **pKVs, int *num_kvs,
size_t idLen = strlen(id->valuestring);
*childTableName = tcalloc(idLen + TS_ESCAPE_CHAR_SIZE + 1, sizeof(char));
memcpy(*childTableName, id->valuestring, idLen);
strntolower_s(*childTableName, *childTableName, (int32_t)idLen);
addEscapeCharToString(*childTableName, (int32_t)idLen);
//check duplicate IDs
......@@ -952,7 +949,6 @@ static int32_t parseTagsFromJSON(cJSON *root, TAOS_SML_KV **pKVs, int *num_kvs,
}
pkv->key = tcalloc(keyLen + TS_ESCAPE_CHAR_SIZE + 1, sizeof(char));
strncpy(pkv->key, tag->string, keyLen);
strntolower_s(pkv->key, pkv->key, (int32_t)keyLen);
addEscapeCharToString(pkv->key, (int32_t)keyLen);
//value
ret = parseValueFromJSON(tag, pkv, info);
......
......@@ -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;
......@@ -1250,6 +1252,12 @@ static void insertBatchClean(STscStmt* pStmt) {
pCmd->insertParam.pDataBlocks = tscDestroyBlockArrayList(pSql, 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);
......@@ -1342,11 +1350,42 @@ 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) && (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);
if (sToken.n <= 0 || sToken.type != TK_LP) {
......@@ -1385,6 +1424,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);
......@@ -1407,7 +1451,13 @@ int stmtGenInsertStatement(SSqlObj* pSql, STscStmt* pStmt, const char* name, TAO
int32_t j = 0;
while (1) {
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);
......@@ -1573,7 +1623,7 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
pRes->qId = 0;
pRes->numOfRows = 0;
strtolower(pSql->sqlstr, sql);
strcpy(pSql->sqlstr, sql);
tscDebugL("0x%"PRIx64" SQL: %s", pSql->self, pSql->sqlstr);
if (tscIsInsertData(pSql->sqlstr)) {
......@@ -1659,6 +1709,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) {
......@@ -1784,7 +1841,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
......@@ -1792,8 +1848,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
......@@ -1801,7 +1855,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) {
......@@ -1868,7 +1921,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;
......@@ -1932,8 +1984,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
......@@ -2086,7 +2136,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;
......@@ -2097,8 +2146,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";
......@@ -2115,4 +2162,3 @@ const char *taos_data_type(int type) {
default: return "UNKNOWN";
}
}
此差异已折叠。
......@@ -916,6 +916,8 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg->window.skey = htobe64(query.window.skey);
pQueryMsg->window.ekey = htobe64(query.window.ekey);
pQueryMsg->range.skey = htobe64(query.range.skey);
pQueryMsg->range.ekey = htobe64(query.range.ekey);
pQueryMsg->order = htons(query.order.order);
pQueryMsg->orderColId = htons(query.order.orderColId);
......@@ -975,7 +977,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
if (pQueryInfo->colCond && taosArrayGetSize(pQueryInfo->colCond) > 0 && !onlyQueryTags(&query) ) {
STblCond *pCond = tsGetTableFilter(pQueryInfo->colCond, pTableMeta->id.uid, 0);
if (pCond != NULL && pCond->cond != NULL) {
pQueryMsg->colCondLen = htons(pCond->len);
pQueryMsg->colCondLen = htonl(pCond->len);
memcpy(pMsg, pCond->cond, pCond->len);
pMsg += pCond->len;
......@@ -1056,7 +1058,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SCond *pCond = tsGetSTableQueryCond(pTagCond, pTableMeta->id.uid);
if (pCond != NULL && pCond->cond != NULL) {
pQueryMsg->tagCondLen = htons(pCond->len);
pQueryMsg->tagCondLen = htonl(pCond->len);
memcpy(pMsg, pCond->cond, pCond->len);
pMsg += pCond->len;
......@@ -1534,7 +1536,17 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pMsg += sizeof(SCreateTableMsg);
SCreatedTableInfo* p = taosArrayGet(list, i);
//what pCreate->tableName point is a fixed char array which size is 237
//what p->fullname point is a char*
//before the time we copy p->fullname to pCreate->tableName , we need to check the length of p->fullname
if (strlen(p->fullname) > 237) {
tscError("failed to write this name, which is over 237, just save the first 237 char here");
strncpy(pCreate->tableName, p->fullname,237);
pCreate->tableName[236]='\0';//I don't know if this line is working properly
}else{
strcpy(pCreate->tableName, p->fullname);
}
pCreate->igExists = (p->igExist)? 1 : 0;
// use dbinfo from table id without modifying current db info
......@@ -3074,12 +3086,16 @@ int tscRenewTableMeta(SSqlObj *pSql, int32_t tableIndex) {
pSql->rootObj->retryReason = pSql->retryReason;
SSqlObj *tmpSql = pSql->rootObj;
tscFreeSubobj(pSql->rootObj);
tfree(tmpSql->pSubs);
SArray* pNameList = taosArrayInit(1, POINTER_BYTES);
SArray* vgroupList = taosArrayInit(1, POINTER_BYTES);
char* n = strdup(name);
taosArrayPush(pNameList, &n);
code = getMultiTableMetaFromMnode(pSql, pNameList, vgroupList, NULL, tscTableMetaCallBack, true);
code = getMultiTableMetaFromMnode(tmpSql, pNameList, vgroupList, NULL, tscTableMetaCallBack, true);
taosArrayDestroyEx(pNameList, freeElem);
taosArrayDestroyEx(vgroupList, freeElem);
......
......@@ -29,6 +29,7 @@
#include "ttimer.h"
#include "tscProfile.h"
static char clusterDefaultId[] = "clusterDefaultId";
static bool validImpl(const char* str, size_t maxsize) {
if (str == NULL) {
return false;
......@@ -193,7 +194,9 @@ TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass,
tscBuildAndSendRequest(pSql, NULL);
tsem_wait(&pSql->rspSem);
if (0 == strlen(pSql->pTscObj->clusterId)) {
memcpy(pSql->pTscObj->clusterId, clusterDefaultId, strlen(clusterDefaultId));
}
pSql->pTscObj->pClusterInfo = (SClusterInfo *)tscAcquireClusterInfo(pSql->pTscObj->clusterId);
if (pSql->res.code != TSDB_CODE_SUCCESS) {
terrno = pSql->res.code;
......
......@@ -468,7 +468,7 @@ SSqlObj* recreateSqlObj(SSub* pSub) {
}
registerSqlObj(pSql);
pSql->rootObj = pSql;
code = tsParseSql(pSql, true);
if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) {
tsem_wait(&pSub->sem);
......
......@@ -394,6 +394,12 @@ SJoinSupporter* tscCreateJoinSupporter(SSqlObj* pSql, int32_t index) {
memcpy(&pSupporter->interval, &pQueryInfo->interval, sizeof(pSupporter->interval));
pSupporter->limit = pQueryInfo->limit;
if (tscIsPointInterpQuery(pQueryInfo)) {
pSupporter->fillType = pQueryInfo->fillType;
pSupporter->fillVal = pQueryInfo->fillVal;
pSupporter->numOfFillVal = pQueryInfo->numOfFillVal;
}
STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, index);
pSupporter->uid = pTableMetaInfo->pTableMeta->id.uid;
assert (pSupporter->uid != 0);
......@@ -580,6 +586,13 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) {
pQueryInfo->groupbyExpr = pSupporter->groupInfo;
pQueryInfo->pUpstream = taosArrayInit(4, sizeof(POINTER_BYTES));
if (tscIsPointInterpQuery(pQueryInfo)) {
pQueryInfo->fillType = pSupporter->fillType;
pQueryInfo->numOfFillVal = pSupporter->numOfFillVal;
pQueryInfo->fillVal = malloc(pQueryInfo->numOfFillVal * sizeof(*pSupporter->fillVal));
memcpy(pQueryInfo->fillVal, pSupporter->fillVal, sizeof(*pSupporter->fillVal) * pQueryInfo->numOfFillVal);
}
assert(pNew->subState.numOfSub == 0 && pQueryInfo->numOfTables == 1);
tscFieldInfoUpdateOffset(pQueryInfo);
......@@ -1987,7 +2000,7 @@ int32_t tscCreateJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSupporter
colIndex.columnIndex = tscGetTagColIndexById(pTableMetaInfo->pTableMeta, tagColId);
int16_t bytes = 0;
int32_t bytes = 0;
int16_t type = 0;
int32_t inter = 0;
......@@ -2623,7 +2636,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
pRes->qId = 0x1; // hack the qhandle check
const uint32_t nBufferSize = (1u << 18u); // 256KB, default buffer size
uint32_t nBufferSize = (1u << 18u); // 256KB, default buffer size
SQueryInfo *pQueryInfo = tscGetQueryInfo(pCmd);
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
......@@ -2639,7 +2652,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
return ret;
}
ret = tscCreateGlobalMergerEnv(pQueryInfo, &pMemoryBuf, pSql->subState.numOfSub, &pDesc, nBufferSize, pSql->self);
ret = tscCreateGlobalMergerEnv(pQueryInfo, &pMemoryBuf, pSql->subState.numOfSub, &pDesc, &nBufferSize, pSql->self);
if (ret != 0) {
pRes->code = ret;
tscAsyncResultOnError(pSql);
......
......@@ -133,8 +133,8 @@ void tscClusterInfoDestroy(SClusterInfo *pObj) {
void *tscAcquireClusterInfo(const char *clusterId) {
pthread_mutex_lock(&clusterMutex);
size_t len = strlen(clusterId);
SClusterInfo *pObj = NULL;
SClusterInfo **ppObj = taosHashGet(tscClusterMap, clusterId, len);
if (ppObj == NULL || *ppObj == NULL) {
......@@ -210,9 +210,9 @@ void taos_init_imp(void) {
taosInitNotes();
rpcInit();
#ifdef LUA_EMBEDDED
scriptEnvPoolInit();
#endif
tscDebug("starting to initialize TAOS client ...");
tscDebug("Local End Point is:%s", tsLocalEp);
}
......@@ -276,7 +276,9 @@ void taos_cleanup(void) {
}
if (tscEmbedded == 0) {
#ifdef LUA_EMBEDDED
scriptEnvPoolCleanup();
#endif
}
int32_t id = tscObjRef;
......
......@@ -187,7 +187,7 @@ bool tscQueryTags(SQueryInfo* pQueryInfo) {
continue;
}
if (functId != TSDB_FUNC_TAGPRJ && functId != TSDB_FUNC_TID_TAG) {
if (functId != TSDB_FUNC_TAGPRJ && functId != TSDB_FUNC_TID_TAG && functId != TSDB_FUNC_BLKINFO) {
return false;
}
}
......@@ -268,10 +268,6 @@ bool tscIsProjectionQueryOnSTable(SQueryInfo* pQueryInfo, int32_t tableIndex) {
functionId != TSDB_FUNC_TS &&
functionId != TSDB_FUNC_ARITHM &&
functionId != TSDB_FUNC_TS_COMP &&
functionId != TSDB_FUNC_DIFF &&
functionId != TSDB_FUNC_DERIVATIVE &&
functionId != TSDB_FUNC_MAVG &&
functionId != TSDB_FUNC_CSUM &&
functionId != TSDB_FUNC_TS_DUMMY &&
functionId != TSDB_FUNC_TID_TAG &&
functionId != TSDB_FUNC_CEIL &&
......@@ -367,7 +363,7 @@ bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo) {
assert(pExpr != NULL);
int32_t functionId = pExpr->base.functionId;
if (functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TS) {
if (functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TAG_DUMMY || functionId == TSDB_FUNC_TS || functionId == TSDB_FUNC_TS_DUMMY) {
continue;
}
......@@ -379,6 +375,23 @@ bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo) {
return true;
}
bool tscGetPointInterpQuery(SQueryInfo* pQueryInfo) {
size_t size = tscNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < size; ++i) {
SExprInfo* pExpr = tscExprGet(pQueryInfo, i);
assert(pExpr != NULL);
int32_t functionId = pExpr->base.functionId;
if (functionId == TSDB_FUNC_INTERP) {
return true;
}
}
return false;
}
bool tsIsArithmeticQueryOnAggResult(SQueryInfo* pQueryInfo) {
if (tscIsProjectionQuery(pQueryInfo)) {
return false;
......@@ -942,23 +955,30 @@ SSDataBlock* doGetDataBlock(void* param, bool* newgroup) {
pBlock->info.rows = pRes->numOfRows;
if (pRes->numOfRows != 0) {
doSetupSDataBlock(pRes, pBlock, pInput->pFilterInfo);
if (pBlock->info.rows > 0) {
*newgroup = false;
return pBlock;
}
}
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;
result = NULL;
break;
}
pBlock->info.rows = pRes->numOfRows;
doSetupSDataBlock(pRes, pBlock, pInput->pFilterInfo);
*newgroup = false;
return pBlock;
result = pBlock;
} while (result->info.rows == 0);
return result;
}
static void fetchNextBlockIfCompleted(SOperatorInfo* pOperator, bool* newgroup) {
......@@ -1838,6 +1858,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;
......@@ -1850,30 +1896,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;
}
......@@ -2953,6 +2985,11 @@ int32_t tscValidateName(SStrToken* pToken, bool escapeEnabled, bool *dbIncluded)
}
} else if (pToken->type == TK_ID) {
tscRmEscapeAndTrimToken(pToken);
if (pToken->n == 0) {
return TSDB_CODE_TSC_INVALID_OPERATION;
}
return TSDB_CODE_SUCCESS;
} else {
if (isNumber(pToken)) {
......@@ -3417,10 +3454,11 @@ int32_t tscQueryInfoCopy(SQueryInfo* pQueryInfo, const SQueryInfo* pSrc) {
pQueryInfo->clauseLimit = pSrc->clauseLimit;
pQueryInfo->prjOffset = pSrc->prjOffset;
pQueryInfo->numOfTables = 0;
pQueryInfo->window = pSrc->window;
pQueryInfo->range = pSrc->range;
pQueryInfo->sessionWindow = pSrc->sessionWindow;
pQueryInfo->pTableMetaInfo = NULL;
pQueryInfo->multigroupResult = pSrc->multigroupResult;
pQueryInfo->stateWindow = pSrc->stateWindow;
pQueryInfo->bufLen = pSrc->bufLen;
pQueryInfo->orderProjectQuery = pSrc->orderProjectQuery;
......@@ -3810,6 +3848,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, __async_cb_func_t
memcpy(&pNewQueryInfo->interval, &pQueryInfo->interval, sizeof(pNewQueryInfo->interval));
pNewQueryInfo->type = pQueryInfo->type;
pNewQueryInfo->window = pQueryInfo->window;
pNewQueryInfo->range = pQueryInfo->range;
pNewQueryInfo->limit = pQueryInfo->limit;
pNewQueryInfo->slimit = pQueryInfo->slimit;
pNewQueryInfo->order = pQueryInfo->order;
......@@ -4824,7 +4863,7 @@ int32_t createProjectionExpr(SQueryInfo* pQueryInfo, STableMetaInfo* pTableMetaI
}
}
pse->colInfo.flag = TSDB_COL_NORMAL;
pse->colInfo.flag = pSource->base.colInfo.flag; //TSDB_COL_NORMAL;
pse->resType = pSource->base.resType;
pse->resBytes = pSource->base.resBytes;
strncpy(pse->colInfo.name, pSource->base.aliasName, tListLen(pse->colInfo.name));
......@@ -5029,6 +5068,7 @@ int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAt
pQueryAttr->fillType = pQueryInfo->fillType;
pQueryAttr->havingNum = pQueryInfo->havingFieldNum;
pQueryAttr->pUdfInfo = pQueryInfo->pUdfInfo;
pQueryAttr->range = pQueryInfo->range;
if (pQueryInfo->order.order == TSDB_ORDER_ASC) { // TODO refactor
pQueryAttr->window = pQueryInfo->window;
......@@ -5328,4 +5368,3 @@ char* cloneCurrentDBName(SSqlObj* pSql) {
return p;
}
......@@ -148,6 +148,7 @@ extern char tsMqttTopic[];
// monitor
extern int8_t tsEnableMonitorModule;
extern int8_t tsMonitorReplica;
extern char tsMonitorDbName[];
extern char tsInternalPass[];
extern int32_t tsMonitorInterval;
......
......@@ -51,7 +51,7 @@ typedef struct SSqlExpr {
int16_t functionId; // function id in aAgg array
int16_t resType; // return value type
int16_t resBytes; // length of return value
int32_t resBytes; // length of return value
int32_t interBytes; // inter result buffer size
int16_t colType; // table column type
......
......@@ -39,7 +39,7 @@ typedef struct tVariant {
bool tVariantIsValid(tVariant *pVar);
void tVariantCreate(tVariant *pVar, SStrToken *token);
void tVariantCreate(tVariant *pVar, SStrToken *token, bool needRmquoteEscape);
void tVariantCreateFromBinary(tVariant *pVar, const char *pz, size_t len, uint32_t type);
......
......@@ -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;
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;
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;
......
......@@ -73,7 +73,7 @@ int32_t tsMaxBinaryDisplayWidth = 30;
* -1: all data are not compressed
* other values: if the message payload size is greater than the tsCompressMsgSize, the message will be compressed.
*/
int32_t tsCompressMsgSize = -1;
int32_t tsCompressMsgSize = 512 * 1024;
/* denote if server needs to compress the retrieved column data before adding to the rpc response message body.
* 0: all data are compressed
......@@ -193,6 +193,7 @@ char tsMqttTopic[TSDB_MQTT_TOPIC_LEN] = "/test"; // #
// monitor
int8_t tsEnableMonitorModule = 1;
int8_t tsMonitorReplica = 1;
char tsMonitorDbName[TSDB_DB_NAME_LEN] = "log";
char tsInternalPass[] = "secretkey";
int32_t tsMonitorInterval = 30; // seconds
......@@ -289,7 +290,7 @@ char Compressor[32] = "ZSTD_COMPRESSOR"; // ZSTD_COMPRESSOR or GZIP_COMPRES
int8_t tsDeadLockKillQuery = 0;
// default JSON string type
char tsDefaultJSONStrType[7] = "binary";
char tsDefaultJSONStrType[7] = "nchar";
char tsSmlChildTableName[TSDB_TABLE_NAME_LEN] = ""; //user defined child table name can be specified in tag value. If set to empty system will generate table name using MD5 hash.
int32_t (*monStartSystemFp)() = NULL;
......@@ -669,6 +670,16 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_SECOND;
taosInitConfigOption(cfg);
cfg.option = "monitorReplica";
cfg.ptr = &tsMonitorReplica;
cfg.valType = TAOS_CFG_VTYPE_INT8;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW;
cfg.minValue = 1;
cfg.maxValue = 3;
cfg.ptrLength = 1;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "offlineThreshold";
cfg.ptr = &tsOfflineThreshold;
cfg.valType = TAOS_CFG_VTYPE_INT32;
......
......@@ -30,7 +30,7 @@
assert(0); \
} while (0)
void tVariantCreate(tVariant *pVar, SStrToken *token) {
void tVariantCreate(tVariant *pVar, SStrToken *token, bool needRmquoteEscape) {
int32_t ret = 0;
int32_t type = token->type;
......@@ -81,7 +81,7 @@ void tVariantCreate(tVariant *pVar, SStrToken *token) {
case TSDB_DATA_TYPE_BINARY: {
pVar->pz = strndup(token->z, token->n);
pVar->nLen = strRmquoteEscape(pVar->pz, token->n);
pVar->nLen = needRmquoteEscape ? strRmquoteEscape(pVar->pz, token->n) : token->n;
break;
}
case TSDB_DATA_TYPE_TIMESTAMP: {
......
......@@ -163,8 +163,12 @@ namespace TDengineDriver
[DllImport("taos", EntryPoint = "taos_close", CallingConvention = CallingConvention.Cdecl)]
static extern public int Close(IntPtr taos);
//get precisionin parameter restultset
//get precision in restultset
[DllImport("taos", EntryPoint = "taos_result_precision", CallingConvention = CallingConvention.Cdecl)]
static extern public int ResultPrecision(IntPtr taos);
//schemaless API
[DllImport("taos",SetLastError = true, EntryPoint = "taos_schemaless_insert", CallingConvention = CallingConvention.Cdecl)]
static extern public IntPtr SchemalessInsert(IntPtr taos, string[] lines, int numLines, int protocol, int precision);
}
}
......@@ -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.config.RequestConfig;
import org.apache.http.client.methods.*;
......
......@@ -29,8 +29,8 @@ extern "C" {
extern int32_t dDebugFlag;
#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", 255, __VA_ARGS__); }}
#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", 255, __VA_ARGS__); }}
#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", 255, __VA_ARGS__); dnodeIncDnodeError(); }}
#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", 255, __VA_ARGS__); dnodeIncDnodeError(); }}
#define dWarn(...) { if (dDebugFlag & DEBUG_WARN) { taosPrintLog("DND WARN ", 255, __VA_ARGS__); }}
#define dInfo(...) { if (dDebugFlag & DEBUG_INFO) { taosPrintLog("DND ", 255, __VA_ARGS__); }}
#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", dDebugFlag, __VA_ARGS__); }}
......
......@@ -54,6 +54,7 @@ void moduleStop() {}
void *tsDnodeTmr = NULL;
static SRunStatus tsRunStatus = TSDB_RUN_STATUS_STOPPED;
static int64_t tsDnodeErrors = 0;
static int32_t dnodeInitStorage();
static void dnodeCleanupStorage();
......@@ -88,7 +89,9 @@ static SStep tsDnodeSteps[] = {
{"dnode-shell", dnodeInitShell, dnodeCleanupShell},
{"dnode-statustmr", dnodeInitStatusTimer,dnodeCleanupStatusTimer},
{"dnode-telemetry", dnodeInitTelemetry, dnodeCleanupTelemetry},
#ifdef LUA_EMBEDDED
{"dnode-script", scriptEnvPoolInit, scriptEnvPoolCleanup},
#endif
{"dnode-grant", grantInit, grantCleanUp},
};
......@@ -225,6 +228,14 @@ static void dnodeSetRunStatus(SRunStatus status) {
tsRunStatus = status;
}
int64_t dnodeGetDnodeError() {
return tsDnodeErrors;
}
void dnodeIncDnodeError() {
atomic_add_fetch_64(&tsDnodeErrors, 1);
}
static void dnodeCheckDataDirOpenned(char *dir) {
char filepath[256] = {0};
sprintf(filepath, "%s/.running", dir);
......
......@@ -28,8 +28,8 @@ static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *);
static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
static void * tsShellRpc = NULL;
static int32_t tsQueryReqNum = 0;
static int32_t tsSubmitReqNum = 0;
static int64_t tsQueryReqNum = 0;
static int64_t tsSubmitReqNum = 0;
int32_t dnodeInitShell() {
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeDispatchToVWriteQueue;
......@@ -136,9 +136,9 @@ static void dnodeProcessMsgFromShell(SRpcMsg *pMsg, SRpcEpSet *pEpSet) {
}
if (pMsg->msgType == TSDB_MSG_TYPE_QUERY) {
atomic_fetch_add_32(&tsQueryReqNum, 1);
atomic_fetch_add_64(&tsQueryReqNum, 1);
} else if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) {
atomic_fetch_add_32(&tsSubmitReqNum, 1);
atomic_fetch_add_64(&tsSubmitReqNum, 1);
} else {}
if ( dnodeProcessShellMsgFp[pMsg->msgType] ) {
......@@ -237,15 +237,31 @@ void *dnodeSendCfgTableToRecv(int32_t vgId, int32_t tid) {
}
}
SStatisInfo dnodeGetStatisInfo() {
SStatisInfo info = {0};
SDnodeStatisInfo dnodeGetStatisInfo() {
SDnodeStatisInfo info = {0};
if (dnodeGetRunStatus() == TSDB_RUN_STATUS_RUNING) {
#ifdef HTTP_EMBEDDED
info.httpReqNum = httpGetReqCount();
#endif
info.queryReqNum = atomic_exchange_32(&tsQueryReqNum, 0);
info.submitReqNum = atomic_exchange_32(&tsSubmitReqNum, 0);
info.queryReqNum = atomic_exchange_64(&tsQueryReqNum, 0);
info.submitReqNum = atomic_exchange_64(&tsSubmitReqNum, 0);
}
return info;
}
int32_t dnodeGetHttpStatusInfo(int32_t index) {
int32_t httpStatus = 0;
#ifdef HTTP_EMBEDDED
httpStatus = httpGetStatusCodeCount(index);
#endif
return httpStatus;
}
void dnodeClearHttpStatusInfo() {
#ifdef HTTP_EMBEDDED
for (int i = 0; i < MAX_HTTP_STATUS_CODE_NUM; ++i) {
httpClearStatusCodeCount(i);
}
#endif
}
......@@ -23,13 +23,16 @@ extern "C" {
#include "trpc.h"
#include "taosmsg.h"
#define MAX_HTTP_STATUS_CODE_NUM 63
typedef struct {
int32_t queryReqNum;
int32_t submitReqNum;
int32_t httpReqNum;
} SStatisInfo;
int64_t queryReqNum;
int64_t submitReqNum;
int64_t httpReqNum;
} SDnodeStatisInfo;
SStatisInfo dnodeGetStatisInfo();
SDnodeStatisInfo dnodeGetStatisInfo();
int32_t dnodeGetHttpStatusInfo(int32_t index);
void dnodeClearHttpStatusInfo();
bool dnodeIsFirstDeploy();
bool dnodeIsMasterEp(char *ep);
......@@ -37,6 +40,8 @@ void dnodeGetEpSetForPeer(SRpcEpSet *epSet);
void dnodeGetEpSetForShell(SRpcEpSet *epSet);
int32_t dnodeGetDnodeId();
void dnodeGetClusterId(char *clusterId);
int64_t dnodeGetDnodeError();
void dnodeIncDnodeError();
void dnodeUpdateEp(int32_t dnodeId, char *ep, char *fqdn, uint16_t *port);
bool dnodeCheckEpChanged(int32_t dnodeId, char *epstr);
......
......@@ -22,7 +22,9 @@ extern "C" {
#include <stdint.h>
int32_t httpGetReqCount();
int64_t httpGetReqCount();
int32_t httpGetStatusCodeCount(int index);
int32_t httpClearStatusCodeCount(int index);
int32_t httpInitSystem();
int32_t httpStartSystem();
void httpStopSystem();
......
......@@ -22,6 +22,17 @@ extern "C" {
#include <stdint.h>
#define monSaveLogs(level, ...) { \
monSaveLog(level, __VA_ARGS__); \
monSaveDnodeLog(level, __VA_ARGS__); \
}
typedef struct {
const char * name;
int32_t code;
int32_t index;
} SMonHttpStatus;
typedef struct {
char * acctId;
int64_t currentPointsPerSecond;
......@@ -53,9 +64,16 @@ void monStopSystem();
void monCleanupSystem();
void monSaveAcctLog(SAcctMonitorObj *pMonObj);
void monSaveLog(int32_t level, const char *const format, ...);
void monSaveDnodeLog(int32_t level, const char *const format, ...);
void monExecuteSQL(char *sql);
typedef void (*MonExecuteSQLCbFP)(void *param, TAOS_RES *, int code);
void monExecuteSQLWithResultCallback(char *sql, MonExecuteSQLCbFP callback, void* param);
void monIncQueryReqCnt();
void monIncSubmitReqCnt();
int32_t monFetchQueryReqCnt();
int32_t monFetchSubmitReqCnt();
SMonHttpStatus *monGetHttpStatusHashTableEntry(int32_t code);
#ifdef __cplusplus
}
#endif
......
......@@ -438,7 +438,7 @@ typedef struct SColumnFilterList {
typedef struct SColumnInfo {
int16_t colId;
int16_t type;
int16_t bytes;
int32_t bytes;
SColumnFilterList flist;
} SColumnInfo;
......@@ -479,14 +479,15 @@ typedef struct {
bool stateWindow; // state window flag
STimeWindow window;
STimeWindow range; // result range for interp query
int32_t numOfTables;
int16_t order;
int16_t orderColId;
int16_t numOfCols; // the number of columns will be load from vnode
SInterval interval;
SSessionWindow sw; // session window
uint16_t tagCondLen; // tag length in current query
uint16_t colCondLen; // column length in current query
uint32_t tagCondLen; // tag length in current query
int32_t colCondLen; // column length in current query
int16_t numOfGroupCols; // num of group by columns
int16_t orderByIdx;
int16_t orderType; // used in group by xx order by xxx
......
......@@ -405,6 +405,7 @@ void tsdbDestroyCommitQueue();
int tsdbSyncCommit(STsdbRepo *repo);
void tsdbIncCommitRef(int vgId);
void tsdbDecCommitRef(int vgId);
void tsdbSwitchTable(TsdbQueryHandleT pQueryHandle);
// For TSDB file sync
int tsdbSyncSend(void *pRepo, SOCKET socketFd);
......
......@@ -142,77 +142,78 @@
#define TK_DISTINCT 124
#define TK_FROM 125
#define TK_VARIABLE 126
#define TK_INTERVAL 127
#define TK_EVERY 128
#define TK_SESSION 129
#define TK_STATE_WINDOW 130
#define TK_FILL 131
#define TK_SLIDING 132
#define TK_ORDER 133
#define TK_BY 134
#define TK_ASC 135
#define TK_GROUP 136
#define TK_HAVING 137
#define TK_LIMIT 138
#define TK_OFFSET 139
#define TK_SLIMIT 140
#define TK_SOFFSET 141
#define TK_WHERE 142
#define TK_RESET 143
#define TK_QUERY 144
#define TK_SYNCDB 145
#define TK_ADD 146
#define TK_COLUMN 147
#define TK_MODIFY 148
#define TK_TAG 149
#define TK_CHANGE 150
#define TK_SET 151
#define TK_KILL 152
#define TK_CONNECTION 153
#define TK_STREAM 154
#define TK_COLON 155
#define TK_ABORT 156
#define TK_AFTER 157
#define TK_ATTACH 158
#define TK_BEFORE 159
#define TK_BEGIN 160
#define TK_CASCADE 161
#define TK_CLUSTER 162
#define TK_CONFLICT 163
#define TK_COPY 164
#define TK_DEFERRED 165
#define TK_DELIMITERS 166
#define TK_DETACH 167
#define TK_EACH 168
#define TK_END 169
#define TK_EXPLAIN 170
#define TK_FAIL 171
#define TK_FOR 172
#define TK_IGNORE 173
#define TK_IMMEDIATE 174
#define TK_INITIALLY 175
#define TK_INSTEAD 176
#define TK_KEY 177
#define TK_OF 178
#define TK_RAISE 179
#define TK_REPLACE 180
#define TK_RESTRICT 181
#define TK_ROW 182
#define TK_STATEMENT 183
#define TK_TRIGGER 184
#define TK_VIEW 185
#define TK_IPTOKEN 186
#define TK_SEMI 187
#define TK_NONE 188
#define TK_PREV 189
#define TK_LINEAR 190
#define TK_IMPORT 191
#define TK_TBNAME 192
#define TK_JOIN 193
#define TK_INSERT 194
#define TK_INTO 195
#define TK_VALUES 196
#define TK_FILE 197
#define TK_RANGE 127
#define TK_INTERVAL 128
#define TK_EVERY 129
#define TK_SESSION 130
#define TK_STATE_WINDOW 131
#define TK_FILL 132
#define TK_SLIDING 133
#define TK_ORDER 134
#define TK_BY 135
#define TK_ASC 136
#define TK_GROUP 137
#define TK_HAVING 138
#define TK_LIMIT 139
#define TK_OFFSET 140
#define TK_SLIMIT 141
#define TK_SOFFSET 142
#define TK_WHERE 143
#define TK_RESET 144
#define TK_QUERY 145
#define TK_SYNCDB 146
#define TK_ADD 147
#define TK_COLUMN 148
#define TK_MODIFY 149
#define TK_TAG 150
#define TK_CHANGE 151
#define TK_SET 152
#define TK_KILL 153
#define TK_CONNECTION 154
#define TK_STREAM 155
#define TK_COLON 156
#define TK_ABORT 157
#define TK_AFTER 158
#define TK_ATTACH 159
#define TK_BEFORE 160
#define TK_BEGIN 161
#define TK_CASCADE 162
#define TK_CLUSTER 163
#define TK_CONFLICT 164
#define TK_COPY 165
#define TK_DEFERRED 166
#define TK_DELIMITERS 167
#define TK_DETACH 168
#define TK_EACH 169
#define TK_END 170
#define TK_EXPLAIN 171
#define TK_FAIL 172
#define TK_FOR 173
#define TK_IGNORE 174
#define TK_IMMEDIATE 175
#define TK_INITIALLY 176
#define TK_INSTEAD 177
#define TK_KEY 178
#define TK_OF 179
#define TK_RAISE 180
#define TK_REPLACE 181
#define TK_RESTRICT 182
#define TK_ROW 183
#define TK_STATEMENT 184
#define TK_TRIGGER 185
#define TK_VIEW 186
#define TK_IPTOKEN 187
#define TK_SEMI 188
#define TK_NONE 189
#define TK_PREV 190
#define TK_LINEAR 191
#define TK_IMPORT 192
#define TK_TBNAME 193
#define TK_JOIN 194
#define TK_INSERT 195
#define TK_INTO 196
#define TK_VALUES 197
#define TK_FILE 198
......
......@@ -50,6 +50,47 @@ typedef struct {
#define TSDB_DATA_TYPE_POINTER_ARRAY (1000)
#define TSDB_DATA_TYPE_VALUE_ARRAY (1001)
#define COPY_DATA(dst, src) *((int64_t *)(dst)) = *((int64_t *)(src))
#define COPY_TYPED_DATA(_v, _type, _data) \
do { \
switch (_type) { \
case TSDB_DATA_TYPE_BOOL: \
case TSDB_DATA_TYPE_TINYINT: \
(*(int8_t *)_v) = GET_INT8_VAL(_data); \
break; \
case TSDB_DATA_TYPE_UTINYINT: \
(*(uint8_t *)_v) = GET_UINT8_VAL(_data); \
break; \
case TSDB_DATA_TYPE_SMALLINT: \
(*(int16_t *)_v) = GET_INT16_VAL(_data); \
break; \
case TSDB_DATA_TYPE_USMALLINT: \
(*(uint16_t *)_v) = GET_UINT16_VAL(_data); \
break; \
case TSDB_DATA_TYPE_TIMESTAMP: \
case TSDB_DATA_TYPE_BIGINT: \
(*(int64_t *)_v) = (GET_INT64_VAL(_data)); \
break; \
case TSDB_DATA_TYPE_UBIGINT: \
(*(uint64_t *)_v) = (GET_UINT64_VAL(_data)); \
break; \
case TSDB_DATA_TYPE_FLOAT: \
(*(float *)_v) = GET_FLOAT_VAL(_data); \
break; \
case TSDB_DATA_TYPE_DOUBLE: \
(*(double *)_v) = GET_DOUBLE_VAL(_data); \
break; \
case TSDB_DATA_TYPE_UINT: \
(*(uint32_t *)_v) = GET_UINT32_VAL(_data); \
break; \
default: \
(*(int32_t *)_v) = GET_INT32_VAL(_data); \
break; \
} \
} while (0)
#define GET_TYPED_DATA(_v, _finalType, _type, _data) \
do { \
switch (_type) { \
......
......@@ -22,6 +22,12 @@ extern "C" {
#include "trpc.h"
#include "twal.h"
typedef struct {
int64_t submitReqSucNum;
int64_t submitRowNum;
int64_t submitRowSucNum;
} SVnodeStatisInfo;
typedef struct {
int32_t len;
void * rsp;
......@@ -80,6 +86,8 @@ int32_t vnodeWriteToWQueue(void *pVnode, void *pHead, int32_t qtype, void *pRpcM
void vnodeFreeFromWQueue(void *pVnode, SVWriteMsg *pWrite);
int32_t vnodeProcessWrite(void *pVnode, void *pHead, int32_t qtype, void *pRspRet);
SVnodeStatisInfo vnodeGetStatisInfo();
// vnodeSync
void vnodeConfirmForward(void *pVnode, uint64_t version, int32_t code, bool force);
......
......@@ -262,7 +262,7 @@ int32_t shellRunCommand(TAOS* con, char* command) {
}
if (c == '\\') {
if (quote != 0 && (*command == '_' || *command == '\\')) {
if (quote != 0 && (*command == '_' || *command == '%' || *command == '\\')) {
//DO nothing
} else {
esc = true;
......
......@@ -2,12 +2,13 @@ CMAKE_MINIMUM_REQUIRED(VERSION 3.0...3.20)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
INCLUDE_DIRECTORIES(inc)
FIND_PACKAGE(Git)
IF (GIT_FOUND)
MESSAGE("Git found")
EXECUTE_PROCESS(
COMMAND ${GIT_EXECUTABLE} log --pretty=oneline -n 1 ${CMAKE_CURRENT_LIST_DIR}/taosdemo.c
COMMAND ${GIT_EXECUTABLE} log --pretty=oneline -n 1 ${CMAKE_CURRENT_LIST_DIR}
WORKING_DIRECTORY ${CMAKE_CURRENT_LIST_DIR}
RESULT_VARIABLE RESULT
OUTPUT_VARIABLE TAOSDEMO_COMMIT_SHA1)
......@@ -18,7 +19,7 @@ IF (GIT_FOUND)
STRING(STRIP "${TAOSDEMO_COMMIT_SHA1}" TAOSDEMO_COMMIT_SHA1)
ENDIF ()
EXECUTE_PROCESS(
COMMAND ${GIT_EXECUTABLE} status -z -s ${CMAKE_CURRENT_LIST_DIR}/taosdemo.c
COMMAND ${GIT_EXECUTABLE} status -z -s ${CMAKE_CURRENT_LIST_DIR}
RESULT_VARIABLE RESULT
OUTPUT_VARIABLE TAOSDEMO_STATUS)
IF (TD_LINUX)
......@@ -64,7 +65,7 @@ ELSE ()
ENDIF ()
IF (TD_LINUX)
AUX_SOURCE_DIRECTORY(. SRC)
AUX_SOURCE_DIRECTORY(./src SRC)
ADD_EXECUTABLE(taosdemo ${SRC})
IF (TD_SOMODE_STATIC)
......@@ -73,9 +74,17 @@ IF (TD_LINUX)
TARGET_LINK_LIBRARIES(taosdemo taos cJson ${LINK_JEMALLOC})
ENDIF ()
ELSEIF (TD_WINDOWS)
AUX_SOURCE_DIRECTORY(. SRC)
AUX_SOURCE_DIRECTORY(./src SRC)
ADD_EXECUTABLE(taosdemo ${SRC})
SET_SOURCE_FILES_PROPERTIES(./taosdemo.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoUtil.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoData.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoInsert.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoCommandOpt.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoQuery.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoMain.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoSubscribe.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoOutput.c PROPERTIES COMPILE_FLAGS -w)
SET_SOURCE_FILES_PROPERTIES(./src/demoJsonOpt.c PROPERTIES COMPILE_FLAGS -w)
IF (TD_SOMODE_STATIC)
TARGET_LINK_LIBRARIES(taosdemo taos_static cJson lua)
ELSE ()
......@@ -83,7 +92,7 @@ ELSEIF (TD_WINDOWS)
ENDIF ()
ELSEIF (TD_DARWIN)
# missing a few dependencies, such as <argp.h>
AUX_SOURCE_DIRECTORY(. SRC)
AUX_SOURCE_DIRECTORY(./src SRC)
ADD_EXECUTABLE(taosdemo ${SRC})
IF (TD_SOMODE_STATIC)
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __DEMO__
#define __DEMO__
#include <stdint.h>
#include <taos.h>
#include <taoserror.h>
#define _GNU_SOURCE
#define CURL_STATICLIB
#ifdef LINUX
#include <argp.h>
#include <inttypes.h>
#ifndef _ALPINE
#include <error.h>
#endif
#include <pthread.h>
#include <regex.h>
#include <semaphore.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <wordexp.h>
#else
#include <regex.h>
#include <stdio.h>
#endif
#include <assert.h>
#include <stdlib.h>
// #include "os.h"
#include "taos.h"
#include "taoserror.h"
#include "tutil.h"
#define REQ_EXTRA_BUF_LEN 1024
#define RESP_BUF_LEN 4096
#define SQL_BUFF_LEN 1024
extern char configDir[];
#define STR_INSERT_INTO "INSERT INTO "
#define MAX_RECORDS_PER_REQ 32766
#define HEAD_BUFF_LEN \
TSDB_MAX_COLUMNS * 24 // 16*MAX_COLUMNS + (192+32)*2 + insert into ..
#define BUFFER_SIZE TSDB_MAX_ALLOWED_SQL_LEN
#define FETCH_BUFFER_SIZE 100 * TSDB_MAX_ALLOWED_SQL_LEN
#define COND_BUF_LEN (BUFFER_SIZE - 30)
#define COL_BUFFER_LEN ((TSDB_COL_NAME_LEN + 15) * TSDB_MAX_COLUMNS)
#define MAX_USERNAME_SIZE 64
#define MAX_HOSTNAME_SIZE \
253 // https://man7.org/linux/man-pages/man7/hostname.7.html
#define MAX_TB_NAME_SIZE 64
#define MAX_DATA_SIZE \
(16 * TSDB_MAX_COLUMNS) + 20 // max record len: 16*MAX_COLUMNS, timestamp
// string and ,('') need extra space
#define OPT_ABORT 1 /* –abort */
#define MAX_FILE_NAME_LEN 256 // max file name length on linux is 255.
#define MAX_PATH_LEN 4096
#define DEFAULT_START_TIME 1500000000000
#define MAX_PREPARED_RAND 1000000
#define INT_BUFF_LEN 12
#define BIGINT_BUFF_LEN 21
#define SMALLINT_BUFF_LEN 7
#define TINYINT_BUFF_LEN 5
#define BOOL_BUFF_LEN 6
#define FLOAT_BUFF_LEN 22
#define DOUBLE_BUFF_LEN 42
#define TIMESTAMP_BUFF_LEN 21
#define PRINT_STAT_INTERVAL 30 * 1000
#define MAX_SAMPLES 10000
#define MAX_NUM_COLUMNS \
(TSDB_MAX_COLUMNS - 1) // exclude first column timestamp
#define MAX_DB_COUNT 8
#define MAX_SUPER_TABLE_COUNT 200
#define MAX_QUERY_SQL_COUNT 100
#define MAX_DATABASE_COUNT 256
#define MAX_JSON_BUFF 6400000
#define INPUT_BUF_LEN 256
#define EXTRA_SQL_LEN 256
#define TBNAME_PREFIX_LEN \
(TSDB_TABLE_NAME_LEN - 20) // 20 characters reserved for seq
#define SMALL_BUFF_LEN 8
#define DATATYPE_BUFF_LEN (SMALL_BUFF_LEN * 3)
#define NOTE_BUFF_LEN (SMALL_BUFF_LEN * 16)
#define DEFAULT_NTHREADS 8
#define DEFAULT_TIMESTAMP_STEP 1
#define DEFAULT_INTERLACE_ROWS 0
#define DEFAULT_DATATYPE_NUM 1
#define DEFAULT_CHILDTABLES 10000
#define DEFAULT_TEST_MODE 0
#define DEFAULT_METAFILE NULL
#define DEFAULT_SQLFILE NULL
#define DEFAULT_HOST "localhost"
#define DEFAULT_PORT 6030
#define DEFAULT_IFACE INTERFACE_BUT
#define DEFAULT_DATABASE "test"
#define DEFAULT_REPLICA 1
#define DEFAULT_TB_PREFIX "d"
#define DEFAULT_ESCAPE_CHAR false
#define DEFAULT_USE_METRIC true
#define DEFAULT_DROP_DB true
#define DEFAULT_AGGR_FUNC false
#define DEFAULT_DEBUG false
#define DEFAULT_VERBOSE false
#define DEFAULT_PERF_STAT false
#define DEFAULT_ANS_YES false
#define DEFAULT_OUTPUT "./output.txt"
#define DEFAULT_SYNC_MODE 0
#define DEFAULT_DATA_TYPE \
{ TSDB_DATA_TYPE_FLOAT, TSDB_DATA_TYPE_INT, TSDB_DATA_TYPE_FLOAT }
#define DEFAULT_DATATYPE \
{ "FLOAT", "INT", "FLOAT" }
#define DEFAULT_DATALENGTH \
{ 4, 4, 4 }
#define DEFAULT_BINWIDTH 64
#define DEFAULT_COL_COUNT 4
#define DEFAULT_LEN_ONE_ROW 76
#define DEFAULT_INSERT_INTERVAL 0
#define DEFAULT_QUERY_TIME 1
#define DEFAULT_PREPARED_RAND 10000
#define DEFAULT_REQ_PER_REQ 30000
#define DEFAULT_INSERT_ROWS 10000
#define DEFAULT_ABORT 0
#define DEFAULT_RATIO 0
#define DEFAULT_DISORDER_RANGE 1000
#define DEFAULT_METHOD_DEL 1
#define DEFAULT_TOTAL_INSERT 0
#define DEFAULT_TOTAL_AFFECT 0
#define DEFAULT_DEMO_MODE true
#define DEFAULT_CREATE_BATCH 10
#define DEFAULT_SUB_INTERVAL 10000
#define DEFAULT_QUERY_INTERVAL 10000
#define SML_LINE_SQL_SYNTAX_OFFSET 7
#if _MSC_VER <= 1900
#define __func__ __FUNCTION__
#endif
#define debugPrint(fmt, ...) \
do { \
if (g_args.debug_print || g_args.verbose_print) \
fprintf(stderr, "DEBG: " fmt, __VA_ARGS__); \
} while (0)
#define verbosePrint(fmt, ...) \
do { \
if (g_args.verbose_print) fprintf(stderr, "VERB: " fmt, __VA_ARGS__); \
} while (0)
#define performancePrint(fmt, ...) \
do { \
if (g_args.performance_print) \
fprintf(stderr, "PERF: " fmt, __VA_ARGS__); \
} while (0)
#define errorPrint(fmt, ...) \
do { \
fprintf(stderr, "\033[31m"); \
fprintf(stderr, "%s(%d) ", __FILE__, __LINE__); \
fprintf(stderr, "ERROR: " fmt, __VA_ARGS__); \
fprintf(stderr, "\033[0m"); \
} while (0)
enum TEST_MODE {
INSERT_TEST, // 0
QUERY_TEST, // 1
SUBSCRIBE_TEST, // 2
INVAID_TEST
};
typedef enum CREATE_SUB_TABLE_MOD_EN {
PRE_CREATE_SUBTBL,
AUTO_CREATE_SUBTBL,
NO_CREATE_SUBTBL
} CREATE_SUB_TABLE_MOD_EN;
typedef enum TABLE_EXISTS_EN {
TBL_NO_EXISTS,
TBL_ALREADY_EXISTS,
TBL_EXISTS_BUTT
} TABLE_EXISTS_EN;
enum enumSYNC_MODE { SYNC_MODE, ASYNC_MODE, MODE_BUT };
enum enum_TAOS_INTERFACE {
TAOSC_IFACE,
REST_IFACE,
STMT_IFACE,
SML_IFACE,
INTERFACE_BUT
};
typedef enum enumQUERY_CLASS {
SPECIFIED_CLASS,
STABLE_CLASS,
CLASS_BUT
} QUERY_CLASS;
typedef enum enum_PROGRESSIVE_OR_INTERLACE {
PROGRESSIVE_INSERT_MODE,
INTERLACE_INSERT_MODE,
INVALID_INSERT_MODE
} PROG_OR_INTERLACE_MODE;
typedef enum enumQUERY_TYPE {
NO_INSERT_TYPE,
INSERT_TYPE,
QUERY_TYPE_BUT
} QUERY_TYPE;
enum _show_db_index {
TSDB_SHOW_DB_NAME_INDEX,
TSDB_SHOW_DB_CREATED_TIME_INDEX,
TSDB_SHOW_DB_NTABLES_INDEX,
TSDB_SHOW_DB_VGROUPS_INDEX,
TSDB_SHOW_DB_REPLICA_INDEX,
TSDB_SHOW_DB_QUORUM_INDEX,
TSDB_SHOW_DB_DAYS_INDEX,
TSDB_SHOW_DB_KEEP_INDEX,
TSDB_SHOW_DB_CACHE_INDEX,
TSDB_SHOW_DB_BLOCKS_INDEX,
TSDB_SHOW_DB_MINROWS_INDEX,
TSDB_SHOW_DB_MAXROWS_INDEX,
TSDB_SHOW_DB_WALLEVEL_INDEX,
TSDB_SHOW_DB_FSYNC_INDEX,
TSDB_SHOW_DB_COMP_INDEX,
TSDB_SHOW_DB_CACHELAST_INDEX,
TSDB_SHOW_DB_PRECISION_INDEX,
TSDB_SHOW_DB_UPDATE_INDEX,
TSDB_SHOW_DB_STATUS_INDEX,
TSDB_MAX_SHOW_DB
};
// -----------------------------------------SHOW TABLES CONFIGURE
// -------------------------------------
enum _show_stables_index {
TSDB_SHOW_STABLES_NAME_INDEX,
TSDB_SHOW_STABLES_CREATED_TIME_INDEX,
TSDB_SHOW_STABLES_COLUMNS_INDEX,
TSDB_SHOW_STABLES_METRIC_INDEX,
TSDB_SHOW_STABLES_UID_INDEX,
TSDB_SHOW_STABLES_TID_INDEX,
TSDB_SHOW_STABLES_VGID_INDEX,
TSDB_MAX_SHOW_STABLES
};
enum _describe_table_index {
TSDB_DESCRIBE_METRIC_FIELD_INDEX,
TSDB_DESCRIBE_METRIC_TYPE_INDEX,
TSDB_DESCRIBE_METRIC_LENGTH_INDEX,
TSDB_DESCRIBE_METRIC_NOTE_INDEX,
TSDB_MAX_DESCRIBE_METRIC
};
typedef struct SArguments_S {
char * metaFile;
uint32_t test_mode;
char * host;
uint16_t port;
uint16_t iface;
char * user;
char password[SHELL_MAX_PASSWORD_LEN];
char * database;
int replica;
char * tb_prefix;
bool escapeChar;
char * sqlFile;
bool use_metric;
bool drop_database;
bool aggr_func;
bool answer_yes;
bool debug_print;
bool verbose_print;
bool performance_print;
char * output_file;
bool async_mode;
char data_type[MAX_NUM_COLUMNS + 1];
char * dataType[MAX_NUM_COLUMNS + 1];
int32_t data_length[MAX_NUM_COLUMNS + 1];
uint32_t binwidth;
uint32_t columnCount;
uint64_t lenOfOneRow;
uint32_t nthreads;
uint64_t insert_interval;
uint64_t timestamp_step;
int64_t query_times;
int64_t prepared_rand;
uint32_t interlaceRows;
uint32_t reqPerReq; // num_of_records_per_req
uint64_t max_sql_len;
int64_t ntables;
int64_t insertRows;
int abort;
uint32_t disorderRatio; // 0: no disorder, >0: x%
int disorderRange; // ms, us or ns. according to database precision
uint32_t method_of_delete;
uint64_t totalInsertRows;
uint64_t totalAffectedRows;
bool demo_mode; // use default column name and semi-random data
} SArguments;
typedef struct SColumn_S {
char field[TSDB_COL_NAME_LEN];
char data_type;
char dataType[DATATYPE_BUFF_LEN];
uint32_t dataLen;
char note[NOTE_BUFF_LEN];
} StrColumn;
typedef struct SSuperTable_S {
char stbName[TSDB_TABLE_NAME_LEN];
char dataSource[SMALL_BUFF_LEN]; // rand_gen or sample
char childTblPrefix[TBNAME_PREFIX_LEN];
uint16_t childTblExists;
int64_t childTblCount;
uint64_t batchCreateTableNum; // 0: no batch, > 0: batch table number in
// one sql
uint8_t autoCreateTable; // 0: create sub table, 1: auto create sub table
uint16_t iface; // 0: taosc, 1: rest, 2: stmt
uint16_t lineProtocol;
int64_t childTblLimit;
uint64_t childTblOffset;
// int multiThreadWriteOneTbl; // 0: no, 1: yes
uint32_t interlaceRows; //
int disorderRatio; // 0: no disorder, >0: x%
int disorderRange; // ms, us or ns. according to database precision
uint64_t maxSqlLen; //
uint64_t insertInterval; // insert interval, will override global insert
// interval
int64_t insertRows;
int64_t timeStampStep;
int tsPrecision;
char startTimestamp[MAX_TB_NAME_SIZE];
char sampleFormat[SMALL_BUFF_LEN]; // csv, json
char sampleFile[MAX_FILE_NAME_LEN];
char tagsFile[MAX_FILE_NAME_LEN];
uint32_t columnCount;
StrColumn columns[TSDB_MAX_COLUMNS];
uint32_t tagCount;
StrColumn tags[TSDB_MAX_TAGS];
char * childTblName;
bool escapeChar;
char * colsOfCreateChildTable;
uint64_t lenOfOneRow;
uint64_t lenOfTagOfOneRow;
char *sampleDataBuf;
bool useSampleTs;
uint32_t tagSource; // 0: rand, 1: tag sample
char * tagDataBuf;
uint32_t tagSampleCount;
uint32_t tagUsePos;
// bind param batch
char *sampleBindBatchArray;
// statistics
uint64_t totalInsertRows;
uint64_t totalAffectedRows;
} SSuperTable;
typedef struct {
char name[TSDB_DB_NAME_LEN];
char create_time[32];
int64_t ntables;
int32_t vgroups;
int16_t replica;
int16_t quorum;
int16_t days;
char keeplist[64];
int32_t cache; // MB
int32_t blocks;
int32_t minrows;
int32_t maxrows;
int8_t wallevel;
int32_t fsync;
int8_t comp;
int8_t cachelast;
char precision[SMALL_BUFF_LEN]; // time resolution
int8_t update;
char status[16];
} SDbInfo;
typedef struct SDbCfg_S {
// int maxtablesPerVnode;
uint32_t minRows; // 0 means default
uint32_t maxRows; // 0 means default
int comp;
int walLevel;
int cacheLast;
int fsync;
int replica;
int update;
int keep;
int days;
int cache;
int blocks;
int quorum;
char precision[SMALL_BUFF_LEN];
} SDbCfg;
typedef struct SDataBase_S {
char dbName[TSDB_DB_NAME_LEN];
bool drop; // 0: use exists, 1: if exists, drop then new create
SDbCfg dbCfg;
uint64_t superTblCount;
SSuperTable *superTbls;
} SDataBase;
typedef struct SDbs_S {
char cfgDir[MAX_FILE_NAME_LEN];
char host[MAX_HOSTNAME_SIZE];
struct sockaddr_in serv_addr;
uint16_t port;
char user[MAX_USERNAME_SIZE];
char password[SHELL_MAX_PASSWORD_LEN];
char resultFile[MAX_FILE_NAME_LEN];
bool use_metric;
bool aggr_func;
bool asyncMode;
uint32_t threadCount;
uint32_t threadCountForCreateTbl;
uint32_t dbCount;
// statistics
uint64_t totalInsertRows;
uint64_t totalAffectedRows;
SDataBase *db;
} SDbs;
typedef struct SpecifiedQueryInfo_S {
uint64_t queryInterval; // 0: unlimited > 0 loop/s
uint32_t concurrent;
int sqlCount;
uint32_t asyncMode; // 0: sync, 1: async
uint64_t subscribeInterval; // ms
uint64_t queryTimes;
bool subscribeRestart;
int subscribeKeepProgress;
char sql[MAX_QUERY_SQL_COUNT][BUFFER_SIZE + 1];
char result[MAX_QUERY_SQL_COUNT][MAX_FILE_NAME_LEN];
int resubAfterConsume[MAX_QUERY_SQL_COUNT];
int endAfterConsume[MAX_QUERY_SQL_COUNT];
TAOS_SUB *tsub[MAX_QUERY_SQL_COUNT];
char topic[MAX_QUERY_SQL_COUNT][32];
int consumed[MAX_QUERY_SQL_COUNT];
TAOS_RES *res[MAX_QUERY_SQL_COUNT];
uint64_t totalQueried;
} SpecifiedQueryInfo;
typedef struct SuperQueryInfo_S {
char stbName[TSDB_TABLE_NAME_LEN];
uint64_t queryInterval; // 0: unlimited > 0 loop/s
uint32_t threadCnt;
uint32_t asyncMode; // 0: sync, 1: async
uint64_t subscribeInterval; // ms
bool subscribeRestart;
int subscribeKeepProgress;
uint64_t queryTimes;
int64_t childTblCount;
char childTblPrefix[TBNAME_PREFIX_LEN]; // 20 characters reserved for seq
int sqlCount;
char sql[MAX_QUERY_SQL_COUNT][BUFFER_SIZE + 1];
char result[MAX_QUERY_SQL_COUNT][MAX_FILE_NAME_LEN];
int resubAfterConsume;
int endAfterConsume;
TAOS_SUB *tsub[MAX_QUERY_SQL_COUNT];
char * childTblName;
uint64_t totalQueried;
} SuperQueryInfo;
typedef struct SQueryMetaInfo_S {
char cfgDir[MAX_FILE_NAME_LEN];
char host[MAX_HOSTNAME_SIZE];
uint16_t port;
struct sockaddr_in serv_addr;
char user[MAX_USERNAME_SIZE];
char password[SHELL_MAX_PASSWORD_LEN];
char dbName[TSDB_DB_NAME_LEN];
char queryMode[SMALL_BUFF_LEN]; // taosc, rest
SpecifiedQueryInfo specifiedQueryInfo;
SuperQueryInfo superQueryInfo;
uint64_t totalQueried;
} SQueryMetaInfo;
typedef struct SThreadInfo_S {
TAOS * taos;
TAOS_STMT * stmt;
int64_t * bind_ts;
int64_t * bind_ts_array;
char * bindParams;
char * is_null;
int threadID;
char db_name[TSDB_DB_NAME_LEN];
uint32_t time_precision;
char filePath[MAX_PATH_LEN];
FILE * fp;
char tb_prefix[TSDB_TABLE_NAME_LEN];
uint64_t start_table_from;
uint64_t end_table_to;
int64_t ntables;
int64_t tables_created;
uint64_t data_of_rate;
int64_t start_time;
char * cols;
bool use_metric;
SSuperTable *stbInfo;
char * buffer; // sql cmd buffer
// for async insert
tsem_t lock_sem;
int64_t counter;
uint64_t st;
uint64_t et;
uint64_t lastTs;
// sample data
int64_t samplePos;
// statistics
uint64_t totalInsertRows;
uint64_t totalAffectedRows;
// insert delay statistics
uint64_t cntDelay;
uint64_t totalDelay;
uint64_t avgDelay;
uint64_t maxDelay;
uint64_t minDelay;
// seq of query or subscribe
uint64_t querySeq; // sequence number of sql command
TAOS_SUB *tsub;
char **lines;
SOCKET sockfd;
} threadInfo;
/* ************ Global variables ************ */
extern char * g_aggreFuncDemo[];
extern char * g_aggreFunc[];
extern SArguments g_args;
extern SDbs g_Dbs;
extern char * g_dupstr;
extern int64_t g_totalChildTables;
extern int64_t g_actualChildTables;
extern SQueryMetaInfo g_queryInfo;
extern FILE * g_fpOfInsertResult;
#define min(a, b) (((a) < (b)) ? (a) : (b))
/* ************ Function declares ************ */
/* demoCommandOpt.c */
int parse_args(int argc, char *argv[]);
void setParaFromArg();
void querySqlFile(TAOS *taos, char *sqlFile);
void testCmdLine();
/* demoJsonOpt.c */
int getInfoFromJsonFile(char *file);
int testMetaFile();
/* demoUtil.c */
int isCommentLine(char *line);
void replaceChildTblName(char *inSql, char *outSql, int tblIndex);
void setupForAnsiEscape(void);
void resetAfterAnsiEscape(void);
int taosRandom();
void tmfree(void *buf);
void tmfclose(FILE *fp);
void fetchResult(TAOS_RES *res, threadInfo *pThreadInfo);
void prompt();
void ERROR_EXIT(const char *msg);
int postProceSql(char *host, uint16_t port, char *sqlstr,
threadInfo *pThreadInfo);
int queryDbExec(TAOS *taos, char *command, QUERY_TYPE type, bool quiet);
int regexMatch(const char *s, const char *reg, int cflags);
int convertHostToServAddr(char *host, uint16_t port,
struct sockaddr_in *serv_addr);
char *formatTimestamp(char *buf, int64_t val, int precision);
void errorWrongValue(char *program, char *wrong_arg, char *wrong_value);
void errorUnrecognized(char *program, char *wrong_arg);
void errorPrintReqArg(char *program, char *wrong_arg);
void errorPrintReqArg2(char *program, char *wrong_arg);
void errorPrintReqArg3(char *program, char *wrong_arg);
bool isStringNumber(char *input);
int getAllChildNameOfSuperTable(TAOS *taos, char *dbName, char *stbName,
char ** childTblNameOfSuperTbl,
int64_t *childTblCountOfSuperTbl);
int getChildNameOfSuperTableWithLimitAndOffset(TAOS *taos, char *dbName,
char * stbName,
char ** childTblNameOfSuperTbl,
int64_t *childTblCountOfSuperTbl,
int64_t limit, uint64_t offset,
bool escapChar);
/* demoInsert.c */
int insertTestProcess();
void postFreeResource();
/* demoOutput.c */
void printVersion();
void printfInsertMeta();
void printfInsertMetaToFile(FILE *fp);
void printStatPerThread(threadInfo *pThreadInfo);
void appendResultBufToFile(char *resultBuf, threadInfo *pThreadInfo);
void printfQueryMeta();
void printHelp();
void printfQuerySystemInfo(TAOS *taos);
/* demoQuery.c */
int queryTestProcess();
/* demoSubscribe.c */
int subscribeTestProcess();
#endif
\ No newline at end of file
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __DEMODATA__
#define __DEMODATA__
#include "cJSON.h"
#include "demo.h"
/***** Global variables ******/
extern char * g_sampleDataBuf;
extern char * g_sampleBindBatchArray;
extern int32_t * g_randint;
extern uint32_t *g_randuint;
extern int64_t * g_randbigint;
extern uint64_t *g_randubigint;
extern float * g_randfloat;
extern double * g_randdouble;
extern char * g_randbool_buff;
extern char * g_randint_buff;
extern char * g_randuint_buff;
extern char * g_rand_voltage_buff;
extern char * g_randbigint_buff;
extern char * g_randubigint_buff;
extern char * g_randsmallint_buff;
extern char * g_randusmallint_buff;
extern char * g_randtinyint_buff;
extern char * g_randutinyint_buff;
extern char * g_randfloat_buff;
extern char * g_rand_current_buff;
extern char * g_rand_phase_buff;
extern char * g_randdouble_buff;
/***** Declare functions *****/
int init_rand_data();
char * rand_bool_str();
int32_t rand_bool();
char * rand_tinyint_str();
int32_t rand_tinyint();
char * rand_utinyint_str();
int32_t rand_utinyint();
char * rand_smallint_str();
int32_t rand_smallint();
char * rand_usmallint_str();
int32_t rand_usmallint();
char * rand_int_str();
int32_t rand_int();
char * rand_uint_str();
int32_t rand_uint();
char * rand_bigint_str();
int64_t rand_bigint();
char * rand_ubigint_str();
int64_t rand_ubigint();
char * rand_float_str();
float rand_float();
char * demo_current_float_str();
float UNUSED_FUNC demo_current_float();
char * demo_voltage_int_str();
int32_t UNUSED_FUNC demo_voltage_int();
char * demo_phase_float_str();
float UNUSED_FUNC demo_phase_float();
void rand_string(char *str, int size);
char * rand_double_str();
double rand_double();
int generateTagValuesForStb(SSuperTable *stbInfo, int64_t tableSeq,
char *tagsValBuf);
int64_t getTSRandTail(int64_t timeStampStep, int32_t seq, int disorderRatio,
int disorderRange);
int32_t prepareStbStmtBindTag(char *bindArray, SSuperTable *stbInfo,
char *tagsVal, int32_t timePrec);
int32_t prepareStmtWithoutStb(threadInfo *pThreadInfo, char *tableName,
uint32_t batch, int64_t insertRows,
int64_t recordFrom, int64_t startTime);
int32_t generateStbInterlaceData(threadInfo *pThreadInfo, char *tableName,
uint32_t batchPerTbl, uint64_t i,
uint32_t batchPerTblTimes, uint64_t tableSeq,
char *buffer, int64_t insertRows,
int64_t startTime, uint64_t *pRemainderBufLen);
int64_t generateInterlaceDataWithoutStb(char *tableName, uint32_t batch,
uint64_t tableSeq, char *dbName,
char *buffer, int64_t insertRows,
int64_t startTime,
uint64_t *pRemainderBufLen);
int32_t generateStbProgressiveData(SSuperTable *stbInfo, char *tableName,
int64_t tableSeq, char *dbName, char *buffer,
int64_t insertRows, uint64_t recordFrom,
int64_t startTime, int64_t *pSamplePos,
int64_t *pRemainderBufLen);
int32_t generateProgressiveDataWithoutStb(
char *tableName, threadInfo *pThreadInfo, char *buffer, int64_t insertRows,
uint64_t recordFrom, int64_t startTime, int64_t *pRemainderBufLen);
int64_t generateStbRowData(SSuperTable *stbInfo, char *recBuf,
int64_t remainderBufLen, int64_t timestamp);
int prepareSampleForStb(SSuperTable *stbInfo);
int prepareSampleForNtb();
int parseSamplefileToStmtBatch(SSuperTable *stbInfo);
int parseStbSampleToStmtBatchForThread(threadInfo * pThreadInfo,
SSuperTable *stbInfo, uint32_t timePrec,
uint32_t batch);
int parseNtbSampleToStmtBatchForThread(threadInfo *pThreadInfo,
uint32_t timePrec, uint32_t batch);
int prepareSampleData();
int32_t generateSmlConstPart(char *sml, SSuperTable *stbInfo,
threadInfo *pThreadInfo, int tbSeq);
int32_t generateSmlMutablePart(char *line, char *sml, SSuperTable *stbInfo,
threadInfo *pThreadInfo, int64_t timestamp);
int32_t generateSmlJsonTags(cJSON *tagsList, SSuperTable *stbInfo,
threadInfo *pThreadInfo, int tbSeq);
int32_t generateSmlJsonCols(cJSON *array, cJSON *tag, SSuperTable *stbInfo,
threadInfo *pThreadInfo, int64_t timestamp);
#endif
\ No newline at end of file
......@@ -21,7 +21,7 @@
"cache": 16,
"blocks": 8,
"precision": "ms",
"keep": 365,
"keep": 36500,
"minRows": 100,
"maxRows": 4096,
"comp":2,
......
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "demo.h"
int64_t g_totalChildTables = DEFAULT_CHILDTABLES;
int64_t g_actualChildTables = 0;
FILE * g_fpOfInsertResult = NULL;
char * g_dupstr = NULL;
SDbs g_Dbs;
SQueryMetaInfo g_queryInfo;
SArguments g_args = {
DEFAULT_METAFILE, // metaFile
DEFAULT_TEST_MODE, // test_mode
DEFAULT_HOST, // host
DEFAULT_PORT, // port
DEFAULT_IFACE, // iface
TSDB_DEFAULT_USER, // user
TSDB_DEFAULT_PASS, // password
DEFAULT_DATABASE, // database
DEFAULT_REPLICA, // replica
DEFAULT_TB_PREFIX, // tb_prefix
DEFAULT_ESCAPE_CHAR, // escapeChar
DEFAULT_SQLFILE, // sqlFile
DEFAULT_USE_METRIC, // use_metric
DEFAULT_DROP_DB, // drop_database
DEFAULT_AGGR_FUNC, // aggr_func
DEFAULT_DEBUG, // debug_print
DEFAULT_VERBOSE, // verbose_print
DEFAULT_PERF_STAT, // performance statistic print
DEFAULT_ANS_YES, // answer_yes;
DEFAULT_OUTPUT, // output_file
DEFAULT_SYNC_MODE, // mode : sync or async
DEFAULT_DATA_TYPE, // data_type
DEFAULT_DATATYPE, // dataType
DEFAULT_DATALENGTH, // data_length
DEFAULT_BINWIDTH, // binwidth
DEFAULT_COL_COUNT, // columnCount, timestamp + float + int + float
DEFAULT_LEN_ONE_ROW, // lenOfOneRow
DEFAULT_NTHREADS, // nthreads
DEFAULT_INSERT_INTERVAL, // insert_interval
DEFAULT_TIMESTAMP_STEP, // timestamp_step
DEFAULT_QUERY_TIME, // query_times
DEFAULT_PREPARED_RAND, // prepared_rand
DEFAULT_INTERLACE_ROWS, // interlaceRows;
DEFAULT_REQ_PER_REQ, // reqPerReq
TSDB_MAX_ALLOWED_SQL_LEN, // max_sql_len
DEFAULT_CHILDTABLES, // ntables
DEFAULT_INSERT_ROWS, // insertRows
DEFAULT_ABORT, // abort
DEFAULT_RATIO, // disorderRatio
DEFAULT_DISORDER_RANGE, // disorderRange
DEFAULT_METHOD_DEL, // method_of_delete
DEFAULT_TOTAL_INSERT, // totalInsertRows;
DEFAULT_TOTAL_AFFECT, // totalAffectedRows;
DEFAULT_DEMO_MODE, // demo_mode;
};
int main(int argc, char *argv[]) {
if (parse_args(argc, argv)) {
exit(EXIT_FAILURE);
}
debugPrint("meta file: %s\n", g_args.metaFile);
if (g_args.metaFile) {
g_totalChildTables = 0;
if (getInfoFromJsonFile(g_args.metaFile)) {
exit(EXIT_FAILURE);
}
if (testMetaFile()) {
exit(EXIT_FAILURE);
}
} else {
memset(&g_Dbs, 0, sizeof(SDbs));
g_Dbs.db = calloc(1, sizeof(SDataBase));
if (NULL == g_Dbs.db) {
errorPrint("%s", "failed to allocate memory\n");
}
g_Dbs.db[0].superTbls = calloc(1, sizeof(SSuperTable));
if (NULL == g_Dbs.db[0].superTbls) {
errorPrint("%s", "failed to allocate memory\n");
}
setParaFromArg();
if (NULL != g_args.sqlFile) {
TAOS *qtaos = taos_connect(g_Dbs.host, g_Dbs.user, g_Dbs.password,
g_Dbs.db[0].dbName, g_Dbs.port);
querySqlFile(qtaos, g_args.sqlFile);
taos_close(qtaos);
} else {
testCmdLine();
}
}
postFreeResource();
return 0;
}
\ No newline at end of file
此差异已折叠。
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "demo.h"
void selectAndGetResult(threadInfo *pThreadInfo, char *command) {
if (0 == strncasecmp(g_queryInfo.queryMode, "taosc", strlen("taosc"))) {
TAOS_RES *res = taos_query(pThreadInfo->taos, command);
if (res == NULL || taos_errno(res) != 0) {
errorPrint("failed to execute sql:%s, reason:%s\n", command,
taos_errstr(res));
taos_free_result(res);
return;
}
fetchResult(res, pThreadInfo);
taos_free_result(res);
} else if (0 ==
strncasecmp(g_queryInfo.queryMode, "rest", strlen("rest"))) {
int retCode = postProceSql(g_queryInfo.host, g_queryInfo.port, command,
pThreadInfo);
if (0 != retCode) {
printf("====restful return fail, threadID[%d]\n",
pThreadInfo->threadID);
}
} else {
errorPrint("unknown query mode: %s\n", g_queryInfo.queryMode);
}
}
void *specifiedTableQuery(void *sarg) {
threadInfo *pThreadInfo = (threadInfo *)sarg;
setThreadName("specTableQuery");
if (pThreadInfo->taos == NULL) {
TAOS *taos = NULL;
taos = taos_connect(g_queryInfo.host, g_queryInfo.user,
g_queryInfo.password, NULL, g_queryInfo.port);
if (taos == NULL) {
errorPrint("[%d] Failed to connect to TDengine, reason:%s\n",
pThreadInfo->threadID, taos_errstr(NULL));
return NULL;
} else {
pThreadInfo->taos = taos;
}
}
char sqlStr[TSDB_DB_NAME_LEN + 5];
sprintf(sqlStr, "use %s", g_queryInfo.dbName);
if (0 != queryDbExec(pThreadInfo->taos, sqlStr, NO_INSERT_TYPE, false)) {
taos_close(pThreadInfo->taos);
errorPrint("use database %s failed!\n\n", g_queryInfo.dbName);
return NULL;
}
uint64_t st = 0;
uint64_t et = 0;
uint64_t queryTimes = g_queryInfo.specifiedQueryInfo.queryTimes;
uint64_t totalQueried = 0;
uint64_t lastPrintTime = taosGetTimestampMs();
uint64_t startTs = taosGetTimestampMs();
if (g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq][0] !=
'\0') {
sprintf(pThreadInfo->filePath, "%s-%d",
g_queryInfo.specifiedQueryInfo.result[pThreadInfo->querySeq],
pThreadInfo->threadID);
}
while (queryTimes--) {
if (g_queryInfo.specifiedQueryInfo.queryInterval &&
(et - st) < (int64_t)g_queryInfo.specifiedQueryInfo.queryInterval) {
taosMsleep((int32_t)(g_queryInfo.specifiedQueryInfo.queryInterval -
(et - st))); // ms
}
st = taosGetTimestampMs();
selectAndGetResult(
pThreadInfo,
g_queryInfo.specifiedQueryInfo.sql[pThreadInfo->querySeq]);
et = taosGetTimestampMs();
printf("=thread[%" PRId64 "] use %s complete one sql, Spent %10.3f s\n",
taosGetSelfPthreadId(), g_queryInfo.queryMode,
(et - st) / 1000.0);
totalQueried++;
g_queryInfo.specifiedQueryInfo.totalQueried++;
uint64_t currentPrintTime = taosGetTimestampMs();
uint64_t endTs = taosGetTimestampMs();
if (currentPrintTime - lastPrintTime > 30 * 1000) {
debugPrint("%s() LN%d, endTs=%" PRIu64 " ms, startTs=%" PRIu64
" ms\n",
__func__, __LINE__, endTs, startTs);
printf("thread[%d] has currently completed queries: %" PRIu64
", QPS: %10.6f\n",
pThreadInfo->threadID, totalQueried,
(double)(totalQueried / ((endTs - startTs) / 1000.0)));
lastPrintTime = currentPrintTime;
}
}
return NULL;
}
void *superTableQuery(void *sarg) {
char *sqlstr = calloc(1, BUFFER_SIZE);
if (NULL == sqlstr) {
errorPrint("%s", "failed to allocate memory\n");
return NULL;
}
threadInfo *pThreadInfo = (threadInfo *)sarg;
setThreadName("superTableQuery");
if (pThreadInfo->taos == NULL) {
TAOS *taos = NULL;
taos = taos_connect(g_queryInfo.host, g_queryInfo.user,
g_queryInfo.password, NULL, g_queryInfo.port);
if (taos == NULL) {
errorPrint("[%d] Failed to connect to TDengine, reason:%s\n",
pThreadInfo->threadID, taos_errstr(NULL));
free(sqlstr);
return NULL;
} else {
pThreadInfo->taos = taos;
}
}
uint64_t st = 0;
uint64_t et = (int64_t)g_queryInfo.superQueryInfo.queryInterval;
uint64_t queryTimes = g_queryInfo.superQueryInfo.queryTimes;
uint64_t totalQueried = 0;
uint64_t startTs = taosGetTimestampMs();
uint64_t lastPrintTime = taosGetTimestampMs();
while (queryTimes--) {
if (g_queryInfo.superQueryInfo.queryInterval &&
(et - st) < (int64_t)g_queryInfo.superQueryInfo.queryInterval) {
taosMsleep((int32_t)(g_queryInfo.superQueryInfo.queryInterval -
(et - st))); // ms
// printf("========sleep duration:%"PRId64 "========inserted
// rows:%d, table range:%d - %d\n", (1000 - (et - st)), i,
// pThreadInfo->start_table_from, pThreadInfo->end_table_to);
}
st = taosGetTimestampMs();
for (int i = (int)pThreadInfo->start_table_from;
i <= pThreadInfo->end_table_to; i++) {
for (int j = 0; j < g_queryInfo.superQueryInfo.sqlCount; j++) {
memset(sqlstr, 0, BUFFER_SIZE);
replaceChildTblName(g_queryInfo.superQueryInfo.sql[j], sqlstr,
i);
if (g_queryInfo.superQueryInfo.result[j][0] != '\0') {
sprintf(pThreadInfo->filePath, "%s-%d",
g_queryInfo.superQueryInfo.result[j],
pThreadInfo->threadID);
}
selectAndGetResult(pThreadInfo, sqlstr);
totalQueried++;
g_queryInfo.superQueryInfo.totalQueried++;
int64_t currentPrintTime = taosGetTimestampMs();
int64_t endTs = taosGetTimestampMs();
if (currentPrintTime - lastPrintTime > 30 * 1000) {
printf(
"thread[%d] has currently completed queries: %" PRIu64
", QPS: %10.3f\n",
pThreadInfo->threadID, totalQueried,
(double)(totalQueried / ((endTs - startTs) / 1000.0)));
lastPrintTime = currentPrintTime;
}
}
}
et = taosGetTimestampMs();
printf("####thread[%" PRId64
"] complete all sqls to allocate all sub-tables[%" PRIu64
" - %" PRIu64 "] once queries duration:%.4fs\n\n",
taosGetSelfPthreadId(), pThreadInfo->start_table_from,
pThreadInfo->end_table_to, (double)(et - st) / 1000.0);
}
free(sqlstr);
return NULL;
}
int queryTestProcess() {
printfQueryMeta();
TAOS *taos = NULL;
taos = taos_connect(g_queryInfo.host, g_queryInfo.user,
g_queryInfo.password, NULL, g_queryInfo.port);
if (taos == NULL) {
errorPrint("Failed to connect to TDengine, reason:%s\n",
taos_errstr(NULL));
exit(EXIT_FAILURE);
}
if (0 != g_queryInfo.superQueryInfo.sqlCount) {
getAllChildNameOfSuperTable(taos, g_queryInfo.dbName,
g_queryInfo.superQueryInfo.stbName,
&g_queryInfo.superQueryInfo.childTblName,
&g_queryInfo.superQueryInfo.childTblCount);
}
prompt();
if (g_args.debug_print || g_args.verbose_print) {
printfQuerySystemInfo(taos);
}
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", strlen("rest"))) {
if (convertHostToServAddr(g_queryInfo.host, g_queryInfo.port,
&g_queryInfo.serv_addr) != 0)
ERROR_EXIT("convert host to server address");
}
pthread_t * pids = NULL;
threadInfo *infos = NULL;
//==== create sub threads for query from specify table
int nConcurrent = g_queryInfo.specifiedQueryInfo.concurrent;
uint64_t nSqlCount = g_queryInfo.specifiedQueryInfo.sqlCount;
uint64_t startTs = taosGetTimestampMs();
if ((nSqlCount > 0) && (nConcurrent > 0)) {
pids = calloc(1, nConcurrent * nSqlCount * sizeof(pthread_t));
infos = calloc(1, nConcurrent * nSqlCount * sizeof(threadInfo));
if ((NULL == pids) || (NULL == infos)) {
taos_close(taos);
ERROR_EXIT("memory allocation failed for create threads\n");
}
for (uint64_t i = 0; i < nSqlCount; i++) {
for (int j = 0; j < nConcurrent; j++) {
uint64_t seq = i * nConcurrent + j;
threadInfo *pThreadInfo = infos + seq;
pThreadInfo->threadID = (int)seq;
pThreadInfo->querySeq = i;
if (0 == strncasecmp(g_queryInfo.queryMode, "taosc", 5)) {
char sqlStr[TSDB_DB_NAME_LEN + 5];
sprintf(sqlStr, "USE %s", g_queryInfo.dbName);
if (0 != queryDbExec(taos, sqlStr, NO_INSERT_TYPE, false)) {
taos_close(taos);
free(infos);
free(pids);
errorPrint("use database %s failed!\n\n",
g_queryInfo.dbName);
return -1;
}
}
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", 4)) {
#ifdef WINDOWS
WSADATA wsaData;
WSAStartup(MAKEWORD(2, 2), &wsaData);
SOCKET sockfd;
#else
int sockfd;
#endif
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0) {
#ifdef WINDOWS
errorPrint("Could not create socket : %d",
WSAGetLastError());
#endif
debugPrint("%s() LN%d, sockfd=%d\n", __func__, __LINE__,
sockfd);
ERROR_EXIT("opening socket");
}
int retConn = connect(
sockfd, (struct sockaddr *)&(g_queryInfo.serv_addr),
sizeof(struct sockaddr));
debugPrint("%s() LN%d connect() return %d\n", __func__,
__LINE__, retConn);
if (retConn < 0) {
ERROR_EXIT("connecting");
}
pThreadInfo->sockfd = sockfd;
}
pThreadInfo->taos =
NULL; // workaround to use separate taos connection;
pthread_create(pids + seq, NULL, specifiedTableQuery,
pThreadInfo);
}
}
} else {
g_queryInfo.specifiedQueryInfo.concurrent = 0;
}
taos_close(taos);
pthread_t * pidsOfSub = NULL;
threadInfo *infosOfSub = NULL;
//==== create sub threads for query from all sub table of the super table
if ((g_queryInfo.superQueryInfo.sqlCount > 0) &&
(g_queryInfo.superQueryInfo.threadCnt > 0)) {
pidsOfSub =
calloc(1, g_queryInfo.superQueryInfo.threadCnt * sizeof(pthread_t));
infosOfSub = calloc(
1, g_queryInfo.superQueryInfo.threadCnt * sizeof(threadInfo));
if ((NULL == pidsOfSub) || (NULL == infosOfSub)) {
free(infos);
free(pids);
ERROR_EXIT("memory allocation failed for create threads\n");
}
int64_t ntables = g_queryInfo.superQueryInfo.childTblCount;
int threads = g_queryInfo.superQueryInfo.threadCnt;
int64_t a = ntables / threads;
if (a < 1) {
threads = (int)ntables;
a = 1;
}
int64_t b = 0;
if (threads != 0) {
b = ntables % threads;
}
uint64_t tableFrom = 0;
for (int i = 0; i < threads; i++) {
threadInfo *pThreadInfo = infosOfSub + i;
pThreadInfo->threadID = i;
pThreadInfo->start_table_from = tableFrom;
pThreadInfo->ntables = i < b ? a + 1 : a;
pThreadInfo->end_table_to =
i < b ? tableFrom + a : tableFrom + a - 1;
tableFrom = pThreadInfo->end_table_to + 1;
pThreadInfo->taos =
NULL; // workaround to use separate taos connection;
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", 4)) {
#ifdef WINDOWS
WSADATA wsaData;
WSAStartup(MAKEWORD(2, 2), &wsaData);
SOCKET sockfd;
#else
int sockfd;
#endif
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0) {
#ifdef WINDOWS
errorPrint("Could not create socket : %d",
WSAGetLastError());
#endif
debugPrint("%s() LN%d, sockfd=%d\n", __func__, __LINE__,
sockfd);
ERROR_EXIT("opening socket");
}
int retConn =
connect(sockfd, (struct sockaddr *)&(g_queryInfo.serv_addr),
sizeof(struct sockaddr));
debugPrint("%s() LN%d connect() return %d\n", __func__,
__LINE__, retConn);
if (retConn < 0) {
ERROR_EXIT("connecting");
}
pThreadInfo->sockfd = sockfd;
}
pthread_create(pidsOfSub + i, NULL, superTableQuery, pThreadInfo);
}
g_queryInfo.superQueryInfo.threadCnt = threads;
} else {
g_queryInfo.superQueryInfo.threadCnt = 0;
}
if ((nSqlCount > 0) && (nConcurrent > 0)) {
for (int i = 0; i < nConcurrent; i++) {
for (int j = 0; j < nSqlCount; j++) {
pthread_join(pids[i * nSqlCount + j], NULL);
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", 4)) {
threadInfo *pThreadInfo = infos + i * nSqlCount + j;
#ifdef WINDOWS
closesocket(pThreadInfo->sockfd);
WSACleanup();
#else
close(pThreadInfo->sockfd);
#endif
}
}
}
}
tmfree((char *)pids);
tmfree((char *)infos);
for (int i = 0; i < g_queryInfo.superQueryInfo.threadCnt; i++) {
pthread_join(pidsOfSub[i], NULL);
if (0 == strncasecmp(g_queryInfo.queryMode, "rest", 4)) {
threadInfo *pThreadInfo = infosOfSub + i;
#ifdef WINDOWS
closesocket(pThreadInfo->sockfd);
WSACleanup();
#else
close(pThreadInfo->sockfd);
#endif
}
}
tmfree((char *)pidsOfSub);
tmfree((char *)infosOfSub);
// taos_close(taos);// workaround to use separate taos connection;
uint64_t endTs = taosGetTimestampMs();
uint64_t totalQueried = g_queryInfo.specifiedQueryInfo.totalQueried +
g_queryInfo.superQueryInfo.totalQueried;
fprintf(stderr,
"==== completed total queries: %" PRIu64
", the QPS of all threads: %10.3f====\n",
totalQueried,
(double)(totalQueried / ((endTs - startTs) / 1000.0)));
return 0;
}
\ No newline at end of file
此差异已折叠。
此差异已折叠。
此差异已折叠。
......@@ -41,9 +41,9 @@ extern int32_t sdbDebugFlag;
#define sdbDebug(...) { if (sdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }}
#define sdbTrace(...) { if (sdbDebugFlag & DEBUG_TRACE) { taosPrintLog("SDB ", sdbDebugFlag, __VA_ARGS__); }}
#define mLError(...) { monSaveLog(2, __VA_ARGS__); mError(__VA_ARGS__) }
#define mLWarn(...) { monSaveLog(1, __VA_ARGS__); mWarn(__VA_ARGS__) }
#define mLInfo(...) { monSaveLog(0, __VA_ARGS__); mInfo(__VA_ARGS__) }
#define mLError(...) { monSaveLogs(2, __VA_ARGS__); mError(__VA_ARGS__) }
#define mLWarn(...) { monSaveLogs(1, __VA_ARGS__); mWarn(__VA_ARGS__) }
#define mLInfo(...) { monSaveLogs(0, __VA_ARGS__); mInfo(__VA_ARGS__) }
#ifdef __cplusplus
}
......
......@@ -1139,7 +1139,7 @@ static int32_t mnodeAlterDbFp(SMnodeMsg *pMsg) {
return sdbUpdateRow(&row);
}
//bnNotify();
bnNotify();
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}
......
......@@ -1178,7 +1178,7 @@ static int32_t mnodeGetVnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pC
SUserObj *pUser = mnodeGetUserFromConn(pConn);
if (pUser == NULL) return 0;
if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0) {
if (strcmp(pUser->pAcct->user, TSDB_DEFAULT_USER) != 0 ) {
mnodeDecUserRef(pUser);
return TSDB_CODE_MND_NO_RIGHTS;
}
......@@ -1256,10 +1256,10 @@ static int32_t mnodeRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, vo
STR_TO_VARSTR(pWrite, syncRole[pVgid->role]);
cols++;
numOfRows++;
}
}
if (numOfRows >= rows) {
mnodeDecVgroupRef(pVgroup);
break;
}
......
......@@ -166,7 +166,7 @@ static void mnodeCancelGetNextConn(void *pIter) {
static int32_t mnodeGetConnsMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
SUserObj *pUser = mnodeGetUserFromConn(pConn);
if (pUser == NULL) return 0;
if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
if (strcmp(pUser->pAcct->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
int32_t cols = 0;
SSchema *pSchema = pMeta->schema;
......@@ -322,7 +322,7 @@ int32_t mnodeSaveQueryStreamList(SConnObj *pConn, SHeartBeatMsg *pHBMsg) {
static int32_t mnodeGetQueryMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
SUserObj *pUser = mnodeGetUserFromConn(pConn);
if (pUser == NULL) return 0;
if (strcmp(pUser->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
if (strcmp(pUser->pAcct->user, TSDB_DEFAULT_USER) != 0) return TSDB_CODE_MND_NO_RIGHTS;
int32_t cols = 0;
SSchema *pSchema = pMeta->schema;
......
......@@ -121,7 +121,7 @@ static int32_t mnodeProcessShowMsg(SMnodeMsg *pMsg) {
}
if (!tsMnodeShowMetaFp[pShowMsg->type] || !tsMnodeShowRetrieveFp[pShowMsg->type]) {
mError("show type:%s is not support", mnodeGetShowType(pShowMsg->type));
mWarn("show type:%s is not support", mnodeGetShowType(pShowMsg->type));
return TSDB_CODE_COM_OPS_NOT_SUPPORT;
}
......
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
此差异已折叠。
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册