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

Merge branch 'develop' of github.com:taosdata/TDengine into dev/chr

This commit is contained in:
tomchon 2021-06-19 10:41:49 +08:00
parent e5f2bab3a6 55fea5300d
commit fd9ca1e964
48 changed files with 1526 additions and 544 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

245
packaging/check_package.sh Executable file
View File

@ -0,0 +1,245 @@
#!/bin/bash
#
# This file is used to install database on linux systems. The operating system
# is required to use systemd to manage services at boot
set -e
#set -x
verMode=edge
pagMode=full
iplist=""
serverFqdn=""
# -----------------------Variables definition---------------------
script_dir="../release"
# Dynamic directory
data_dir="/var/lib/taos"
log_dir="/var/log/taos"
data_link_dir="/usr/local/taos/data"
log_link_dir="/usr/local/taos/log"
cfg_install_dir="/etc/taos"
bin_link_dir="/usr/bin"
lib_link_dir="/usr/lib"
lib64_link_dir="/usr/lib64"
inc_link_dir="/usr/include"
#install main path
install_main_dir="/usr/local/taos"
# old bin dir
sbin_dir="/usr/local/taos/bin"
temp_version=""
fin_result=""
service_config_dir="/etc/systemd/system"
nginx_port=6060
nginx_dir="/usr/local/nginxd"
# Color setting
RED='\033[0;31m'
GREEN='\033[1;32m'
GREEN_DARK='\033[0;32m'
GREEN_UNDERLINE='\033[4;32m'
NC='\033[0m'
csudo=""
if command -v sudo > /dev/null; then
csudo="sudo"
fi
# ============================= get input parameters =================================================
# install.sh -v [server | client] -e [yes | no] -i [systemd | service | ...]
# set parameters by default value
interactiveFqdn=yes # [yes | no]
verType=server # [server | client]
initType=systemd # [systemd | service | ...]
while getopts "hv:d:" arg
do
case $arg in
d)
#echo "interactiveFqdn=$OPTARG"
script_dir=$( echo $OPTARG )
;;
h)
echo "Usage: `basename $0` -d scripy_path"
exit 0
;;
?) #unknow option
echo "unkonw argument"
exit 1
;;
esac
done
#echo "verType=${verType} interactiveFqdn=${interactiveFqdn}"
function kill_process() {
pid=$(ps -ef | grep "$1" | grep -v "grep" | awk '{print $2}')
if [ -n "$pid" ]; then
${csudo} kill -9 $pid || :
fi
}
function check_file() {
#check file whether exists
if [ ! -e $1/$2 ];then
echo -e "$1/$2 \033[31mnot exists\033[0m!quit"
fin_result=$fin_result"\033[31m$temp_version\033[0m test failed!\n"
echo -e $fin_result
exit 8
fi
}
function get_package_name() {
var=$1
if [[ $1 =~ 'aarch' ]];then
echo ${var::-21}
else
echo ${var::-17}
fi
}
function check_link() {
#check Link whether exists or broken
if [ -L $1 ] ; then
if [ ! -e $1 ] ; then
echo -e "$1 \033[31Broken link\033[0m"
fin_result=$fin_result"\033[31m$temp_version\033[0m test failed!\n"
echo -e $fin_result
exit 8
fi
else
echo -e "$1 \033[31mnot exists\033[0m!quit"
fin_result=$fin_result"\033[31m$temp_version\033[0m test failed!\n"
echo -e $fin_result
exit 8
fi
}
function check_main_path() {
#check install main dir and all sub dir
main_dir=("" "cfg" "bin" "connector" "driver" "examples" "include" "init.d")
for i in ${main_dir[@]};do
check_file ${install_main_dir} $i
done
if [ "$verMode" == "cluster" ]; then
nginx_main_dir=("admin" "conf" "html" "sbin" "logs")
for i in ${nginx_main_dir[@]};do
check_file ${nginx_dir} $i
done
fi
echo -e "Check main path:\033[32mOK\033[0m!"
}
function check_bin_path() {
# check install bin dir and all sub dir
bin_dir=("taos" "taosd" "taosdemo" "taosdump" "remove.sh" "tarbitrator" "set_core.sh")
for i in ${bin_dir[@]};do
check_file ${sbin_dir} $i
done
lbin_dir=("taos" "taosd" "taosdemo" "taosdump" "rmtaos" "tarbitrator" "set_core")
for i in ${lbin_dir[@]};do
check_link ${bin_link_dir}/$i
done
if [ "$verMode" == "cluster" ]; then
check_file ${nginx_dir}/sbin nginx
fi
echo -e "Check bin path:\033[32mOK\033[0m!"
}
function check_lib_path() {
# check all links
check_link ${lib_link_dir}/libtaos.so
check_link ${lib_link_dir}/libtaos.so.1
if [[ -d ${lib64_link_dir} ]]; then
check_link ${lib64_link_dir}/libtaos.so
check_link ${lib64_link_dir}/libtaos.so.1
fi
echo -e "Check lib path:\033[32mOK\033[0m!"
}
function check_header_path() {
# check all header
header_dir=("taos.h" "taoserror.h")
for i in ${header_dir[@]};do
check_link ${inc_link_dir}/$i
done
echo -e "Check bin path:\033[32mOK\033[0m!"
}
function check_config_dir() {
# check all config
check_file ${cfg_install_dir} taos.cfg
check_file ${install_main_dir}/cfg taos.cfg.org
echo -e "Check conf path:\033[32mOK\033[0m!"
}
function check_log_path() {
# check log path
check_file ${log_dir}
echo -e "Check log path:\033[32mOK\033[0m!"
}
function check_data_path() {
# check data path
check_file ${data_dir}
echo -e "Check data path:\033[32mOK\033[0m!"
}
function install_TDengine() {
cd ${script_dir}
tar zxf $1
temp_version=$(get_package_name $1)
cd $(get_package_name $1)
echo -e "\033[32muninstall TDengine && install TDengine...\033[0m"
rmtaos >/dev/null 2>&1 || echo 'taosd not installed' && echo -e '\n\n' |./install.sh >/dev/null 2>&1
echo -e "\033[32mTDengine has been installed!\033[0m"
echo -e "\033[32mTDengine is starting...\033[0m"
kill_process taos && systemctl start taosd && sleep 10
}
function test_TDengine() {
check_main_path
check_bin_path
check_lib_path
check_header_path
check_config_dir
check_log_path
check_data_path
result=`taos -s 'create database test ;create table test.tt(ts timestamp ,i int);insert into test.tt values(now,11);select * from test.tt' 2>&1 ||:`
if [[ $result =~ "Unable to establish" ]];then
echo -e "\033[31mTDengine connect failed\033[0m"
fin_result=$fin_result"\033[31m$temp_version\033[0m test failed!\n"
echo -e $fin_result
exit 8
fi
echo -e "Check TDengine connect:\033[32mOK\033[0m!"
fin_result=$fin_result"\033[32m$temp_version\033[0m test OK!\n"
}
# ## ==============================Main program starts from here============================
TD_package_name=`ls ${script_dir}/*server*gz |awk -F '/' '{print $NF}' `
temp=`pwd`
for i in $TD_package_name;do
if [[ $i =~ 'enterprise' ]];then
verMode="cluster"
else
verMode=""
fi
cd $temp
install_TDengine $i
test_TDengine
done
echo "============================================================"
echo -e $fin_result

1
src/balance/src/bnMain.c
View File

@ -367,6 +367,7 @@ static bool bnMonitorBalance() {
for (int32_t dest = 0; dest < src; dest++) { for (int32_t dest = 0; dest < src; dest++) {
SDnodeObj *pDestDnode = tsBnDnodes.list[dest]; SDnodeObj *pDestDnode = tsBnDnodes.list[dest];
if (bnCheckDnodeInVgroup(pDestDnode, pVgroup)) continue; if (bnCheckDnodeInVgroup(pDestDnode, pVgroup)) continue;
if (taosGetTimestampMs() - pDestDnode->createdTime < 2000) continue;
float destScore = bnTryCalcDnodeScore(pDestDnode, 1); float destScore = bnTryCalcDnodeScore(pDestDnode, 1);
if (srcScore + 0.0001 < destScore) continue; if (srcScore + 0.0001 < destScore) continue;

2
src/client/inc/tscUtil.h
View File

@ -123,6 +123,8 @@ int32_t tscGetDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, i
*/ */
bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo); bool tscIsPointInterpQuery(SQueryInfo* pQueryInfo);
bool tscIsTWAQuery(SQueryInfo* pQueryInfo); bool tscIsTWAQuery(SQueryInfo* pQueryInfo);
bool tscIsIrateQuery(SQueryInfo* pQueryInfo);
bool tscIsSessionWindowQuery(SQueryInfo* pQueryInfo); bool tscIsSessionWindowQuery(SQueryInfo* pQueryInfo);
bool tscIsSecondStageQuery(SQueryInfo* pQueryInfo); bool tscIsSecondStageQuery(SQueryInfo* pQueryInfo);
bool tsIsArithmeticQueryOnAggResult(SQueryInfo* pQueryInfo); bool tsIsArithmeticQueryOnAggResult(SQueryInfo* pQueryInfo);

10
src/client/src/tscPrepare.c
View File

@ -906,7 +906,7 @@ static int insertStmtBindParam(STscStmt* stmt, TAOS_BIND* bind) {
int code = doBindParam(pBlock, data, param, &bind[param->idx], 1); int code = doBindParam(pBlock, data, param, &bind[param->idx], 1);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tscDebug("0x%"PRIx64" bind column %d: type mismatch or invalid", pStmt->pSql->self, param->idx); tscDebug("0x%"PRIx64" bind column %d: type mismatch or invalid", pStmt->pSql->self, param->idx);
return invalidOperationMsg(tscGetErrorMsgPayload(&stmt->pSql->cmd), "bind column type mismatch or invalid"); return invalidOperationMsg(tscGetErrorMsgPayload(&stmt->pSql->cmd), "bind column type mismatch or invalid");
} }
} }
@ -1256,7 +1256,7 @@ int stmtParseInsertTbTags(SSqlObj* pSql, STscStmt* pStmt) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
if (sToken.n <= 0 || sToken.type != TK_USING) { if (sToken.n <= 0 || sToken.type != TK_USING) {
tscError("keywords USING is expected, sql:%s", pCmd->insertParam.sql); tscError("keywords USING is expected, sql:%s", pCmd->insertParam.sql);
return tscSQLSyntaxErrMsg(pCmd->payload, "keywords USING is expected", sToken.z ? sToken.z : pCmd->insertParam.sql); return tscSQLSyntaxErrMsg(pCmd->payload, "keywords USING is expected", sToken.z ? sToken.z : pCmd->insertParam.sql);
} }
@ -1403,7 +1403,7 @@ int stmtGenInsertStatement(SSqlObj* pSql, STscStmt* pStmt, const char* name, TAO
} else { } else {
tfree(pSql->sqlstr); tfree(pSql->sqlstr);
} }
pSql->sqlstr = str; pSql->sqlstr = str;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1415,7 +1415,7 @@ int stmtGenInsertStatement(SSqlObj* pSql, STscStmt* pStmt, const char* name, TAO
TAOS_STMT* taos_stmt_init(TAOS* taos) { TAOS_STMT* taos_stmt_init(TAOS* taos) {
STscObj* pObj = (STscObj*)taos; STscObj* pObj = (STscObj*)taos;
STscStmt* pStmt = NULL; STscStmt* pStmt = NULL;
if (pObj == NULL || pObj->signature != pObj) { if (pObj == NULL || pObj->signature != pObj) {
terrno = TSDB_CODE_TSC_DISCONNECTED; terrno = TSDB_CODE_TSC_DISCONNECTED;
tscError("connection disconnected"); tscError("connection disconnected");
@ -1469,7 +1469,7 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
tscError("sql is NULL"); tscError("sql is NULL");
STMT_RET(invalidOperationMsg(tscGetErrorMsgPayload(&pStmt->pSql->cmd), "sql is NULL")); STMT_RET(invalidOperationMsg(tscGetErrorMsgPayload(&pStmt->pSql->cmd), "sql is NULL"));
} }
if (pStmt->last != STMT_INIT) { if (pStmt->last != STMT_INIT) {
tscError("prepare status error, last:%d", pStmt->last); tscError("prepare status error, last:%d", pStmt->last);
STMT_RET(invalidOperationMsg(tscGetErrorMsgPayload(&pStmt->pSql->cmd), "prepare status error")); STMT_RET(invalidOperationMsg(tscGetErrorMsgPayload(&pStmt->pSql->cmd), "prepare status error"));

153
src/client/src/tscSQLParser.c
View File

@ -147,57 +147,57 @@ int16_t getNewResColId(SSqlCmd* pCmd) {
return pCmd->resColumnId--; return pCmd->resColumnId--;
} }
// serialize expr in exprlist to binary // serialize expr in exprlist to binary
// formate "type | size | value" // formate "type | size | value"
bool serializeExprListToVariant(SArray* pList, tVariant **dst, int16_t colType) { bool serializeExprListToVariant(SArray* pList, tVariant **dst, int16_t colType) {
bool ret = false; bool ret = false;
if (!pList || pList->size <= 0 || colType < 0) { if (!pList || pList->size <= 0 || colType < 0) {
return ret; return ret;
} }
tSqlExprItem* item = (tSqlExprItem *)taosArrayGet(pList, 0); tSqlExprItem* item = (tSqlExprItem *)taosArrayGet(pList, 0);
int32_t firstTokenType = item->pNode->token.type; int32_t firstTokenType = item->pNode->token.type;
int32_t type = firstTokenType; int32_t type = firstTokenType;
//nchar to binary and //nchar to binary and
toTSDBType(type); toTSDBType(type);
if (type != colType && (type != TSDB_DATA_TYPE_BINARY || colType != TSDB_DATA_TYPE_NCHAR)) { if (type != colType && (type != TSDB_DATA_TYPE_BINARY || colType != TSDB_DATA_TYPE_NCHAR)) {
return false; return false;
} }
type = colType; type = colType;
SBufferWriter bw = tbufInitWriter( NULL, false ); SBufferWriter bw = tbufInitWriter( NULL, false );
tbufEnsureCapacity(&bw, 512); tbufEnsureCapacity(&bw, 512);
int32_t size = (int32_t)(pList->size); int32_t size = (int32_t)(pList->size);
tbufWriteUint32(&bw, type); tbufWriteUint32(&bw, type);
tbufWriteInt32(&bw, size); tbufWriteInt32(&bw, size);
for (int32_t i = 0; i < size; i++) { for (int32_t i = 0; i < size; i++) {
tSqlExpr* pSub = ((tSqlExprItem*)(taosArrayGet(pList, i)))->pNode; tSqlExpr* pSub = ((tSqlExprItem*)(taosArrayGet(pList, i)))->pNode;
// check all the token type in expr list same or not // check all the token type in expr list same or not
if (firstTokenType != pSub->token.type) { if (firstTokenType != pSub->token.type) {
break; break;
} }
toTSDBType(pSub->token.type); toTSDBType(pSub->token.type);
tVariant var; tVariant var;
tVariantCreate(&var, &pSub->token); tVariantCreate(&var, &pSub->token);
if (type == TSDB_DATA_TYPE_BOOL || type == TSDB_DATA_TYPE_TINYINT || type == TSDB_DATA_TYPE_SMALLINT if (type == TSDB_DATA_TYPE_BOOL || type == TSDB_DATA_TYPE_TINYINT || type == TSDB_DATA_TYPE_SMALLINT
|| type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_INT) { || type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_INT) {
tbufWriteInt64(&bw, var.i64); tbufWriteInt64(&bw, var.i64);
} else if (type == TSDB_DATA_TYPE_DOUBLE || type == TSDB_DATA_TYPE_FLOAT) { } else if (type == TSDB_DATA_TYPE_DOUBLE || type == TSDB_DATA_TYPE_FLOAT) {
tbufWriteDouble(&bw, var.dKey); tbufWriteDouble(&bw, var.dKey);
} else if (type == TSDB_DATA_TYPE_BINARY){ } else if (type == TSDB_DATA_TYPE_BINARY){
tbufWriteBinary(&bw, var.pz, var.nLen); tbufWriteBinary(&bw, var.pz, var.nLen);
} else if (type == TSDB_DATA_TYPE_NCHAR) { } else if (type == TSDB_DATA_TYPE_NCHAR) {
char *buf = (char *)calloc(1, (var.nLen + 1)*TSDB_NCHAR_SIZE); char *buf = (char *)calloc(1, (var.nLen + 1)*TSDB_NCHAR_SIZE);
if (tVariantDump(&var, buf, type, false) != TSDB_CODE_SUCCESS) { if (tVariantDump(&var, buf, type, false) != TSDB_CODE_SUCCESS) {
free(buf); free(buf);
tVariantDestroy(&var); tVariantDestroy(&var);
break; break;
} }
tbufWriteBinary(&bw, buf, twcslen((wchar_t *)buf) * TSDB_NCHAR_SIZE); tbufWriteBinary(&bw, buf, twcslen((wchar_t *)buf) * TSDB_NCHAR_SIZE);
free(buf); free(buf);
@ -209,21 +209,21 @@ bool serializeExprListToVariant(SArray* pList, tVariant **dst, int16_t colType)
if (ret == true) { if (ret == true) {
if ((*dst = calloc(1, sizeof(tVariant))) != NULL) { if ((*dst = calloc(1, sizeof(tVariant))) != NULL) {
tVariantCreateFromBinary(*dst, tbufGetData(&bw, false), tbufTell(&bw), TSDB_DATA_TYPE_BINARY); tVariantCreateFromBinary(*dst, tbufGetData(&bw, false), tbufTell(&bw), TSDB_DATA_TYPE_BINARY);
} else { } else {
ret = false; ret = false;
} }
} }
tbufCloseWriter(&bw); tbufCloseWriter(&bw);
return ret; return ret;
} }
static int32_t validateParamOfRelationIn(tVariant *pVar, int32_t colType) { static int32_t validateParamOfRelationIn(tVariant *pVar, int32_t colType) {
if (pVar->nType != TSDB_DATA_TYPE_BINARY) { if (pVar->nType != TSDB_DATA_TYPE_BINARY) {
return -1; return -1;
} }
SBufferReader br = tbufInitReader(pVar->pz, pVar->nLen, false); SBufferReader br = tbufInitReader(pVar->pz, pVar->nLen, false);
return colType == TSDB_DATA_TYPE_NCHAR ? 0 : (tbufReadUint32(&br) == colType ? 0: -1); return colType == TSDB_DATA_TYPE_NCHAR ? 0 : (tbufReadUint32(&br) == colType ? 0: -1);
} }
static uint8_t convertOptr(SStrToken *pToken) { static uint8_t convertOptr(SStrToken *pToken) {
@ -423,18 +423,6 @@ int32_t tscValidateSqlInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
if(code != TSDB_CODE_SUCCESS) { if(code != TSDB_CODE_SUCCESS) {
return code; return code;
} }
if (pInfo->pMiscInfo->tableType == TSDB_SUPER_TABLE) {
//// code = tscGetTableMeta(pSql, pTableMetaInfo);
//// if (code != TSDB_CODE_SUCCESS) {
//// return code;
//// }
//
// if (!UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
// return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg4);
// }
}
} else if (pInfo->type == TSDB_SQL_DROP_DNODE) { } else if (pInfo->type == TSDB_SQL_DROP_DNODE) {
if (pzName->type == TK_STRING) { if (pzName->type == TK_STRING) {
pzName->n = strdequote(pzName->z); pzName->n = strdequote(pzName->z);
@ -507,7 +495,7 @@ int32_t tscValidateSqlInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
break; break;
} }
case TSDB_SQL_CREATE_DNODE: { case TSDB_SQL_CREATE_DNODE: {
const char* msg = "invalid host name (ip address)"; const char* msg = "invalid host name (ip address)";
@ -772,8 +760,9 @@ int32_t tscValidateSqlInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
pCmd->active = pCmd->pQueryInfo; pCmd->active = pCmd->pQueryInfo;
pCmd->command = pCmd->pQueryInfo->command; pCmd->command = pCmd->pQueryInfo->command;
if (pTableMetaInfo->pTableMeta != NULL) { STableMetaInfo* pTableMetaInfo1 = tscGetMetaInfo(pCmd->active, 0);
pSql->res.precision = tscGetTableInfo(pTableMetaInfo->pTableMeta).precision; if (pTableMetaInfo1->pTableMeta != NULL) {
pSql->res.precision = tscGetTableInfo(pTableMetaInfo1->pTableMeta).precision;
} }
return TSDB_CODE_SUCCESS; // do not build query message here return TSDB_CODE_SUCCESS; // do not build query message here
@ -807,12 +796,12 @@ int32_t tscValidateSqlInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
} }
break; break;
} }
case TSDB_SQL_COMPACT_VNODE:{ case TSDB_SQL_COMPACT_VNODE:{
const char* msg = "invalid compact"; const char* msg = "invalid compact";
if (setCompactVnodeInfo(pSql, pInfo) != TSDB_CODE_SUCCESS) { if (setCompactVnodeInfo(pSql, pInfo) != TSDB_CODE_SUCCESS) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg);
} }
break; break;
} }
default: default:
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), "not support sql expression"); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), "not support sql expression");
@ -960,20 +949,22 @@ int32_t validateIntervalNode(SSqlObj* pSql, SQueryInfo* pQueryInfo, SSqlNode* pS
static int32_t validateStateWindowNode(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SSqlNode* pSqlNode, bool isStable) { static int32_t validateStateWindowNode(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SSqlNode* pSqlNode, bool isStable) {
const char* msg1 = "invalid column name"; const char* msg1 = "invalid column name";
const char* msg2 = "invalid column type";
const char* msg3 = "not support state_window with group by "; const char* msg3 = "not support state_window with group by ";
const char* msg4 = "function not support for super table query"; const char* msg4 = "function not support for super table query";
const char* msg5 = "not support state_window on tag column";
SStrToken *col = &(pSqlNode->windowstateVal.col) ; SStrToken *col = &(pSqlNode->windowstateVal.col) ;
if (col->z == NULL || col->n <= 0) { if (col->z == NULL || col->n <= 0) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
if (pQueryInfo->colList == NULL) { if (pQueryInfo->colList == NULL) {
pQueryInfo->colList = taosArrayInit(4, POINTER_BYTES); pQueryInfo->colList = taosArrayInit(4, POINTER_BYTES);
} }
if (pQueryInfo->groupbyExpr.numOfGroupCols > 0) { if (pQueryInfo->groupbyExpr.numOfGroupCols > 0) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg3); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg3);
} }
pQueryInfo->groupbyExpr.numOfGroupCols = 1; pQueryInfo->groupbyExpr.numOfGroupCols = 1;
//TODO(dengyihao): check tag column //TODO(dengyihao): check tag column
@ -984,23 +975,27 @@ static int32_t validateStateWindowNode(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SS
SColumnIndex index = COLUMN_INDEX_INITIALIZER; SColumnIndex index = COLUMN_INDEX_INITIALIZER;
if (getColumnIndexByName(pCmd, col, pQueryInfo, &index) != TSDB_CODE_SUCCESS) { if (getColumnIndexByName(pCmd, col, pQueryInfo, &index) != TSDB_CODE_SUCCESS) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg1); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg1);
} }
STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex); STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex);
STableMeta* pTableMeta = pTableMetaInfo->pTableMeta; STableMeta* pTableMeta = pTableMetaInfo->pTableMeta;
int32_t numOfCols = tscGetNumOfColumns(pTableMeta); int32_t numOfCols = tscGetNumOfColumns(pTableMeta);
if (index.columnIndex == TSDB_TBNAME_COLUMN_INDEX || index.columnIndex >= numOfCols) { if (index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg3); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg3);
} else if (index.columnIndex >= numOfCols) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg5);
} }
SGroupbyExpr* pGroupExpr = &pQueryInfo->groupbyExpr; SGroupbyExpr* pGroupExpr = &pQueryInfo->groupbyExpr;
if (pGroupExpr->columnInfo == NULL) { if (pGroupExpr->columnInfo == NULL) {
pGroupExpr->columnInfo = taosArrayInit(4, sizeof(SColIndex)); pGroupExpr->columnInfo = taosArrayInit(4, sizeof(SColIndex));
} }
SSchema* pSchema = tscGetTableColumnSchema(pTableMeta, index.columnIndex); SSchema* pSchema = tscGetTableColumnSchema(pTableMeta, index.columnIndex);
if (pSchema->type == TSDB_DATA_TYPE_TIMESTAMP || pSchema->type == TSDB_DATA_TYPE_FLOAT || pSchema->type == TSDB_DATA_TYPE_DOUBLE) { if (pSchema->type == TSDB_DATA_TYPE_TIMESTAMP || pSchema->type == TSDB_DATA_TYPE_FLOAT
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg1); || pSchema->type == TSDB_DATA_TYPE_DOUBLE || pSchema->type == TSDB_DATA_TYPE_NCHAR
|| pSchema->type == TSDB_DATA_TYPE_BINARY) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg2);
} }
tscColumnListInsert(pQueryInfo->colList, index.columnIndex, pTableMeta->id.uid, pSchema); tscColumnListInsert(pQueryInfo->colList, index.columnIndex, pTableMeta->id.uid, pSchema);
@ -1049,7 +1044,7 @@ int32_t validateSessionNode(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, SSqlNode * pS
} }
if (index.columnIndex != PRIMARYKEY_TIMESTAMP_COL_INDEX) { if (index.columnIndex != PRIMARYKEY_TIMESTAMP_COL_INDEX) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg3); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg3);
} }
pQueryInfo->sessionWindow.primaryColId = PRIMARYKEY_TIMESTAMP_COL_INDEX; pQueryInfo->sessionWindow.primaryColId = PRIMARYKEY_TIMESTAMP_COL_INDEX;
@ -2026,8 +2021,10 @@ int32_t addProjectionExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, t
if (index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) { if (index.columnIndex == TSDB_TBNAME_COLUMN_INDEX) {
SSchema colSchema = *tGetTbnameColumnSchema(); SSchema colSchema = *tGetTbnameColumnSchema();
getColumnName(pItem, colSchema.name, colSchema.name, sizeof(colSchema.name) - 1); char name[TSDB_COL_NAME_LEN] = {0};
getColumnName(pItem, name, colSchema.name, sizeof(colSchema.name) - 1);
tstrncpy(colSchema.name, name, TSDB_COL_NAME_LEN);
/*SExprInfo* pExpr = */tscAddFuncInSelectClause(pQueryInfo, startPos, TSDB_FUNC_TAGPRJ, &index, &colSchema, TSDB_COL_TAG, getNewResColId(pCmd)); /*SExprInfo* pExpr = */tscAddFuncInSelectClause(pQueryInfo, startPos, TSDB_FUNC_TAGPRJ, &index, &colSchema, TSDB_COL_TAG, getNewResColId(pCmd));
} else { } else {
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex); STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index.tableIndex);
@ -2341,7 +2338,7 @@ int32_t addExprAndResultField(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, int32_t col
tickPerSec /= 1000000; tickPerSec /= 1000000;
} else if (info.precision == TSDB_TIME_PRECISION_MICRO) { } else if (info.precision == TSDB_TIME_PRECISION_MICRO) {
tickPerSec /= 1000; tickPerSec /= 1000;
} }
if (tickPerSec <= 0 || tickPerSec < TSDB_TICK_PER_SECOND(info.precision)) { if (tickPerSec <= 0 || tickPerSec < TSDB_TICK_PER_SECOND(info.precision)) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg10); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg10);
@ -2978,7 +2975,7 @@ int32_t setKillInfo(SSqlObj* pSql, struct SSqlInfo* pInfo, int32_t killType) {
static int32_t setCompactVnodeInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) { static int32_t setCompactVnodeInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
SSqlCmd* pCmd = &pSql->cmd; SSqlCmd* pCmd = &pSql->cmd;
pCmd->command = pInfo->type; pCmd->command = pInfo->type;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
bool validateIpAddress(const char* ip, size_t size) { bool validateIpAddress(const char* ip, size_t size) {
@ -3066,8 +3063,8 @@ void tscRestoreFuncForSTableQuery(SQueryInfo* pQueryInfo) {
} }
bool hasUnsupportFunctionsForSTableQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo) { bool hasUnsupportFunctionsForSTableQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo) {
const char* msg1 = "TWA/Diff not allowed to apply to super table directly"; const char* msg1 = "TWA/Diff/Derivative/Irate not allowed to apply to super table directly";
const char* msg2 = "TWA/Diff only support group by tbname for super table query"; const char* msg2 = "TWA/Diff/Derivative/Irate only support group by tbname for super table query";
const char* msg3 = "function not support for super table query"; const char* msg3 = "function not support for super table query";
// filter sql function not supported by metric query yet. // filter sql function not supported by metric query yet.
@ -3080,7 +3077,7 @@ bool hasUnsupportFunctionsForSTableQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo)
} }
} }
if (tscIsTWAQuery(pQueryInfo) || tscIsDiffDerivQuery(pQueryInfo)) { if (tscIsTWAQuery(pQueryInfo) || tscIsDiffDerivQuery(pQueryInfo) || tscIsIrateQuery(pQueryInfo)) {
if (pQueryInfo->groupbyExpr.numOfGroupCols == 0) { if (pQueryInfo->groupbyExpr.numOfGroupCols == 0) {
invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg1); invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg1);
return true; return true;
@ -3098,7 +3095,7 @@ bool hasUnsupportFunctionsForSTableQuery(SSqlCmd* pCmd, SQueryInfo* pQueryInfo)
} }
} else if (tscIsSessionWindowQuery(pQueryInfo)) { } else if (tscIsSessionWindowQuery(pQueryInfo)) {
invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg3); invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg3);
return true; return true;
} }
return false; return false;
@ -3340,24 +3337,24 @@ static int32_t doExtractColumnFilterInfo(SSqlCmd* pCmd, SQueryInfo* pQueryInfo,
// TK_GT,TK_GE,TK_EQ,TK_NE are based on the pColumn->lowerBndd // TK_GT,TK_GE,TK_EQ,TK_NE are based on the pColumn->lowerBndd
} else if (pExpr->tokenId == TK_IN) { } else if (pExpr->tokenId == TK_IN) {
tVariant *pVal; tVariant *pVal;
if (pRight->tokenId != TK_SET || !serializeExprListToVariant(pRight->pParam, &pVal, colType) || colType == TSDB_DATA_TYPE_TIMESTAMP) { if (pRight->tokenId != TK_SET || !serializeExprListToVariant(pRight->pParam, &pVal, colType) || colType == TSDB_DATA_TYPE_TIMESTAMP) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg);
} }
if (validateParamOfRelationIn(pVal, colType) != TSDB_CODE_SUCCESS) { if (validateParamOfRelationIn(pVal, colType) != TSDB_CODE_SUCCESS) {
tVariantDestroy(pVal); tVariantDestroy(pVal);
free(pVal); free(pVal);
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg);
} }
pColumnFilter->pz = (int64_t)calloc(1, pVal->nLen + 1); pColumnFilter->pz = (int64_t)calloc(1, pVal->nLen + 1);
pColumnFilter->len = pVal->nLen; pColumnFilter->len = pVal->nLen;
pColumnFilter->filterstr = 1; pColumnFilter->filterstr = 1;
memcpy((char *)(pColumnFilter->pz), (char *)(pVal->pz), pVal->nLen); memcpy((char *)(pColumnFilter->pz), (char *)(pVal->pz), pVal->nLen);
//retVal = tVariantDump(pVal, (char *)(pColumnFilter->pz), TSDB_DATA_TYPE_BINARY, false); //retVal = tVariantDump(pVal, (char *)(pColumnFilter->pz), TSDB_DATA_TYPE_BINARY, false);
tVariantDestroy(pVal); tVariantDestroy(pVal);
free(pVal); free(pVal);
} else if (colType == TSDB_DATA_TYPE_BINARY) { } else if (colType == TSDB_DATA_TYPE_BINARY) {
pColumnFilter->pz = (int64_t)calloc(1, bufLen * TSDB_NCHAR_SIZE); pColumnFilter->pz = (int64_t)calloc(1, bufLen * TSDB_NCHAR_SIZE);
pColumnFilter->len = pRight->value.nLen; pColumnFilter->len = pRight->value.nLen;
@ -4884,7 +4881,7 @@ int32_t getTimeRange(STimeWindow* win, tSqlExpr* pRight, int32_t optr, int16_t t
if (pRight->flags & (1 << EXPR_FLAG_NS_TIMESTAMP)) { if (pRight->flags & (1 << EXPR_FLAG_NS_TIMESTAMP)) {
pRight->value.i64 = convertTimePrecision(pRight->value.i64, TSDB_TIME_PRECISION_NANO, timePrecision); pRight->value.i64 = convertTimePrecision(pRight->value.i64, TSDB_TIME_PRECISION_NANO, timePrecision);
} }
tVariantDump(&pRight->value, (char*)&val, TSDB_DATA_TYPE_BIGINT, true); tVariantDump(&pRight->value, (char*)&val, TSDB_DATA_TYPE_BIGINT, true);
} }
@ -5542,7 +5539,7 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
if (pItem->type != TSDB_DATA_TYPE_BINARY && pItem->type != TSDB_DATA_TYPE_NCHAR) { if (pItem->type != TSDB_DATA_TYPE_BINARY && pItem->type != TSDB_DATA_TYPE_NCHAR) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg21); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg21);
} }
SColumnIndex columnIndex = COLUMN_INDEX_INITIALIZER; SColumnIndex columnIndex = COLUMN_INDEX_INITIALIZER;
SStrToken name = {.type = TK_STRING, .z = pItem->name, .n = (uint32_t)strlen(pItem->name)}; SStrToken name = {.type = TK_STRING, .z = pItem->name, .n = (uint32_t)strlen(pItem->name)};
if (getColumnIndexByName(pCmd, &name, pQueryInfo, &columnIndex) != TSDB_CODE_SUCCESS) { if (getColumnIndexByName(pCmd, &name, pQueryInfo, &columnIndex) != TSDB_CODE_SUCCESS) {
@ -5563,11 +5560,11 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
(pItem->type == TSDB_DATA_TYPE_NCHAR && (pItem->bytes <= 0 || pItem->bytes > TSDB_MAX_NCHAR_LEN))) { (pItem->type == TSDB_DATA_TYPE_NCHAR && (pItem->bytes <= 0 || pItem->bytes > TSDB_MAX_NCHAR_LEN))) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg24); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg24);
} }
if (pItem->bytes <= pColSchema->bytes) { if (pItem->bytes <= pColSchema->bytes) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg22); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg22);
} }
TAOS_FIELD f = tscCreateField(pColSchema->type, name.z, pItem->bytes); TAOS_FIELD f = tscCreateField(pColSchema->type, name.z, pItem->bytes);
tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f); tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
}else if (pAlterSQL->type == TSDB_ALTER_TABLE_MODIFY_TAG_COLUMN) { }else if (pAlterSQL->type == TSDB_ALTER_TABLE_MODIFY_TAG_COLUMN) {
@ -5579,7 +5576,7 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
if (pItem->type != TSDB_DATA_TYPE_BINARY && pItem->type != TSDB_DATA_TYPE_NCHAR) { if (pItem->type != TSDB_DATA_TYPE_BINARY && pItem->type != TSDB_DATA_TYPE_NCHAR) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg21); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg21);
} }
SColumnIndex columnIndex = COLUMN_INDEX_INITIALIZER; SColumnIndex columnIndex = COLUMN_INDEX_INITIALIZER;
SStrToken name = {.type = TK_STRING, .z = pItem->name, .n = (uint32_t)strlen(pItem->name)}; SStrToken name = {.type = TK_STRING, .z = pItem->name, .n = (uint32_t)strlen(pItem->name)};
if (getColumnIndexByName(pCmd, &name, pQueryInfo, &columnIndex) != TSDB_CODE_SUCCESS) { if (getColumnIndexByName(pCmd, &name, pQueryInfo, &columnIndex) != TSDB_CODE_SUCCESS) {
@ -5604,11 +5601,11 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
(pItem->type == TSDB_DATA_TYPE_NCHAR && (pItem->bytes <= 0 || pItem->bytes > TSDB_MAX_NCHAR_LEN))) { (pItem->type == TSDB_DATA_TYPE_NCHAR && (pItem->bytes <= 0 || pItem->bytes > TSDB_MAX_NCHAR_LEN))) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg24); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg24);
} }
if (pItem->bytes <= pColSchema->bytes) { if (pItem->bytes <= pColSchema->bytes) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg22); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg22);
} }
TAOS_FIELD f = tscCreateField(pColSchema->type, name.z, pItem->bytes); TAOS_FIELD f = tscCreateField(pColSchema->type, name.z, pItem->bytes);
tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f); tscFieldInfoAppend(&pQueryInfo->fieldsInfo, &f);
} }
@ -5974,7 +5971,7 @@ static int32_t setKeepOption(SSqlCmd* pCmd, SCreateDbMsg* pMsg, SCreateDbInfo* p
tVariantListItem* p0 = taosArrayGet(pKeep, 0); tVariantListItem* p0 = taosArrayGet(pKeep, 0);
tVariantListItem* p1 = (s > 1) ? taosArrayGet(pKeep, 1) : p0; tVariantListItem* p1 = (s > 1) ? taosArrayGet(pKeep, 1) : p0;
tVariantListItem* p2 = (s > 2) ? taosArrayGet(pKeep, 2) : p1; tVariantListItem* p2 = (s > 2) ? taosArrayGet(pKeep, 2) : p1;
if ((int32_t)p0->pVar.i64 <= 0 || (int32_t)p1->pVar.i64 <= 0 || (int32_t)p2->pVar.i64 <= 0) { if ((int32_t)p0->pVar.i64 <= 0 || (int32_t)p1->pVar.i64 <= 0 || (int32_t)p2->pVar.i64 <= 0) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg2); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg2);
} }
@ -7068,7 +7065,7 @@ int32_t doCheckForStream(SSqlObj* pSql, SSqlInfo* pInfo) {
} else { } else {
if (pQueryInfo->interval.interval == 0) { if (pQueryInfo->interval.interval == 0) {
return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg7); return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg7);
} }
} }
// set the created table[stream] name // set the created table[stream] name
@ -7791,7 +7788,7 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf
} }
} }
// todo derivate funtion requires ts column exists in subquery // todo derivative function requires ts column exists in subquery
STableMeta* pTableMeta = tscGetMetaInfo(pQueryInfo, 0)->pTableMeta; STableMeta* pTableMeta = tscGetMetaInfo(pQueryInfo, 0)->pTableMeta;
SSchema* pSchema = tscGetTableColumnSchema(pTableMeta, 0); SSchema* pSchema = tscGetTableColumnSchema(pTableMeta, 0);
@ -7884,7 +7881,7 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf
TSDB_CODE_SUCCESS) { TSDB_CODE_SUCCESS) {
return TSDB_CODE_TSC_INVALID_OPERATION; return TSDB_CODE_TSC_INVALID_OPERATION;
} }
// parse the window_state // parse the window_state
if (validateStateWindowNode(pCmd, pQueryInfo, pSqlNode, isSTable) != TSDB_CODE_SUCCESS) { if (validateStateWindowNode(pCmd, pQueryInfo, pSqlNode, isSTable) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_TSC_INVALID_OPERATION; return TSDB_CODE_TSC_INVALID_OPERATION;
} }
@ -8099,7 +8096,7 @@ int32_t exprTreeFromSqlExpr(SSqlCmd* pCmd, tExprNode **pExpr, const tSqlExpr* pS
SColIndex* idx = taosArrayGet(pCols, 0); SColIndex* idx = taosArrayGet(pCols, 0);
SSchema* pSchema = tscGetTableColumnSchema(pTableMeta, idx->colIndex); SSchema* pSchema = tscGetTableColumnSchema(pTableMeta, idx->colIndex);
if (pSchema != NULL) { if (pSchema != NULL) {
colType = pSchema->type; colType = pSchema->type;
} }
} }

24
src/client/src/tscSub.c
View File

@ -283,6 +283,7 @@ static int tscUpdateSubscription(STscObj* pObj, SSub* pSub) {
SArray* tables = getTableList(pSql); SArray* tables = getTableList(pSql);
if (tables == NULL) { if (tables == NULL) {
pSub->lastSyncTime = 0; //force to get table list next time
return 0; return 0;
} }
size_t numOfTables = taosArrayGetSize(tables); size_t numOfTables = taosArrayGetSize(tables);
@ -489,7 +490,15 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
SSub *pSub = (SSub *)tsub; SSub *pSub = (SSub *)tsub;
if (pSub == NULL) return NULL; if (pSub == NULL) return NULL;
if (pSub->pSql->cmd.command == TSDB_SQL_RETRIEVE_EMPTY_RESULT) { if (pSub->pTimer == NULL) {
int64_t duration = taosGetTimestampMs() - pSub->lastConsumeTime;
if (duration < (int64_t)(pSub->interval)) {
tscDebug("subscription consume too frequently, blocking...");
taosMsleep(pSub->interval - (int32_t)duration);
}
}
if (pSub->pSql->cmd.command == TSDB_SQL_RETRIEVE_EMPTY_RESULT) { //may reach here when retireve stable vgroup failed
SSqlObj* pSql = recreateSqlObj(pSub); SSqlObj* pSql = recreateSqlObj(pSub);
if (pSql == NULL) { if (pSql == NULL) {
return NULL; return NULL;
@ -503,6 +512,11 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
pSub->pSql = pSql; pSub->pSql = pSql;
pSql->pSubscription = pSub; pSql->pSubscription = pSub;
// no table list now, force to update it
tscDebug("begin table synchronization");
if (!tscUpdateSubscription(pSub->taos, pSub)) return NULL;
tscDebug("table synchronization completed");
} }
tscSaveSubscriptionProgress(pSub); tscSaveSubscriptionProgress(pSub);
@ -527,14 +541,6 @@ TAOS_RES *taos_consume(TAOS_SUB *tsub) {
tscDebug("subscribe:%s set next round subscribe skey:%"PRId64, pSub->topic, pQueryInfo->window.skey); tscDebug("subscribe:%s set next round subscribe skey:%"PRId64, pSub->topic, pQueryInfo->window.skey);
} }
if (pSub->pTimer == NULL) {
int64_t duration = taosGetTimestampMs() - pSub->lastConsumeTime;
if (duration < (int64_t)(pSub->interval)) {
tscDebug("subscription consume too frequently, blocking...");
taosMsleep(pSub->interval - (int32_t)duration);
}
}
size_t size = taosArrayGetSize(pSub->progress) * sizeof(STableIdInfo); size_t size = taosArrayGetSize(pSub->progress) * sizeof(STableIdInfo);
size += sizeof(SQueryTableMsg) + 4096; size += sizeof(SQueryTableMsg) + 4096;
int code = tscAllocPayload(&pSql->cmd, (int)size); int code = tscAllocPayload(&pSql->cmd, (int)size);

8
src/client/src/tscSubquery.c
View File

@ -109,7 +109,7 @@ bool subAndCheckDone(SSqlObj *pSql, SSqlObj *pParentSql, int idx) {
// pthread_mutex_unlock(&subState->mutex); // pthread_mutex_unlock(&subState->mutex);
// return false; // return false;
// } // }
tscDebug("0x%"PRIx64" subquery:0x%"PRIx64", index:%d state set to 1", pParentSql->self, pSql->self, idx); tscDebug("0x%"PRIx64" subquery:0x%"PRIx64", index:%d state set to 1", pParentSql->self, pSql->self, idx);
subState->states[idx] = 1; subState->states[idx] = 1;
@ -622,7 +622,7 @@ static int32_t tscLaunchRealSubqueries(SSqlObj* pSql) {
int16_t colId = tscGetJoinTagColIdByUid(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->id.uid); int16_t colId = tscGetJoinTagColIdByUid(&pQueryInfo->tagCond, pTableMetaInfo->pTableMeta->id.uid);
// set the tag column id for executor to extract correct tag value // set the tag column id for executor to extract correct tag value
#ifndef _TD_NINGSI_60 #ifndef _TD_NINGSI_60
pExpr->base.param[0] = (tVariant) {.i64 = colId, .nType = TSDB_DATA_TYPE_BIGINT, .nLen = sizeof(int64_t)}; pExpr->base.param[0] = (tVariant) {.i64 = colId, .nType = TSDB_DATA_TYPE_BIGINT, .nLen = sizeof(int64_t)};
#else #else
pExpr->base.param[0].i64 = colId; pExpr->base.param[0].i64 = colId;
@ -1843,7 +1843,7 @@ int32_t tscCreateJoinSubquery(SSqlObj *pSql, int16_t tableIndex, SJoinSupporter
// refactor as one method // refactor as one method
SQueryInfo *pNewQueryInfo = tscGetQueryInfo(&pNew->cmd); SQueryInfo *pNewQueryInfo = tscGetQueryInfo(&pNew->cmd);
assert(pNewQueryInfo != NULL); assert(pNewQueryInfo != NULL);
pSupporter->colList = pNewQueryInfo->colList; pSupporter->colList = pNewQueryInfo->colList;
pNewQueryInfo->colList = NULL; pNewQueryInfo->colList = NULL;
@ -3158,7 +3158,7 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
pRes->code = TSDB_CODE_TSC_APP_ERROR; pRes->code = TSDB_CODE_TSC_APP_ERROR;
return pRes->code; return pRes->code;
} }
for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) { for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
SSqlObj* pSub = pSql->pSubs[i]; SSqlObj* pSub = pSql->pSubs[i];
SInsertSupporter* pSup = calloc(1, sizeof(SInsertSupporter)); SInsertSupporter* pSup = calloc(1, sizeof(SInsertSupporter));

17
src/client/src/tscUtil.c
View File

@ -460,6 +460,22 @@ bool tscIsTWAQuery(SQueryInfo* pQueryInfo) {
return false; return false;
} }
bool tscIsIrateQuery(SQueryInfo* pQueryInfo) {
size_t numOfExprs = tscNumOfExprs(pQueryInfo);
for (int32_t i = 0; i < numOfExprs; ++i) {
SExprInfo* pExpr = tscExprGet(pQueryInfo, i);
if (pExpr == NULL) {
continue;
}
if (pExpr->base.functionId == TSDB_FUNC_IRATE) {
return true;
}
}
return false;
}
bool tscIsSessionWindowQuery(SQueryInfo* pQueryInfo) { bool tscIsSessionWindowQuery(SQueryInfo* pQueryInfo) {
return pQueryInfo->sessionWindow.gap > 0; return pQueryInfo->sessionWindow.gap > 0;
} }
@ -3477,6 +3493,7 @@ static void tscSubqueryRetrieveCallback(void* param, TAOS_RES* tres, int code) {
if (pSql->res.code == TSDB_CODE_SUCCESS) { if (pSql->res.code == TSDB_CODE_SUCCESS) {
(*pSql->fp)(pParentSql->param, pParentSql, pParentSql->res.numOfRows); (*pSql->fp)(pParentSql->param, pParentSql, pParentSql->res.numOfRows);
} else { } else {
pParentSql->res.code = pSql->res.code;
tscAsyncResultOnError(pParentSql); tscAsyncResultOnError(pParentSql);
} }
} }

2
src/connector/jdbc/CMakeLists.txt
View File

@ -12,4 +12,4 @@ IF (TD_MVN_INSTALLED)
COMMAND mvn -Dmaven.test.skip=true clean -f ${CMAKE_CURRENT_SOURCE_DIR}/pom.xml COMMAND mvn -Dmaven.test.skip=true clean -f ${CMAKE_CURRENT_SOURCE_DIR}/pom.xml
COMMENT "build jdbc driver") COMMENT "build jdbc driver")
ADD_CUSTOM_TARGET(${JDBC_TARGET_NAME} ALL WORKING_DIRECTORY ${EXECUTABLE_OUTPUT_PATH} DEPENDS ${JDBC_CMD_NAME}) ADD_CUSTOM_TARGET(${JDBC_TARGET_NAME} ALL WORKING_DIRECTORY ${EXECUTABLE_OUTPUT_PATH} DEPENDS ${JDBC_CMD_NAME})
ENDIF () ENDIF ()

1
src/connector/jdbc/pom.xml
View File

@ -122,7 +122,6 @@
<exclude>**/FailOverTest.java</exclude> <exclude>**/FailOverTest.java</exclude>
<exclude>**/InvalidResultSetPointerTest.java</exclude> <exclude>**/InvalidResultSetPointerTest.java</exclude>
<exclude>**/RestfulConnectionTest.java</exclude> <exclude>**/RestfulConnectionTest.java</exclude>
<exclude>**/TD4144Test.java</exclude>
<exclude>**/ConnectMultiTaosdByRestfulWithDifferentTokenTest.java</exclude> <exclude>**/ConnectMultiTaosdByRestfulWithDifferentTokenTest.java</exclude>
</excludes> </excludes>
<testFailureIgnore>true</testFailureIgnore> <testFailureIgnore>true</testFailureIgnore>

5
src/connector/jdbc/src/main/java/com/taosdata/jdbc/AbstractConnection.java
View File

@ -1,5 +1,7 @@
package com.taosdata.jdbc; package com.taosdata.jdbc;
import com.taosdata.jdbc.rs.enums.TimestampFormat;
import java.sql.*; import java.sql.*;
import java.util.Enumeration; import java.util.Enumeration;
import java.util.Map; import java.util.Map;
@ -18,7 +20,7 @@ public abstract class AbstractConnection extends WrapperImpl implements Connecti
for (String propName : propNames) { for (String propName : propNames) {
clientInfoProps.setProperty(propName, properties.getProperty(propName)); clientInfoProps.setProperty(propName, properties.getProperty(propName));
} }
String timestampFormat = properties.getProperty(TSDBDriver.PROPERTY_KEY_TIMESTAMP_FORMAT, "STRING"); String timestampFormat = properties.getProperty(TSDBDriver.PROPERTY_KEY_TIMESTAMP_FORMAT, String.valueOf(TimestampFormat.STRING));
clientInfoProps.setProperty(TSDBDriver.PROPERTY_KEY_TIMESTAMP_FORMAT, timestampFormat); clientInfoProps.setProperty(TSDBDriver.PROPERTY_KEY_TIMESTAMP_FORMAT, timestampFormat);
} }
@ -516,7 +518,6 @@ public abstract class AbstractConnection extends WrapperImpl implements Connecti
public void abort(Executor executor) throws SQLException { public void abort(Executor executor) throws SQLException {
if (isClosed()) if (isClosed())
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_CONNECTION_CLOSED); throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_CONNECTION_CLOSED);
// do nothing // do nothing
} }

2
src/connector/jdbc/src/main/java/com/taosdata/jdbc/AbstractStatement.java
View File

@ -9,8 +9,6 @@ public abstract class AbstractStatement extends WrapperImpl implements Statement
protected List<String> batchedArgs; protected List<String> batchedArgs;
private int fetchSize; private int fetchSize;
@Override @Override
public abstract ResultSet executeQuery(String sql) throws SQLException; public abstract ResultSet executeQuery(String sql) throws SQLException;

1
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBError.java
View File

@ -32,6 +32,7 @@ public class TSDBError {
TSDBErrorMap.put(TSDBErrorNumbers.ERROR_INVALID_SQL, "invalid sql"); TSDBErrorMap.put(TSDBErrorNumbers.ERROR_INVALID_SQL, "invalid sql");
TSDBErrorMap.put(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE, "numeric value out of range"); TSDBErrorMap.put(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE, "numeric value out of range");
TSDBErrorMap.put(TSDBErrorNumbers.ERROR_UNKNOWN_TAOS_TYPE_IN_TDENGINE, "unknown taos type in tdengine"); TSDBErrorMap.put(TSDBErrorNumbers.ERROR_UNKNOWN_TAOS_TYPE_IN_TDENGINE, "unknown taos type in tdengine");
TSDBErrorMap.put(TSDBErrorNumbers.ERROR_UNKNOWN_TIMESTAMP_PERCISION, "unknown timestamp precision");
/**************************************************/ /**************************************************/
TSDBErrorMap.put(TSDBErrorNumbers.ERROR_UNKNOWN, "unknown error"); TSDBErrorMap.put(TSDBErrorNumbers.ERROR_UNKNOWN, "unknown error");

2
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBErrorNumbers.java
View File

@ -25,6 +25,7 @@ public class TSDBErrorNumbers {
public static final int ERROR_INVALID_SQL = 0x2313; // invalid sql public static final int ERROR_INVALID_SQL = 0x2313; // invalid sql
public static final int ERROR_NUMERIC_VALUE_OUT_OF_RANGE = 0x2314; // numeric value out of range public static final int ERROR_NUMERIC_VALUE_OUT_OF_RANGE = 0x2314; // numeric value out of range
public static final int ERROR_UNKNOWN_TAOS_TYPE_IN_TDENGINE = 0x2315; //unknown taos type in tdengine public static final int ERROR_UNKNOWN_TAOS_TYPE_IN_TDENGINE = 0x2315; //unknown taos type in tdengine
public static final int ERROR_UNKNOWN_TIMESTAMP_PERCISION = 0x2316; // unknown timestamp precision
public static final int ERROR_UNKNOWN = 0x2350; //unknown error public static final int ERROR_UNKNOWN = 0x2350; //unknown error
@ -62,6 +63,7 @@ public class TSDBErrorNumbers {
errorNumbers.add(ERROR_INVALID_SQL); errorNumbers.add(ERROR_INVALID_SQL);
errorNumbers.add(ERROR_NUMERIC_VALUE_OUT_OF_RANGE); errorNumbers.add(ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
errorNumbers.add(ERROR_UNKNOWN_TAOS_TYPE_IN_TDENGINE); errorNumbers.add(ERROR_UNKNOWN_TAOS_TYPE_IN_TDENGINE);
errorNumbers.add(ERROR_UNKNOWN_TIMESTAMP_PERCISION);
/*****************************************************/ /*****************************************************/
errorNumbers.add(ERROR_SUBSCRIBE_FAILED); errorNumbers.add(ERROR_SUBSCRIBE_FAILED);

44
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBPreparedStatement.java
View File

@ -38,7 +38,6 @@ import java.util.regex.Pattern;
public class TSDBPreparedStatement extends TSDBStatement implements PreparedStatement { public class TSDBPreparedStatement extends TSDBStatement implements PreparedStatement {
private String rawSql; private String rawSql;
private Object[] parameters; private Object[] parameters;
private boolean isPrepared;
private ArrayList<ColumnInfo> colData; private ArrayList<ColumnInfo> colData;
private ArrayList<TableTagInfo> tableTags; private ArrayList<TableTagInfo> tableTags;
@ -47,8 +46,6 @@ public class TSDBPreparedStatement extends TSDBStatement implements PreparedStat
private String tableName; private String tableName;
private long nativeStmtHandle = 0; private long nativeStmtHandle = 0;
private volatile TSDBParameterMetaData parameterMetaData;
TSDBPreparedStatement(TSDBConnection connection, String sql) { TSDBPreparedStatement(TSDBConnection connection, String sql) {
super(connection); super(connection);
init(sql); init(sql);
@ -61,7 +58,6 @@ public class TSDBPreparedStatement extends TSDBStatement implements PreparedStat
} }
} }
parameters = new Object[parameterCnt]; parameters = new Object[parameterCnt];
this.isPrepared = true;
} }
if (parameterCnt > 1) { if (parameterCnt > 1) {
@ -76,11 +72,6 @@ public class TSDBPreparedStatement extends TSDBStatement implements PreparedStat
preprocessSql(); preprocessSql();
} }
@Override
public int[] executeBatch() throws SQLException {
return super.executeBatch();
}
/* /*
* Some of the SQLs sent by other popular frameworks or tools like Spark, contains syntax that cannot be parsed by * Some of the SQLs sent by other popular frameworks or tools like Spark, contains syntax that cannot be parsed by
* the TDengine client. Thus, some simple parsers/filters are intentionally added in this JDBC implementation in * the TDengine client. Thus, some simple parsers/filters are intentionally added in this JDBC implementation in
@ -137,29 +128,15 @@ public class TSDBPreparedStatement extends TSDBStatement implements PreparedStat
/***** for inner queries *****/ /***** for inner queries *****/
} }
/**
* Populate parameters into prepared sql statements
*
* @return a string of the native sql statement for TSDB
*/
private String getNativeSql(String rawSql) throws SQLException {
return Utils.getNativeSql(rawSql, this.parameters);
}
@Override @Override
public ResultSet executeQuery() throws SQLException { public ResultSet executeQuery() throws SQLException {
if (!isPrepared) final String sql = Utils.getNativeSql(this.rawSql, this.parameters);
return executeQuery(this.rawSql);
final String sql = getNativeSql(this.rawSql);
return executeQuery(sql); return executeQuery(sql);
} }
@Override @Override
public int executeUpdate() throws SQLException { public int executeUpdate() throws SQLException {
if (!isPrepared) String sql = Utils.getNativeSql(this.rawSql, this.parameters);
return executeUpdate(this.rawSql);
String sql = getNativeSql(this.rawSql);
return executeUpdate(sql); return executeUpdate(sql);
} }
@ -282,25 +259,14 @@ public class TSDBPreparedStatement extends TSDBStatement implements PreparedStat
@Override @Override
public boolean execute() throws SQLException { public boolean execute() throws SQLException {
if (!isPrepared) final String sql = Utils.getNativeSql(this.rawSql, this.parameters);
return execute(this.rawSql);
final String sql = getNativeSql(this.rawSql);
return execute(sql); return execute(sql);
} }
@Override @Override
public void addBatch() throws SQLException { public void addBatch() throws SQLException {
if (this.batchedArgs == null) { String sql = Utils.getNativeSql(this.rawSql, this.parameters);
batchedArgs = new ArrayList<>(); addBatch(sql);
}
if (!isPrepared) {
addBatch(this.rawSql);
} else {
String sql = this.getConnection().nativeSQL(this.rawSql);
addBatch(sql);
}
} }
@Override @Override

63
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBResultSetRowData.java
View File

@ -147,30 +147,14 @@ public class TSDBResultSetRowData {
case TSDBConstants.TSDB_DATA_TYPE_NCHAR: case TSDBConstants.TSDB_DATA_TYPE_NCHAR:
case TSDBConstants.TSDB_DATA_TYPE_BINARY: case TSDBConstants.TSDB_DATA_TYPE_BINARY:
return Integer.parseInt((String) obj); return Integer.parseInt((String) obj);
case TSDBConstants.TSDB_DATA_TYPE_UTINYINT: { case TSDBConstants.TSDB_DATA_TYPE_UTINYINT:
Byte value = (byte) obj; return parseUnsignedTinyIntToInt(obj);
if (value < 0) case TSDBConstants.TSDB_DATA_TYPE_USMALLINT:
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE); return parseUnsignedSmallIntToInt(obj);
return value; case TSDBConstants.TSDB_DATA_TYPE_UINT:
} return parseUnsignedIntegerToInt(obj);
case TSDBConstants.TSDB_DATA_TYPE_USMALLINT: { case TSDBConstants.TSDB_DATA_TYPE_UBIGINT:
short value = (short) obj; return parseUnsignedBigIntToInt(obj);
if (value < 0)
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
return value;
}
case TSDBConstants.TSDB_DATA_TYPE_UINT: {
int value = (int) obj;
if (value < 0)
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
return value;
}
case TSDBConstants.TSDB_DATA_TYPE_UBIGINT: {
long value = (long) obj;
if (value < 0)
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
return Long.valueOf(value).intValue();
}
case TSDBConstants.TSDB_DATA_TYPE_FLOAT: case TSDBConstants.TSDB_DATA_TYPE_FLOAT:
return ((Float) obj).intValue(); return ((Float) obj).intValue();
case TSDBConstants.TSDB_DATA_TYPE_DOUBLE: case TSDBConstants.TSDB_DATA_TYPE_DOUBLE:
@ -180,6 +164,35 @@ public class TSDBResultSetRowData {
} }
} }
private byte parseUnsignedTinyIntToInt(Object obj) throws SQLException {
byte value = (byte) obj;
if (value < 0)
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
return value;
}
private short parseUnsignedSmallIntToInt(Object obj) throws SQLException {
short value = (short) obj;
if (value < 0)
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
return value;
}
private int parseUnsignedIntegerToInt(Object obj) throws SQLException {
int value = (int) obj;
if (value < 0)
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
return value;
}
private int parseUnsignedBigIntToInt(Object obj) throws SQLException {
long value = (long) obj;
if (value < 0)
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
return Long.valueOf(value).intValue();
}
/** /**
* $$$ this method is invoked by databaseMetaDataResultSet and so on which use a index start from 1 in JDBC api * $$$ this method is invoked by databaseMetaDataResultSet and so on which use a index start from 1 in JDBC api
*/ */
@ -216,7 +229,7 @@ public class TSDBResultSetRowData {
case TSDBConstants.TSDB_DATA_TYPE_BINARY: case TSDBConstants.TSDB_DATA_TYPE_BINARY:
return Long.parseLong((String) obj); return Long.parseLong((String) obj);
case TSDBConstants.TSDB_DATA_TYPE_UTINYINT: { case TSDBConstants.TSDB_DATA_TYPE_UTINYINT: {
Byte value = (byte) obj; byte value = (byte) obj;
if (value < 0) if (value < 0)
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE); throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_NUMERIC_VALUE_OUT_OF_RANGE);
return value; return value;

8
src/connector/jdbc/src/main/java/com/taosdata/jdbc/enums/TimestampPrecision.java Normal file
View File

@ -0,0 +1,8 @@
package com.taosdata.jdbc.enums;
public enum TimestampPrecision {
MS,
US,
NS,
UNKNOWN
}

21
src/connector/jdbc/src/main/java/com/taosdata/jdbc/rs/RestfulPreparedStatement.java
View File

@ -44,7 +44,7 @@ public class RestfulPreparedStatement extends RestfulStatement implements Prepar
if (!isPrepared) if (!isPrepared)
return executeQuery(this.rawSql); return executeQuery(this.rawSql);
final String sql = getNativeSql(this.rawSql); final String sql = Utils.getNativeSql(this.rawSql, this.parameters);
return executeQuery(sql); return executeQuery(sql);
} }
@ -55,20 +55,10 @@ public class RestfulPreparedStatement extends RestfulStatement implements Prepar
if (!isPrepared) if (!isPrepared)
return executeUpdate(this.rawSql); return executeUpdate(this.rawSql);
final String sql = getNativeSql(this.rawSql); final String sql = Utils.getNativeSql(rawSql, this.parameters);
return executeUpdate(sql); return executeUpdate(sql);
} }
/****
* 将rawSql转换成一条可执行的sql语句，使用属性parameters中的变脸进行替换
* 对于insert into ?.? (?,?,?) using ?.? (?,?,?) tags(?, ?, ?) values(?, ?, ?)
* @param rawSql，可能是insert、select或其他，使用?做占位符
* @return
*/
private String getNativeSql(String rawSql) {
return Utils.getNativeSql(rawSql, this.parameters);
}
@Override @Override
public void setNull(int parameterIndex, int sqlType) throws SQLException { public void setNull(int parameterIndex, int sqlType) throws SQLException {
if (isClosed()) if (isClosed())
@ -224,16 +214,13 @@ public class RestfulPreparedStatement extends RestfulStatement implements Prepar
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_STATEMENT_CLOSED); throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_STATEMENT_CLOSED);
if (!isPrepared) if (!isPrepared)
return execute(this.rawSql); return execute(this.rawSql);
final String sql = getNativeSql(rawSql); final String sql = Utils.getNativeSql(rawSql, this.parameters);
return execute(sql); return execute(sql);
} }
@Override @Override
public void addBatch() throws SQLException { public void addBatch() throws SQLException {
if (isClosed()) final String sql = Utils.getNativeSql(rawSql, this.parameters);
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_STATEMENT_CLOSED);
final String sql = getNativeSql(this.rawSql);
addBatch(sql); addBatch(sql);
} }

107
src/connector/jdbc/src/main/java/com/taosdata/jdbc/rs/RestfulResultSet.java
View File

@ -6,6 +6,8 @@ import com.google.common.primitives.Ints;
import com.google.common.primitives.Longs; import com.google.common.primitives.Longs;
import com.google.common.primitives.Shorts; import com.google.common.primitives.Shorts;
import com.taosdata.jdbc.*; import com.taosdata.jdbc.*;
import com.taosdata.jdbc.enums.TimestampPrecision;
import com.taosdata.jdbc.rs.enums.TimestampFormat;
import com.taosdata.jdbc.utils.Utils; import com.taosdata.jdbc.utils.Utils;
import java.math.BigDecimal; import java.math.BigDecimal;
@ -46,6 +48,7 @@ public class RestfulResultSet extends AbstractResultSet implements ResultSet {
columnNames.clear(); columnNames.clear();
columns.clear(); columns.clear();
this.resultSet.clear(); this.resultSet.clear();
this.metaData = new RestfulResultSetMetaData(this.database, null, this);
return; return;
} }
// get head // get head
@ -131,7 +134,6 @@ public class RestfulResultSet extends AbstractResultSet implements ResultSet {
} }
} }
private Object parseColumnData(JSONArray row, int colIndex, int taosType) throws SQLException { private Object parseColumnData(JSONArray row, int colIndex, int taosType) throws SQLException {
switch (taosType) { switch (taosType) {
case TSDBConstants.TSDB_DATA_TYPE_NULL: case TSDBConstants.TSDB_DATA_TYPE_NULL:
@ -150,44 +152,8 @@ public class RestfulResultSet extends AbstractResultSet implements ResultSet {
return row.getFloat(colIndex); return row.getFloat(colIndex);
case TSDBConstants.TSDB_DATA_TYPE_DOUBLE: case TSDBConstants.TSDB_DATA_TYPE_DOUBLE:
return row.getDouble(colIndex); return row.getDouble(colIndex);
case TSDBConstants.TSDB_DATA_TYPE_TIMESTAMP: { case TSDBConstants.TSDB_DATA_TYPE_TIMESTAMP:
if (row.get(colIndex) == null) return parseTimestampColumnData(row, colIndex);
return null;
String timestampFormat = this.statement.getConnection().getClientInfo(TSDBDriver.PROPERTY_KEY_TIMESTAMP_FORMAT);
if ("TIMESTAMP".equalsIgnoreCase(timestampFormat)) {
Long value = row.getLong(colIndex);
//TODO: this implementation has bug if the timestamp bigger than 9999_9999_9999_9
if (value < 1_0000_0000_0000_0L)
return new Timestamp(value);
long epochSec = value / 1000_000l;
long nanoAdjustment = value % 1000_000l * 1000l;
return Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
}
if ("UTC".equalsIgnoreCase(timestampFormat)) {
String value = row.getString(colIndex);
long epochSec = Timestamp.valueOf(value.substring(0, 19).replace("T", " ")).getTime() / 1000;
int fractionalSec = Integer.parseInt(value.substring(20, value.length() - 5));
long nanoAdjustment = 0;
if (value.length() > 28) {
// ms timestamp: yyyy-MM-ddTHH:mm:ss.SSSSSS+0x00
nanoAdjustment = fractionalSec * 1000l;
} else {
// ms timestamp: yyyy-MM-ddTHH:mm:ss.SSS+0x00
nanoAdjustment = fractionalSec * 1000_000l;
}
ZoneOffset zoneOffset = ZoneOffset.of(value.substring(value.length() - 5));
Instant instant = Instant.ofEpochSecond(epochSec, nanoAdjustment).atOffset(zoneOffset).toInstant();
return Timestamp.from(instant);
}
String value = row.getString(colIndex);
if (value.length() <= 23) // ms timestamp: yyyy-MM-dd HH:mm:ss.SSS
return row.getTimestamp(colIndex);
// us timestamp: yyyy-MM-dd HH:mm:ss.SSSSSS
long epochSec = Timestamp.valueOf(value.substring(0, 19)).getTime() / 1000;
long nanoAdjustment = Integer.parseInt(value.substring(20)) * 1000l;
Timestamp timestamp = Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
return timestamp;
}
case TSDBConstants.TSDB_DATA_TYPE_BINARY: case TSDBConstants.TSDB_DATA_TYPE_BINARY:
return row.getString(colIndex) == null ? null : row.getString(colIndex).getBytes(); return row.getString(colIndex) == null ? null : row.getString(colIndex).getBytes();
case TSDBConstants.TSDB_DATA_TYPE_NCHAR: case TSDBConstants.TSDB_DATA_TYPE_NCHAR:
@ -197,7 +163,66 @@ public class RestfulResultSet extends AbstractResultSet implements ResultSet {
} }
} }
public class Field { private Timestamp parseTimestampColumnData(JSONArray row, int colIndex) throws SQLException {
if (row.get(colIndex) == null)
return null;
String tsFormatUpperCase = this.statement.getConnection().getClientInfo(TSDBDriver.PROPERTY_KEY_TIMESTAMP_FORMAT).toUpperCase();
TimestampFormat timestampFormat = TimestampFormat.valueOf(tsFormatUpperCase);
switch (timestampFormat) {
case TIMESTAMP: {
Long value = row.getLong(colIndex);
//TODO: this implementation has bug if the timestamp bigger than 9999_9999_9999_9
if (value < 1_0000_0000_0000_0L)
return new Timestamp(value);
long epochSec = value / 1000_000l;
long nanoAdjustment = value % 1000_000l * 1000l;
return Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
}
case UTC: {
String value = row.getString(colIndex);
long epochSec = Timestamp.valueOf(value.substring(0, 19).replace("T", " ")).getTime() / 1000;
int fractionalSec = Integer.parseInt(value.substring(20, value.length() - 5));
long nanoAdjustment;
if (value.length() > 31) {
// ns timestamp: yyyy-MM-ddTHH:mm:ss.SSSSSSSSS+0x00
nanoAdjustment = fractionalSec;
} else if (value.length() > 28) {
// ms timestamp: yyyy-MM-ddTHH:mm:ss.SSSSSS+0x00
nanoAdjustment = fractionalSec * 1000L;
} else {
// ms timestamp: yyyy-MM-ddTHH:mm:ss.SSS+0x00
nanoAdjustment = fractionalSec * 1000_000L;
}
ZoneOffset zoneOffset = ZoneOffset.of(value.substring(value.length() - 5));
Instant instant = Instant.ofEpochSecond(epochSec, nanoAdjustment).atOffset(zoneOffset).toInstant();
return Timestamp.from(instant);
}
case STRING:
default: {
String value = row.getString(colIndex);
TimestampPrecision precision = Utils.guessTimestampPrecision(value);
if (precision == TimestampPrecision.MS) {
// ms timestamp: yyyy-MM-dd HH:mm:ss.SSS
return row.getTimestamp(colIndex);
}
if (precision == TimestampPrecision.US) {
// us timestamp: yyyy-MM-dd HH:mm:ss.SSSSSS
long epochSec = Timestamp.valueOf(value.substring(0, 19)).getTime() / 1000;
long nanoAdjustment = Integer.parseInt(value.substring(20)) * 1000L;
return Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
}
if (precision == TimestampPrecision.NS) {
// ms timestamp: yyyy-MM-dd HH:mm:ss.SSSSSSSSS
long epochSec = Timestamp.valueOf(value.substring(0, 19)).getTime() / 1000;
long nanoAdjustment = Integer.parseInt(value.substring(20));
return Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
}
throw TSDBError.createSQLException(TSDBErrorNumbers.ERROR_UNKNOWN_TIMESTAMP_PERCISION);
}
}
}
public static class Field {
String name; String name;
int type; int type;
int length; int length;
@ -211,6 +236,7 @@ public class RestfulResultSet extends AbstractResultSet implements ResultSet {
this.note = note; this.note = note;
this.taos_type = taos_type; this.taos_type = taos_type;
} }
} }
@Override @Override
@ -334,6 +360,7 @@ public class RestfulResultSet extends AbstractResultSet implements ResultSet {
wasNull = true; wasNull = true;
return 0; return 0;
} }
wasNull = false; wasNull = false;
if (value instanceof Timestamp) { if (value instanceof Timestamp) {
return ((Timestamp) value).getTime(); return ((Timestamp) value).getTime();

8
src/connector/jdbc/src/main/java/com/taosdata/jdbc/rs/RestfulResultSetMetaData.java
View File

@ -8,20 +8,22 @@ import java.sql.SQLException;
import java.sql.Timestamp; import java.sql.Timestamp;
import java.sql.Types; import java.sql.Types;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
public class RestfulResultSetMetaData extends WrapperImpl implements ResultSetMetaData { public class RestfulResultSetMetaData extends WrapperImpl implements ResultSetMetaData {
private final String database; private final String database;
private ArrayList<RestfulResultSet.Field> fields; private List<RestfulResultSet.Field> fields;
private final RestfulResultSet resultSet; private final RestfulResultSet resultSet;
public RestfulResultSetMetaData(String database, ArrayList<RestfulResultSet.Field> fields, RestfulResultSet resultSet) { public RestfulResultSetMetaData(String database, ArrayList<RestfulResultSet.Field> fields, RestfulResultSet resultSet) {
this.database = database; this.database = database;
this.fields = fields; this.fields = fields == null ? Collections.emptyList() : fields;
this.resultSet = resultSet; this.resultSet = resultSet;
} }
public ArrayList<RestfulResultSet.Field> getFields() { public List<RestfulResultSet.Field> getFields() {
return fields; return fields;
} }

7
src/connector/jdbc/src/main/java/com/taosdata/jdbc/rs/enums/TimestampFormat.java Normal file
View File

@ -0,0 +1,7 @@
package com.taosdata.jdbc.rs.enums;
public enum TimestampFormat {
STRING,
TIMESTAMP,
UTC
}

17
src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/HttpClientPoolUtil.java
View File

@ -16,13 +16,12 @@ import org.apache.http.protocol.HTTP;
import org.apache.http.protocol.HttpContext; import org.apache.http.protocol.HttpContext;
import org.apache.http.util.EntityUtils; import org.apache.http.util.EntityUtils;
import java.nio.charset.Charset; import java.nio.charset.StandardCharsets;
public class HttpClientPoolUtil { public class HttpClientPoolUtil {
private static final String DEFAULT_CONTENT_TYPE = "application/json"; private static final String DEFAULT_CONTENT_TYPE = "application/json";
private static final String DEFAULT_TOKEN = "cm9vdDp0YW9zZGF0YQ==";
private static final int DEFAULT_TIME_OUT = 15000; private static final int DEFAULT_TIME_OUT = 15000;
private static final int DEFAULT_MAX_PER_ROUTE = 32; private static final int DEFAULT_MAX_PER_ROUTE = 32;
private static final int DEFAULT_MAX_TOTAL = 1000; private static final int DEFAULT_MAX_TOTAL = 1000;
@ -80,7 +79,7 @@ public class HttpClientPoolUtil {
method.setHeader("Connection", "keep-alive"); method.setHeader("Connection", "keep-alive");
method.setHeader("Authorization", "Taosd " + token); method.setHeader("Authorization", "Taosd " + token);
method.setEntity(new StringEntity(data, Charset.forName("UTF-8"))); method.setEntity(new StringEntity(data, StandardCharsets.UTF_8));
HttpContext context = HttpClientContext.create(); HttpContext context = HttpClientContext.create();
CloseableHttpResponse httpResponse = httpClient.execute(method, context); CloseableHttpResponse httpResponse = httpClient.execute(method, context);
httpEntity = httpResponse.getEntity(); httpEntity = httpResponse.getEntity();
@ -165,28 +164,18 @@ public class HttpClientPoolUtil {
httpEntity = httpResponse.getEntity(); httpEntity = httpResponse.getEntity();
if (httpEntity != null) { if (httpEntity != null) {
responseBody = EntityUtils.toString(httpEntity, "UTF-8"); responseBody = EntityUtils.toString(httpEntity, "UTF-8");
// logger.info("请求URL: " + uri + "+ 返回状态码：" + httpResponse.getStatusLine().getStatusCode());
} }
} catch (Exception e) { } catch (Exception e) {
if (method != null) { if (method != null) {
method.abort(); method.abort();
} }
e.printStackTrace(); e.printStackTrace();
// logger.error("execute get request exception, url:" + uri + ", exception:" + e.toString() + ",cost time(ms):"
// + (System.currentTimeMillis() - startTime));
System.out.println("log:调用 HttpClientPoolUtil execute get request exception, url:" + uri + ", exception:" + e.toString() + ",cost time(ms):"
+ (System.currentTimeMillis() - startTime));
} finally { } finally {
if (httpEntity != null) { if (httpEntity != null) {
try { try {
EntityUtils.consumeQuietly(httpEntity); EntityUtils.consumeQuietly(httpEntity);
} catch (Exception e) { } catch (Exception e) {
// e.printStackTrace(); new Exception("close response exception, url:" + uri + ", exception:" + e.toString() + ",cost time(ms):" + (System.currentTimeMillis() - startTime)).printStackTrace();
// logger.error("close response exception, url:" + uri + ", exception:" + e.toString()
// + ",cost time(ms):" + (System.currentTimeMillis() - startTime));
new Exception("close response exception, url:" + uri + ", exception:" + e.toString()
+ ",cost time(ms):" + (System.currentTimeMillis() - startTime))
.printStackTrace();
} }
} }
} }

104
src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/Utils.java
View File

@ -3,14 +3,13 @@ package com.taosdata.jdbc.utils;
import com.google.common.collect.Range; import com.google.common.collect.Range;
import com.google.common.collect.RangeSet; import com.google.common.collect.RangeSet;
import com.google.common.collect.TreeRangeSet; import com.google.common.collect.TreeRangeSet;
import com.taosdata.jdbc.enums.TimestampPrecision;
import java.nio.charset.Charset; import java.nio.charset.Charset;
import java.sql.Date; import java.sql.Date;
import java.sql.Time; import java.sql.Time;
import java.sql.Timestamp; import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder; import java.time.format.DateTimeFormatterBuilder;
import java.time.format.DateTimeParseException; import java.time.format.DateTimeParseException;
@ -25,39 +24,52 @@ public class Utils {
private static Pattern ptn = Pattern.compile(".*?'"); private static Pattern ptn = Pattern.compile(".*?'");
private static final DateTimeFormatter formatter = new DateTimeFormatterBuilder() private static final DateTimeFormatter milliSecFormatter = new DateTimeFormatterBuilder().appendPattern("yyyy-MM-dd HH:mm:ss.SSS").toFormatter();
.appendPattern("yyyy-MM-dd HH:mm:ss.SSS").toFormatter(); private static final DateTimeFormatter microSecFormatter = new DateTimeFormatterBuilder().appendPattern("yyyy-MM-dd HH:mm:ss.SSSSSS").toFormatter();
private static final DateTimeFormatter formatter2 = new DateTimeFormatterBuilder() private static final DateTimeFormatter nanoSecFormatter = new DateTimeFormatterBuilder().appendPattern("yyyy-MM-dd HH:mm:ss.SSSSSSSSS").toFormatter();
.appendPattern("yyyy-MM-dd HH:mm:ss.SSSSSS").toFormatter();
public static Time parseTime(String timestampStr) throws DateTimeParseException { public static Time parseTime(String timestampStr) throws DateTimeParseException {
LocalTime time; LocalDateTime dateTime = parseLocalDateTime(timestampStr);
try { return dateTime != null ? Time.valueOf(dateTime.toLocalTime()) : null;
time = LocalTime.parse(timestampStr, formatter);
} catch (DateTimeParseException e) {
time = LocalTime.parse(timestampStr, formatter2);
}
return Time.valueOf(time);
} }
public static Date parseDate(String timestampStr) throws DateTimeParseException { public static Date parseDate(String timestampStr) {
LocalDate date; LocalDateTime dateTime = parseLocalDateTime(timestampStr);
try { return dateTime != null ? Date.valueOf(String.valueOf(dateTime)) : null;
date = LocalDate.parse(timestampStr, formatter);
} catch (DateTimeParseException e) {
date = LocalDate.parse(timestampStr, formatter2);
}
return Date.valueOf(date);
} }
public static Timestamp parseTimestamp(String timeStampStr) { public static Timestamp parseTimestamp(String timeStampStr) {
LocalDateTime dateTime; LocalDateTime dateTime = parseLocalDateTime(timeStampStr);
return dateTime != null ? Timestamp.valueOf(dateTime) : null;
}
private static LocalDateTime parseLocalDateTime(String timeStampStr) {
try { try {
dateTime = LocalDateTime.parse(timeStampStr, formatter); return parseMilliSecTimestamp(timeStampStr);
} catch (DateTimeParseException e) { } catch (DateTimeParseException e) {
dateTime = LocalDateTime.parse(timeStampStr, formatter2); try {
return parseMicroSecTimestamp(timeStampStr);
} catch (DateTimeParseException ee) {
try {
return parseNanoSecTimestamp(timeStampStr);
} catch (DateTimeParseException eee) {
eee.printStackTrace();
}
}
} }
return Timestamp.valueOf(dateTime); return null;
}
private static LocalDateTime parseMilliSecTimestamp(String timeStampStr) throws DateTimeParseException {
return LocalDateTime.parse(timeStampStr, milliSecFormatter);
}
private static LocalDateTime parseMicroSecTimestamp(String timeStampStr) throws DateTimeParseException {
return LocalDateTime.parse(timeStampStr, microSecFormatter);
}
private static LocalDateTime parseNanoSecTimestamp(String timeStampStr) throws DateTimeParseException {
return LocalDateTime.parse(timeStampStr, nanoSecFormatter);
} }
public static String escapeSingleQuota(String origin) { public static String escapeSingleQuota(String origin) {
@ -93,6 +105,8 @@ public class Utils {
} }
public static String getNativeSql(String rawSql, Object[] parameters) { public static String getNativeSql(String rawSql, Object[] parameters) {
if (parameters == null || !rawSql.contains("?"))
return rawSql;
// toLowerCase // toLowerCase
String preparedSql = rawSql.trim().toLowerCase(); String preparedSql = rawSql.trim().toLowerCase();
String[] clause = new String[]{"values\\s*\\(.*?\\)", "tags\\s*\\(.*?\\)", "where\\s*.*"}; String[] clause = new String[]{"values\\s*\\(.*?\\)", "tags\\s*\\(.*?\\)", "where\\s*.*"};
@ -167,13 +181,47 @@ public class Utils {
}).collect(Collectors.joining()); }).collect(Collectors.joining());
} }
public static String formatTimestamp(Timestamp timestamp) { public static String formatTimestamp(Timestamp timestamp) {
int nanos = timestamp.getNanos(); int nanos = timestamp.getNanos();
if (nanos % 1000000l != 0) if (nanos % 1000000l != 0)
return timestamp.toLocalDateTime().format(formatter2); return timestamp.toLocalDateTime().format(microSecFormatter);
return timestamp.toLocalDateTime().format(formatter); return timestamp.toLocalDateTime().format(milliSecFormatter);
} }
public static TimestampPrecision guessTimestampPrecision(String value) {
if (isMilliSecFormat(value))
return TimestampPrecision.MS;
if (isMicroSecFormat(value))
return TimestampPrecision.US;
if (isNanoSecFormat(value))
return TimestampPrecision.NS;
return TimestampPrecision.UNKNOWN;
}
private static boolean isMilliSecFormat(String timestampStr) {
try {
milliSecFormatter.parse(timestampStr);
} catch (DateTimeParseException e) {
return false;
}
return true;
}
private static boolean isMicroSecFormat(String timestampStr) {
try {
microSecFormatter.parse(timestampStr);
} catch (DateTimeParseException e) {
return false;
}
return true;
}
private static boolean isNanoSecFormat(String timestampStr) {
try {
nanoSecFormatter.parse(timestampStr);
} catch (DateTimeParseException e) {
return false;
}
return true;
}
} }

218
src/connector/jdbc/src/test/java/com/taosdata/jdbc/TSDBPreparedStatementTest.java
View File

@ -5,7 +5,6 @@ import org.junit.*;
import java.io.IOException; import java.io.IOException;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.sql.*; import java.sql.*;
import java.time.LocalTime;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Random; import java.util.Random;
@ -15,6 +14,7 @@ public class TSDBPreparedStatementTest {
private static Connection conn; private static Connection conn;
private static final String sql_insert = "insert into t1 values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; private static final String sql_insert = "insert into t1 values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String sql_select = "select * from t1 where ts >= ? and ts < ? and f1 >= ?"; private static final String sql_select = "select * from t1 where ts >= ? and ts < ? and f1 >= ?";
private static final String dbname = "test_pstmt_jni";
private PreparedStatement pstmt_insert; private PreparedStatement pstmt_insert;
private PreparedStatement pstmt_select; private PreparedStatement pstmt_select;
@ -299,53 +299,53 @@ public class TSDBPreparedStatementTest {
} }
@Test @Test
public void executeTest() throws SQLException { public void executeTest() throws SQLException {
Statement stmt = conn.createStatement(); Statement stmt = conn.createStatement();
int numOfRows = 1000; int numOfRows = 1000;
for (int loop = 0; loop < 10; loop++){ for (int loop = 0; loop < 10; loop++) {
stmt.execute("drop table if exists weather_test"); stmt.execute("drop table if exists weather_test");
stmt.execute("create table weather_test(ts timestamp, f1 nchar(4), f2 float, f3 double, f4 timestamp, f5 int, f6 bool, f7 binary(10))"); stmt.execute("create table weather_test(ts timestamp, f1 nchar(4), f2 float, f3 double, f4 timestamp, f5 int, f6 bool, f7 binary(10))");
TSDBPreparedStatement s = (TSDBPreparedStatement) conn.prepareStatement("insert into ? values(?, ?, ?, ?, ?, ?, ?, ?)"); TSDBPreparedStatement s = (TSDBPreparedStatement) conn.prepareStatement("insert into ? values(?, ?, ?, ?, ?, ?, ?, ?)");
Random r = new Random(); Random r = new Random();
s.setTableName("weather_test"); s.setTableName("weather_test");
ArrayList<Long> ts = new ArrayList<Long>(); ArrayList<Long> ts = new ArrayList<Long>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
ts.add(System.currentTimeMillis() + i); ts.add(System.currentTimeMillis() + i);
} }
s.setTimestamp(0, ts); s.setTimestamp(0, ts);
int random = 10 + r.nextInt(5); int random = 10 + r.nextInt(5);
ArrayList<String> s2 = new ArrayList<String>(); ArrayList<String> s2 = new ArrayList<String>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
s2.add(null); s2.add(null);
}else{ } else {
s2.add("分支" + i % 4); s2.add("分支" + i % 4);
} }
} }
s.setNString(1, s2, 4); s.setNString(1, s2, 4);
random = 10 + r.nextInt(5); random = 10 + r.nextInt(5);
ArrayList<Float> s3 = new ArrayList<Float>(); ArrayList<Float> s3 = new ArrayList<Float>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
s3.add(null); s3.add(null);
}else{ } else {
s3.add(r.nextFloat()); s3.add(r.nextFloat());
} }
} }
s.setFloat(2, s3); s.setFloat(2, s3);
random = 10 + r.nextInt(5); random = 10 + r.nextInt(5);
ArrayList<Double> s4 = new ArrayList<Double>(); ArrayList<Double> s4 = new ArrayList<Double>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
s4.add(null); s4.add(null);
}else{ } else {
s4.add(r.nextDouble()); s4.add(r.nextDouble());
} }
} }
@ -353,47 +353,47 @@ public class TSDBPreparedStatementTest {
random = 10 + r.nextInt(5); random = 10 + r.nextInt(5);
ArrayList<Long> ts2 = new ArrayList<Long>(); ArrayList<Long> ts2 = new ArrayList<Long>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
ts2.add(null); ts2.add(null);
}else{ } else {
ts2.add(System.currentTimeMillis() + i); ts2.add(System.currentTimeMillis() + i);
} }
} }
s.setTimestamp(4, ts2); s.setTimestamp(4, ts2);
random = 10 + r.nextInt(5); random = 10 + r.nextInt(5);
ArrayList<Integer> vals = new ArrayList<>(); ArrayList<Integer> vals = new ArrayList<>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
vals.add(null); vals.add(null);
}else{ } else {
vals.add(r.nextInt()); vals.add(r.nextInt());
} }
} }
s.setInt(5, vals); s.setInt(5, vals);
random = 10 + r.nextInt(5); random = 10 + r.nextInt(5);
ArrayList<Boolean> sb = new ArrayList<>(); ArrayList<Boolean> sb = new ArrayList<>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
sb.add(null); sb.add(null);
}else{ } else {
sb.add(i % 2 == 0 ? true : false); sb.add(i % 2 == 0 ? true : false);
} }
} }
s.setBoolean(6, sb); s.setBoolean(6, sb);
random = 10 + r.nextInt(5); random = 10 + r.nextInt(5);
ArrayList<String> s5 = new ArrayList<String>(); ArrayList<String> s5 = new ArrayList<String>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
s5.add(null); s5.add(null);
}else{ } else {
s5.add("test" + i % 10); s5.add("test" + i % 10);
} }
} }
s.setString(7, s5, 10); s.setString(7, s5, 10);
s.columnDataAddBatch(); s.columnDataAddBatch();
s.columnDataExecuteBatch(); s.columnDataExecuteBatch();
@ -403,54 +403,54 @@ public class TSDBPreparedStatementTest {
PreparedStatement statement = conn.prepareStatement(sql); PreparedStatement statement = conn.prepareStatement(sql);
ResultSet rs = statement.executeQuery(); ResultSet rs = statement.executeQuery();
int rows = 0; int rows = 0;
while(rs.next()) { while (rs.next()) {
rows++; rows++;
} }
Assert.assertEquals(numOfRows, rows); Assert.assertEquals(numOfRows, rows);
} }
} }
@Test @Test
public void bindDataSelectColumnTest() throws SQLException { public void bindDataSelectColumnTest() throws SQLException {
Statement stmt = conn.createStatement(); Statement stmt = conn.createStatement();
int numOfRows = 1000; int numOfRows = 1000;
for (int loop = 0; loop < 10; loop++){ for (int loop = 0; loop < 10; loop++) {
stmt.execute("drop table if exists weather_test"); stmt.execute("drop table if exists weather_test");
stmt.execute("create table weather_test(ts timestamp, f1 nchar(4), f2 float, f3 double, f4 timestamp, f5 int, f6 bool, f7 binary(10))"); stmt.execute("create table weather_test(ts timestamp, f1 nchar(4), f2 float, f3 double, f4 timestamp, f5 int, f6 bool, f7 binary(10))");
TSDBPreparedStatement s = (TSDBPreparedStatement) conn.prepareStatement("insert into ? (ts, f1, f7) values(?, ?, ?)"); TSDBPreparedStatement s = (TSDBPreparedStatement) conn.prepareStatement("insert into ? (ts, f1, f7) values(?, ?, ?)");
Random r = new Random(); Random r = new Random();
s.setTableName("weather_test"); s.setTableName("weather_test");
ArrayList<Long> ts = new ArrayList<Long>(); ArrayList<Long> ts = new ArrayList<Long>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
ts.add(System.currentTimeMillis() + i); ts.add(System.currentTimeMillis() + i);
} }
s.setTimestamp(0, ts); s.setTimestamp(0, ts);
int random = 10 + r.nextInt(5); int random = 10 + r.nextInt(5);
ArrayList<String> s2 = new ArrayList<String>(); ArrayList<String> s2 = new ArrayList<String>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
s2.add(null); s2.add(null);
}else{ } else {
s2.add("分支" + i % 4); s2.add("分支" + i % 4);
} }
} }
s.setNString(1, s2, 4); s.setNString(1, s2, 4);
random = 10 + r.nextInt(5); random = 10 + r.nextInt(5);
ArrayList<String> s3 = new ArrayList<String>(); ArrayList<String> s3 = new ArrayList<String>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
if(i % random == 0) { if (i % random == 0) {
s3.add(null); s3.add(null);
}else{ } else {
s3.add("test" + i % 10); s3.add("test" + i % 10);
} }
} }
s.setString(2, s3, 10); s.setString(2, s3, 10);
s.columnDataAddBatch(); s.columnDataAddBatch();
s.columnDataExecuteBatch(); s.columnDataExecuteBatch();
@ -460,30 +460,30 @@ public class TSDBPreparedStatementTest {
PreparedStatement statement = conn.prepareStatement(sql); PreparedStatement statement = conn.prepareStatement(sql);
ResultSet rs = statement.executeQuery(); ResultSet rs = statement.executeQuery();
int rows = 0; int rows = 0;
while(rs.next()) { while (rs.next()) {
rows++; rows++;
} }
Assert.assertEquals(numOfRows, rows); Assert.assertEquals(numOfRows, rows);
} }
} }
@Test @Test
public void bindDataWithSingleTagTest() throws SQLException { public void bindDataWithSingleTagTest() throws SQLException {
Statement stmt = conn.createStatement(); Statement stmt = conn.createStatement();
String types[] = new String[] {"tinyint", "smallint", "int", "bigint", "bool", "float", "double", "binary(10)", "nchar(10)"}; String types[] = new String[]{"tinyint", "smallint", "int", "bigint", "bool", "float", "double", "binary(10)", "nchar(10)"};
for (String type : types) { for (String type : types) {
stmt.execute("drop table if exists weather_test"); stmt.execute("drop table if exists weather_test");
stmt.execute("create table weather_test(ts timestamp, f1 nchar(10), f2 binary(10)) tags (t " + type + ")"); stmt.execute("create table weather_test(ts timestamp, f1 nchar(10), f2 binary(10)) tags (t " + type + ")");
int numOfRows = 1; int numOfRows = 1;
TSDBPreparedStatement s = (TSDBPreparedStatement) conn.prepareStatement("insert into ? using weather_test tags(?) values(?, ?, ?)"); TSDBPreparedStatement s = (TSDBPreparedStatement) conn.prepareStatement("insert into ? using weather_test tags(?) values(?, ?, ?)");
Random r = new Random(); Random r = new Random();
s.setTableName("w1"); s.setTableName("w1");
switch(type) { switch (type) {
case "tinyint": case "tinyint":
case "smallint": case "smallint":
case "int": case "int":
@ -508,37 +508,37 @@ public class TSDBPreparedStatementTest {
default: default:
break; break;
} }
ArrayList<Long> ts = new ArrayList<Long>(); ArrayList<Long> ts = new ArrayList<Long>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
ts.add(System.currentTimeMillis() + i); ts.add(System.currentTimeMillis() + i);
} }
s.setTimestamp(0, ts); s.setTimestamp(0, ts);
int random = 10 + r.nextInt(5); int random = 10 + r.nextInt(5);
ArrayList<String> s2 = new ArrayList<String>(); ArrayList<String> s2 = new ArrayList<String>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
s2.add("分支" + i % 4); s2.add("分支" + i % 4);
} }
s.setNString(1, s2, 10); s.setNString(1, s2, 10);
random = 10 + r.nextInt(5); random = 10 + r.nextInt(5);
ArrayList<String> s3 = new ArrayList<String>(); ArrayList<String> s3 = new ArrayList<String>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
s3.add("test" + i % 4); s3.add("test" + i % 4);
} }
s.setString(2, s3, 10); s.setString(2, s3, 10);
s.columnDataAddBatch(); s.columnDataAddBatch();
s.columnDataExecuteBatch(); s.columnDataExecuteBatch();
s.columnDataCloseBatch(); s.columnDataCloseBatch();
String sql = "select * from weather_test"; String sql = "select * from weather_test";
PreparedStatement statement = conn.prepareStatement(sql); PreparedStatement statement = conn.prepareStatement(sql);
ResultSet rs = statement.executeQuery(); ResultSet rs = statement.executeQuery();
int rows = 0; int rows = 0;
while(rs.next()) { while (rs.next()) {
rows++; rows++;
} }
Assert.assertEquals(numOfRows, rows); Assert.assertEquals(numOfRows, rows);
@ -547,32 +547,32 @@ public class TSDBPreparedStatementTest {
@Test @Test
public void bindDataWithMultipleTagsTest() throws SQLException { public void bindDataWithMultipleTagsTest() throws SQLException {
Statement stmt = conn.createStatement(); Statement stmt = conn.createStatement();
stmt.execute("drop table if exists weather_test"); stmt.execute("drop table if exists weather_test");
stmt.execute("create table weather_test(ts timestamp, f1 nchar(10), f2 binary(10)) tags (t1 int, t2 binary(10))"); stmt.execute("create table weather_test(ts timestamp, f1 nchar(10), f2 binary(10)) tags (t1 int, t2 binary(10))");
int numOfRows = 1; int numOfRows = 1;
TSDBPreparedStatement s = (TSDBPreparedStatement) conn.prepareStatement("insert into ? using weather_test tags(?,?) (ts, f2) values(?, ?)"); TSDBPreparedStatement s = (TSDBPreparedStatement) conn.prepareStatement("insert into ? using weather_test tags(?,?) (ts, f2) values(?, ?)");
s.setTableName("w2"); s.setTableName("w2");
s.setTagInt(0, 1); s.setTagInt(0, 1);
s.setTagString(1, "test"); s.setTagString(1, "test");
ArrayList<Long> ts = new ArrayList<Long>(); ArrayList<Long> ts = new ArrayList<Long>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
ts.add(System.currentTimeMillis() + i); ts.add(System.currentTimeMillis() + i);
} }
s.setTimestamp(0, ts); s.setTimestamp(0, ts);
ArrayList<String> s2 = new ArrayList<String>(); ArrayList<String> s2 = new ArrayList<String>();
for(int i = 0; i < numOfRows; i++) { for (int i = 0; i < numOfRows; i++) {
s2.add("test" + i % 4); s2.add("test" + i % 4);
} }
s.setString(1, s2, 10); s.setString(1, s2, 10);
s.columnDataAddBatch(); s.columnDataAddBatch();
s.columnDataExecuteBatch(); s.columnDataExecuteBatch();
s.columnDataCloseBatch(); s.columnDataCloseBatch();
@ -581,21 +581,20 @@ public class TSDBPreparedStatementTest {
PreparedStatement statement = conn.prepareStatement(sql); PreparedStatement statement = conn.prepareStatement(sql);
ResultSet rs = statement.executeQuery(); ResultSet rs = statement.executeQuery();
int rows = 0; int rows = 0;
while(rs.next()) { while (rs.next()) {
rows++; rows++;
} }
Assert.assertEquals(numOfRows, rows); Assert.assertEquals(numOfRows, rows);
} }
@Test @Test(expected = SQLException.class)
public void createTwoSameDbTest() throws SQLException { public void createTwoSameDbTest() throws SQLException {
Statement stmt = conn.createStatement(); // when
Statement stmt = conn.createStatement();
stmt.execute("create database dbtest");
stmt.execute("create database dbtest"); stmt.execute("create database dbtest");
Assert.assertThrows(SQLException.class, () -> stmt.execute("create database dbtest"));
} }
@Test @Test
public void setBoolean() throws SQLException { public void setBoolean() throws SQLException {
// given // given
@ -1097,9 +1096,9 @@ public class TSDBPreparedStatementTest {
try { try {
conn = DriverManager.getConnection("jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata"); conn = DriverManager.getConnection("jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata");
try (Statement stmt = conn.createStatement()) { try (Statement stmt = conn.createStatement()) {
stmt.execute("drop database if exists test_pstmt_jni"); stmt.execute("drop database if exists " + dbname);
stmt.execute("create database if not exists test_pstmt_jni"); stmt.execute("create database if not exists " + dbname);
stmt.execute("use test_pstmt_jni"); stmt.execute("use " + dbname);
} }
} catch (SQLException e) { } catch (SQLException e) {
e.printStackTrace(); e.printStackTrace();
@ -1109,6 +1108,9 @@ public class TSDBPreparedStatementTest {
@AfterClass @AfterClass
public static void afterClass() { public static void afterClass() {
try { try {
Statement statement = conn.createStatement();
statement.execute("drop database if exists " + dbname);
statement.close();
if (conn != null) if (conn != null)
conn.close(); conn.close();
} catch (SQLException e) { } catch (SQLException e) {

62
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/TD4174Test.java → src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/DoubleQuoteInSqlTest.java
View File

@ -7,56 +7,64 @@ import org.junit.*;
import java.sql.*; import java.sql.*;
import java.util.Properties; import java.util.Properties;
public class TD4174Test { public class DoubleQuoteInSqlTest {
private Connection conn;
private static final String host = "127.0.0.1"; private static final String host = "127.0.0.1";
private static final String dbname = "td4174";
private Connection conn;
@Test @Test
public void test() { public void test() {
// given
long ts = System.currentTimeMillis(); long ts = System.currentTimeMillis();
try (PreparedStatement pstmt = conn.prepareStatement("insert into weather values(" + ts + ", ?)")) {
JSONObject value = new JSONObject();
value.put("name", "John Smith");
value.put("age", 20);
Assert.assertEquals("{\"name\":\"John Smith\",\"age\":20}",value.toJSONString());
pstmt.setString(1, value.toJSONString());
int ret = pstmt.executeUpdate();
Assert.assertEquals(1, ret);
} catch (SQLException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
JSONObject value = new JSONObject(); JSONObject value = new JSONObject();
value.put("name", "John Smith"); value.put("name", "John Smith");
value.put("age", 20); value.put("age", 20);
System.out.println(value.toJSONString());
// when
int ret = 0;
try (PreparedStatement pstmt = conn.prepareStatement("insert into weather values(" + ts + ", ?)")) {
pstmt.setString(1, value.toJSONString());
ret = pstmt.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertEquals("{\"name\":\"John Smith\",\"age\":20}", value.toJSONString());
Assert.assertEquals(1, ret);
} }
@Before @Before
public void before() throws SQLException { public void before() {
String url = "jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata"; String url = "jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata";
Properties properties = new Properties(); Properties properties = new Properties();
properties.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8"); properties.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
properties.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "en_US.UTF-8"); properties.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "en_US.UTF-8");
properties.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8"); properties.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
conn = DriverManager.getConnection(url, properties); try {
try (Statement stmt = conn.createStatement()) { conn = DriverManager.getConnection(url, properties);
stmt.execute("drop database if exists td4174"); Statement stmt = conn.createStatement();
stmt.execute("create database if not exists td4174"); stmt.execute("drop database if exists " + dbname);
stmt.execute("use td4174"); stmt.execute("create database if not exists " + dbname);
stmt.execute("use " + dbname);
stmt.execute("create table weather(ts timestamp, text binary(64))"); stmt.execute("create table weather(ts timestamp, text binary(64))");
} catch (SQLException e) {
e.printStackTrace();
} }
} }
@After @After
public void after() throws SQLException { public void after() {
if (conn != null) try {
Statement stmt = conn.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.close();
conn.close(); conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
} }
} }

2
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/TwoTypeTimestampPercisionInJniTest.java → src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/MicroSecondPrecisionJNITest.java
View File

@ -10,7 +10,7 @@ import org.junit.Test;
import java.sql.*; import java.sql.*;
import java.util.Properties; import java.util.Properties;
public class TwoTypeTimestampPercisionInJniTest { public class MicroSecondPrecisionJNITest {
private static final String host = "127.0.0.1"; private static final String host = "127.0.0.1";
private static final String ms_timestamp_db = "ms_precision_test"; private static final String ms_timestamp_db = "ms_precision_test";

3
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/TwoTypeTimestampPercisionInRestfulTest.java → src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/MicroSecondPrecisionRestfulTest.java
View File

@ -10,10 +10,9 @@ import org.junit.Test;
import java.sql.*; import java.sql.*;
import java.util.Properties; import java.util.Properties;
public class TwoTypeTimestampPercisionInRestfulTest { public class MicroSecondPrecisionRestfulTest {
private static final String host = "127.0.0.1"; private static final String host = "127.0.0.1";
private static final String ms_timestamp_db = "ms_precision_test"; private static final String ms_timestamp_db = "ms_precision_test";
private static final String us_timestamp_db = "us_precision_test"; private static final String us_timestamp_db = "us_precision_test";
private static final long timestamp1 = System.currentTimeMillis(); private static final long timestamp1 = System.currentTimeMillis();

182
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/NanoSecondTimestampJNITest.java Normal file
View File

@ -0,0 +1,182 @@
package com.taosdata.jdbc.cases;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import java.sql.*;
import java.time.Instant;
import java.util.Random;
public class NanoSecondTimestampJNITest {
private static final String host = "127.0.0.1";
private static final String dbname = "nano_sec_test";
private static final Random random = new Random(System.currentTimeMillis());
private static Connection conn;
@Test
public void insertUsingLongValue() {
// given
long ms = System.currentTimeMillis();
long ns = ms * 1000_000 + random.nextInt(1000_000);
// when
int ret = 0;
try (Statement stmt = conn.createStatement()) {
ret = stmt.executeUpdate("insert into weather(ts, temperature, humidity) values(" + ns + ", 12.3, 4)");
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertEquals(1, ret);
}
@Test
public void insertUsingStringValue() {
// given
// when
int ret = 0;
try (Statement stmt = conn.createStatement()) {
ret = stmt.executeUpdate("insert into weather(ts, temperature, humidity) values('2021-01-01 12:00:00.123456789', 12.3, 4)");
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertEquals(1, ret);
}
@Test
public void insertUsingTimestampValue() {
// given
long epochSec = System.currentTimeMillis() / 1000;
long nanoAdjustment = random.nextInt(1000_000_000);
Timestamp ts = Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
// when
int ret = 0;
String sql = "insert into weather(ts, temperature, humidity) values( ?, ?, ?)";
try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
pstmt.setTimestamp(1, ts);
pstmt.setFloat(2, 12.34f);
pstmt.setInt(3, 55);
ret = pstmt.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertEquals(1, ret);
}
@Test
public void selectUsingLongValue() throws SQLException {
// given
long ms = System.currentTimeMillis();
long ns = ms * 1000_000L + random.nextInt(1000_000);
// when
ResultSet rs = null;
try (Statement stmt = conn.createStatement()) {
stmt.executeUpdate("insert into weather(ts, temperature, humidity) values(" + ns + ", 12.3, 4)");
rs = stmt.executeQuery("select * from weather");
rs.next();
} catch (SQLException e) {
e.printStackTrace();
}
// then
long actual = rs.getLong(1);
Assert.assertEquals(ms, actual);
actual = rs.getLong("ts");
Assert.assertEquals(ms, actual);
}
@Test
public void selectUsingStringValue() throws SQLException {
// given
String timestampStr = "2021-01-01 12:00:00.123456789";
// when
ResultSet rs = null;
try (Statement stmt = conn.createStatement()) {
stmt.executeUpdate("insert into weather(ts, temperature, humidity) values('" + timestampStr + "', 12.3, 4)");
rs = stmt.executeQuery("select * from weather");
rs.next();
} catch (SQLException e) {
e.printStackTrace();
}
// then
String actual = rs.getString(1);
Assert.assertEquals(timestampStr, actual);
actual = rs.getString("ts");
Assert.assertEquals(timestampStr, actual);
}
@Test
public void selectUsingTimestampValue() throws SQLException {
// given
long timeMillis = System.currentTimeMillis();
long epochSec = timeMillis / 1000;
long nanoAdjustment = (timeMillis % 1000) * 1000_000L + random.nextInt(1000_000);
Timestamp ts = Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
// insert one row
String sql = "insert into weather(ts, temperature, humidity) values( ?, ?, ?)";
try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
pstmt.setTimestamp(1, ts);
pstmt.setFloat(2, 12.34f);
pstmt.setInt(3, 55);
pstmt.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
}
// when
ResultSet rs = null;
try (Statement stmt = conn.createStatement()) {
rs = stmt.executeQuery("select * from weather");
rs.next();
} catch (SQLException e) {
e.printStackTrace();
}
// then
Timestamp actual = rs.getTimestamp(1);
Assert.assertEquals(ts, actual);
actual = rs.getTimestamp("ts");
Assert.assertEquals(ts, actual);
Assert.assertEquals(timeMillis, actual.getTime());
Assert.assertEquals(nanoAdjustment, actual.getNanos());
}
@Before
public void before() {
try (Statement stmt = conn.createStatement()) {
stmt.execute("drop table if exists weather");
stmt.execute("create table weather(ts timestamp, temperature float, humidity int)");
} catch (SQLException e) {
e.printStackTrace();
}
}
@BeforeClass
public static void beforeClass() {
final String url = "jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata";
try {
conn = DriverManager.getConnection(url);
Statement stmt = conn.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.execute("create database if not exists " + dbname + " precision 'ns'");
stmt.execute("use " + dbname);
} catch (SQLException e) {
e.printStackTrace();
}
}
}

182
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/NanoSecondTimestampRestfulTest.java Normal file
View File

@ -0,0 +1,182 @@
package com.taosdata.jdbc.cases;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import java.sql.*;
import java.time.Instant;
import java.util.Random;
public class NanoSecondTimestampRestfulTest {
private static final String host = "127.0.0.1";
private static final String dbname = "nano_sec_test";
private static final Random random = new Random(System.currentTimeMillis());
private static Connection conn;
@Test
public void insertUsingLongValue() {
// given
long ms = System.currentTimeMillis();
long ns = ms * 1000_000 + random.nextInt(1000_000);
// when
int ret = 0;
try (Statement stmt = conn.createStatement()) {
ret = stmt.executeUpdate("insert into weather(ts, temperature, humidity) values(" + ns + ", 12.3, 4)");
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertEquals(1, ret);
}
@Test
public void insertUsingStringValue() {
// given
// when
int ret = 0;
try (Statement stmt = conn.createStatement()) {
ret = stmt.executeUpdate("insert into weather(ts, temperature, humidity) values('2021-01-01 12:00:00.123456789', 12.3, 4)");
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertEquals(1, ret);
}
@Test
public void insertUsingTimestampValue() {
// given
long epochSec = System.currentTimeMillis() / 1000;
long nanoAdjustment = random.nextInt(1000_000_000);
Timestamp ts = Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
// when
int ret = 0;
String sql = "insert into weather(ts, temperature, humidity) values( ?, ?, ?)";
try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
pstmt.setTimestamp(1, ts);
pstmt.setFloat(2, 12.34f);
pstmt.setInt(3, 55);
ret = pstmt.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertEquals(1, ret);
}
@Test
public void selectUsingLongValue() throws SQLException {
// given
long ms = System.currentTimeMillis();
long ns = ms * 1000_000L + random.nextInt(1000_000);
// when
ResultSet rs = null;
try (Statement stmt = conn.createStatement()) {
stmt.executeUpdate("insert into weather(ts, temperature, humidity) values(" + ns + ", 12.3, 4)");
rs = stmt.executeQuery("select * from weather");
rs.next();
} catch (SQLException e) {
e.printStackTrace();
}
// then
long actual = rs.getLong(1);
Assert.assertEquals(ms, actual);
actual = rs.getLong("ts");
Assert.assertEquals(ms, actual);
}
@Test
public void selectUsingStringValue() throws SQLException {
// given
String timestampStr = "2021-01-01 12:00:00.123456789";
// when
ResultSet rs = null;
try (Statement stmt = conn.createStatement()) {
stmt.executeUpdate("insert into weather(ts, temperature, humidity) values('" + timestampStr + "', 12.3, 4)");
rs = stmt.executeQuery("select * from weather");
rs.next();
} catch (SQLException e) {
e.printStackTrace();
}
// then
String actual = rs.getString(1);
Assert.assertEquals(timestampStr, actual);
actual = rs.getString("ts");
Assert.assertEquals(timestampStr, actual);
}
@Test
public void selectUsingTimestampValue() throws SQLException {
// given
long timeMillis = System.currentTimeMillis();
long epochSec = timeMillis / 1000;
long nanoAdjustment = (timeMillis % 1000) * 1000_000L + random.nextInt(1000_000);
Timestamp ts = Timestamp.from(Instant.ofEpochSecond(epochSec, nanoAdjustment));
// insert one row
String sql = "insert into weather(ts, temperature, humidity) values( ?, ?, ?)";
try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
pstmt.setTimestamp(1, ts);
pstmt.setFloat(2, 12.34f);
pstmt.setInt(3, 55);
pstmt.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
}
// when
ResultSet rs = null;
try (Statement stmt = conn.createStatement()) {
rs = stmt.executeQuery("select * from weather");
rs.next();
} catch (SQLException e) {
e.printStackTrace();
}
// then
Timestamp actual = rs.getTimestamp(1);
Assert.assertEquals(ts, actual);
actual = rs.getTimestamp("ts");
Assert.assertEquals(ts, actual);
Assert.assertEquals(timeMillis, actual.getTime());
Assert.assertEquals(nanoAdjustment, actual.getNanos());
}
@Before
public void before() {
try (Statement stmt = conn.createStatement()) {
stmt.execute("drop table if exists weather");
stmt.execute("create table weather(ts timestamp, temperature float, humidity int)");
} catch (SQLException e) {
e.printStackTrace();
}
}
@BeforeClass
public static void beforeClass() {
final String url = "jdbc:TAOS-RS://" + host + ":6041/?user=root&password=taosdata";
try {
conn = DriverManager.getConnection(url);
Statement stmt = conn.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.execute("create database if not exists " + dbname + " precision 'ns'");
stmt.execute("use " + dbname);
} catch (SQLException e) {
e.printStackTrace();
}
}
}

2
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/NullValueInResultSetForJdbcJniTest.java → src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/NullValueInResultSetJNITest.java
View File

@ -6,7 +6,7 @@ import org.junit.Test;
import java.sql.*; import java.sql.*;
public class NullValueInResultSetForJdbcJniTest { public class NullValueInResultSetJNITest {
private static final String host = "127.0.0.1"; private static final String host = "127.0.0.1";
Connection conn; Connection conn;

2
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/NullValueInResultSetForJdbcRestfulTest.java → src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/NullValueInResultSetRestfulTest.java
View File

@ -6,7 +6,7 @@ import org.junit.Test;
import java.sql.*; import java.sql.*;
public class NullValueInResultSetForJdbcRestfulTest { public class NullValueInResultSetRestfulTest {
private static final String host = "127.0.0.1"; private static final String host = "127.0.0.1";
Connection conn; Connection conn;

2
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/TD3841Test.java → src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/NullValueInResultSetTest.java
View File

@ -7,7 +7,7 @@ import org.junit.*;
import java.sql.*; import java.sql.*;
import java.util.Properties; import java.util.Properties;
public class TD3841Test { public class NullValueInResultSetTest {
private static final String host = "127.0.0.1"; private static final String host = "127.0.0.1";
private static Properties properties; private static Properties properties;
private static Connection conn_restful; private static Connection conn_restful;

98
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/PreparedStatementBatchInsertJNITest.java Normal file
View File

@ -0,0 +1,98 @@
package com.taosdata.jdbc.cases;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.sql.*;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class PreparedStatementBatchInsertJNITest {
private static final String host = "127.0.0.1";
private static final String dbname = "td4668";
private final Random random = new Random(System.currentTimeMillis());
private Connection conn;
@Test
public void test() {
// given
long ts = System.currentTimeMillis();
List<Object[]> rows = IntStream.range(0, 10).mapToObj(i -> {
Object[] row = new Object[6];
final String groupId = String.format("%02d", random.nextInt(100));
// table name (d + groupId)组合
row[0] = "d" + groupId;
// tag
row[1] = groupId;
// ts
row[2] = ts + i;
// current 电流
row[3] = random.nextFloat();
// voltage 电压
row[4] = Math.random() > 0.5 ? 220 : 380;
// phase 相位
row[5] = random.nextInt(10);
return row;
}).collect(Collectors.toList());
final String sql = "INSERT INTO ? (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (?) VALUES (?,?,?,?)";
// when
try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
for (Object[] row : rows) {
for (int i = 0; i < row.length; i++) {
pstmt.setObject(i + 1, row[i]);
}
pstmt.addBatch();
}
pstmt.executeBatch();
} catch (SQLException e) {
e.printStackTrace();
Assert.fail();
}
// then
try (Statement stmt = conn.createStatement()) {
ResultSet rs = stmt.executeQuery("select * from meters");
int count = 0;
while (rs.next()) {
count++;
}
Assert.assertEquals(10, count);
} catch (SQLException e) {
e.printStackTrace();
}
}
@Before
public void before() {
try {
conn = DriverManager.getConnection("jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata");
Statement stmt = conn.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.execute("create database if not exists " + dbname);
stmt.execute("use " + dbname);
stmt.execute("create table meters(ts timestamp, current float, voltage int, phase int) tags(groupId int)");
} catch (SQLException e) {
e.printStackTrace();
}
}
@After
public void after() {
try {
Statement stmt = conn.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.close();
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}

98
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/PreparedStatementBatchInsertRestfulTest.java Normal file
View File

@ -0,0 +1,98 @@
package com.taosdata.jdbc.cases;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.sql.*;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class PreparedStatementBatchInsertRestfulTest {
private static final String host = "127.0.0.1";
private static final String dbname = "td4668";
private final Random random = new Random(System.currentTimeMillis());
private Connection conn;
@Test
public void test() {
// given
long ts = System.currentTimeMillis();
List<Object[]> rows = IntStream.range(0, 10).mapToObj(i -> {
Object[] row = new Object[6];
final String groupId = String.format("%02d", random.nextInt(100));
// table name (d + groupId)组合
row[0] = "d" + groupId;
// tag
row[1] = groupId;
// ts
row[2] = ts + i;
// current 电流
row[3] = random.nextFloat();
// voltage 电压
row[4] = Math.random() > 0.5 ? 220 : 380;
// phase 相位
row[5] = random.nextInt(10);
return row;
}).collect(Collectors.toList());
final String sql = "INSERT INTO ? (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (?) VALUES (?,?,?,?)";
// when
try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
for (Object[] row : rows) {
for (int i = 0; i < row.length; i++) {
pstmt.setObject(i + 1, row[i]);
}
pstmt.addBatch();
}
pstmt.executeBatch();
} catch (SQLException e) {
e.printStackTrace();
Assert.fail();
}
// then
try (Statement stmt = conn.createStatement()) {
ResultSet rs = stmt.executeQuery("select * from meters");
int count = 0;
while (rs.next()) {
count++;
}
Assert.assertEquals(10, count);
} catch (SQLException e) {
e.printStackTrace();
}
}
@Before
public void before() {
try {
conn = DriverManager.getConnection("jdbc:TAOS-RS://" + host + ":6041/?user=root&password=taosdata");
Statement stmt = conn.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.execute("create database if not exists " + dbname);
stmt.execute("use " + dbname);
stmt.execute("create table meters(ts timestamp, current float, voltage int, phase int) tags(groupId int)");
} catch (SQLException e) {
e.printStackTrace();
}
}
@After
public void after() {
try {
Statement stmt = conn.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.close();
conn.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}

86
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/ResultSetMetaShouldNotBeNullRestfulTest.java Normal file
View File

@ -0,0 +1,86 @@
package com.taosdata.jdbc.cases;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import java.sql.*;
public class ResultSetMetaShouldNotBeNullRestfulTest {
private static final String host = "127.0.0.1";
private static final String dbname = "td4745";
private Connection connection;
@Test
public void testExecuteQuery() {
// given
ResultSetMetaData metaData = null;
int columnCount = -1;
// when
try {
Statement statement = connection.createStatement();
metaData = statement.executeQuery("select * from weather").getMetaData();
columnCount = metaData.getColumnCount();
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertNotNull(metaData);
Assert.assertEquals(0, columnCount);
}
@Test
public void testExecute() {
// given
ResultSetMetaData metaData = null;
int columnCount = -1;
boolean execute = false;
// when
try {
Statement statement = connection.createStatement();
execute = statement.execute("select * from weather");
metaData = statement.getResultSet().getMetaData();
columnCount = metaData.getColumnCount();
} catch (SQLException e) {
e.printStackTrace();
}
// then
Assert.assertEquals(true, execute);
Assert.assertNotNull(metaData);
Assert.assertEquals(0, columnCount);
}
@Before
public void before() {
try {
connection = DriverManager.getConnection("jdbc:TAOS-RS://" + host + ":6041/?user=root&password=taosdata");
Statement stmt = connection.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.execute("create database if not exists " + dbname);
stmt.execute("use " + dbname);
stmt.execute("create table weather (ts timestamp, temperature float)");
stmt.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
@After
public void after() {
try {
Statement stmt = connection.createStatement();
stmt.execute("drop database if exists " + dbname);
stmt.close();
connection.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
}

105
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/TD4144Test.java
View File

@ -1,105 +0,0 @@
package com.taosdata.jdbc.cases;
import com.taosdata.jdbc.TSDBConnection;
import com.taosdata.jdbc.TSDBDriver;
import com.taosdata.jdbc.TSDBResultSet;
import com.taosdata.jdbc.TSDBSubscribe;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.sql.DriverManager;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
public class TD4144Test {
private static TSDBConnection connection;
private static final String host = "127.0.0.1";
private static final String topic = "topic-meter-current-bg-10";
private static final String sql = "select * from meters where current > 10";
private static final String sql2 = "select * from meters where ts >= '2020-08-15 12:20:00.000'";
@Test
public void test() throws SQLException {
TSDBSubscribe subscribe = null;
TSDBResultSet res = null;
boolean hasNext = false;
try {
subscribe = connection.subscribe(topic, sql, false);
int count = 0;
while (true) {
// 等待1秒，避免频繁调用 consume，给服务端造成压力
TimeUnit.SECONDS.sleep(1);
if (res == null) {
// 消费数据
res = subscribe.consume();
hasNext = res.next();
}
if (res == null) {
continue;
}
ResultSetMetaData metaData = res.getMetaData();
int number = 0;
while (hasNext) {
int columnCount = metaData.getColumnCount();
for (int i = 1; i <= columnCount; i++) {
System.out.print(metaData.getColumnLabel(i) + ": " + res.getString(i) + "\t");
}
System.out.println();
count++;
number++;
hasNext = res.next();
if (!hasNext) {
res.close();
res = null;
System.out.println("rows： " + count);
}
if (hasNext == true && number >= 10) {
System.out.println("batch" + number);
break;
}
}
}
} catch (SQLException | InterruptedException throwables) {
throwables.printStackTrace();
} finally {
if (subscribe != null)
subscribe.close(true);
}
}
@BeforeClass
public static void beforeClass() throws SQLException {
Properties properties = new Properties();
properties.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
properties.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
String url = "jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata";
connection = (DriverManager.getConnection(url, properties)).unwrap(TSDBConnection.class);
try (Statement stmt = connection.createStatement()) {
stmt.execute("drop database if exists power");
stmt.execute("create database if not exists power");
stmt.execute("use power");
stmt.execute("create table meters(ts timestamp, current float, voltage int, phase int) tags(location binary(64), groupId int)");
stmt.execute("create table d1001 using meters tags(\"Beijing.Chaoyang\", 2)");
stmt.execute("create table d1002 using meters tags(\"Beijing.Haidian\", 2)");
stmt.execute("insert into d1001 values(\"2020-08-15 12:00:00.000\", 12, 220, 1),(\"2020-08-15 12:10:00.000\", 12.3, 220, 2),(\"2020-08-15 12:20:00.000\", 12.2, 220, 1)");
stmt.execute("insert into d1002 values(\"2020-08-15 12:00:00.000\", 9.9, 220, 1),(\"2020-08-15 12:10:00.000\", 10.3, 220, 1),(\"2020-08-15 12:20:00.000\", 11.2, 220, 1)");
}
}
@AfterClass
public static void afterClass() throws SQLException {
if (connection != null)
connection.close();
}
}

78
src/connector/jdbc/src/test/java/com/taosdata/jdbc/utils/UtilsTest.java
View File

@ -3,22 +3,98 @@ package com.taosdata.jdbc.utils;
import org.junit.Assert; import org.junit.Assert;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*; import java.util.stream.Stream;
public class UtilsTest { public class UtilsTest {
@Test @Test
public void escapeSingleQuota() { public void escapeSingleQuota() {
// given
String s = "'''''a\\'"; String s = "'''''a\\'";
// when
String news = Utils.escapeSingleQuota(s); String news = Utils.escapeSingleQuota(s);
// then
Assert.assertEquals("\\'\\'\\'\\'\\'a\\'", news); Assert.assertEquals("\\'\\'\\'\\'\\'a\\'", news);
// given
s = "\'''''a\\'"; s = "\'''''a\\'";
// when
news = Utils.escapeSingleQuota(s); news = Utils.escapeSingleQuota(s);
// then
Assert.assertEquals("\\'\\'\\'\\'\\'a\\'", news); Assert.assertEquals("\\'\\'\\'\\'\\'a\\'", news);
// given
s = "\'\'\'\''a\\'"; s = "\'\'\'\''a\\'";
// when
news = Utils.escapeSingleQuota(s); news = Utils.escapeSingleQuota(s);
// then
Assert.assertEquals("\\'\\'\\'\\'\\'a\\'", news); Assert.assertEquals("\\'\\'\\'\\'\\'a\\'", news);
} }
@Test
public void getNativeSqlReplaceQuestionMarks() {
// given
String nativeSql = "insert into ?.? (ts, temperature, humidity) using ?.? tags(?,?) values(now, ?, ?)";
Object[] parameters = Stream.of("test", "t1", "test", "weather", "beijing", 1, 12.2, 4).toArray();
// when
String actual = Utils.getNativeSql(nativeSql, parameters);
// then
String expected = "insert into test.t1 (ts, temperature, humidity) using test.weather tags('beijing',1) values(now, 12.2, 4)";
Assert.assertEquals(expected, actual);
}
@Test
public void getNativeSqlReplaceQuestionMarks2() {
// given
String nativeSql = "INSERT INTO ? (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (?) VALUES (?,?,?,?)";
Object[] parameters = Stream.of("d1", 1, 123, 3.14, 220, 4).toArray();
// when
String actual = Utils.getNativeSql(nativeSql, parameters);
// then
String expected = "INSERT INTO d1 (TS,CURRENT,VOLTAGE,PHASE) USING METERS TAGS (1) VALUES (123,3.14,220,4)";
Assert.assertEquals(expected, actual);
}
@Test
public void getNativeSqlReplaceNothing() {
// given
String nativeSql = "insert into test.t1 (ts, temperature, humidity) using test.weather tags('beijing',1) values(now, 12.2, 4)";
// when
String actual = Utils.getNativeSql(nativeSql, null);
// then
Assert.assertEquals(nativeSql, actual);
}
@Test
public void getNativeSqlReplaceNothing2() {
// given
String nativeSql = "insert into test.t1 (ts, temperature, humidity) using test.weather tags('beijing',1) values(now, 12.2, 4)";
Object[] parameters = Stream.of("test", "t1", "test", "weather", "beijing", 1, 12.2, 4).toArray();
// when
String actual = Utils.getNativeSql(nativeSql, parameters);
// then
Assert.assertEquals(nativeSql, actual);
}
@Test
public void getNativeSqlReplaceNothing3() {
// given
String nativeSql = "insert into ?.? (ts, temperature, humidity) using ?.? tags(?,?) values(now, ?, ?)";
// when
String actual = Utils.getNativeSql(nativeSql, null);
// then
Assert.assertEquals(nativeSql, actual);
}
} }

2
src/connector/python/taos/cinterface.py
View File

@ -15,7 +15,7 @@ def _convert_microsecond_to_datetime(micro):
def _convert_nanosecond_to_datetime(nanosec): def _convert_nanosecond_to_datetime(nanosec):
return datetime.datetime.fromtimestamp(nanosec / 1000000000.0) return nanosec
def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, precision=FieldType.C_TIMESTAMP_UNKNOWN): def _crow_timestamp_to_python(data, num_of_rows, nbytes=None, precision=FieldType.C_TIMESTAMP_UNKNOWN):

1
src/dnode/inc/dnodeCfg.h
View File

@ -27,6 +27,7 @@ void dnodeUpdateCfg(SDnodeCfg *cfg);
int32_t dnodeGetDnodeId(); int32_t dnodeGetDnodeId();
void dnodeGetClusterId(char *clusterId); void dnodeGetClusterId(char *clusterId);
void dnodeGetCfg(int32_t *dnodeId, char *clusterId); void dnodeGetCfg(int32_t *dnodeId, char *clusterId);
void dnodeSetDropped();
#ifdef __cplusplus #ifdef __cplusplus
} }

22
src/dnode/src/dnodeCfg.c
View File

@ -21,6 +21,7 @@
static SDnodeCfg tsCfg = {0}; static SDnodeCfg tsCfg = {0};
static pthread_mutex_t tsCfgMutex; static pthread_mutex_t tsCfgMutex;
static int32_t tsDnodeDropped;
static int32_t dnodeReadCfg(); static int32_t dnodeReadCfg();
static int32_t dnodeWriteCfg(); static int32_t dnodeWriteCfg();
@ -34,6 +35,10 @@ int32_t dnodeInitCfg() {
if (ret == 0) { if (ret == 0) {
dInfo("dnode cfg is initialized"); dInfo("dnode cfg is initialized");
} }
if (tsDnodeDropped) {
dInfo("dnode is dropped, exiting");
return -1;
}
return ret; return ret;
} }
@ -44,6 +49,14 @@ void dnodeUpdateCfg(SDnodeCfg *cfg) {
dnodeResetCfg(cfg); dnodeResetCfg(cfg);
} }
void dnodeSetDropped() {
pthread_mutex_lock(&tsCfgMutex);
tsDnodeDropped = 1;
dnodeWriteCfg();
pthread_mutex_unlock(&tsCfgMutex);
}
int32_t dnodeGetDnodeId() { int32_t dnodeGetDnodeId() {
int32_t dnodeId = 0; int32_t dnodeId = 0;
pthread_mutex_lock(&tsCfgMutex); pthread_mutex_lock(&tsCfgMutex);
@ -119,6 +132,14 @@ static int32_t dnodeReadCfg() {
} }
cfg.dnodeId = (int32_t)dnodeId->valueint; cfg.dnodeId = (int32_t)dnodeId->valueint;
cJSON *dnodeDropped = cJSON_GetObjectItem(root, "dnodeDropped");
if (!dnodeDropped || dnodeDropped->type != cJSON_Number) {
dError("failed to read %s, dnodeDropped not found", file);
//goto PARSE_CFG_OVER;
} else {
tsDnodeDropped = (int32_t)dnodeDropped->valueint;
}
cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId"); cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId");
if (!clusterId || clusterId->type != cJSON_String) { if (!clusterId || clusterId->type != cJSON_String) {
dError("failed to read %s, clusterId not found", file); dError("failed to read %s, clusterId not found", file);
@ -154,6 +175,7 @@ static int32_t dnodeWriteCfg() {
len += snprintf(content + len, maxLen - len, "{\n"); len += snprintf(content + len, maxLen - len, "{\n");
len += snprintf(content + len, maxLen - len, " \"dnodeId\": %d,\n", tsCfg.dnodeId); len += snprintf(content + len, maxLen - len, " \"dnodeId\": %d,\n", tsCfg.dnodeId);
len += snprintf(content + len, maxLen - len, " \"dnodeDropped\": %d,\n", tsDnodeDropped);
len += snprintf(content + len, maxLen - len, " \"clusterId\": \"%s\"\n", tsCfg.clusterId); len += snprintf(content + len, maxLen - len, " \"clusterId\": \"%s\"\n", tsCfg.clusterId);
len += snprintf(content + len, maxLen - len, "}\n"); len += snprintf(content + len, maxLen - len, "}\n");

1
src/dnode/src/dnodeVnodes.c
View File

@ -202,6 +202,7 @@ static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
char clusterId[TSDB_CLUSTER_ID_LEN]; char clusterId[TSDB_CLUSTER_ID_LEN];
dnodeGetClusterId(clusterId); dnodeGetClusterId(clusterId);
if (clusterId[0] != '\0') { if (clusterId[0] != '\0') {
dnodeSetDropped();
dError("exit zombie dropped dnode"); dError("exit zombie dropped dnode");
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }

1
src/query/inc/qTableMeta.h
View File

@ -138,7 +138,6 @@ typedef struct SQueryInfo {
bool hasFilter; bool hasFilter;
bool onlyTagQuery; bool onlyTagQuery;
bool orderProjectQuery; bool orderProjectQuery;
// bool diffQuery;
bool stateWindow; bool stateWindow;
} SQueryInfo; } SQueryInfo;

2
src/query/inc/sql.y
View File

@ -176,7 +176,7 @@ cmd ::= ALTER ACCOUNT ids(X) PASS ids(Y) acct_optr(Z). { setCreateAcctSql(p
////////////////////////////// COMPACT STATEMENT ////////////////////////////////////////////// ////////////////////////////// COMPACT STATEMENT //////////////////////////////////////////////
cmd ::= COMPACT VNODES IN LP exprlist(Y) RP. { setCompactVnodeSql(pInfo, TSDB_SQL_COMPACT_VNODE, Y);} cmd ::= COMPACT VNODES IN LP exprlist(Y) RP. { setCompactVnodeSql(pInfo, TSDB_SQL_COMPACT_VNODE, Y);}
// An IDENTIFIER can be a generic identifier, or one of several keywords. // An IDENTIFIER can be a generic identifier, or one of several keywords.
// Any non-standard keyword can also be an identifier. // Any non-standard keyword can also be an identifier.

11
src/query/src/qAggMain.c
View File

@ -3613,16 +3613,7 @@ static void deriv_function(SQLFunctionCtx *pCtx) {
qError("error input type"); qError("error input type");
} }
// initial value is not set yet, all data block are null GET_RES_INFO(pCtx)->numOfRes += notNullElems;
if (!pDerivInfo->valueSet || notNullElems <= 0) {
/*
* 1. current block and blocks before are full of null
* 2. current block may be null value
*/
assert(pCtx->hasNull);
} else {
GET_RES_INFO(pCtx)->numOfRes += notNullElems;
}
} }
#define DIFF_IMPL(ctx, d, type) \ #define DIFF_IMPL(ctx, d, type) \

2
src/query/src/qExecutor.c
View File

@ -1312,6 +1312,8 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SGroupbyOperatorInfo *pIn
} }
} }
// todo opt perf
for (int32_t k = 0; k < pOperator->numOfOutput; ++k) { for (int32_t k = 0; k < pOperator->numOfOutput; ++k) {
pInfo->binfo.pCtx[k].size = 1; pInfo->binfo.pCtx[k].size = 1;
int32_t functionId = pInfo->binfo.pCtx[k].functionId; int32_t functionId = pInfo->binfo.pCtx[k].functionId;

9
tests/pytest/functions/function_derivative.py
View File

@ -128,7 +128,14 @@ class TDTestCase:
def run(self): def run(self):
tdSql.prepare() tdSql.prepare()
self.insertAndCheckData() self.insertAndCheckData()
tdSql.execute("create table st(ts timestamp, c1 int, c2 int) tags(id int)")
tdSql.execute("insert into dev1(ts, c1) using st tags(1) values(now, 1)")
tdSql.error("select derivative(c1, 10s, 0) from (select c1 from st)")
tdSql.query("select diff(c1) from (select derivative(c1, 1s, 0) c1 from dev1)")
tdSql.checkRows(0)
def stop(self): def stop(self):
tdSql.close() tdSql.close()

38
tests/pytest/util/sql.py
View File

@ -18,6 +18,7 @@ import datetime
import inspect import inspect
import psutil import psutil
import shutil import shutil
import pandas as pd
from util.log import * from util.log import *
@ -134,25 +135,32 @@ class TDSql:
return self.cursor.istype(col, dataType) return self.cursor.istype(col, dataType)
def checkData(self, row, col, data): def checkData(self, row, col, data):
self.checkRowCol(row, col) self.checkRowCol(row, col)
if self.queryResult[row][col] != data: if self.queryResult[row][col] != data:
if self.cursor.istype(col, "TIMESTAMP") and self.queryResult[row][col] == datetime.datetime.fromisoformat(data): if self.cursor.istype(col, "TIMESTAMP"):
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % # suppose user want to check nanosecond timestamp if a longer data passed
if (len(data) >= 28):
if pd.to_datetime(self.queryResult[row][col]) == pd.to_datetime(data):
tdLog.info("sql:%s, row:%d col:%d data:%d == expect:%s" %
(self.sql, row, col, self.queryResult[row][col], data))
else:
if self.queryResult[row][col] == datetime.datetime.fromisoformat(data):
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" %
(self.sql, row, col, self.queryResult[row][col], data)) (self.sql, row, col, self.queryResult[row][col], data))
return return
if str(self.queryResult[row][col]) == str(data): if str(self.queryResult[row][col]) == str(data):
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" %
(self.sql, row, col, self.queryResult[row][col], data)) (self.sql, row, col, self.queryResult[row][col], data))
return return
elif isinstance(data, float) and abs(self.queryResult[row][col] - data) <= 0.000001: elif isinstance(data, float) and abs(self.queryResult[row][col] - data) <= 0.000001:
tdLog.info("sql:%s, row:%d col:%d data:%f == expect:%f" % tdLog.info("sql:%s, row:%d col:%d data:%f == expect:%f" %
(self.sql, row, col, self.queryResult[row][col], data)) (self.sql, row, col, self.queryResult[row][col], data))
return return
else: else:
caller = inspect.getframeinfo(inspect.stack()[1][0]) caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, self.sql, row, col, self.queryResult[row][col], data) args = (caller.filename, caller.lineno, self.sql, row, col, self.queryResult[row][col], data)
tdLog.exit("%s(%d) failed: sql:%s row:%d col:%d data:%s != expect:%s" % args) tdLog.exit("%s(%d) failed: sql:%s row:%d col:%d data:%s != expect:%s" % args)
if data is None: if data is None:
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" %
@ -162,11 +170,11 @@ class TDSql:
(self.sql, row, col, self.queryResult[row][col], data)) (self.sql, row, col, self.queryResult[row][col], data))
elif isinstance(data, datetime.date): elif isinstance(data, datetime.date):
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" %
(self.sql, row, col, self.queryResult[row][col], data)) (self.sql, row, col, self.queryResult[row][col], data))
elif isinstance(data, float): elif isinstance(data, float):
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" % tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%s" %
(self.sql, row, col, self.queryResult[row][col], data)) (self.sql, row, col, self.queryResult[row][col], data))
else: else:
tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%d" % tdLog.info("sql:%s, row:%d col:%d data:%s == expect:%d" %
(self.sql, row, col, self.queryResult[row][col], data)) (self.sql, row, col, self.queryResult[row][col], data))
@ -200,7 +208,7 @@ class TDSql:
tdLog.exit("%s(%d) failed: sql:%s, affectedRows:%d != expect:%d" % args) tdLog.exit("%s(%d) failed: sql:%s, affectedRows:%d != expect:%d" % args)
tdLog.info("sql:%s, affectedRows:%d == expect:%d" % (self.sql, self.affectedRows, expectAffectedRows)) tdLog.info("sql:%s, affectedRows:%d == expect:%d" % (self.sql, self.affectedRows, expectAffectedRows))
def taosdStatus(self, state): def taosdStatus(self, state):
tdLog.sleep(5) tdLog.sleep(5)
pstate = 0 pstate = 0
@ -221,7 +229,7 @@ class TDSql:
continue continue
pstate = 0 pstate = 0
break break
args=(pstate,state) args=(pstate,state)
if pstate == state: if pstate == state:
tdLog.info("taosd state is %d == expect:%d" %args) tdLog.info("taosd state is %d == expect:%d" %args)
@ -236,11 +244,11 @@ class TDSql:
tdLog.exit("dir: %s is empty, expect: not empty" %dir) tdLog.exit("dir: %s is empty, expect: not empty" %dir)
else: else:
tdLog.info("dir: %s is empty, expect: empty" %dir) tdLog.info("dir: %s is empty, expect: empty" %dir)
else: else:
if state : if state :
tdLog.info("dir: %s is not empty, expect: not empty" %dir) tdLog.info("dir: %s is not empty, expect: not empty" %dir)
else: else:
tdLog.exit("dir: %s is not empty, expect: empty" %dir) tdLog.exit("dir: %s is not empty, expect: empty" %dir)
else: else:
tdLog.exit("dir: %s doesn't exist" %dir) tdLog.exit("dir: %s doesn't exist" %dir)
def createDir(self, dir): def createDir(self, dir):
@ -250,5 +258,5 @@ class TDSql:
os.makedirs( dir, 755 ) os.makedirs( dir, 755 )
tdLog.info("dir: %s is created" %dir) tdLog.info("dir: %s is created" %dir)
pass pass
tdSql = TDSql() tdSql = TDSql()

9
tests/script/general/parser/function.sim
View File

@ -817,6 +817,9 @@ print ====================> TODO stddev + normal column filter
print ====================> irate print ====================> irate
sql_error select irate(f1) from st1;
sql select irate(f1) from st1 group by tbname;
sql select irate(k) from t1 sql select irate(k) from t1
if $rows != 1 then if $rows != 1 then
return -1 return -1
@ -1073,6 +1076,12 @@ sql insert into t0 values('2020-1-1 1:3:9', 9);
sql insert into t0 values('2020-1-1 1:4:10', 10); sql insert into t0 values('2020-1-1 1:4:10', 10);
sql insert into t1 values('2020-1-1 1:1:2', 2); sql insert into t1 values('2020-1-1 1:1:2', 2);
print ===========================>td-4739
sql select diff(val) from (select derivative(k, 1s, 0) val from t1);
if $rows != 0 then
return -1
endi
sql insert into t1 values('2020-1-1 1:1:4', 20); sql insert into t1 values('2020-1-1 1:1:4', 20);
sql insert into t1 values('2020-1-1 1:1:6', 200); sql insert into t1 values('2020-1-1 1:1:6', 200);
sql insert into t1 values('2020-1-1 1:1:8', 2000); sql insert into t1 values('2020-1-1 1:1:8', 2000);