xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch '3.0' of https://github.com/taosdata/TDengine into test/3.0/TD-26638

This commit is contained in:
chenhaoran 2023-11-14 14:30:39 +08:00
parent 526d766fac ba752adc40
commit 383e00b709
42 changed files with 1438 additions and 885 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/cmake.define
View File

@ -170,7 +170,7 @@ ELSE ()
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mavx2")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mavx2")
ENDIF()
MESSAGE(STATUS "SIMD instructions (FMA/AVX/AVX2/AVX512) is ACTIVATED")
MESSAGE(STATUS "SIMD instructions (FMA/AVX/AVX2) is ACTIVATED")
IF (COMPILER_SUPPORT_AVX512F AND COMPILER_SUPPORT_AVX512BMI)
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mavx512f -mavx512vbmi")

4
include/common/tglobal.h
View File

@ -217,13 +217,13 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs,
bool tsc);
void taosCleanupCfg();
void taosCfgDynamicOptions(const char *option, const char *value);
int32_t taosCfgDynamicOptions(SConfig *pCfg, char *name, bool forServer);
struct SConfig *taosGetCfg();
void taosSetAllDebugFlag(int32_t flag, bool rewrite);
void taosSetDebugFlag(int32_t *pFlagPtr, const char *flagName, int32_t flagVal, bool rewrite);
int32_t taosApplyLocalCfg(SConfig *pCfg, char *name);
void taosLocalCfgForbiddenToChange(char *name, bool *forbidden);
int8_t taosGranted();

1
include/libs/executor/storageapi.h
View File

@ -119,6 +119,7 @@ typedef struct SRowBuffPos {
bool beFlushed;
bool beUsed;
bool needFree;
bool beUpdated;
} SRowBuffPos;
// tq

1
include/libs/stream/tstream.h
View File

@ -825,6 +825,7 @@ int32_t streamMetaGetNumOfTasks(SStreamMeta* pMeta);
SStreamTask* streamMetaAcquireTask(SStreamMeta* pMeta, int64_t streamId, int32_t taskId);
void streamMetaReleaseTask(SStreamMeta* pMeta, SStreamTask* pTask);
int32_t streamMetaReopen(SStreamMeta* pMeta);
void streamMetaInitBackend(SStreamMeta* pMeta);
int32_t streamMetaCommit(SStreamMeta* pMeta);
int32_t streamMetaLoadAllTasks(SStreamMeta* pMeta);
int64_t streamMetaGetLatestCheckpointId(SStreamMeta* pMeta);

2
include/util/taoserror.h
View File

@ -172,7 +172,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_TSC_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0X0231)
// mnode-common
// #define TSDB_CODE_MND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0300) // 2.x
#define TSDB_CODE_MND_REQ_REJECTED TAOS_DEF_ERROR_CODE(0, 0x0300)
// #define TSDB_CODE_MND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0301) // 2.x
// #define TSDB_CODE_MND_ACTION_NEED_REPROCESSEDTAOS_DEF_ERROR_CODE(0, 0x0302) // 2.x
#define TSDB_CODE_MND_NO_RIGHTS TAOS_DEF_ERROR_CODE(0, 0x0303)

10
include/util/tcompression.h
View File

@ -30,6 +30,10 @@ extern "C" {
#define INT64MASK(_x) ((((uint64_t)1) << _x) - 1)
#define INT32MASK(_x) (((uint32_t)1 << _x) - 1)
#define INT8MASK(_x) (((uint8_t)1 << _x) - 1)
#define ZIGZAG_ENCODE(T, v) (((u##T)((v) >> (sizeof(T) * 8 - 1))) ^ (((u##T)(v)) << 1)) // zigzag encode
#define ZIGZAG_DECODE(T, v) (((v) >> 1) ^ -((T)((v)&1))) // zigzag decode
// Compression algorithm
#define NO_COMPRESSION 0
#define ONE_STAGE_COMP 1
@ -129,6 +133,12 @@ int32_t tsCompressBigint(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32
int32_t nBuf);
int32_t tsDecompressBigint(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf,
int32_t nBuf);
// for internal usage
int32_t getWordLength(char type);
int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, char *const output, const char type);
int32_t tsDecompressFloatImplAvx512(const char *const input, const int32_t nelements, char *const output);
int32_t tsDecompressFloatImplAvx2(const char *const input, const int32_t nelements, char *const output);
/*************************************************************************
* STREAM COMPRESSION

1
include/util/tdef.h
View File

@ -34,7 +34,6 @@ extern "C" {
// Bytes for each type.
extern const int32_t TYPE_BYTES[21];
// TODO: replace and remove code below
#define CHAR_BYTES sizeof(char)
#define SHORT_BYTES sizeof(int16_t)
#define INT_BYTES sizeof(int32_t)

35
include/util/tunit.h Normal file
View File

@ -0,0 +1,35 @@
/*
* 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 _TD_UNIT_H_
#define _TD_UNIT_H_
#include "os.h"
#ifdef __cplusplus
extern "C" {
#endif
int64_t taosStrHumanToInt64(const char* str);
void taosInt64ToHumanStr(int64_t val, char* outStr);
int32_t taosStrHumanToInt32(const char* str);
void taosInt32ToHumanStr(int32_t val, char* outStr);
#ifdef __cplusplus
}
#endif
#endif /*_TD_UNIT_H_*/

331
packaging/tools/install.sh
View File

@ -34,6 +34,7 @@ benchmarkName="taosBenchmark"
dumpName="taosdump"
demoName="taosdemo"
xname="taosx"
keeperName="taoskeeper"
clientName2="taos"
serverName2="${clientName2}d"
@ -42,6 +43,7 @@ productName2="TDengine"
emailName2="taosdata.com"
xname2="${clientName2}x"
adapterName2="${clientName2}adapter"
keeperName2="${clientName2}keeper"
explorerName="${clientName2}-explorer"
benchmarkName2="${clientName2}Benchmark"
@ -154,7 +156,7 @@ interactiveFqdn=yes # [yes | no]
verType=server # [server | client]
initType=systemd # [systemd | service | ...]
while getopts "hv:e:i:" arg; do
while getopts "hv:e:" arg; do
case $arg in
e)
#echo "interactiveFqdn=$OPTARG"
@ -164,10 +166,6 @@ while getopts "hv:e:i:" arg; do
#echo "verType=$OPTARG"
verType=$(echo $OPTARG)
;;
i)
#echo "initType=$OPTARG"
initType=$(echo $OPTARG)
;;
h)
echo "Usage: $(basename $0) -v [server | client] -e [yes | no]"
exit 0
@ -218,6 +216,7 @@ function install_bin() {
${csudo}rm -f ${bin_link_dir}/${demoName2} || :
${csudo}rm -f ${bin_link_dir}/${benchmarkName2} || :
${csudo}rm -f ${bin_link_dir}/${dumpName2} || :
${csudo}rm -f ${bin_link_dir}/${keeperName2} || :
${csudo}rm -f ${bin_link_dir}/set_core || :
${csudo}rm -f ${bin_link_dir}/TDinsight.sh || :
@ -231,6 +230,7 @@ function install_bin() {
[ -x ${install_main_dir}/bin/${benchmarkName2} ] && ${csudo}ln -sf ${install_main_dir}/bin/${benchmarkName2} ${bin_link_dir}/${demoName2} || :
[ -x ${install_main_dir}/bin/${benchmarkName2} ] && ${csudo}ln -sf ${install_main_dir}/bin/${benchmarkName2} ${bin_link_dir}/${benchmarkName2} || :
[ -x ${install_main_dir}/bin/${dumpName2} ] && ${csudo}ln -sf ${install_main_dir}/bin/${dumpName2} ${bin_link_dir}/${dumpName2} || :
[ -x ${install_main_dir}/bin/${keeperName2} ] && ${csudo}ln -sf ${install_main_dir}/bin/${keeperName2} ${bin_link_dir}/${keeperName2} || :
[ -x ${install_main_dir}/bin/TDinsight.sh ] && ${csudo}ln -sf ${install_main_dir}/bin/TDinsight.sh ${bin_link_dir}/TDinsight.sh || :
if [ "$clientName2" == "${clientName}" ]; then
[ -x ${install_main_dir}/bin/remove.sh ] && ${csudo}ln -s ${install_main_dir}/bin/remove.sh ${bin_link_dir}/${uninstallScript} || :
@ -373,42 +373,56 @@ function add_newHostname_to_hosts() {
return
fi
done
${csudo}echo "127.0.0.1 $1" >>/etc/hosts || :
if grep -q "127.0.0.1 $1" /etc/hosts; then
return
else
${csudo}chmod 666 /etc/hosts
${csudo}echo "127.0.0.1 $1" >>/etc/hosts
fi
}
function set_hostname() {
echo -e -n "${GREEN}Please enter one hostname(must not be 'localhost')${NC}:"
read newHostname
echo -e -n "${GREEN}Host name or IP (assigned to this machine) which can be accessed by your tools or apps (must not be 'localhost')${NC}"
read -e -p " : " -i "$(hostname)" newHostname
while true; do
if [[ ! -z "$newHostname" && "$newHostname" != "localhost" ]]; then
if [ -z "$newHostname" ]; then
newHostname=$(hostname)
break
elif [ "$newHostname" != "localhost" ]; then
break
else
read -p "Please enter one hostname(must not be 'localhost'):" newHostname
echo -e -n "${GREEN}Host name or IP (assigned to this machine) which can be accessed by your tools or apps (must not be 'localhost')${NC}"
read -e -p " : " -i "$(hostname)" newHostname
fi
done
${csudo}hostname $newHostname || :
retval=$(echo $?)
if [[ $retval != 0 ]]; then
echo
echo "set hostname fail!"
return
fi
# ${csudo}hostname $newHostname || :
# retval=$(echo $?)
# if [[ $retval != 0 ]]; then
# echo
# echo "set hostname fail!"
# return
# fi
#ubuntu/centos /etc/hostname
if [[ -e /etc/hostname ]]; then
${csudo}echo $newHostname >/etc/hostname || :
fi
# #ubuntu/centos /etc/hostname
# if [[ -e /etc/hostname ]]; then
# ${csudo}echo $newHostname >/etc/hostname || :
# fi
#debian: #HOSTNAME=yourname
if [[ -e /etc/sysconfig/network ]]; then
${csudo}sed -i -r "s/#*\s*(HOSTNAME=\s*).*/\1$newHostname/" /etc/sysconfig/network || :
fi
# #debian: #HOSTNAME=yourname
# if [[ -e /etc/sysconfig/network ]]; then
# ${csudo}sed -i -r "s/#*\s*(HOSTNAME=\s*).*/\1$newHostname/" /etc/sysconfig/network || :
# fi
if [ -f ${cfg_install_dir}/${configFile2} ]; then
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$newHostname/" ${cfg_install_dir}/${configFile2}
else
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$newHostname/" ${script_dir}/cfg/${configFile2}
fi
serverFqdn=$newHostname
if [[ -e /etc/hosts ]]; then
if [[ -e /etc/hosts ]] && [[ ! $newHostname =~ ^[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}$ ]]; then
add_newHostname_to_hosts $newHostname
fi
}
@ -439,7 +453,12 @@ function set_ipAsFqdn() {
echo -e -n "${GREEN}Unable to get local ip, use 127.0.0.1${NC}"
localFqdn="127.0.0.1"
# Write the local FQDN to configuration file
if [ -f ${cfg_install_dir}/${configFile2} ]; then
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${cfg_install_dir}/${configFile2}
else
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${script_dir}/cfg/${configFile2}
fi
serverFqdn=$localFqdn
echo
return
@ -461,7 +480,11 @@ function set_ipAsFqdn() {
read -p "Please choose an IP from local IP list:" localFqdn
else
# Write the local FQDN to configuration file
if [ -f ${cfg_install_dir}/${configFile2} ]; then
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${cfg_install_dir}/${configFile2}
else
${csudo}sed -i -r "s/#*\s*(fqdn\s*).*/\1$localFqdn/" ${script_dir}/cfg/${configFile2}
fi
serverFqdn=$localFqdn
break
fi
@ -476,37 +499,13 @@ function local_fqdn_check() {
echo
echo -e -n "System hostname is: ${GREEN}$serverFqdn${NC}"
echo
if [[ "$serverFqdn" == "" ]] || [[ "$serverFqdn" == "localhost" ]]; then
echo -e -n "${GREEN}It is strongly recommended to configure a hostname for this machine ${NC}"
echo
while true; do
read -r -p "Set hostname now? [Y/n] " input
if [ ! -n "$input" ]; then
set_hostname
break
else
case $input in
[yY][eE][sS] | [yY])
set_hostname
break
;;
[nN][oO] | [nN])
set_ipAsFqdn
break
;;
*)
echo "Invalid input..."
;;
esac
fi
done
fi
}
function install_adapter_config() {
if [ -f ${script_dir}/cfg/${adapterName}.toml ]; then
${csudo}sed -i -r "s/localhost/${serverFqdn}/g" ${script_dir}/cfg/${adapterName}.toml
fi
if [ ! -f "${cfg_install_dir}/${adapterName}.toml" ]; then
${csudo}mkdir -p ${cfg_install_dir}
[ -f ${script_dir}/cfg/${adapterName}.toml ] && ${csudo}cp ${script_dir}/cfg/${adapterName}.toml ${cfg_install_dir}
@ -523,13 +522,38 @@ function install_adapter_config() {
}
function install_keeper_config() {
if [ -f ${script_dir}/cfg/${keeperName2}.toml ]; then
${csudo}sed -i -r "s/127.0.0.1/${serverFqdn}/g" ${script_dir}/cfg/${keeperName2}.toml
fi
if [ -f "${configDir}/keeper.toml" ]; then
echo "The file keeper.toml will be renamed to ${keeperName2}.toml"
${csudo}cp ${script_dir}/cfg/${keeperName2}.toml ${configDir}/${keeperName2}.toml.new
${csudo}mv ${configDir}/keeper.toml ${configDir}/${keeperName2}.toml
elif [ -f "${configDir}/${keeperName2}.toml" ]; then
# "taoskeeper.toml exists,new config is taoskeeper.toml.new"
${csudo}cp ${script_dir}/cfg/${keeperName2}.toml ${configDir}/${keeperName2}.toml.new
else
${csudo}cp ${script_dir}/cfg/${keeperName2}.toml ${configDir}/${keeperName2}.toml
fi
command -v systemctl >/dev/null 2>&1 && ${csudo}systemctl daemon-reload >/dev/null 2>&1 || true
}
function install_config() {
if [ ! -f "${cfg_install_dir}/${configFile2}" ]; then
${csudo}mkdir -p ${cfg_install_dir}
[ -f ${script_dir}/cfg/${configFile2} ] && ${csudo}cp ${script_dir}/cfg/${configFile2} ${cfg_install_dir}
if [ -f ${script_dir}/cfg/${configFile2} ]; then
${csudo} echo "monitor 1" >> ${script_dir}/cfg/${configFile2}
${csudo} echo "monitorFQDN ${serverFqdn}" >> ${script_dir}/cfg/${configFile2}
${csudo} echo "audit 1" >> ${script_dir}/cfg/${configFile2}
${csudo}cp ${script_dir}/cfg/${configFile2} ${cfg_install_dir}
fi
${csudo}chmod 644 ${cfg_install_dir}/*
else
${csudo} echo "monitor 1" >> ${script_dir}/cfg/${configFile2}
${csudo} echo "monitorFQDN ${serverFqdn}" >> ${script_dir}/cfg/${configFile2}
${csudo} echo "audit 1" >> ${script_dir}/cfg/${configFile2}
${csudo}cp -f ${script_dir}/cfg/${configFile2} ${cfg_install_dir}/${configFile2}.new
fi
@ -537,6 +561,8 @@ function install_config() {
[ ! -z $1 ] && return 0 || : # only install client
if ((${update_flag} == 1)); then
return 0
fi
@ -554,7 +580,11 @@ function install_config() {
read firstEp
while true; do
if [ ! -z "$firstEp" ]; then
if [ -f ${cfg_install_dir}/${configFile2} ]; then
${csudo}sed -i -r "s/#*\s*(firstEp\s*).*/\1$firstEp/" ${cfg_install_dir}/${configFile2}
else
${csudo}sed -i -r "s/#*\s*(firstEp\s*).*/\1$firstEp/" ${script_dir}/cfg/${configFile2}
fi
break
else
break
@ -607,6 +637,9 @@ function install_data() {
function install_connector() {
if [ -d "${script_dir}/connector/" ]; then
${csudo}cp -rf ${script_dir}/connector/ ${install_main_dir}/ || echo "failed to copy connector"
${csudo}cp ${script_dir}/start-all.sh ${install_main_dir}/ || echo "failed to copy start-all.sh"
${csudo}cp ${script_dir}/stop-all.sh ${install_main_dir}/ || echo "failed to copy stop-all.sh"
${csudo}cp ${script_dir}/README.md ${install_main_dir}/ || echo "failed to copy README.md"
fi
}
@ -622,6 +655,14 @@ function install_web() {
fi
}
function install_taosx() {
if [ -f "${script_dir}/taosx/install_taosx.sh" ]; then
cd ${script_dir}/taosx
chmod a+x install_taosx.sh
bash install_taosx.sh -e $serverFqdn
fi
}
function clean_service_on_sysvinit() {
if ps aux | grep -v grep | grep ${serverName2} &>/dev/null; then
${csudo}service ${serverName2} stop || :
@ -702,29 +743,6 @@ function clean_service_on_systemd() {
fi
${csudo}systemctl disable tarbitratord &>/dev/null || echo &>/dev/null
${csudo}rm -f ${tarbitratord_service_config}
if [ "$verMode" == "cluster" ] && [ "$clientName" != "$clientName2" ]; then
x_service_config="${service_config_dir}/${xName2}.service"
if [ -e "$x_service_config" ]; then
if systemctl is-active --quiet ${xName2}; then
echo "${productName2} ${xName2} is running, stopping it..."
${csudo}systemctl stop ${xName2} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${xName2} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${x_service_config}
fi
explorer_service_config="${service_config_dir}/${explorerName2}.service"
if [ -e "$explorer_service_config" ]; then
if systemctl is-active --quiet ${explorerName2}; then
echo "${productName2} ${explorerName2} is running, stopping it..."
${csudo}systemctl stop ${explorerName2} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${explorerName2} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${explorer_service_config}
${csudo}rm -f /etc/${clientName2}/explorer.toml
fi
fi
}
function install_service_on_systemd() {
@ -745,15 +763,27 @@ function install_service_on_systemd() {
${csudo}systemctl daemon-reload
${csudo}systemctl enable ${serverName2}
${csudo}systemctl daemon-reload
}
function install_adapter_service() {
if ((${service_mod} == 0)); then
[ -f ${script_dir}/cfg/${adapterName}.service ] &&
${csudo}cp ${script_dir}/cfg/${adapterName}.service \
[ -f ${script_dir}/cfg/${adapterName2}.service ] &&
${csudo}cp ${script_dir}/cfg/${adapterName2}.service \
${service_config_dir}/ || :
${csudo}systemctl enable ${adapterName2}
${csudo}systemctl daemon-reload
fi
}
function install_keeper_service() {
if ((${service_mod} == 0)); then
[ -f ${script_dir}/cfg/${clientName2}keeper.service ] &&
${csudo}cp ${script_dir}/cfg/${clientName2}keeper.service \
${service_config_dir}/ || :
${csudo}systemctl enable ${clientName2}keeper
${csudo}systemctl daemon-reload
fi
}
@ -872,7 +902,7 @@ function updateProduct() {
tar -zxf ${tarName}
install_jemalloc
echo -e "${GREEN}Start to update ${productName2}...${NC}"
echo "Start to update ${productName2}..."
# Stop the service if running
if ps aux | grep -v grep | grep ${serverName2} &>/dev/null; then
if ((${service_mod} == 0)); then
@ -890,9 +920,11 @@ function updateProduct() {
install_log
install_header
install_lib
install_config
if [ "$verMode" == "cluster" ]; then
install_connector
install_taosx
fi
install_examples
@ -901,52 +933,69 @@ function updateProduct() {
install_bin
install_service
install_adapter_service
install_config
install_adapter_config
install_keeper_service
install_keeper_config
openresty_work=false
echo
echo -e "${GREEN_DARK}To configure ${productName2} ${NC}\t: edit ${cfg_install_dir}/${configFile2}"
echo -e "${GREEN_DARK}To configure ${productName2} ${NC}\t\t: edit ${cfg_install_dir}/${configFile2}"
[ -f ${configDir}/${clientName2}adapter.toml ] && [ -f ${installDir}/bin/${clientName2}adapter ] && \
echo -e "${GREEN_DARK}To configure ${clientName2}Adapter ${NC}\t: edit ${configDir}/${clientName2}adapter.toml"
if [ "$verMode" == "cluster" ]; then
echo -e "${GREEN_DARK}To configure ${clientName2}-explorer ${NC}\t: edit ${configDir}/explorer.toml"
fi
if ((${service_mod} == 0)); then
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t: ${csudo}systemctl start ${serverName2}${NC}"
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t\t: ${csudo}systemctl start ${serverName2}${NC}"
[ -f ${service_config_dir}/${clientName2}adapter.service ] && [ -f ${installDir}/bin/${clientName2}adapter ] && \
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t: ${csudo}systemctl start ${clientName2}adapter ${NC}"
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t\t: ${csudo}systemctl start ${clientName2}adapter ${NC}"
elif ((${service_mod} == 1)); then
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t: ${csudo}service ${serverName2} start${NC}"
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t\t: ${csudo}service ${serverName2} start${NC}"
[ -f ${service_config_dir}/${clientName2}adapter.service ] && [ -f ${installDir}/bin/${clientName2}adapter ] && \
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t: ${csudo}service ${clientName2}adapter start${NC}"
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t\t: ${csudo}service ${clientName2}adapter start${NC}"
else
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t: ./${serverName2}${NC}"
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t\t: ./${serverName2}${NC}"
[ -f ${installDir}/bin/${clientName2}adapter ] && \
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t: ${clientName2}adapter ${NC}"
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t\t: ${clientName2}adapter ${NC}"
fi
echo -e "${GREEN_DARK}To enable ${clientName2}keeper ${NC}\t: sudo systemctl enable ${clientName2}keeper ${NC}"
if [ ${openresty_work} = 'true' ]; then
echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t: use ${GREEN_UNDERLINE}${clientName2} -h $serverFqdn${NC} in shell OR from ${GREEN_UNDERLINE}http://127.0.0.1:${web_port}${NC}"
else
echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t: use ${GREEN_UNDERLINE}${clientName2} -h $serverFqdn${NC} in shell${NC}"
echo -e "${GREEN_DARK}To enable ${clientName2}keeper ${NC}\t\t: sudo systemctl enable ${clientName2}keeper ${NC}"
if [ "$verMode" == "cluster" ];then
echo -e "${GREEN_DARK}To start ${clientName2}x ${NC}\t\t\t: sudo systemctl start ${clientName2}x ${NC}"
echo -e "${GREEN_DARK}To start ${clientName2}-explorer ${NC}\t\t: sudo systemctl start ${clientName2}-explorer ${NC}"
fi
if ((${prompt_force} == 1)); then
echo ""
echo -e "${RED}Please run '${serverName2} --force-keep-file' at first time for the exist ${productName2} $exist_version!${NC}"
fi
# if [ ${openresty_work} = 'true' ]; then
# echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t\t: use ${GREEN_UNDERLINE}${clientName2} -h $serverFqdn${NC} in shell OR from ${GREEN_UNDERLINE}http://127.0.0.1:${web_port}${NC}"
# else
# echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t\t: use ${GREEN_UNDERLINE}${clientName2} -h $serverFqdn${NC} in shell${NC}"
# fi
# if ((${prompt_force} == 1)); then
# echo ""
# echo -e "${RED}Please run '${serverName2} --force-keep-file' at first time for the exist ${productName2} $exist_version!${NC}"
# fi
echo
echo -e "\033[44;32;1m${productName2} is updated successfully!${NC}"
echo -e "\033[44;32;1mTo manage ${productName2} instance, view documentation or explorer features, please install ${clientName2}Explorer ${NC}"
echo "${productName2} is updated successfully!"
echo
if [ "$verMode" == "cluster" ];then
echo -e "\033[44;32;1mTo start all the components : ./start-all.sh${NC}"
fi
echo -e "\033[44;32;1mTo access ${productName2} : ${clientName2} -h $serverFqdn${NC}"
if [ "$verMode" == "cluster" ];then
echo -e "\033[44;32;1mTo access the management system : http://$serverFqdn:6060${NC}"
echo -e "\033[44;32;1mTo read the user manual : http://$serverFqdn:6060/docs${NC}"
fi
else
install_bin
install_config
echo
echo -e "\033[44;32;1m${productName2} client is updated successfully!${NC}"
fi
cd $script_dir
rm -rf $(tar -tf ${tarName} | grep -Ev "^\./$|^\/")
}
@ -958,7 +1007,7 @@ function installProduct() {
fi
tar -zxf ${tarName}
echo -e "${GREEN}Start to install ${productName2}...${NC}"
echo "Start to install ${productName2}..."
install_main_path
@ -972,9 +1021,11 @@ function installProduct() {
install_jemalloc
#install_avro lib
#install_avro lib64
install_config
if [ "$verMode" == "cluster" ]; then
install_connector
install_taosx
fi
install_examples
install_web
@ -984,62 +1035,80 @@ function installProduct() {
install_service
install_adapter_service
install_adapter_config
install_keeper_service
install_keeper_config
openresty_work=false
install_config
# Ask if to start the service
echo
echo -e "${GREEN_DARK}To configure ${productName2} ${NC}\t: edit ${cfg_install_dir}/${configFile2}"
echo -e "${GREEN_DARK}To configure ${productName2} ${NC}\t\t: edit ${cfg_install_dir}/${configFile2}"
[ -f ${configDir}/${clientName2}adapter.toml ] && [ -f ${installDir}/bin/${clientName2}adapter ] && \
echo -e "${GREEN_DARK}To configure ${clientName2}Adapter ${NC}\t: edit ${configDir}/${clientName2}adapter.toml"
if [ "$verMode" == "cluster" ]; then
echo -e "${GREEN_DARK}To configure ${clientName2}-explorer ${NC}\t: edit ${configDir}/explorer.toml"
fi
if ((${service_mod} == 0)); then
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t: ${csudo}systemctl start ${serverName2}${NC}"
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t\t: ${csudo}systemctl start ${serverName2}${NC}"
[ -f ${service_config_dir}/${clientName2}adapter.service ] && [ -f ${installDir}/bin/${clientName2}adapter ] && \
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t: ${csudo}systemctl start ${clientName2}adapter ${NC}"
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t\t: ${csudo}systemctl start ${clientName2}adapter ${NC}"
elif ((${service_mod} == 1)); then
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t: ${csudo}service ${serverName2} start${NC}"
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t\t: ${csudo}service ${serverName2} start${NC}"
[ -f ${service_config_dir}/${clientName2}adapter.service ] && [ -f ${installDir}/bin/${clientName2}adapter ] && \
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t: ${csudo}service ${clientName2}adapter start${NC}"
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t\t: ${csudo}service ${clientName2}adapter start${NC}"
else
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t: ${serverName2}${NC}"
echo -e "${GREEN_DARK}To start ${productName2} ${NC}\t\t: ${serverName2}${NC}"
[ -f ${installDir}/bin/${clientName2}adapter ] && \
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t: ${clientName2}adapter ${NC}"
echo -e "${GREEN_DARK}To start ${clientName2}Adapter ${NC}\t\t: ${clientName2}adapter ${NC}"
fi
echo -e "${GREEN_DARK}To enable ${clientName2}keeper ${NC}\t: sudo systemctl enable ${clientName2}keeper ${NC}"
echo -e "${GREEN_DARK}To enable ${clientName2}keeper ${NC}\t\t: sudo systemctl enable ${clientName2}keeper ${NC}"
if [ ! -z "$firstEp" ]; then
tmpFqdn=${firstEp%%:*}
substr=":"
if [[ $firstEp =~ $substr ]]; then
tmpPort=${firstEp#*:}
else
tmpPort=""
if [ "$verMode" == "cluster" ];then
echo -e "${GREEN_DARK}To start ${clientName2}x ${NC}\t\t\t: sudo systemctl start ${clientName2}x ${NC}"
echo -e "${GREEN_DARK}To start ${clientName2}-explorer ${NC}\t\t: sudo systemctl start ${clientName2}-explorer ${NC}"
fi
if [[ "$tmpPort" != "" ]]; then
echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t: ${clientName2} -h $tmpFqdn -P $tmpPort${GREEN_DARK} to login into cluster, then${NC}"
else
echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t: ${clientName2} -h $tmpFqdn${GREEN_DARK} to login into cluster, then${NC}"
fi
echo -e "${GREEN_DARK}execute ${NC}: create dnode 'newDnodeFQDN:port'; ${GREEN_DARK}to add this new node${NC}"
# if [ ! -z "$firstEp" ]; then
# tmpFqdn=${firstEp%%:*}
# substr=":"
# if [[ $firstEp =~ $substr ]]; then
# tmpPort=${firstEp#*:}
# else
# tmpPort=""
# fi
# if [[ "$tmpPort" != "" ]]; then
# echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t\t: ${clientName2} -h $tmpFqdn -P $tmpPort${GREEN_DARK} to login into cluster, then${NC}"
# else
# echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t\t: ${clientName2} -h $tmpFqdn${GREEN_DARK} to login into cluster, then${NC}"
# fi
# echo -e "${GREEN_DARK}execute ${NC}: create dnode 'newDnodeFQDN:port'; ${GREEN_DARK}to add this new node${NC}"
# echo
# elif [ ! -z "$serverFqdn" ]; then
# echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t\t: ${clientName2} -h $serverFqdn${GREEN_DARK} to login into ${productName2} server${NC}"
# echo
# fi
echo
elif [ ! -z "$serverFqdn" ]; then
echo -e "${GREEN_DARK}To access ${productName2} ${NC}\t: ${clientName2} -h $serverFqdn${GREEN_DARK} to login into ${productName2} server${NC}"
echo "${productName2} is installed successfully!"
echo
if [ "$verMode" == "cluster" ];then
echo -e "\033[44;32;1mTo start all the components : sudo ./start-all.sh${NC}"
fi
echo -e "\033[44;32;1mTo access ${productName2} : ${clientName2} -h $serverFqdn${NC}"
if [ "$verMode" == "cluster" ];then
echo -e "\033[44;32;1mTo access the management system : http://$serverFqdn:6060${NC}"
echo -e "\033[44;32;1mTo read the user manual : http://$serverFqdn:6060/docs-en${NC}"
fi
echo -e "\033[44;32;1m${productName2} is installed successfully!${NC}"
echo -e "\033[44;32;1mTo manage ${productName2} instance, view documentation or explorer features, please install ${clientName2}Explorer ${NC}"
echo
else # Only install client
install_bin
install_config
echo
echo -e "\033[44;32;1m${productName2} client is installed successfully!${NC}"
fi
cd $script_dir
touch ~/.${historyFile}
rm -rf $(tar -tf ${tarName} | grep -Ev "^\./$|^\/")
}

1
packaging/tools/install_client.sh
View File

@ -129,6 +129,7 @@ function install_bin() {
if [ "$osType" != "Darwin" ]; then
[ -x ${install_main_dir}/bin/${demoName2} ] && ${csudo}ln -s ${install_main_dir}/bin/${demoName2} ${bin_link_dir}/${demoName2} || :
[ -x ${install_main_dir}/bin/${benchmarkName2} ] && ${csudo}ln -s ${install_main_dir}/bin/${benchmarkName2} ${bin_link_dir}/${benchmarkName2} || :
[ -x ${install_main_dir}/bin/${dumpName2} ] && ${csudo}ln -s ${install_main_dir}/bin/${dumpName2} ${bin_link_dir}/${dumpName2} || :
fi
[ -x ${install_main_dir}/bin/remove_client.sh ] && ${csudo}ln -sf ${install_main_dir}/bin/remove_client.sh ${bin_link_dir}/${uninstallScript2} || :
fi

3
packaging/tools/makeclient.sh
View File

@ -24,10 +24,12 @@ clientName2="${12}"
productName="TDengine"
clientName="taos"
benchmarkName="taosBenchmark"
dumpName="taosdump"
configFile="taos.cfg"
tarName="package.tar.gz"
benchmarkName2="${clientName2}Benchmark"
dumpName2="${clientName2}dump"
if [ "$osType" != "Darwin" ]; then
script_dir="$(dirname $(readlink -f $0))"
@ -71,6 +73,7 @@ if [ "$osType" != "Darwin" ]; then
else
bin_files="${build_dir}/bin/${clientName} \
${build_dir}/bin/${benchmarkName} \
${build_dir}/bin/${dumpName} \
${script_dir}/remove_client.sh \
${script_dir}/set_core.sh \
${script_dir}/get_client.sh"

23
packaging/tools/makepkg.sh
View File

@ -42,7 +42,7 @@ release_dir="${top_dir}/release"
#package_name='linux'
if [ "$verMode" == "cluster" ]; then
install_dir="${release_dir}/${productName2}-enterprise-server-${version}"
install_dir="${release_dir}/${productName2}-enterprise-${version}"
elif [ "$verMode" == "cloud" ]; then
install_dir="${release_dir}/${productName2}-cloud-server-${version}"
else
@ -92,14 +92,10 @@ else
${build_dir}/bin/tdengine-datasource.zip.md5"
fi
[ -f ${build_dir}/bin/taosx ] && taosx_bin="${build_dir}/bin/taosx"
explorer_bin_files=$(find ${build_dir}/bin/ -name '*-explorer')
bin_files="${build_dir}/bin/${serverName} \
${build_dir}/bin/${clientName} \
${taostools_bin_files} \
${taosx_bin} \
${explorer_bin_files} \
${build_dir}/bin/${clientName}adapter \
${build_dir}/bin/udfd \
${script_dir}/remove.sh \
@ -330,7 +326,7 @@ fi
mkdir -p ${install_dir}/driver && cp ${lib_files} ${install_dir}/driver && echo "${versionComp}" >${install_dir}/driver/vercomp.txt
[ -f ${wslib_files} ] && cp ${wslib_files} ${install_dir}/driver || :
# Copy connector
# Copy connector && taosx
if [ "$verMode" == "cluster" ]; then
connector_dir="${code_dir}/connector"
mkdir -p ${install_dir}/connector
@ -364,8 +360,19 @@ if [ "$verMode" == "cluster" ]; then
git clone --depth 1 https://github.com/taosdata/taos-connector-rust ${install_dir}/connector/rust
rm -rf ${install_dir}/connector/rust/.git ||:
# cp -r ${connector_dir}/python ${install_dir}/connector
# cp -r ${connector_dir}/nodejs ${install_dir}/connector
cp ${top_dir}/../enterprise/packaging/start-all.sh ${install_dir}
cp ${top_dir}/../enterprise/packaging/stop-all.sh ${install_dir}
cp ${top_dir}/../enterprise/packaging/README.md ${install_dir}
chmod a+x ${install_dir}/start-all.sh
chmod a+x ${install_dir}/stop-all.sh
# copy taosx
if [ -d ${top_dir}/../enterprise/src/plugins/taosx/release/taosx ]; then
cp -r ${top_dir}/../enterprise/src/plugins/taosx/release/taosx ${install_dir}
cp ${top_dir}/../enterprise/packaging/install_taosx.sh ${install_dir}/taosx
cp ${top_dir}/../enterprise/src/plugins/taosx/packaging/uninstall.sh ${install_dir}/taosx
sed -i 's/target=\"\"/target=\"taosx\"/g' ${install_dir}/taosx/uninstall.sh
fi
fi
fi

135
packaging/tools/remove.sh
View File

@ -63,6 +63,10 @@ service_config_dir="/etc/systemd/system"
taos_service_name=${serverName2}
taosadapter_service_name="${clientName2}adapter"
tarbitrator_service_name="tarbitratord"
config_dir="/etc/${clientName2}"
csudo=""
if command -v sudo >/dev/null; then
csudo="sudo "
@ -113,8 +117,10 @@ function clean_bin() {
# Remove link
${csudo}rm -f ${bin_link_dir}/${clientName} || :
${csudo}rm -f ${bin_link_dir}/${serverName} || :
echo "${serverName} is removed successfully"
${csudo}rm -f ${bin_link_dir}/udfd || :
${csudo}rm -f ${bin_link_dir}/${adapterName2} || :
echo "${adapterName2} is removed successfully"
${csudo}rm -f ${bin_link_dir}/${benchmarkName2} || :
${csudo}rm -f ${bin_link_dir}/${demoName2} || :
${csudo}rm -f ${bin_link_dir}/${dumpName2} || :
@ -175,7 +181,7 @@ function clean_log() {
function clean_service_on_systemd() {
taosd_service_config="${service_config_dir}/${taos_service_name}.service"
if systemctl is-active --quiet ${taos_service_name}; then
echo "${productName2} ${serverName2} is running, stopping it..."
echo "${taos_service_name} is running, stopping it..."
${csudo}systemctl stop ${taos_service_name} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${taos_service_name} &>/dev/null || echo &>/dev/null
@ -183,7 +189,7 @@ function clean_service_on_systemd() {
taosadapter_service_config="${service_config_dir}/${clientName2}adapter.service"
if systemctl is-active --quiet ${taosadapter_service_name}; then
echo "${productName2} ${clientName2}Adapter is running, stopping it..."
echo "${clientName2}Adapter is running, stopping it..."
${csudo}systemctl stop ${taosadapter_service_name} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${taosadapter_service_name} &>/dev/null || echo &>/dev/null
@ -196,33 +202,11 @@ function clean_service_on_systemd() {
fi
${csudo}systemctl disable ${tarbitrator_service_name} &>/dev/null || echo &>/dev/null
if [ "$verMode" == "cluster" ] && [ "$clientName" != "$clientName2" ]; then
x_service_config="${service_config_dir}/${xName2}.service"
if [ -e "$x_service_config" ]; then
if systemctl is-active --quiet ${xName2}; then
echo "${productName2} ${xName2} is running, stopping it..."
${csudo}systemctl stop ${xName2} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${xName2} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${x_service_config}
fi
explorer_service_config="${service_config_dir}/${explorerName2}.service"
if [ -e "$explorer_service_config" ]; then
if systemctl is-active --quiet ${explorerName2}; then
echo "${productName2} ${explorerName2} is running, stopping it..."
${csudo}systemctl stop ${explorerName2} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${explorerName2} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${explorer_service_config}
${csudo}rm -f /etc/${clientName2}/explorer.toml
fi
fi
}
function clean_service_on_sysvinit() {
if ps aux | grep -v grep | grep ${serverName} &>/dev/null; then
echo "${productName2} ${serverName2} is running, stopping it..."
echo "${serverName2} is running, stopping it..."
${csudo}service ${serverName} stop || :
fi
@ -284,6 +268,97 @@ function clean_service() {
fi
}
function remove_data_and_config() {
data_dir=`grep dataDir /etc/taos/taos.cfg | grep -v '#' | tail -n 1 | awk {'print $2'}`
if [ X"$data_dir" == X"" ]; then
data_dir="/var/lib/taos"
fi
log_dir=`grep logDir /etc/taos/taos.cfg | grep -v '#' | tail -n 1 | awk {'print $2'}`
if [ X"$log_dir" == X"" ]; then
log_dir="/var/log/taos"
fi
[ -d "${config_dir}" ] && ${csudo}rm -rf ${config_dir}/*
[ -d "${data_dir}" ] && ${csudo}rm -rf ${data_dir}/*
[ -d "${log_dir}" ] && ${csudo}rm -rf ${log_dir}/*
}
_kill_service_of() {
_service=$1
pid=$(ps -ef | grep "$_service" | grep -v "grep" | awk '{print $2}')
if [ -n "$pid" ]; then
${csudo}kill -9 $pid || :
fi
}
_clean_service_on_systemd_of() {
_service=$1
_service_config="${service_config_dir}/${_service}.service"
if systemctl is-active --quiet ${_service}; then
echo "taoskeeper is running, stopping it..."
${csudo}systemctl stop ${_service} &>/dev/null || echo &>/dev/null
fi
${csudo}systemctl disable ${_service} &>/dev/null || echo &>/dev/null
${csudo}rm -f ${_service_config}
}
_clean_service_on_sysvinit_of() {
_service=$1
if pidof ${_service} &>/dev/null; then
echo "${_service} is running, stopping it..."
${csudo}service ${_service} stop || :
fi
if ((${initd_mod} == 1)); then
if [ -e ${service_config_dir}/${_service} ]; then
${csudo}chkconfig --del ${_service} || :
fi
elif ((${initd_mod} == 2)); then
if [ -e ${service_config_dir}/${_service} ]; then
${csudo}insserv -r ${_service} || :
fi
elif ((${initd_mod} == 3)); then
if [ -e ${service_config_dir}/${_service} ]; then
${csudo}update-rc.d -f ${_service} remove || :
fi
fi
${csudo}rm -f ${service_config_dir}/${_service} || :
if $(which init &>/dev/null); then
${csudo}init q || :
fi
}
_clean_service_of() {
_service=$1
if ((${service_mod} == 0)); then
_clean_service_on_systemd_of $_service
elif ((${service_mod} == 1)); then
_clean_service_on_sysvinit_of $_service
else
_kill_service_of $_service
fi
}
remove_taoskeeper() {
# remove taoskeeper bin
_clean_service_of taoskeeper
[ -e "${bin_link_dir}/taoskeeper" ] && ${csudo}rm -rf ${bin_link_dir}/taoskeeper
[ -e "${installDir}/taoskeeper" ] && ${csudo}rm -rf ${installDir}/taoskeeper
[ -e "${cfg_link_dir}/metrics.toml" ] || ${csudo}rm -rf ${cfg_link_dir}/metrics.toml
echo "taosKeeper is removed successfully!"
}
function uninstall_taosx() {
if [ -f ${installDir}/uninstall.sh ]; then
cd ${installDir}
bash uninstall.sh
fi
}
if [ "$verMode" == "cluster" ]; then
uninstall_taosx
fi
remove_taoskeeper
# Stop service and disable booting start.
clean_service
# Remove binary file and links
@ -322,5 +397,13 @@ if [ "$osType" = "Darwin" ]; then
${csudo}rm -rf /Applications/TDengine.app
fi
echo -e "${GREEN}${productName2} is removed successfully!${NC}"
echo
echo "Do you want to remove all the data, log and configuration files? [y/n]"
read answer
if [ X$answer == X"y" ] || [ X$answer == X"Y" ]; then
remove_data_and_config
fi
echo
echo "${productName2} is removed successfully!"
echo

2
source/client/src/clientEnv.c
View File

@ -776,7 +776,7 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
} else {
tscInfo("set cfg:%s to %s", pItem->name, str);
if (TSDB_OPTION_SHELL_ACTIVITY_TIMER == option || TSDB_OPTION_USE_ADAPTER == option) {
code = taosApplyLocalCfg(pCfg, pItem->name);
code = taosCfgDynamicOptions(pCfg, pItem->name, false);
}
}

21
source/common/src/cos.c
View File

@ -267,8 +267,6 @@ typedef struct list_parts_callback_data {
} list_parts_callback_data;
typedef struct MultipartPartData {
char err_msg[512];
S3Status status;
put_object_callback_data put_object_data;
int seq;
UploadManager *manager;
@ -276,11 +274,12 @@ typedef struct MultipartPartData {
static int putObjectDataCallback(int bufferSize, char *buffer, void *callbackData) {
put_object_callback_data *data = (put_object_callback_data *)callbackData;
/*
if (data->infileFD == 0) {
MultipartPartData *mpd = (MultipartPartData *)callbackData;
data = &mpd->put_object_data;
}
*/
int ret = 0;
if (data->contentLength) {
@ -458,13 +457,13 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object) {
int metaPropertiesCount = 0;
S3NameValue metaProperties[S3_MAX_METADATA_COUNT];
char useServerSideEncryption = 0;
int noStatus = 0;
put_object_callback_data data;
put_object_callback_data data = {0};
// int noStatus = 0;
// data.infile = 0;
data.infileFD = NULL;
data.gb = 0;
data.noStatus = noStatus;
// data.gb = 0;
// data.infileFD = NULL;
// data.noStatus = noStatus;
if (taosStatFile(file, &contentLength, NULL, NULL) < 0) {
uError("ERROR: %s Failed to stat file %s: ", __func__, file);
@ -581,9 +580,9 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object) {
do {
S3_upload_part(&bucketContext, key, &putProperties, &putObjectHandler, seq, manager.upload_id,
partContentLength, 0, timeoutMsG, &partData);
} while (S3_status_is_retryable(partData.status) && should_retry());
if (partData.status != S3StatusOK) {
s3PrintError(__func__, partData.status, partData.err_msg);
} while (S3_status_is_retryable(partData.put_object_data.status) && should_retry());
if (partData.put_object_data.status != S3StatusOK) {
s3PrintError(__func__, partData.put_object_data.status, partData.put_object_data.err_msg);
code = TAOS_SYSTEM_ERROR(EIO);
goto clean;
}

647
source/common/src/tglobal.c
View File

@ -21,6 +21,7 @@
#include "tgrant.h"
#include "tlog.h"
#include "tmisce.h"
#include "tunit.h"
#if defined(CUS_NAME) || defined(CUS_PROMPT) || defined(CUS_EMAIL)
#include "cus_name.h"
@ -721,7 +722,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddBool(pCfg, "disableStream", tsDisableStream, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0) return -1;
if (cfgAddInt64(pCfg, "streamBufferSize", tsStreamBufferSize, 0, INT64_MAX, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0)
return -1;
if (cfgAddInt64(pCfg, "checkpointInterval", tsStreamCheckpointInterval, 60, 1200, CFG_SCOPE_SERVER,
if (cfgAddInt32(pCfg, "checkpointInterval", tsStreamCheckpointInterval, 60, 1200, CFG_SCOPE_SERVER,
CFG_DYN_ENT_SERVER) != 0)
return -1;
if (cfgAddFloat(pCfg, "streamSinkDataRate", tsSinkDataRate, 0.1, 5, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
@ -752,6 +753,10 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
if (cfgAddString(pCfg, "s3BucketName", tsS3BucketName, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0) return -1;
if (cfgAddInt32(pCfg, "s3BlockSize", tsS3BlockSize, -1, 1024 * 1024, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) != 0)
return -1;
if (tsS3BlockSize > -1 && tsS3BlockSize < 1024) {
uError("failed to config s3blocksize since value:%d. Valid range: -1 or [1024, 1024 * 1024]", tsS3BlockSize);
return -1;
}
if (cfgAddInt32(pCfg, "s3BlockCacheSize", tsS3BlockCacheSize, 4, 1024 * 1024, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER) !=
0)
return -1;
@ -1194,302 +1199,6 @@ static int32_t taosSetReleaseCfg(SConfig *pCfg) { return 0; }
int32_t taosSetReleaseCfg(SConfig *pCfg);
#endif
int32_t taosApplyLocalCfg(SConfig *pCfg, char *name) {
int32_t len = strlen(name);
char lowcaseName[CFG_NAME_MAX_LEN + 1] = {0};
strntolower(lowcaseName, name, TMIN(CFG_NAME_MAX_LEN, len));
bool matchItem = true;
switch (lowcaseName[0]) {
case 'a': {
if (strcasecmp("asyncLog", name) == 0) {
tsAsyncLog = cfgGetItem(pCfg, "asyncLog")->bval;
} else if (strcasecmp("assert", name) == 0) {
tsAssert = cfgGetItem(pCfg, "assert")->bval;
} else {
matchItem = false;
}
break;
}
case 'c': {
if (strcasecmp("compressMsgSize", name) == 0) {
tsCompressMsgSize = cfgGetItem(pCfg, "compressMsgSize")->i32;
} else if (strcasecmp("countAlwaysReturnValue", name) == 0) {
tsCountAlwaysReturnValue = cfgGetItem(pCfg, "countAlwaysReturnValue")->i32;
} else if (strcasecmp("cDebugFlag", name) == 0) {
cDebugFlag = cfgGetItem(pCfg, "cDebugFlag")->i32;
} else if (strcasecmp("crashReporting", name) == 0) {
tsEnableCrashReport = cfgGetItem(pCfg, "crashReporting")->bval;
} else {
matchItem = false;
}
break;
}
case 'd': {
if (strcasecmp("dDebugFlag", name) == 0) {
dDebugFlag = cfgGetItem(pCfg, "dDebugFlag")->i32;
} else if (strcasecmp("debugFlag", name) == 0) {
int32_t flag = cfgGetItem(pCfg, "debugFlag")->i32;
taosSetAllDebugFlag(flag, true);
} else {
matchItem = false;
}
break;
}
case 'e': {
if (strcasecmp("enableCoreFile", name) == 0) {
bool enableCore = cfgGetItem(pCfg, "enableCoreFile")->bval;
taosSetCoreDump(enableCore);
} else if (strcasecmp("enableQueryHb", name) == 0) {
tsEnableQueryHb = cfgGetItem(pCfg, "enableQueryHb")->bval;
} else {
matchItem = false;
}
break;
}
case 'f': {
if (strcasecmp("fqdn", name) == 0) {
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_FQDN_LEN);
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
char defaultFirstEp[TSDB_EP_LEN] = {0};
snprintf(defaultFirstEp, TSDB_EP_LEN, "%s:%u", tsLocalFqdn, tsServerPort);
SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp");
SEp firstEp = {0};
taosGetFqdnPortFromEp(strlen(pFirstEpItem->str) == 0 ? defaultFirstEp : pFirstEpItem->str, &firstEp);
snprintf(tsFirst, sizeof(tsFirst), "%s:%u", firstEp.fqdn, firstEp.port);
cfgSetItem(pCfg, "firstEp", tsFirst, pFirstEpItem->stype);
} else if (strcasecmp("firstEp", name) == 0) {
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_FQDN_LEN);
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
char defaultFirstEp[TSDB_EP_LEN] = {0};
snprintf(defaultFirstEp, TSDB_EP_LEN, "%s:%u", tsLocalFqdn, tsServerPort);
SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp");
SEp firstEp = {0};
taosGetFqdnPortFromEp(strlen(pFirstEpItem->str) == 0 ? defaultFirstEp : pFirstEpItem->str, &firstEp);
snprintf(tsFirst, sizeof(tsFirst), "%s:%u", firstEp.fqdn, firstEp.port);
cfgSetItem(pCfg, "firstEp", tsFirst, pFirstEpItem->stype);
} else if (strcasecmp("fsDebugFlag", name) == 0) {
fsDebugFlag = cfgGetItem(pCfg, "fsDebugFlag")->i32;
} else {
matchItem = false;
}
break;
}
case 'i': {
if (strcasecmp("idxDebugFlag", name) == 0) {
idxDebugFlag = cfgGetItem(pCfg, "idxDebugFlag")->i32;
} else {
matchItem = false;
}
break;
}
case 'j': {
if (strcasecmp("jniDebugFlag", name) == 0) {
jniDebugFlag = cfgGetItem(pCfg, "jniDebugFlag")->i32;
} else {
matchItem = false;
}
break;
}
case 'k': {
if (strcasecmp("keepColumnName", name) == 0) {
tsKeepColumnName = cfgGetItem(pCfg, "keepColumnName")->bval;
} else if (strcasecmp("keepAliveIdle", name) == 0) {
tsKeepAliveIdle = cfgGetItem(pCfg, "keepAliveIdle")->i32;
} else {
matchItem = false;
}
break;
}
case 'l': {
if (strcasecmp("locale", name) == 0) {
const char *locale = cfgGetItem(pCfg, "locale")->str;
const char *charset = cfgGetItem(pCfg, "charset")->str;
taosSetSystemLocale(locale, charset);
osSetSystemLocale(locale, charset);
} else if (strcasecmp("logDir", name) == 0) {
tstrncpy(tsLogDir, cfgGetItem(pCfg, "logDir")->str, PATH_MAX);
taosExpandDir(tsLogDir, tsLogDir, PATH_MAX);
} else if (strcasecmp("logKeepDays", name) == 0) {
tsLogKeepDays = cfgGetItem(pCfg, "logKeepDays")->i32;
} else {
matchItem = false;
}
break;
}
case 'm': {
switch (lowcaseName[1]) {
case 'a': {
if (strcasecmp("maxInsertBatchRows", name) == 0) {
tsMaxInsertBatchRows = cfgGetItem(pCfg, "maxInsertBatchRows")->i32;
} else if (strcasecmp("maxRetryWaitTime", name) == 0) {
tsMaxRetryWaitTime = cfgGetItem(pCfg, "maxRetryWaitTime")->i32;
} else {
matchItem = false;
}
break;
}
case 'e': {
if (strcasecmp("metaCacheMaxSize", name) == 0) {
atomic_store_32(&tsMetaCacheMaxSize, cfgGetItem(pCfg, "metaCacheMaxSize")->i32);
} else {
matchItem = false;
}
break;
}
case 'i': {
if (strcasecmp("minimalTmpDirGB", name) == 0) {
tsTempSpace.reserved = (int64_t)(((double)cfgGetItem(pCfg, "minimalTmpDirGB")->fval) * 1024 * 1024 * 1024);
} else if (strcasecmp("minimalDataDirGB", name) == 0) {
tsDataSpace.reserved = (int64_t)(((double)cfgGetItem(pCfg, "minimalDataDirGB")->fval) * 1024 * 1024 * 1024);
} else if (strcasecmp("minSlidingTime", name) == 0) {
tsMinSlidingTime = cfgGetItem(pCfg, "minSlidingTime")->i32;
} else if (strcasecmp("minIntervalTime", name) == 0) {
tsMinIntervalTime = cfgGetItem(pCfg, "minIntervalTime")->i32;
} else if (strcasecmp("minimalLogDirGB", name) == 0) {
tsLogSpace.reserved = (int64_t)(((double)cfgGetItem(pCfg, "minimalLogDirGB")->fval) * 1024 * 1024 * 1024);
} else {
matchItem = false;
}
break;
}
default:
terrno = TSDB_CODE_CFG_NOT_FOUND;
return -1;
}
break;
}
case 'n': {
if (strcasecmp("numOfLogLines", name) == 0) {
tsNumOfLogLines = cfgGetItem(pCfg, "numOfLogLines")->i32;
} else {
matchItem = false;
}
break;
}
case 'q': {
if (strcasecmp("querySmaOptimize", name) == 0) {
tsQuerySmaOptimize = cfgGetItem(pCfg, "querySmaOptimize")->i32;
} else if (strcasecmp("queryPolicy", name) == 0) {
tsQueryPolicy = cfgGetItem(pCfg, "queryPolicy")->i32;
} else if (strcasecmp("qDebugFlag", name) == 0) {
qDebugFlag = cfgGetItem(pCfg, "qDebugFlag")->i32;
} else if (strcasecmp("queryPlannerTrace", name) == 0) {
tsQueryPlannerTrace = cfgGetItem(pCfg, "queryPlannerTrace")->bval;
} else if (strcasecmp("queryNodeChunkSize", name) == 0) {
tsQueryNodeChunkSize = cfgGetItem(pCfg, "queryNodeChunkSize")->i32;
} else if (strcasecmp("queryUseNodeAllocator", name) == 0) {
tsQueryUseNodeAllocator = cfgGetItem(pCfg, "queryUseNodeAllocator")->bval;
} else {
matchItem = false;
}
break;
}
case 'r': {
if (strcasecmp("rpcDebugFlag", name) == 0) {
rpcDebugFlag = cfgGetItem(pCfg, "rpcDebugFlag")->i32;
} else {
matchItem = false;
}
break;
}
case 's': {
if (strcasecmp("secondEp", name) == 0) {
SConfigItem *pSecondpItem = cfgGetItem(pCfg, "secondEp");
SEp secondEp = {0};
taosGetFqdnPortFromEp(strlen(pSecondpItem->str) == 0 ? tsFirst : pSecondpItem->str, &secondEp);
snprintf(tsSecond, sizeof(tsSecond), "%s:%u", secondEp.fqdn, secondEp.port);
cfgSetItem(pCfg, "secondEp", tsSecond, pSecondpItem->stype);
} else if (strcasecmp("smlChildTableName", name) == 0) {
tstrncpy(tsSmlChildTableName, cfgGetItem(pCfg, "smlChildTableName")->str, TSDB_TABLE_NAME_LEN);
} else if (strcasecmp("smlAutoChildTableNameDelimiter", name) == 0) {
tstrncpy(tsSmlAutoChildTableNameDelimiter, cfgGetItem(pCfg, "smlAutoChildTableNameDelimiter")->str,
TSDB_TABLE_NAME_LEN);
} else if (strcasecmp("smlTagName", name) == 0) {
tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN);
// } else if (strcasecmp("smlDataFormat", name) == 0) {
// tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval;
// } else if (strcasecmp("smlBatchSize", name) == 0) {
// tsSmlBatchSize = cfgGetItem(pCfg, "smlBatchSize")->i32;
} else if (strcasecmp("smlTsDefaultName", name) == 0) {
tstrncpy(tsSmlTsDefaultName, cfgGetItem(pCfg, "smlTsDefaultName")->str, TSDB_COL_NAME_LEN);
} else if (strcasecmp("smlDot2Underline", name) == 0) {
tsSmlDot2Underline = cfgGetItem(pCfg, "smlDot2Underline")->bval;
} else if (strcasecmp("shellActivityTimer", name) == 0) {
tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32;
} else if (strcasecmp("serverPort", name) == 0) {
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_FQDN_LEN);
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
char defaultFirstEp[TSDB_EP_LEN] = {0};
snprintf(defaultFirstEp, TSDB_EP_LEN, "%s:%u", tsLocalFqdn, tsServerPort);
SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp");
SEp firstEp = {0};
taosGetFqdnPortFromEp(strlen(pFirstEpItem->str) == 0 ? defaultFirstEp : pFirstEpItem->str, &firstEp);
snprintf(tsFirst, sizeof(tsFirst), "%s:%u", firstEp.fqdn, firstEp.port);
cfgSetItem(pCfg, "firstEp", tsFirst, pFirstEpItem->stype);
} else if (strcasecmp("smaDebugFlag", name) == 0) {
smaDebugFlag = cfgGetItem(pCfg, "smaDebugFlag")->i32;
} else if (strcasecmp("slowLogThreshold", name) == 0) {
tsSlowLogThreshold = cfgGetItem(pCfg, "slowLogThreshold")->i32;
} else if (strcasecmp("slowLogScope", name) == 0) {
if (taosSetSlowLogScope(cfgGetItem(pCfg, "slowLogScope")->str)) {
return -1;
}
} else {
matchItem = false;
}
break;
}
case 't': {
if (strcasecmp("timezone", name) == 0) {
SConfigItem *pItem = cfgGetItem(pCfg, "timezone");
osSetTimezone(pItem->str);
uDebug("timezone format changed from %s to %s", pItem->str, tsTimezoneStr);
cfgSetItem(pCfg, "timezone", tsTimezoneStr, pItem->stype);
} else if (strcasecmp("tempDir", name) == 0) {
tstrncpy(tsTempDir, cfgGetItem(pCfg, "tempDir")->str, PATH_MAX);
taosExpandDir(tsTempDir, tsTempDir, PATH_MAX);
if (taosMulMkDir(tsTempDir) != 0) {
uError("failed to create tempDir:%s since %s", tsTempDir, terrstr());
return -1;
}
} else if (strcasecmp("telemetryServer", name) == 0) {
tstrncpy(tsTelemServer, cfgGetItem(pCfg, "telemetryServer")->str, TSDB_FQDN_LEN);
} else if (strcasecmp("tmrDebugFlag", name) == 0) {
tmrDebugFlag = cfgGetItem(pCfg, "tmrDebugFlag")->i32;
} else {
matchItem = false;
}
break;
}
case 'u': {
if (strcasecmp("uDebugFlag", name) == 0) {
uDebugFlag = cfgGetItem(pCfg, "uDebugFlag")->i32;
} else if (strcasecmp("useAdapter", name) == 0) {
tsUseAdapter = cfgGetItem(pCfg, "useAdapter")->bval;
} else {
matchItem = false;
}
break;
}
default:
terrno = TSDB_CODE_CFG_NOT_FOUND;
return -1;
}
if (!matchItem) terrno = TSDB_CODE_CFG_NOT_FOUND;
return matchItem ? 0 : -1;
}
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd,
const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc) {
if (tsCfg == NULL) osDefaultInit();
@ -1627,48 +1336,98 @@ void taosCleanupCfg() {
tsCfg = NULL;
}
}
typedef struct {
const char *optionName;
void *optionVar;
} OptionNameAndVar;
void taosCfgDynamicOptions(const char *option, const char *value) {
if (strncasecmp(option, "debugFlag", 9) == 0) {
int32_t flag = atoi(value);
taosSetAllDebugFlag(flag, true);
return;
static int32_t taosCfgSetOption(OptionNameAndVar *pOptions, int32_t optionSize, SConfigItem *pItem, bool isDebugflag) {
terrno = TSDB_CODE_INVALID_CFG;
char *name = pItem->name;
for (int32_t d = 0; d < optionSize; ++d) {
const char *optName = pOptions[d].optionName;
int32_t optLen = strlen(optName);
if (strncasecmp(name, optName, optLen) != 0) continue;
switch (pItem->dtype) {
case CFG_DTYPE_BOOL: {
int32_t flag = pItem->i32;
bool *pVar = pOptions[d].optionVar;
uInfo("%s set from %d to %d", optName, *pVar, flag);
*pVar = flag;
terrno = TSDB_CODE_SUCCESS;
} break;
case CFG_DTYPE_INT32: {
int32_t flag = pItem->i32;
int32_t *pVar = pOptions[d].optionVar;
uInfo("%s set from %d to %d", optName, *pVar, flag);
*pVar = flag;
if (isDebugflag) {
taosSetDebugFlag(pOptions[d].optionVar, optName, flag, true);
}
terrno = TSDB_CODE_SUCCESS;
} break;
case CFG_DTYPE_INT64: {
int64_t flag = pItem->i64;
int64_t *pVar = pOptions[d].optionVar;
uInfo("%s set from %" PRId64 " to %" PRId64, optName, *pVar, flag);
*pVar = flag;
terrno = TSDB_CODE_SUCCESS;
} break;
case CFG_DTYPE_FLOAT:
case CFG_DTYPE_DOUBLE: {
float flag = pItem->fval;
float *pVar = pOptions[d].optionVar;
uInfo("%s set from %f to %f", optName, *pVar, flag);
*pVar = flag;
terrno = TSDB_CODE_SUCCESS;
} break;
default:
terrno = TSDB_CODE_INVALID_CFG;
break;
}
if (strcasecmp(option, "resetlog") == 0) {
break;
}
return terrno == TSDB_CODE_SUCCESS ? 0 : -1;
}
static int32_t taosCfgDynamicOptionsForServer(SConfig *pCfg, char *name) {
terrno = TSDB_CODE_SUCCESS;
SConfigItem *pItem = cfgGetItem(pCfg, name);
if (!pItem || (pItem->dynScope & CFG_DYN_SERVER) == 0) {
uError("failed to config:%s, not support", name);
terrno = TSDB_CODE_INVALID_CFG;
return -1;
}
if (strncasecmp(name, "debugFlag", 9) == 0) {
int32_t flag = pItem->i32;
taosSetAllDebugFlag(flag, true);
return 0;
}
if (strcasecmp(name, "resetlog") == 0) {
taosResetLog();
cfgDumpCfg(tsCfg, 0, false);
return;
return 0;
}
{ // 'bool/int32_t/int64_t' variables with general modification function
const int32_t nDebugFlag = 20;
static OptionNameAndVar options[] = {
{"dDebugFlag", &dDebugFlag},
{"vDebugFlag", &vDebugFlag},
{"mDebugFlag", &mDebugFlag},
{"wDebugFlag", &wDebugFlag},
{"sDebugFlag", &sDebugFlag},
{"tsdbDebugFlag", &tsdbDebugFlag},
{"tqDebugFlag", &tqDebugFlag},
{"fsDebugFlag", &fsDebugFlag},
{"udfDebugFlag", &udfDebugFlag},
{"smaDebugFlag", &smaDebugFlag},
{"idxDebugFlag", &idxDebugFlag},
{"tdbDebugFlag", &tdbDebugFlag},
{"tmrDebugFlag", &tmrDebugFlag},
{"uDebugFlag", &uDebugFlag},
{"smaDebugFlag", &smaDebugFlag},
{"rpcDebugFlag", &rpcDebugFlag},
{"qDebugFlag", &qDebugFlag},
{"metaDebugFlag", &metaDebugFlag},
{"jniDebugFlag", &jniDebugFlag},
{"stDebugFlag", &stDebugFlag},
{ // 'bool/int32_t/int64_t/float/double' variables with general modification function
static OptionNameAndVar debugOptions[] = {
{"dDebugFlag", &dDebugFlag}, {"vDebugFlag", &vDebugFlag}, {"mDebugFlag", &mDebugFlag},
{"wDebugFlag", &wDebugFlag}, {"sDebugFlag", &sDebugFlag}, {"tsdbDebugFlag", &tsdbDebugFlag},
{"tqDebugFlag", &tqDebugFlag}, {"fsDebugFlag", &fsDebugFlag}, {"udfDebugFlag", &udfDebugFlag},
{"smaDebugFlag", &smaDebugFlag}, {"idxDebugFlag", &idxDebugFlag}, {"tdbDebugFlag", &tdbDebugFlag},
{"tmrDebugFlag", &tmrDebugFlag}, {"uDebugFlag", &uDebugFlag}, {"smaDebugFlag", &smaDebugFlag},
{"rpcDebugFlag", &rpcDebugFlag}, {"qDebugFlag", &qDebugFlag}, {"metaDebugFlag", &metaDebugFlag},
{"jniDebugFlag", &jniDebugFlag}, {"stDebugFlag", &stDebugFlag},
};
static OptionNameAndVar options[] = {
{"audit", &tsEnableAudit},
{"asynclog", &tsAsyncLog},
{"disableStream", &tsDisableStream},
@ -1701,54 +1460,216 @@ void taosCfgDynamicOptions(const char *option, const char *value) {
{"supportVnodes", &tsNumOfSupportVnodes},
};
int32_t optionSize = tListLen(options);
for (int32_t d = 0; d < optionSize; ++d) {
const char *optName = options[d].optionName;
int32_t optLen = strlen(optName);
if (strncasecmp(option, optName, optLen) != 0) continue;
if (taosCfgSetOption(debugOptions, tListLen(debugOptions), pItem, true) != 0) {
taosCfgSetOption(options, tListLen(options), pItem, false);
}
}
SConfig *pCfg = taosGetCfg();
SConfigItem *pItem = NULL;
return terrno == TSDB_CODE_SUCCESS ? 0 : -1;
}
pItem = cfgGetItem(pCfg, optName);
if (!pItem || (pItem->dynScope & CFG_DYN_SERVER) == 0) {
uError("failed to config:%s, not support", optName);
static int32_t taosCfgDynamicOptionsForClient(SConfig *pCfg, char *name) {
terrno = TSDB_CODE_SUCCESS;
SConfigItem *pItem = cfgGetItem(pCfg, name);
if (!pItem || (pItem->dynScope & CFG_DYN_CLIENT) == 0) {
uError("failed to config:%s, not support", name);
terrno = TSDB_CODE_INVALID_CFG;
return -1;
}
int32_t len = strlen(name);
char lowcaseName[CFG_NAME_MAX_LEN + 1] = {0};
strntolower(lowcaseName, name, TMIN(CFG_NAME_MAX_LEN, len));
switch (lowcaseName[0]) {
case 'd': {
if (strcasecmp("debugFlag", name) == 0) {
int32_t flag = pItem->i32;
taosSetAllDebugFlag(flag, true);
}
break;
}
switch (pItem->dtype) {
case CFG_DTYPE_BOOL: {
int32_t flag = atoi(value);
bool *pVar = options[d].optionVar;
uInfo("%s set from %d to %d", optName, *pVar, flag);
*pVar = flag;
} break;
case CFG_DTYPE_INT32: {
int32_t flag = atoi(value);
int32_t *pVar = options[d].optionVar;
uInfo("%s set from %d to %d", optName, *pVar, flag);
*pVar = flag;
if (d < nDebugFlag) {
// debug flags
taosSetDebugFlag(options[d].optionVar, optName, flag, true);
case 'e': {
if (strcasecmp("enableCoreFile", name) == 0) {
bool enableCore = pItem->bval;
taosSetCoreDump(enableCore);
uInfo("%s set to %d", name, enableCore);
}
break;
}
case 'f': {
if (strcasecmp("fqdn", name) == 0) {
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_FQDN_LEN);
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
char defaultFirstEp[TSDB_EP_LEN] = {0};
snprintf(defaultFirstEp, TSDB_EP_LEN, "%s:%u", tsLocalFqdn, tsServerPort);
SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp");
SEp firstEp = {0};
taosGetFqdnPortFromEp(strlen(pFirstEpItem->str) == 0 ? defaultFirstEp : pFirstEpItem->str, &firstEp);
snprintf(tsFirst, sizeof(tsFirst), "%s:%u", firstEp.fqdn, firstEp.port);
cfgSetItem(pCfg, "firstEp", tsFirst, pFirstEpItem->stype);
uInfo("localEp set to '%s', tsFirst set to '%s'", tsLocalEp, tsFirst);
} else if (strcasecmp("firstEp", name) == 0) {
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_FQDN_LEN);
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
char defaultFirstEp[TSDB_EP_LEN] = {0};
snprintf(defaultFirstEp, TSDB_EP_LEN, "%s:%u", tsLocalFqdn, tsServerPort);
SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp");
SEp firstEp = {0};
taosGetFqdnPortFromEp(strlen(pFirstEpItem->str) == 0 ? defaultFirstEp : pFirstEpItem->str, &firstEp);
snprintf(tsFirst, sizeof(tsFirst), "%s:%u", firstEp.fqdn, firstEp.port);
cfgSetItem(pCfg, "firstEp", tsFirst, pFirstEpItem->stype);
uInfo("localEp set to '%s', tsFirst set to '%s'", tsLocalEp, tsFirst);
}
break;
}
case 'l': {
if (strcasecmp("locale", name) == 0) {
const char *locale = cfgGetItem(pCfg, "locale")->str;
const char *charset = cfgGetItem(pCfg, "charset")->str;
taosSetSystemLocale(locale, charset);
osSetSystemLocale(locale, charset);
uInfo("locale set to '%s', charset set to '%s'", locale, charset);
} else if (strcasecmp("logDir", name) == 0) {
uInfo("%s set from '%s' to '%s'", name, tsLogDir, pItem->str);
tstrncpy(tsLogDir, pItem->str, PATH_MAX);
taosExpandDir(tsLogDir, tsLogDir, PATH_MAX);
}
break;
}
case 'm': {
if (strcasecmp("metaCacheMaxSize", name) == 0) {
atomic_store_32(&tsMetaCacheMaxSize, pItem->i32);
uInfo("%s set to %d", name, atomic_load_32(&tsMetaCacheMaxSize));
} else if (strcasecmp("minimalTmpDirGB", name) == 0) {
tsTempSpace.reserved = (int64_t)(((double)pItem->fval) * 1024 * 1024 * 1024);
uInfo("%s set to %"PRId64, name, tsTempSpace.reserved);
} else if (strcasecmp("minimalDataDirGB", name) == 0) {
tsDataSpace.reserved = (int64_t)(((double)pItem->fval) * 1024 * 1024 * 1024);
uInfo("%s set to %"PRId64, name, tsDataSpace.reserved);
} else if (strcasecmp("minimalLogDirGB", name) == 0) {
tsLogSpace.reserved = (int64_t)(((double)pItem->fval) * 1024 * 1024 * 1024);
uInfo("%s set to %"PRId64, name, tsLogSpace.reserved);
}
break;
}
case 's': {
if (strcasecmp("secondEp", name) == 0) {
SEp secondEp = {0};
taosGetFqdnPortFromEp(strlen(pItem->str) == 0 ? tsFirst : pItem->str, &secondEp);
snprintf(tsSecond, sizeof(tsSecond), "%s:%u", secondEp.fqdn, secondEp.port);
cfgSetItem(pCfg, "secondEp", tsSecond, pItem->stype);
uInfo("%s set to %s", name, tsSecond);
} else if (strcasecmp("smlChildTableName", name) == 0) {
uInfo("%s set from %s to %s", name, tsSmlChildTableName, pItem->str);
tstrncpy(tsSmlChildTableName, pItem->str, TSDB_TABLE_NAME_LEN);
} else if (strcasecmp("smlAutoChildTableNameDelimiter", name) == 0) {
uInfo("%s set from %s to %s", name, tsSmlAutoChildTableNameDelimiter, pItem->str);
tstrncpy(tsSmlAutoChildTableNameDelimiter, pItem->str, TSDB_TABLE_NAME_LEN);
} else if (strcasecmp("smlTagName", name) == 0) {
uInfo("%s set from %s to %s", name, tsSmlTagName, pItem->str);
tstrncpy(tsSmlTagName, pItem->str, TSDB_COL_NAME_LEN);
} else if (strcasecmp("smlTsDefaultName", name) == 0) {
uInfo("%s set from %s to %s", name, tsSmlTsDefaultName, pItem->str);
tstrncpy(tsSmlTsDefaultName, pItem->str, TSDB_COL_NAME_LEN);
} else if (strcasecmp("serverPort", name) == 0) {
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_FQDN_LEN);
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
char defaultFirstEp[TSDB_EP_LEN] = {0};
snprintf(defaultFirstEp, TSDB_EP_LEN, "%s:%u", tsLocalFqdn, tsServerPort);
SConfigItem *pFirstEpItem = cfgGetItem(pCfg, "firstEp");
SEp firstEp = {0};
taosGetFqdnPortFromEp(strlen(pFirstEpItem->str) == 0 ? defaultFirstEp : pFirstEpItem->str, &firstEp);
snprintf(tsFirst, sizeof(tsFirst), "%s:%u", firstEp.fqdn, firstEp.port);
cfgSetItem(pCfg, "firstEp", tsFirst, pFirstEpItem->stype);
uInfo("localEp set to '%s', tsFirst set to '%s'", tsLocalEp, tsFirst);
} else if (strcasecmp("slowLogScope", name) == 0) {
if (taosSetSlowLogScope(pItem->str)) {
return -1;
}
uInfo("%s set to %s", name, pItem->str);
}
break;
}
case 't': {
if (strcasecmp("timezone", name) == 0) {
osSetTimezone(pItem->str);
uInfo("%s set from %s to %s", name, tsTimezoneStr, pItem->str);
cfgSetItem(pCfg, "timezone", tsTimezoneStr, pItem->stype);
} else if (strcasecmp("tempDir", name) == 0) {
uInfo("%s set from %s to %s", name, tsTempDir, pItem->str);
tstrncpy(tsTempDir, pItem->str, PATH_MAX);
taosExpandDir(tsTempDir, tsTempDir, PATH_MAX);
if (taosMulMkDir(tsTempDir) != 0) {
uError("failed to create tempDir:%s since %s", tsTempDir, terrstr());
return -1;
}
} else if (strcasecmp("telemetryServer", name) == 0) {
uInfo("%s set from %s to %s", name, pItem->str, tsTelemServer);
tstrncpy(tsTelemServer, pItem->str, TSDB_FQDN_LEN);
}
break;
}
} break;
case CFG_DTYPE_INT64: {
int64_t flag = atoll(value);
int64_t *pVar = options[d].optionVar;
uInfo("%s set from %" PRId64 " to %" PRId64, optName, *pVar, flag);
*pVar = flag;
} break;
default:
terrno = TSDB_CODE_CFG_NOT_FOUND;
break;
}
return;
{ // 'bool/int32_t/int64_t/float/double' variables with general modification function
static OptionNameAndVar debugOptions[] = {
{"cDebugFlag", &cDebugFlag}, {"dDebugFlag", &dDebugFlag}, {"fsDebugFlag", &fsDebugFlag},
{"idxDebugFlag", &idxDebugFlag}, {"jniDebugFlag", &jniDebugFlag}, {"qDebugFlag", &qDebugFlag},
{"rpcDebugFlag", &rpcDebugFlag}, {"smaDebugFlag", &smaDebugFlag}, {"tmrDebugFlag", &tmrDebugFlag},
{"uDebugFlag", &uDebugFlag},
};
static OptionNameAndVar options[] = {
{"asyncLog", &tsAsyncLog},
{"assert", &tsAssert},
{"compressMsgSize", &tsCompressMsgSize},
{"countAlwaysReturnValue", &tsCountAlwaysReturnValue},
{"crashReporting", &tsEnableCrashReport},
{"enableCoreFile", &tsAsyncLog},
{"enableQueryHb", &tsEnableQueryHb},
{"keepColumnName", &tsKeepColumnName},
{"keepAliveIdle", &tsKeepAliveIdle},
{"logKeepDays", &tsLogKeepDays},
{"maxInsertBatchRows", &tsMaxInsertBatchRows},
{"maxRetryWaitTime", &tsMaxRetryWaitTime},
{"minSlidingTime", &tsMinSlidingTime},
{"minIntervalTime", &tsMinIntervalTime},
{"numOfLogLines", &tsNumOfLogLines},
{"querySmaOptimize", &tsQuerySmaOptimize},
{"queryPolicy", &tsQueryPolicy},
{"queryPlannerTrace", &tsQueryPlannerTrace},
{"queryNodeChunkSize", &tsQueryNodeChunkSize},
{"queryUseNodeAllocator", &tsQueryUseNodeAllocator},
{"smlDot2Underline", &tsSmlDot2Underline},
{"shellActivityTimer", &tsShellActivityTimer},
{"slowLogThreshold", &tsSlowLogThreshold},
{"useAdapter", &tsUseAdapter},
};
if (taosCfgSetOption(debugOptions, tListLen(debugOptions), pItem, true) != 0) {
taosCfgSetOption(options, tListLen(options), pItem, false);
}
}
uError("failed to cfg dynamic option:%s value:%s", option, value);
return terrno == TSDB_CODE_SUCCESS ? 0 : -1;
}
int32_t taosCfgDynamicOptions(SConfig *pCfg, char *name, bool forServer) {
if (forServer) return taosCfgDynamicOptionsForServer(pCfg, name);
return taosCfgDynamicOptionsForClient(pCfg, name);
}
void taosSetDebugFlag(int32_t *pFlagPtr, const char *flagName, int32_t flagVal, bool rewrite) {

5
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
View File

@ -218,7 +218,10 @@ int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
}
dInfo("start to config, option:%s, value:%s", cfgReq.config, cfgReq.value);
taosCfgDynamicOptions(cfgReq.config, cfgReq.value);
SConfig *pCfg = taosGetCfg();
cfgSetItem(pCfg, cfgReq.config, cfgReq.value, CFG_STYPE_ALTER_CMD);
taosCfgDynamicOptions(pCfg, cfgReq.config, true);
return 0;
}

7
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -27,6 +27,7 @@
#include "mndUser.h"
#include "mndVgroup.h"
#include "tmisce.h"
#include "tunit.h"
#define TSDB_DNODE_VER_NUMBER 2
#define TSDB_DNODE_RESERVE_SIZE 64
@ -1316,8 +1317,8 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
int32_t code = mndMCfgGetValInt32(&cfgReq, optLen, &flag);
if (code < 0) return code;
if (flag > 1024 * 1024 || (flag > -1 && flag < 4) || flag < -1) {
mError("dnode:%d, failed to config s3blocksize since value:%d. Valid range: -1 or [4, 1024 * 1024]",
if (flag > 1024 * 1024 || (flag > -1 && flag < 1024) || flag < -1) {
mError("dnode:%d, failed to config s3blocksize since value:%d. Valid range: -1 or [1024, 1024 * 1024]",
cfgReq.dnodeId, flag);
terrno = TSDB_CODE_INVALID_CFG;
tFreeSMCfgDnodeReq(&cfgReq);
@ -1328,7 +1329,7 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
snprintf(dcfgReq.value, TSDB_DNODE_VALUE_LEN, "%d", flag);
#endif
} else {
mndMCfg2DCfg(&cfgReq, &dcfgReq);
if (mndMCfg2DCfg(&cfgReq, &dcfgReq)) goto _err_out;
if (strlen(dcfgReq.config) > TSDB_DNODE_CONFIG_LEN) {
mError("dnode:%d, failed to config since config is too long", cfgReq.dnodeId);
terrno = TSDB_CODE_INVALID_CFG;

2
source/dnode/mnode/impl/src/mndProfile.c
View File

@ -313,7 +313,7 @@ _CONNECT:
code = 0;
char detail[1000] = {0};
sprintf(detail, "%s:%d, app:%s", ip, pConn->port, connReq.app);
sprintf(detail, "app:%s", connReq.app);
auditRecord(pReq, pMnode->clusterId, "login", "", "", detail, strlen(detail));

6
source/dnode/mnode/impl/src/mndUser.c
View File

@ -1561,7 +1561,11 @@ static int32_t mndProcessCreateUserReq(SRpcMsg *pReq) {
code = mndCreateUser(pMnode, pOperUser->acct, &createReq, pReq);
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
auditRecord(pReq, pMnode->clusterId, "createUser", "", createReq.user, createReq.sql, createReq.sqlLen);
char detail[1000] = {0};
sprintf(detail, "enable:%d, superUser:%d, sysInfo:%d, password:xxx",
createReq.enable, createReq.superUser, createReq.sysInfo);
auditRecord(pReq, pMnode->clusterId, "createUser", "", createReq.user, detail, strlen(detail));
_OVER:
if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {

2
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -2176,7 +2176,7 @@ static int32_t mndProcessRedistributeVgroupMsg(SRpcMsg *pReq) {
code = mndRedistributeVgroup(pMnode, pReq, pDb, pVgroup, pNew1, pOld1, pNew2, pOld2, pNew3, pOld3);
} else {
terrno = TSDB_CODE_MND_INVALID_REPLICA;
terrno = TSDB_CODE_MND_REQ_REJECTED;
goto _OVER;
}

2
source/dnode/vnode/src/tq/tq.c
View File

@ -1974,6 +1974,8 @@ int32_t tqProcessTaskUpdateReq(STQ* pTq, SRpcMsg* pMsg) {
return -1;
}
streamMetaInitBackend(pMeta);
if (streamMetaLoadAllTasks(pTq->pStreamMeta) < 0) {
tqError("vgId:%d failed to load stream tasks", vgId);
streamMetaWUnLock(pMeta);

5
source/dnode/vnode/src/tq/tqStreamStateSnap.c
View File

@ -169,10 +169,15 @@ int32_t streamStateSnapWriterClose(SStreamStateWriter* pWriter, int8_t rollback)
}
int32_t streamStateRebuildFromSnap(SStreamStateWriter* pWriter, int64_t chkpId) {
tqDebug("vgId:%d, vnode %s start to rebuild stream-state", TD_VID(pWriter->pTq->pVnode), STREAM_STATE_TRANSFER);
streamMetaWLock(pWriter->pTq->pStreamMeta);
int32_t code = streamMetaReopen(pWriter->pTq->pStreamMeta);
if (code == 0) {
streamMetaInitBackend(pWriter->pTq->pStreamMeta);
code = streamStateLoadTasks(pWriter);
}
streamMetaWUnLock(pWriter->pTq->pStreamMeta);
tqDebug("vgId:%d, vnode %s succ to rebuild stream-state", TD_VID(pWriter->pTq->pVnode), STREAM_STATE_TRANSFER);
taosMemoryFree(pWriter);
return code;

2
source/dnode/vnode/src/tq/tqStreamTask.c
View File

@ -144,7 +144,6 @@ int32_t tqRestartStreamTasks(STQ* pTq) {
}
streamMetaWLock(pMeta);
code = streamMetaReopen(pMeta);
if (code != TSDB_CODE_SUCCESS) {
tqError("vgId:%d failed to reopen stream meta", vgId);
@ -153,6 +152,7 @@ int32_t tqRestartStreamTasks(STQ* pTq) {
return code;
}
streamMetaInitBackend(pMeta);
int64_t el = taosGetTimestampMs() - st;
tqInfo("vgId:%d close&reload state elapsed time:%.3fms", vgId, el/1000.);

23
source/dnode/vnode/src/tsdb/tsdbRetention.c
View File

@ -74,6 +74,8 @@ static int32_t tsdbDoCopyFile(SRTNer *rtner, const STFileObj *from, const STFile
if (fdFrom == NULL) code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
tsdbInfo("vgId: %d, open tofile: %s size: %" PRId64, TD_VID(rtner->tsdb->pVnode), fname, from->f->size);
fdTo = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
if (fdTo == NULL) code = terrno;
TSDB_CHECK_CODE(code, lino, _exit);
@ -333,6 +335,7 @@ static int32_t tsdbDoRetentionOnFileSet(SRTNer *rtner, STFileSet *fset) {
int32_t mtime = 0;
taosStatFile(fobj->fname, NULL, &mtime, NULL);
if (mtime < rtner->now - tsS3UploadDelaySec) {
tsdbInfo("file:%s size: %" PRId64 " do migrate s3", fobj->fname, fobj->f->size);
code = tsdbMigrateDataFileS3(rtner, fobj, &did);
TSDB_CHECK_CODE(code, lino, _exit);
}
@ -356,6 +359,8 @@ static int32_t tsdbDoRetentionOnFileSet(SRTNer *rtner, STFileSet *fset) {
s3EvictCache(fobj->fname, fsize * 2);
}
*/
tsdbInfo("file:%s size: %" PRId64 " do migrate", fobj->fname, fobj->f->size);
code = tsdbDoMigrateFileObj(rtner, fobj, &did);
TSDB_CHECK_CODE(code, lino, _exit);
//}
@ -375,14 +380,6 @@ static int32_t tsdbDoRetentionOnFileSet(SRTNer *rtner, STFileSet *fset) {
_exit:
if (code) {
if (TARRAY2_DATA(rtner->fopArr)) {
TARRAY2_DESTROY(rtner->fopArr, NULL);
}
TFileSetArray **fsetArr = &rtner->fsetArr;
if (fsetArr[0]) {
tsdbFSDestroyCopySnapshot(&rtner->fsetArr);
}
TSDB_ERROR_LOG(TD_VID(rtner->tsdb->pVnode), lino, code);
}
return code;
@ -437,13 +434,19 @@ static int32_t tsdbDoRetentionAsync(void *arg) {
_exit:
if (code) {
if (TARRAY2_DATA(rtner->fopArr)) {
TARRAY2_DESTROY(rtner->fopArr, NULL);
}
TFileSetArray **fsetArr = &rtner->fsetArr;
if (fsetArr[0]) {
tsdbFSDestroyCopySnapshot(&rtner->fsetArr);
}
TSDB_ERROR_LOG(TD_VID(rtner->tsdb->pVnode), lino, code);
}
return code;
}
int32_t tsdbRetention(STsdb *tsdb, int64_t now, int32_t sync) {
int32_t code = 0;

24
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -25,9 +25,11 @@
static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessDropStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc);
static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc);
static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessAlterConfirmReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp);
@ -509,13 +511,13 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg
if (vnodeProcessDropStbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;
break;
case TDMT_VND_CREATE_TABLE:
if (vnodeProcessCreateTbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;
if (vnodeProcessCreateTbReq(pVnode, ver, pReq, len, pRsp, pMsg) < 0) goto _err;
break;
case TDMT_VND_ALTER_TABLE:
if (vnodeProcessAlterTbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;
break;
case TDMT_VND_DROP_TABLE:
if (vnodeProcessDropTbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;
if (vnodeProcessDropTbReq(pVnode, ver, pReq, len, pRsp, pMsg) < 0) goto _err;
break;
case TDMT_VND_DROP_TTL_TABLE:
if (vnodeProcessDropTtlTbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;
@ -878,7 +880,8 @@ _err:
return -1;
}
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc) {
SDecoder decoder = {0};
SEncoder encoder = {0};
int32_t rcode = 0;
@ -928,7 +931,7 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq,
goto _exit;
}
if(tsEnableAuditCreateTable){
if(tsEnableAudit && tsEnableAuditCreateTable){
char* str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
if (str == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -983,7 +986,7 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq,
tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
if (tsEnableAuditCreateTable) {
if(tsEnableAudit && tsEnableAuditCreateTable){
int64_t clusterId = pVnode->config.syncCfg.nodeInfo[0].clusterId;
SName name = {0};
@ -1002,7 +1005,7 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq,
size_t len = 0;
char *keyJoined = taosStringBuilderGetResult(&sb, &len);
auditRecord(NULL, clusterId, "createTable", name.dbname, "", keyJoined, len);
auditRecord(pOriginRpc, clusterId, "createTable", name.dbname, "", keyJoined, len);
taosStringBuilderDestroy(&sb);
}
@ -1144,7 +1147,8 @@ _exit:
return 0;
}
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,
SRpcMsg *pOriginRpc) {
SVDropTbBatchReq req = {0};
SVDropTbBatchRsp rsp = {0};
SDecoder decoder = {0};
@ -1223,7 +1227,7 @@ static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, in
size_t len = 0;
char *keyJoined = taosStringBuilderGetResult(&sb, &len);
auditRecord(NULL, clusterId, "dropTable", name.dbname, "", keyJoined, len);
auditRecord(pOriginRpc, clusterId, "dropTable", name.dbname, "", keyJoined, len);
taosStringBuilderDestroy(&sb);
}

2
source/libs/command/src/command.c
View File

@ -842,7 +842,7 @@ static int32_t execAlterLocal(SAlterLocalStmt* pStmt) {
return terrno;
}
if (taosApplyLocalCfg(tsCfg, pStmt->config)) {
if (taosCfgDynamicOptions(tsCfg, pStmt->config, false)) {
return terrno;
}

8
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -279,7 +279,12 @@ static int32_t getAllIntervalWindow(SSHashObj* pHashMap, SSHashObj* resWins) {
SWinKey* pKey = tSimpleHashGetKey(pIte, NULL);
uint64_t groupId = pKey->groupId;
TSKEY ts = pKey->ts;
int32_t code = saveWinResultInfo(ts, groupId, *(SRowBuffPos**)pIte, resWins);
SRowBuffPos* pPos = *(SRowBuffPos**)pIte;
if (!pPos->beUpdated) {
continue;
}
pPos->beUpdated = false;
int32_t code = saveWinResultInfo(ts, groupId, pPos, resWins);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
@ -866,6 +871,7 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDat
}
if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
pResPos->beUpdated = true;
tSimpleHashPut(pInfo->aggSup.pResultRowHashTable, &key, sizeof(SWinKey), &pResPos, POINTER_BYTES);
}

27
source/libs/stream/src/streamMeta.c
View File

@ -262,18 +262,33 @@ int32_t streamMetaReopen(SStreamMeta* pMeta) {
}
}
// todo: not wait in a critical region
while ((pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId, pMeta->vgId)) == NULL) {
taosMemoryFree(defaultPath);
taosMemoryFree(newPath);
return 0;
}
// todo refactor: the lock shoud be restricted in one function
void streamMetaInitBackend(SStreamMeta* pMeta) {
pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId, pMeta->vgId);
if (pMeta->streamBackend == NULL) {
streamMetaWUnLock(pMeta);
while (1) {
streamMetaWLock(pMeta);
pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId, pMeta->vgId);
if (pMeta->streamBackend != NULL) {
break;
}
streamMetaWUnLock(pMeta);
stInfo("vgId:%d failed to init stream backend, retry in 100ms", pMeta->vgId);
taosMsleep(100);
}
}
pMeta->streamBackendRid = taosAddRef(streamBackendId, pMeta->streamBackend);
streamBackendLoadCheckpointInfo(pMeta);
taosMemoryFree(defaultPath);
taosMemoryFree(newPath);
return 0;
}
void streamMetaClear(SStreamMeta* pMeta) {

4
source/libs/stream/src/streamSnapshot.c
View File

@ -195,7 +195,7 @@ int32_t streamSnapHandleInit(SStreamSnapHandle* pHandle, char* path, int64_t chk
}
}
if (qDebugFlag & DEBUG_TRACE) {
char* buf = taosMemoryCalloc(1, 128 + taosArrayGetSize(pFile->pSst) * 16);
char* buf = taosMemoryCalloc(1, 128 + taosArrayGetSize(pFile->pSst) * 64);
sprintf(buf, "[current: %s,", pFile->pCurrent);
sprintf(buf + strlen(buf), "MANIFEST: %s,", pFile->pMainfest);
sprintf(buf + strlen(buf), "options: %s,", pFile->pOptions);
@ -483,7 +483,7 @@ int32_t streamSnapWrite(SStreamSnapWriter* pWriter, uint8_t* pData, uint32_t nDa
int32_t streamSnapWriterClose(SStreamSnapWriter* pWriter, int8_t rollback) {
SStreamSnapHandle* handle = &pWriter->handle;
if (qDebugFlag & DEBUG_TRACE) {
char* buf = (char*)taosMemoryMalloc(128 + taosArrayGetSize(handle->pFileList) * 16);
char* buf = (char*)taosMemoryMalloc(128 + taosArrayGetSize(handle->pFileList) * 64);
int n = sprintf(buf, "[");
for (int i = 0; i < taosArrayGetSize(handle->pFileList); i++) {
SBackendFileItem* item = taosArrayGet(handle->pFileList, i);

1
source/libs/stream/src/tstreamFileState.c
View File

@ -428,6 +428,7 @@ SRowBuffPos* getNewRowPosForWrite(SStreamFileState* pFileState) {
newPos->beUsed = true;
newPos->beFlushed = false;
newPos->needFree = false;
newPos->beUpdated = true;
return newPos;
}

307
source/util/src/tcompression.c
View File

@ -52,6 +52,7 @@
#include "lz4.h"
#include "tRealloc.h"
#include "tlog.h"
#include "ttypes.h"
#ifdef TD_TSZ
#include "td_sz.h"
@ -62,8 +63,6 @@ static const int32_t TEST_NUMBER = 1;
#define SIMPLE8B_MAX_INT64 ((uint64_t)1152921504606846974LL)
#define safeInt64Add(a, b) (((a >= 0) && (b <= INT64_MAX - a)) || ((a < 0) && (b >= INT64_MIN - a)))
#define ZIGZAG_ENCODE(T, v) (((u##T)((v) >> (sizeof(T) * 8 - 1))) ^ (((u##T)(v)) << 1)) // zigzag encode
#define ZIGZAG_DECODE(T, v) (((v) >> 1) ^ -((T)((v)&1))) // zigzag decode
#ifdef TD_TSZ
bool lossyFloat = false;
@ -99,24 +98,7 @@ int32_t tsCompressINTImp(const char *const input, const int32_t nelements, char
15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15, 15};
// get the byte limit.
int32_t word_length = 0;
switch (type) {
case TSDB_DATA_TYPE_BIGINT:
word_length = LONG_BYTES;
break;
case TSDB_DATA_TYPE_INT:
word_length = INT_BYTES;
break;
case TSDB_DATA_TYPE_SMALLINT:
word_length = SHORT_BYTES;
break;
case TSDB_DATA_TYPE_TINYINT:
word_length = CHAR_BYTES;
break;
default:
uError("Invalid compress integer type:%d", type);
return -1;
}
int32_t word_length = getWordLength(type);
int32_t byte_limit = nelements * word_length + 1;
int32_t opos = 1;
@ -221,24 +203,9 @@ int32_t tsCompressINTImp(const char *const input, const int32_t nelements, char
}
int32_t tsDecompressINTImp(const char *const input, const int32_t nelements, char *const output, const char type) {
int32_t word_length = 0;
switch (type) {
case TSDB_DATA_TYPE_BIGINT:
word_length = LONG_BYTES;
break;
case TSDB_DATA_TYPE_INT:
word_length = INT_BYTES;
break;
case TSDB_DATA_TYPE_SMALLINT:
word_length = SHORT_BYTES;
break;
case TSDB_DATA_TYPE_TINYINT:
word_length = CHAR_BYTES;
break;
default:
uError("Invalid decompress integer type:%d", type);
return -1;
int32_t word_length = getWordLength(type);
if (word_length == -1) {
return word_length;
}
// If not compressed.
@ -247,8 +214,11 @@ int32_t tsDecompressINTImp(const char *const input, const int32_t nelements, cha
return nelements * word_length;
}
// Selector value: 0 1 2 3 4 5 6 7 8 9 10 11
// 12 13 14 15
#if __AVX2__
tsDecompressIntImpl_Hw(input, nelements, output, type);
return nelements * word_length;
#else
// Selector value: 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
char bit_per_integer[] = {0, 0, 1, 2, 3, 4, 5, 6, 7, 8, 10, 12, 15, 20, 30, 60};
int32_t selector_to_elems[] = {240, 120, 60, 30, 20, 15, 12, 10, 8, 7, 6, 5, 4, 3, 2, 1};
@ -257,185 +227,6 @@ int32_t tsDecompressINTImp(const char *const input, const int32_t nelements, cha
int32_t _pos = 0;
int64_t prev_value = 0;
#if __AVX2__
while (1) {
if (_pos == nelements) break;
uint64_t w = 0;
memcpy(&w, ip, LONG_BYTES);
char selector = (char)(w & INT64MASK(4)); // selector = 4
char bit = bit_per_integer[(int32_t)selector]; // bit = 3
int32_t elems = selector_to_elems[(int32_t)selector];
// Optimize the performance, by remove the constantly switch operation.
int32_t v = 4;
uint64_t zigzag_value = 0;
uint64_t mask = INT64MASK(bit);
switch (type) {
case TSDB_DATA_TYPE_BIGINT: {
int64_t* p = (int64_t*) output;
int32_t gRemainder = (nelements - _pos);
int32_t num = (gRemainder > elems)? elems:gRemainder;
int32_t batch = num >> 2;
int32_t remain = num & 0x03;
if (selector == 0 || selector == 1) {
if (tsAVX2Enable && tsSIMDEnable) {
for (int32_t i = 0; i < batch; ++i) {
__m256i prev = _mm256_set1_epi64x(prev_value);
_mm256_storeu_si256((__m256i *)&p[_pos], prev);
_pos += 4;
}
for (int32_t i = 0; i < remain; ++i) {
p[_pos++] = prev_value;
}
} else {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
p[_pos++] = prev_value;
v += bit;
}
}
} else {
if (tsAVX2Enable && tsSIMDEnable) {
__m256i base = _mm256_set1_epi64x(w);
__m256i maskVal = _mm256_set1_epi64x(mask);
__m256i shiftBits = _mm256_set_epi64x(bit * 3 + 4, bit * 2 + 4, bit + 4, 4);
__m256i inc = _mm256_set1_epi64x(bit << 2);
for (int32_t i = 0; i < batch; ++i) {
__m256i after = _mm256_srlv_epi64(base, shiftBits);
__m256i zigzagVal = _mm256_and_si256(after, maskVal);
// ZIGZAG_DECODE(T, v) (((v) >> 1) ^ -((T)((v)&1)))
__m256i signmask = _mm256_and_si256(_mm256_set1_epi64x(1), zigzagVal);
signmask = _mm256_sub_epi64(_mm256_setzero_si256(), signmask);
// get the four zigzag values here
__m256i delta = _mm256_xor_si256(_mm256_srli_epi64(zigzagVal, 1), signmask);
// calculate the cumulative sum (prefix sum) for each number
// decode[0] = prev_value + final[0]
// decode[1] = decode[0] + final[1] -----> prev_value + final[0] + final[1]
// decode[2] = decode[1] + final[2] -----> prev_value + final[0] + final[1] + final[2]
// decode[3] = decode[2] + final[3] -----> prev_value + final[0] + final[1] + final[2] + final[3]
// 1, 2, 3, 4
//+ 0, 1, 0, 3
// 1, 3, 3, 7
// shift and add for the first round
__m128i prev = _mm_set1_epi64x(prev_value);
__m256i x = _mm256_slli_si256(delta, 8);
delta = _mm256_add_epi64(delta, x);
_mm256_storeu_si256((__m256i *)&p[_pos], delta);
// 1, 3, 3, 7
//+ 0, 0, 3, 3
// 1, 3, 6, 10
// shift and add operation for the second round
__m128i firstPart = _mm_loadu_si128((__m128i *)&p[_pos]);
__m128i secondItem = _mm_set1_epi64x(p[_pos + 1]);
__m128i secPart = _mm_add_epi64(_mm_loadu_si128((__m128i *)&p[_pos + 2]), secondItem);
firstPart = _mm_add_epi64(firstPart, prev);
secPart = _mm_add_epi64(secPart, prev);
// save it in the memory
_mm_storeu_si128((__m128i *)&p[_pos], firstPart);
_mm_storeu_si128((__m128i *)&p[_pos + 2], secPart);
shiftBits = _mm256_add_epi64(shiftBits, inc);
prev_value = p[_pos + 3];
// uDebug("_pos:%d %"PRId64", %"PRId64", %"PRId64", %"PRId64, _pos, p[_pos], p[_pos+1], p[_pos+2], p[_pos+3]);
_pos += 4;
}
// handle the remain value
for (int32_t i = 0; i < remain; i++) {
zigzag_value = ((w >> (v + (batch * bit * 4))) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = prev_value;
// uDebug("_pos:%d %"PRId64, _pos-1, p[_pos-1]);
v += bit;
}
} else {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
zigzag_value = ((w >> v) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = prev_value;
// uDebug("_pos:%d %"PRId64, _pos-1, p[_pos-1]);
v += bit;
}
}
}
} break;
case TSDB_DATA_TYPE_INT: {
int32_t* p = (int32_t*) output;
if (selector == 0 || selector == 1) {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
p[_pos++] = (int32_t)prev_value;
}
} else {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
zigzag_value = ((w >> v) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = (int32_t)prev_value;
v += bit;
}
}
} break;
case TSDB_DATA_TYPE_SMALLINT: {
int16_t* p = (int16_t*) output;
if (selector == 0 || selector == 1) {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
p[_pos++] = (int16_t)prev_value;
}
} else {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
zigzag_value = ((w >> v) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = (int16_t)prev_value;
v += bit;
}
}
} break;
case TSDB_DATA_TYPE_TINYINT: {
int8_t *p = (int8_t *)output;
if (selector == 0 || selector == 1) {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
p[_pos++] = (int8_t)prev_value;
}
} else {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
zigzag_value = ((w >> v) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = (int8_t)prev_value;
v += bit;
}
}
} break;
}
ip += LONG_BYTES;
}
return nelements * word_length;
#else
while (1) {
if (count == nelements) break;
@ -644,6 +435,8 @@ int32_t tsDecompressStringImp(const char *const input, int32_t compressedSize, c
// TODO: Take care here, we assumes little endian encoding.
int32_t tsCompressTimestampImp(const char *const input, const int32_t nelements, char *const output) {
int32_t _pos = 1;
int32_t longBytes = LONG_BYTES;
ASSERTS(nelements >= 0, "nelements is negative");
if (nelements == 0) return 0;
@ -684,25 +477,25 @@ int32_t tsCompressTimestampImp(const char *const input, const int32_t nelements,
}
flags = flag1 | (flag2 << 4);
// Encode the flag.
if ((_pos + CHAR_BYTES - 1) >= nelements * LONG_BYTES) goto _exit_over;
if ((_pos + CHAR_BYTES - 1) >= nelements * longBytes) goto _exit_over;
memcpy(output + _pos, &flags, CHAR_BYTES);
_pos += CHAR_BYTES;
/* Here, we assume it is little endian encoding method. */
// Encode dd1
if (is_bigendian()) {
if ((_pos + flag1 - 1) >= nelements * LONG_BYTES) goto _exit_over;
memcpy(output + _pos, (char *)(&dd1) + LONG_BYTES - flag1, flag1);
if ((_pos + flag1 - 1) >= nelements * longBytes) goto _exit_over;
memcpy(output + _pos, (char *)(&dd1) + longBytes - flag1, flag1);
} else {
if ((_pos + flag1 - 1) >= nelements * LONG_BYTES) goto _exit_over;
if ((_pos + flag1 - 1) >= nelements * longBytes) goto _exit_over;
memcpy(output + _pos, (char *)(&dd1), flag1);
}
_pos += flag1;
// Encode dd2;
if (is_bigendian()) {
if ((_pos + flag2 - 1) >= nelements * LONG_BYTES) goto _exit_over;
memcpy(output + _pos, (char *)(&dd2) + LONG_BYTES - flag2, flag2);
if ((_pos + flag2 - 1) >= nelements * longBytes) goto _exit_over;
memcpy(output + _pos, (char *)(&dd2) + longBytes - flag2, flag2);
} else {
if ((_pos + flag2 - 1) >= nelements * LONG_BYTES) goto _exit_over;
if ((_pos + flag2 - 1) >= nelements * longBytes) goto _exit_over;
memcpy(output + _pos, (char *)(&dd2), flag2);
}
_pos += flag2;
@ -715,15 +508,15 @@ int32_t tsCompressTimestampImp(const char *const input, const int32_t nelements,
flag2 = 0;
flags = flag1 | (flag2 << 4);
// Encode the flag.
if ((_pos + CHAR_BYTES - 1) >= nelements * LONG_BYTES) goto _exit_over;
if ((_pos + CHAR_BYTES - 1) >= nelements * longBytes) goto _exit_over;
memcpy(output + _pos, &flags, CHAR_BYTES);
_pos += CHAR_BYTES;
// Encode dd1;
if (is_bigendian()) {
if ((_pos + flag1 - 1) >= nelements * LONG_BYTES) goto _exit_over;
memcpy(output + _pos, (char *)(&dd1) + LONG_BYTES - flag1, flag1);
if ((_pos + flag1 - 1) >= nelements * longBytes) goto _exit_over;
memcpy(output + _pos, (char *)(&dd1) + longBytes - flag1, flag1);
} else {
if ((_pos + flag1 - 1) >= nelements * LONG_BYTES) goto _exit_over;
if ((_pos + flag1 - 1) >= nelements * longBytes) goto _exit_over;
memcpy(output + _pos, (char *)(&dd1), flag1);
}
_pos += flag1;
@ -734,17 +527,19 @@ int32_t tsCompressTimestampImp(const char *const input, const int32_t nelements,
_exit_over:
output[0] = 0; // Means the string is not compressed
memcpy(output + 1, input, nelements * LONG_BYTES);
return nelements * LONG_BYTES + 1;
memcpy(output + 1, input, nelements * longBytes);
return nelements * longBytes + 1;
}
int32_t tsDecompressTimestampImp(const char *const input, const int32_t nelements, char *const output) {
int64_t longBytes = LONG_BYTES;
ASSERTS(nelements >= 0, "nelements is negative");
if (nelements == 0) return 0;
if (input[0] == 0) {
memcpy(output, input + 1, nelements * LONG_BYTES);
return nelements * LONG_BYTES;
memcpy(output, input + 1, nelements * longBytes);
return nelements * longBytes;
} else if (input[0] == 1) { // Decompress
int64_t *ostream = (int64_t *)output;
@ -763,7 +558,7 @@ int32_t tsDecompressTimestampImp(const char *const input, const int32_t nelement
delta_of_delta = 0;
} else {
if (is_bigendian()) {
memcpy(((char *)(&dd1)) + LONG_BYTES - nbytes, input + ipos, nbytes);
memcpy(((char *)(&dd1)) + longBytes - nbytes, input + ipos, nbytes);
} else {
memcpy(&dd1, input + ipos, nbytes);
}
@ -779,7 +574,7 @@ int32_t tsDecompressTimestampImp(const char *const input, const int32_t nelement
prev_value = prev_value + prev_delta;
ostream[opos++] = prev_value;
}
if (opos == nelements) return nelements * LONG_BYTES;
if (opos == nelements) return nelements * longBytes;
// Decode dd2
uint64_t dd2 = 0;
@ -788,7 +583,7 @@ int32_t tsDecompressTimestampImp(const char *const input, const int32_t nelement
delta_of_delta = 0;
} else {
if (is_bigendian()) {
memcpy(((char *)(&dd2)) + LONG_BYTES - nbytes, input + ipos, nbytes);
memcpy(((char *)(&dd2)) + longBytes - nbytes, input + ipos, nbytes);
} else {
memcpy(&dd2, input + ipos, nbytes);
}
@ -799,7 +594,7 @@ int32_t tsDecompressTimestampImp(const char *const input, const int32_t nelement
prev_delta = delta_of_delta + prev_delta;
prev_value = prev_value + prev_delta;
ostream[opos++] = prev_value;
if (opos == nelements) return nelements * LONG_BYTES;
if (opos == nelements) return nelements * longBytes;
}
} else {
@ -807,11 +602,13 @@ int32_t tsDecompressTimestampImp(const char *const input, const int32_t nelement
return -1;
}
}
/* --------------------------------------------Double Compression
* ---------------------------------------------- */
/* --------------------------------------------Double Compression ---------------------------------------------- */
void encodeDoubleValue(uint64_t diff, uint8_t flag, char *const output, int32_t *const pos) {
int32_t longBytes = LONG_BYTES;
uint8_t nbytes = (flag & INT8MASK(3)) + 1;
int32_t nshift = (LONG_BYTES * BITS_PER_BYTE - nbytes * BITS_PER_BYTE) * (flag >> 3);
int32_t nshift = (longBytes * BITS_PER_BYTE - nbytes * BITS_PER_BYTE) * (flag >> 3);
diff >>= nshift;
while (nbytes) {
@ -906,12 +703,14 @@ int32_t tsCompressDoubleImp(const char *const input, const int32_t nelements, ch
}
FORCE_INLINE uint64_t decodeDoubleValue(const char *const input, int32_t *const ipos, uint8_t flag) {
int32_t longBytes = LONG_BYTES;
uint64_t diff = 0ul;
int32_t nbytes = (flag & 0x7) + 1;
for (int32_t i = 0; i < nbytes; i++) {
diff |= (((uint64_t)0xff & input[(*ipos)++]) << BITS_PER_BYTE * i);
}
int32_t shift_width = (LONG_BYTES * BITS_PER_BYTE - nbytes * BITS_PER_BYTE) * (flag >> 3);
int32_t shift_width = (longBytes * BITS_PER_BYTE - nbytes * BITS_PER_BYTE) * (flag >> 3);
diff <<= shift_width;
return diff;
@ -1061,14 +860,7 @@ uint32_t decodeFloatValue(const char *const input, int32_t *const ipos, uint8_t
return diff;
}
int32_t tsDecompressFloatImp(const char *const input, const int32_t nelements, char *const output) {
float *ostream = (float *)output;
if (input[0] == 1) {
memcpy(output, input + 1, nelements * FLOAT_BYTES);
return nelements * FLOAT_BYTES;
}
static void tsDecompressFloatHelper(const char *const input, const int32_t nelements, float* ostream) {
uint8_t flags = 0;
int32_t ipos = 1;
int32_t opos = 0;
@ -1094,6 +886,21 @@ int32_t tsDecompressFloatImp(const char *const input, const int32_t nelements, c
ostream[opos++] = curr.real;
}
}
int32_t tsDecompressFloatImp(const char *const input, const int32_t nelements, char *const output) {
if (input[0] == 1) {
memcpy(output, input + 1, nelements * FLOAT_BYTES);
return nelements * FLOAT_BYTES;
}
if (tsSIMDEnable && tsAVX2Enable) {
tsDecompressFloatImplAvx2(input, nelements, output);
} else if (tsSIMDEnable && tsAVX512Enable) {
tsDecompressFloatImplAvx512(input, nelements, output);
} else { // alternative implementation without SIMD instructions.
tsDecompressFloatHelper(input, nelements, (float*)output);
}
return nelements * FLOAT_BYTES;
}

9
source/util/src/tconfig.c
View File

@ -22,6 +22,7 @@
#include "tjson.h"
#include "tlog.h"
#include "tutil.h"
#include "tunit.h"
#define CFG_NAME_PRINT_LEN 24
#define CFG_SRC_PRINT_LEN 12
@ -173,7 +174,7 @@ static int32_t cfgSetBool(SConfigItem *pItem, const char *value, ECfgSrcType sty
}
static int32_t cfgSetInt32(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
int32_t ival = (int32_t)atoi(value);
int32_t ival = taosStrHumanToInt32(value);
if (ival < pItem->imin || ival > pItem->imax) {
uError("cfg:%s, type:%s src:%s value:%d out of range[%" PRId64 ", %" PRId64 "]", pItem->name,
cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), ival, pItem->imin, pItem->imax);
@ -187,7 +188,7 @@ static int32_t cfgSetInt32(SConfigItem *pItem, const char *value, ECfgSrcType st
}
static int32_t cfgSetInt64(SConfigItem *pItem, const char *value, ECfgSrcType stype) {
int64_t ival = (int64_t)atoll(value);
int64_t ival = taosStrHumanToInt64(value);
if (ival < pItem->imin || ival > pItem->imax) {
uError("cfg:%s, type:%s src:%s value:%" PRId64 " out of range[%" PRId64 ", %" PRId64 "]", pItem->name,
cfgDtypeStr(pItem->dtype), cfgStypeStr(stype), ival, pItem->imin, pItem->imax);
@ -379,7 +380,7 @@ int32_t cfgCheckRangeForDynUpdate(SConfig *pCfg, const char *name, const char *p
}
} break;
case CFG_DTYPE_INT32: {
int32_t ival = (int32_t)atoi(pVal);
int32_t ival = (int32_t)taosStrHumanToInt32(pVal);
if (ival < pItem->imin || ival > pItem->imax) {
uError("cfg:%s, type:%s value:%d out of range[%" PRId64 ", %" PRId64 "]", pItem->name,
cfgDtypeStr(pItem->dtype), ival, pItem->imin, pItem->imax);
@ -388,7 +389,7 @@ int32_t cfgCheckRangeForDynUpdate(SConfig *pCfg, const char *name, const char *p
}
} break;
case CFG_DTYPE_INT64: {
int64_t ival = (int64_t)atoll(pVal);
int64_t ival = (int64_t)taosStrHumanToInt64(pVal);
if (ival < pItem->imin || ival > pItem->imax) {
uError("cfg:%s, type:%s value:%" PRId64 " out of range[%" PRId64 ", %" PRId64 "]", pItem->name,
cfgDtypeStr(pItem->dtype), ival, pItem->imin, pItem->imax);

251
source/util/src/tdecompress.c Normal file
View File

@ -0,0 +1,251 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "ttypes.h"
#include "tcompression.h"
int32_t getWordLength(char type) {
int32_t wordLength = 0;
switch (type) {
case TSDB_DATA_TYPE_BIGINT:
wordLength = LONG_BYTES;
break;
case TSDB_DATA_TYPE_INT:
wordLength = INT_BYTES;
break;
case TSDB_DATA_TYPE_SMALLINT:
wordLength = SHORT_BYTES;
break;
case TSDB_DATA_TYPE_TINYINT:
wordLength = CHAR_BYTES;
break;
default:
uError("Invalid decompress integer type:%d", type);
return -1;
}
return wordLength;
}
int32_t tsDecompressIntImpl_Hw(const char *const input, const int32_t nelements, char *const output, const char type) {
int32_t word_length = getWordLength(type);
// Selector value: 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
char bit_per_integer[] = {0, 0, 1, 2, 3, 4, 5, 6, 7, 8, 10, 12, 15, 20, 30, 60};
int32_t selector_to_elems[] = {240, 120, 60, 30, 20, 15, 12, 10, 8, 7, 6, 5, 4, 3, 2, 1};
const char *ip = input + 1;
int32_t count = 0;
int32_t _pos = 0;
int64_t prev_value = 0;
#if __AVX2__
while (1) {
if (_pos == nelements) break;
uint64_t w = 0;
memcpy(&w, ip, LONG_BYTES);
char selector = (char)(w & INT64MASK(4)); // selector = 4
char bit = bit_per_integer[(int32_t)selector]; // bit = 3
int32_t elems = selector_to_elems[(int32_t)selector];
// Optimize the performance, by remove the constantly switch operation.
int32_t v = 4;
uint64_t zigzag_value = 0;
uint64_t mask = INT64MASK(bit);
switch (type) {
case TSDB_DATA_TYPE_BIGINT: {
int64_t* p = (int64_t*) output;
int32_t gRemainder = (nelements - _pos);
int32_t num = (gRemainder > elems)? elems:gRemainder;
int32_t batch = num >> 2;
int32_t remain = num & 0x03;
if (selector == 0 || selector == 1) {
if (tsSIMDEnable && tsAVX2Enable) {
for (int32_t i = 0; i < batch; ++i) {
__m256i prev = _mm256_set1_epi64x(prev_value);
_mm256_storeu_si256((__m256i *)&p[_pos], prev);
_pos += 4;
}
for (int32_t i = 0; i < remain; ++i) {
p[_pos++] = prev_value;
}
} else if (tsSIMDEnable && tsAVX512Enable) {
#if __AVX512F__
// todo add avx512 impl
#endif
} else { // alternative implementation without SIMD instructions.
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
p[_pos++] = prev_value;
v += bit;
}
}
} else {
if (tsSIMDEnable && tsAVX2Enable) {
__m256i base = _mm256_set1_epi64x(w);
__m256i maskVal = _mm256_set1_epi64x(mask);
__m256i shiftBits = _mm256_set_epi64x(bit * 3 + 4, bit * 2 + 4, bit + 4, 4);
__m256i inc = _mm256_set1_epi64x(bit << 2);
for (int32_t i = 0; i < batch; ++i) {
__m256i after = _mm256_srlv_epi64(base, shiftBits);
__m256i zigzagVal = _mm256_and_si256(after, maskVal);
// ZIGZAG_DECODE(T, v) (((v) >> 1) ^ -((T)((v)&1)))
__m256i signmask = _mm256_and_si256(_mm256_set1_epi64x(1), zigzagVal);
signmask = _mm256_sub_epi64(_mm256_setzero_si256(), signmask);
// get the four zigzag values here
__m256i delta = _mm256_xor_si256(_mm256_srli_epi64(zigzagVal, 1), signmask);
// calculate the cumulative sum (prefix sum) for each number
// decode[0] = prev_value + final[0]
// decode[1] = decode[0] + final[1] -----> prev_value + final[0] + final[1]
// decode[2] = decode[1] + final[2] -----> prev_value + final[0] + final[1] + final[2]
// decode[3] = decode[2] + final[3] -----> prev_value + final[0] + final[1] + final[2] + final[3]
// 1, 2, 3, 4
//+ 0, 1, 0, 3
// 1, 3, 3, 7
// shift and add for the first round
__m128i prev = _mm_set1_epi64x(prev_value);
__m256i x = _mm256_slli_si256(delta, 8);
delta = _mm256_add_epi64(delta, x);
_mm256_storeu_si256((__m256i *)&p[_pos], delta);
// 1, 3, 3, 7
//+ 0, 0, 3, 3
// 1, 3, 6, 10
// shift and add operation for the second round
__m128i firstPart = _mm_loadu_si128((__m128i *)&p[_pos]);
__m128i secondItem = _mm_set1_epi64x(p[_pos + 1]);
__m128i secPart = _mm_add_epi64(_mm_loadu_si128((__m128i *)&p[_pos + 2]), secondItem);
firstPart = _mm_add_epi64(firstPart, prev);
secPart = _mm_add_epi64(secPart, prev);
// save it in the memory
_mm_storeu_si128((__m128i *)&p[_pos], firstPart);
_mm_storeu_si128((__m128i *)&p[_pos + 2], secPart);
shiftBits = _mm256_add_epi64(shiftBits, inc);
prev_value = p[_pos + 3];
_pos += 4;
}
// handle the remain value
for (int32_t i = 0; i < remain; i++) {
zigzag_value = ((w >> (v + (batch * bit * 4))) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = prev_value;
v += bit;
}
} else if (tsSIMDEnable && tsAVX512Enable) {
#if __AVX512F__
// todo add avx512 impl
#endif
} else { // alternative implementation without SIMD instructions.
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
zigzag_value = ((w >> v) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = prev_value;
v += bit;
}
}
}
} break;
case TSDB_DATA_TYPE_INT: {
int32_t* p = (int32_t*) output;
if (selector == 0 || selector == 1) {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
p[_pos++] = (int32_t)prev_value;
}
} else {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
zigzag_value = ((w >> v) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = (int32_t)prev_value;
v += bit;
}
}
} break;
case TSDB_DATA_TYPE_SMALLINT: {
int16_t* p = (int16_t*) output;
if (selector == 0 || selector == 1) {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
p[_pos++] = (int16_t)prev_value;
}
} else {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
zigzag_value = ((w >> v) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = (int16_t)prev_value;
v += bit;
}
}
} break;
case TSDB_DATA_TYPE_TINYINT: {
int8_t *p = (int8_t *)output;
if (selector == 0 || selector == 1) {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
p[_pos++] = (int8_t)prev_value;
}
} else {
for (int32_t i = 0; i < elems && count < nelements; i++, count++) {
zigzag_value = ((w >> v) & mask);
prev_value += ZIGZAG_DECODE(int64_t, zigzag_value);
p[_pos++] = (int8_t)prev_value;
v += bit;
}
}
} break;
}
ip += LONG_BYTES;
}
#endif
return nelements * word_length;
}
int32_t tsDecompressFloatImplAvx512(const char *const input, const int32_t nelements, char *const output) {
#if __AVX512F__
// todo add it
#endif
return 0;
}
// todo add later
int32_t tsDecompressFloatImplAvx2(const char *const input, const int32_t nelements, char *const output) {
#if __AVX2__
#endif
return 0;
}

1
source/util/src/terror.c
View File

@ -147,6 +147,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSC_STMT_CACHE_ERROR, "Stmt cache error")
TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INTERNAL_ERROR, "Internal error")
// mnode-common
TAOS_DEFINE_ERROR(TSDB_CODE_MND_REQ_REJECTED, "Request rejected")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_NO_RIGHTS, "Insufficient privilege for operation")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SHOWOBJ, "Data expired")
TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_QUERY_ID, "Invalid query id")

124
source/util/src/tunit.c Normal file
View File

@ -0,0 +1,124 @@
/*
* 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 "tunit.h"
#define UNIT_SIZE_CONVERT_FACTOR 1024LL
#define UNIT_ONE_KIBIBYTE UNIT_SIZE_CONVERT_FACTOR
#define UNIT_ONE_MEBIBYTE (UNIT_ONE_KIBIBYTE * UNIT_SIZE_CONVERT_FACTOR)
#define UNIT_ONE_GIBIBYTE (UNIT_ONE_MEBIBYTE * UNIT_SIZE_CONVERT_FACTOR)
#define UNIT_ONE_TEBIBYTE (UNIT_ONE_GIBIBYTE * UNIT_SIZE_CONVERT_FACTOR)
#define UNIT_ONE_PEBIBYTE (UNIT_ONE_TEBIBYTE * UNIT_SIZE_CONVERT_FACTOR)
#define UNIT_ONE_EXBIBYTE (UNIT_ONE_PEBIBYTE * UNIT_SIZE_CONVERT_FACTOR)
int64_t taosStrHumanToInt64(const char* str) {
size_t sLen = strlen(str);
if (sLen < 2) return atoll(str);
int64_t val = 0;
char* strNoUnit = NULL;
char unit = str[sLen - 1];
if ((unit == 'P') || (unit == 'p')) {
strNoUnit = taosMemoryCalloc(sLen, 1);
memcpy(strNoUnit, str, sLen - 1);
val = atof(strNoUnit) * UNIT_ONE_PEBIBYTE;
} else if ((unit == 'T') || (unit == 't')) {
strNoUnit = taosMemoryCalloc(sLen, 1);
memcpy(strNoUnit, str, sLen - 1);
val = atof(strNoUnit) * UNIT_ONE_TEBIBYTE;
} else if ((unit == 'G') || (unit == 'g')) {
strNoUnit = taosMemoryCalloc(sLen, 1);
memcpy(strNoUnit, str, sLen - 1);
val = atof(strNoUnit) * UNIT_ONE_GIBIBYTE;
} else if ((unit == 'M') || (unit == 'm')) {
strNoUnit = taosMemoryCalloc(sLen, 1);
memcpy(strNoUnit, str, sLen - 1);
val = atof(strNoUnit) * UNIT_ONE_MEBIBYTE;
} else if ((unit == 'K') || (unit == 'k')) {
strNoUnit = taosMemoryCalloc(sLen, 1);
memcpy(strNoUnit, str, sLen - 1);
val = atof(strNoUnit) * UNIT_ONE_KIBIBYTE;
} else {
val = atoll(str);
}
taosMemoryFree(strNoUnit);
return val;
}
void taosInt64ToHumanStr(int64_t val, char* outStr) {
if (((val >= UNIT_ONE_EXBIBYTE) || (-val >= UNIT_ONE_EXBIBYTE)) && ((val % UNIT_ONE_EXBIBYTE) == 0)) {
sprintf(outStr, "%qdE", (long long)val / UNIT_ONE_EXBIBYTE);
} else if (((val >= UNIT_ONE_PEBIBYTE) || (-val >= UNIT_ONE_PEBIBYTE)) && ((val % UNIT_ONE_PEBIBYTE) == 0)) {
sprintf(outStr, "%qdP", (long long)val / UNIT_ONE_PEBIBYTE);
} else if (((val >= UNIT_ONE_TEBIBYTE) || (-val >= UNIT_ONE_TEBIBYTE)) && ((val % UNIT_ONE_TEBIBYTE) == 0)) {
sprintf(outStr, "%qdT", (long long)val / UNIT_ONE_TEBIBYTE);
} else if (((val >= UNIT_ONE_GIBIBYTE) || (-val >= UNIT_ONE_GIBIBYTE)) && ((val % UNIT_ONE_GIBIBYTE) == 0)) {
sprintf(outStr, "%qdG", (long long)val / UNIT_ONE_GIBIBYTE);
} else if (((val >= UNIT_ONE_MEBIBYTE) || (-val >= UNIT_ONE_MEBIBYTE)) && ((val % UNIT_ONE_MEBIBYTE) == 0)) {
sprintf(outStr, "%qdM", (long long)val / UNIT_ONE_MEBIBYTE);
} else if (((val >= UNIT_ONE_KIBIBYTE) || (-val >= UNIT_ONE_KIBIBYTE)) && ((val % UNIT_ONE_KIBIBYTE) == 0)) {
sprintf(outStr, "%qdK", (long long)val / UNIT_ONE_KIBIBYTE);
} else
sprintf(outStr, "%qd", (long long)val);
}
int32_t taosStrHumanToInt32(const char* str) {
size_t sLen = strlen(str);
if (sLen < 2) return atoll(str);
int32_t val = 0;
char* strNoUnit = NULL;
char unit = str[sLen - 1];
if ((unit == 'G') || (unit == 'g')) {
strNoUnit = taosMemoryCalloc(sLen, 1);
memcpy(strNoUnit, str, sLen - 1);
val = atof(strNoUnit) * UNIT_ONE_GIBIBYTE;
} else if ((unit == 'M') || (unit == 'm')) {
strNoUnit = taosMemoryCalloc(sLen, 1);
memcpy(strNoUnit, str, sLen - 1);
val = atof(strNoUnit) * UNIT_ONE_MEBIBYTE;
} else if ((unit == 'K') || (unit == 'k')) {
strNoUnit = taosMemoryCalloc(sLen, 1);
memcpy(strNoUnit, str, sLen - 1);
val = atof(strNoUnit) * UNIT_ONE_KIBIBYTE;
} else {
val = atoll(str);
}
taosMemoryFree(strNoUnit);
return val;
}
void taosInt32ToHumanStr(int32_t val, char* outStr) {
if (((val >= UNIT_ONE_GIBIBYTE) || (-val >= UNIT_ONE_GIBIBYTE)) && ((val % UNIT_ONE_GIBIBYTE) == 0)) {
sprintf(outStr, "%qdG", (long long)val / UNIT_ONE_GIBIBYTE);
} else if (((val >= UNIT_ONE_MEBIBYTE) || (-val >= UNIT_ONE_MEBIBYTE)) && ((val % UNIT_ONE_MEBIBYTE) == 0)) {
sprintf(outStr, "%qdM", (long long)val / UNIT_ONE_MEBIBYTE);
} else if (((val >= UNIT_ONE_KIBIBYTE) || (-val >= UNIT_ONE_KIBIBYTE)) && ((val % UNIT_ONE_KIBIBYTE) == 0)) {
sprintf(outStr, "%qdK", (long long)val / UNIT_ONE_KIBIBYTE);
} else
sprintf(outStr, "%qd", (long long)val);
}

4
tests/parallel_test/cases.task
View File

@ -120,6 +120,10 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/slimit.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/slimit.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/slimit.py -Q 4
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ts-4233.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ts-4233.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ts-4233.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ts-4233.py -Q 4
,,y,system-test,./pytest.sh python3 ./test.py -f 3-enterprise/restore/restoreDnode.py -N 5 -M 3 -i False
,,y,system-test,./pytest.sh python3 ./test.py -f 3-enterprise/restore/restoreVnode.py -N 5 -M 3 -i False

2
tests/script/tsim/stream/fillIntervalRange.sim
View File

@ -64,7 +64,7 @@ endi
sql select count(*) from streamt;
if $data00 != 9098 then
print =====rows=$rows
print =====data00=$data00
goto loop1
endi

156
tests/script/tsim/stream/windowClose.sim
View File

@ -134,6 +134,162 @@ if $rows != 2 then
goto loop1
endi
print step 1 max delay 2s
sql create database test3 vgroups 4;
sql use test3;
sql create table t1(ts timestamp, a int, b int , c int, d double);
sql create stream stream13 trigger max_delay 2s into streamt13 as select _wstart, sum(a), now from t1 interval(10s);
sleep 1000
sql insert into t1 values(1648791213000,1,2,3,1.0);
sql insert into t1 values(1648791223001,2,2,3,1.1);
$loop_count = 0
loop2:
sleep 1000
$loop_count = $loop_count + 1
if $loop_count == 20 then
return -1
endi
sql select * from streamt13;
if $rows != 2 then
print ======rows=$rows
goto loop2
endi
$now02 = $data02
$now12 = $data12
print step1 max delay 2s......... sleep 3s
sleep 3000
sql select * from streamt13;
if $data02 != $now02 then
print ======data02=$data02
return -1
endi
if $data12 != $now12 then
print ======data12=$data12
return -1
endi
print step 2 max delay 2s
sql create database test4 vgroups 4;
sql use test4;
sql create stable st(ts timestamp, a int, b int , c int, d double) tags(ta int,tb int,tc int);
sql create table t1 using st tags(1,1,1);
sql create table t2 using st tags(2,2,2);
sql create stream stream14 trigger max_delay 2s into streamt14 as select _wstart, sum(a), now from st partition by tbname interval(10s);
sleep 1000
sql insert into t1 values(1648791213000,1,2,3,1.0);
sql insert into t1 values(1648791223000,2,2,3,1.1);
sql insert into t2 values(1648791213000,3,2,3,1.0);
sql insert into t2 values(1648791223000,4,2,3,1.1);
$loop_count = 0
loop3:
sleep 1000
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt14 order by 2;
print $data00 $data01 $data02
print $data10 $data11 $data12
print $data20 $data21 $data22
print $data30 $data31 $data32
if $rows != 4 then
print ======rows=$rows
goto loop3
endi
$now02 = $data02
$now12 = $data12
$now22 = $data22
$now32 = $data32
print step2 max delay 2s......... sleep 3s
sleep 3000
sql select * from streamt14 order by 2;
print $data00 $data01 $data02
print $data10 $data11 $data12
print $data20 $data21 $data22
print $data30 $data31 $data32
if $data02 != $now02 then
print ======data02=$data02
return -1
endi
if $data12 != $now12 then
print ======data12=$data12
return -1
endi
if $data22 != $now22 then
print ======data22=$data22
return -1
endi
if $data32 != $now32 then
print ======data32=$data32
return -1
endi
print step2 max delay 2s......... sleep 3s
sleep 3000
sql select * from streamt14 order by 2;
print $data00 $data01 $data02
print $data10 $data11 $data12
print $data20 $data21 $data22
print $data30 $data31 $data32
if $data02 != $now02 then
print ======data02=$data02
return -1
endi
if $data12 != $now12 then
print ======data12=$data12
return -1
endi
if $data22 != $now22 then
print ======data22=$data22
return -1
endi
if $data32 != $now32 then
print ======data32=$data32
return -1
endi
print ======over
system sh/exec.sh -n dnode1 -s stop -x SIGINT

46
tests/system-test/2-query/ts-4233.py Normal file
View File

@ -0,0 +1,46 @@
import taos
from util.log import *
from util.sql import *
from util.cases import *
from util.dnodes import *
from util.common import *
class TDTestCase:
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug(f"start to excute {__file__}")
tdSql.init(conn.cursor(), True)
def checksql(self, sql):
result = os.popen("taos -s '%s'" %sql)
res = result.read()
print(res)
if ("Query OK" in res):
tdLog.info(f"checkEqual success")
else :
tdLog.exit(f"checkEqual error")
def check(self):
conn = taos.connect()
sql = "select 'a;b' as x"
tdSql.query(f"%s" %sql)
tdSql.checkRows(1)
self.checksql('select "a;b" as x\G')
self.checksql('select "a;b" as x >> /tmp/res.txt')
return
def run(self):
tdSql.prepare()
self.check()
def stop(self):
tdSql.close()
tdLog.success(f"{__file__} successfully executed")
tdCases.addLinux(__file__, TDTestCase())
tdCases.addWindows(__file__, TDTestCase())

11
tools/shell/src/shellEngine.c
View File

@ -199,22 +199,17 @@ void shellRunSingleCommandImp(char *command) {
bool printMode = false;
if ((sptr = strstr(command, ">>")) != NULL) {
cptr = strstr(command, ";");
if (cptr != NULL) {
*cptr = '\0';
}
fname = sptr + 2;
while (*fname == ' ') fname++;
*sptr = '\0';
}
if ((sptr = strstr(command, "\\G")) != NULL) {
cptr = strstr(command, ";");
cptr = strstr(fname, ";");
if (cptr != NULL) {
*cptr = '\0';
}
}
if ((sptr = strstr(command, "\\G")) != NULL) {
*sptr = '\0';
printMode = true; // When output to a file, the switch does not work.
}

11
tools/shell/src/shellWebsocket.c
View File

@ -236,22 +236,17 @@ void shellRunSingleCommandWebsocketImp(char *command) {
bool printMode = false;
if ((sptr = strstr(command, ">>")) != NULL) {
cptr = strstr(command, ";");
if (cptr != NULL) {
*cptr = '\0';
}
fname = sptr + 2;
while (*fname == ' ') fname++;
*sptr = '\0';
}
if ((sptr = strstr(command, "\\G")) != NULL) {
cptr = strstr(command, ";");
cptr = strstr(fname, ";");
if (cptr != NULL) {
*cptr = '\0';
}
}
if ((sptr = strstr(command, "\\G")) != NULL) {
*sptr = '\0';
printMode = true; // When output to a file, the switch does not work.
}