Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 219e0994 编写于 作者: H Hongze Cheng
浏览文件
操作

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

上级 dee48f94 1fdc1232
隐藏空白更改
内联 并排
Showing with 192 addition and 357 deletion
Jenkinsfile
浏览文件 @ 219e0994
......@@ -45,7 +45,7 @@ def pre_test(){
git pull
git fetch origin +refs/pull/${CHANGE_ID}/merge
git checkout -qf FETCH_HEAD
git --no-pager diff --name-only FETCH_HEAD $(git merge-base FETCH_HEAD develop)|grep -v -E '.*md|//src//connector|Jenkinsfile' || exit 0
git --no-pager diff --name-only FETCH_HEAD $(git merge-base FETCH_HEAD develop)|grep -v -E '.*md|.*src/connector|Jenkinsfile' || exit 0
cd ${WK}
git reset --hard HEAD~10
git checkout develop
......
src/client/src/tscSQLParser.c
浏览文件 @ 219e0994
......@@ -3440,6 +3440,7 @@ static int32_t getTagCondString(tSQLExpr* pExpr, char** str) {
static int32_t getTablenameCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSQLExpr* pTableCond, SStringBuilder* sb) {
const char* msg0 = "invalid table name list";
const char* msg1 = "not string following like";
if (pTableCond == NULL) {
return TSDB_CODE_SUCCESS;
......@@ -3457,6 +3458,10 @@ static int32_t getTablenameCond(SSqlCmd* pCmd, SQueryInfo* pQueryInfo, tSQLExpr*
if (pTableCond->nSQLOptr == TK_IN) {
ret = tablenameListToString(pRight, sb);
} else if (pTableCond->nSQLOptr == TK_LIKE) {
if (pRight->nSQLOptr != TK_STRING) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
}
ret = tablenameCondToString(pRight, sb);
}
......
src/client/src/tscSubquery.c
浏览文件 @ 219e0994
......@@ -2410,7 +2410,7 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
// record the total inserted rows
if (numOfRows > 0) {
pParentObj->res.numOfRows += numOfRows;
atomic_add_fetch_32(&pParentObj->res.numOfRows, numOfRows);
}
if (taos_errno(tres) != TSDB_CODE_SUCCESS) {
......
src/client/src/tscSystem.c
浏览文件 @ 219e0994
......@@ -87,7 +87,7 @@ int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncry
rpcInit.sessions = tsMaxConnections;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.user = (char *)user;
rpcInit.idleTime = 2000;
rpcInit.idleTime = tsShellActivityTimer * 1000;
rpcInit.ckey = "key";
rpcInit.spi = 1;
rpcInit.secret = (char *)secretEncrypt;
......
src/connector/jdbc/pom.xml
浏览文件 @ 219e0994
......@@ -36,6 +36,7 @@
<maven-compiler-plugin.version>3.6.0</maven-compiler-plugin.version>
<commons-logging.version>1.1.2</commons-logging.version>
<commons-lang3.version>3.5</commons-lang3.version>
<maven.test.jvmargs></maven.test.jvmargs>
</properties>
<dependencies>
<dependency>
......@@ -122,11 +123,14 @@
<artifactId>maven-surefire-plugin</artifactId>
<version>2.12.4</version>
<configuration>
<forkMode>pertest</forkMode>
<argLine>${maven.test.jvmargs}</argLine>
<includes>
<include>**/*Test.java</include>
</includes>
<excludes>
<exclude>**/AppMemoryLeakTest.java</exclude>
<exclude>**/TaosInfoMonitorTest.java</exclude>
<exclude>**/FailOverTest.java</exclude>
</excludes>
<testFailureIgnore>true</testFailureIgnore>
......
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBJNIConnector.java
浏览文件 @ 219e0994
......@@ -14,6 +14,8 @@
*****************************************************************************/
package com.taosdata.jdbc;
import com.taosdata.jdbc.utils.TaosInfo;
import java.sql.SQLException;
import java.sql.SQLWarning;
import java.util.List;
......@@ -21,6 +23,8 @@ import java.util.List;
public class TSDBJNIConnector {
private static volatile Boolean isInitialized = false;
private TaosInfo taosInfo = TaosInfo.getInstance();
static {
System.loadLibrary("taos");
System.out.println("java.library.path:" + System.getProperty("java.library.path"));
......@@ -91,7 +95,8 @@ public class TSDBJNIConnector {
*/
public boolean connect(String host, int port, String dbName, String user, String password) throws SQLException {
if (this.taos != TSDBConstants.JNI_NULL_POINTER) {
this.closeConnectionImp(this.taos);
// this.closeConnectionImp(this.taos);
closeConnection();
this.taos = TSDBConstants.JNI_NULL_POINTER;
}
......@@ -99,7 +104,8 @@ public class TSDBJNIConnector {
if (this.taos == TSDBConstants.JNI_NULL_POINTER) {
throw new SQLException(TSDBConstants.WrapErrMsg(this.getErrMsg(0L)), "", this.getErrCode(0l));
}
// invoke connectImp only here
taosInfo.conn_open_increment();
return true;
}
......@@ -120,6 +126,7 @@ public class TSDBJNIConnector {
Long pSql = 0l;
try {
pSql = this.executeQueryImp(sql.getBytes(TaosGlobalConfig.getCharset()), this.taos);
taosInfo.stmt_count_increment();
} catch (Exception e) {
e.printStackTrace();
this.freeResultSetImp(this.taos, pSql);
......@@ -244,10 +251,11 @@ public class TSDBJNIConnector {
private native int fetchRowImp(long connection, long resultSet, TSDBResultSetRowData rowData);
public int fetchBlock(long resultSet, TSDBResultSetBlockData blockData) {
return this.fetchBlockImp(this.taos, resultSet, blockData);
return this.fetchBlockImp(this.taos, resultSet, blockData);
}
private native int fetchBlockImp(long connection, long resultSet, TSDBResultSetBlockData blockData);
/**
* Execute close operation from C to release connection pointer by JNI
*
......@@ -262,6 +270,8 @@ public class TSDBJNIConnector {
} else {
throw new SQLException("Undefined error code returned by TDengine when closing a connection");
}
// invoke closeConnectionImpl only here
taosInfo.connect_close_increment();
}
private native int closeConnectionImp(long connection);
......
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBStatement.java
浏览文件 @ 219e0994
......@@ -14,13 +14,15 @@
*****************************************************************************/
package com.taosdata.jdbc;
import com.taosdata.jdbc.utils.TaosInfo;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
public class TSDBStatement implements Statement {
private TSDBJNIConnector connector = null;
private TSDBJNIConnector connector;
private TaosInfo taosInfo = TaosInfo.getInstance();
/**
* To store batched commands
......
src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/TDNode.java 已删除 100644 → 0
浏览文件 @ dee48f94
package com.taosdata.jdbc.utils;
import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
import java.util.*;
import java.util.concurrent.TimeUnit;
public class TDNode {
private int index;
private int running;
private int deployed;
private boolean testCluster;
private String path;
private String cfgDir;
private String dataDir;
private String logDir;
private String cfgPath;
public TDNode(int index) {
this.index = index;
running = 0;
deployed = 0;
testCluster = false;
}
public void setPath(String path) {
this.path = path;
}
public void setTestCluster(boolean testCluster) {
this.testCluster = testCluster;
}
public void setRunning(int running) {
this.running = running;
}
public void searchTaosd(File dir, ArrayList<String> taosdPath) {
File[] fileList = dir.listFiles();
if(fileList == null || fileList.length == 0) {
return;
}
for(File file : fileList) {
if(file.isFile()) {
if(file.getName().equals("taosd")) {
taosdPath.add(file.getAbsolutePath());
}
} else {
searchTaosd(file, taosdPath);
}
}
}
public void start() {
String selfPath = System.getProperty("user.dir");
String binPath = "";
String projDir = selfPath + "/../../../";
try {
ArrayList<String> taosdPath = new ArrayList<>();
File dir = new File(projDir);
String realProjDir = dir.getCanonicalPath();
dir = new File(realProjDir);
System.out.println("project Dir: " + projDir);
searchTaosd(dir, taosdPath);
if(taosdPath.size() == 0) {
System.out.println("The project path doens't exist");
return;
} else {
for(String p : taosdPath) {
if(!p.contains("packaging")) {
binPath = p;
}
}
}
} catch (Exception e) {
e.printStackTrace();
}
if(binPath.isEmpty()) {
System.out.println("taosd not found");
return;
} else {
System.out.println("taosd found in " + binPath);
}
if(this.deployed == 0) {
System.out.println("dnode" + index + "is not deployed");
return;
}
String cmd = "nohup " + binPath + " -c " + cfgDir + " > /dev/null 2>&1 & ";
System.out.println("start taosd cmd: " + cmd);
try{
Runtime.getRuntime().exec(cmd);
TimeUnit.SECONDS.sleep(5);
} catch (Exception e) {
e.printStackTrace();
}
this.running = 1;
}
public Integer getTaosdPid() {
String cmd = "ps -ef|grep -w taosd| grep -v grep | awk '{print $2}'";
String[] cmds = {"sh", "-c", cmd};
try {
Process process = Runtime.getRuntime().exec(cmds);
BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
String line = null;
Integer res = null;
while((line = reader.readLine()) != null) {
if(!line.isEmpty()) {
res = Integer.valueOf(line);
break;
}
}
return res;
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
public void stop() {
if (this.running != 0) {
Integer pid = null;
while((pid = getTaosdPid()) != null) {
String killCmd = "kill -term " + pid;
String[] killCmds = {"sh", "-c", killCmd};
try {
Runtime.getRuntime().exec(killCmds).waitFor();
TimeUnit.SECONDS.sleep(2);
} catch (Exception e) {
e.printStackTrace();
}
}
try {
for(int port = 6030; port < 6041; port ++) {
String fuserCmd = "fuser -k -n tcp " + port;
Runtime.getRuntime().exec(fuserCmd).waitFor();
}
} catch (Exception e) {
e.printStackTrace();
}
this.running = 0;
System.out.println("dnode:" + this.index + " is stopped by kill -term");
}
}
public void startIP() {
try{
String cmd = "sudo ifconfig lo:" + index + "192.168.0." + index + " up";
Runtime.getRuntime().exec(cmd).waitFor();
} catch (Exception e) {
e.printStackTrace();
}
}
public void stopIP() {
try{
String cmd = "sudo ifconfig lo:" + index + "192.168.0." + index + " down";
Runtime.getRuntime().exec(cmd).waitFor();
} catch (Exception e) {
e.printStackTrace();
}
}
public void setCfgConfig(String option, String value) {
try{
String cmd = "echo " + option + " " + value + " >> " + this.cfgPath;
String[] cmdLine = {"sh", "-c", cmd};
Process ps = Runtime.getRuntime().exec(cmdLine);
ps.waitFor();
} catch (Exception e) {
e.printStackTrace();
}
}
public String getDnodeRootDir() {
String dnodeRootDir = this.path + "/sim/psim/dnode" + this.index;
return dnodeRootDir;
}
public String getDnodesRootDir() {
String dnodesRootDir = this.path + "/sim/psim" + this.index;
return dnodesRootDir;
}
public void deploy() {
this.logDir = this.path + "/sim/dnode" + this.index + "/log";
this.dataDir = this.path + "/sim/dnode" + this.index + "/data";
this.cfgDir = this.path + "/sim/dnode" + this.index + "/cfg";
this.cfgPath = this.path + "/sim/dnode" + this.index + "/cfg/taos.cfg";
try {
String cmd = "rm -rf " + this.logDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "rm -rf " + this.cfgDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "rm -rf " + this.dataDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "mkdir -p " + this.logDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "mkdir -p " + this.cfgDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "mkdir -p " + this.dataDir;
Runtime.getRuntime().exec(cmd).waitFor();
cmd = "touch " + this.cfgPath;
Runtime.getRuntime().exec(cmd).waitFor();
} catch (Exception e) {
e.printStackTrace();
}
if(this.testCluster) {
startIP();
setCfgConfig("masterIp", "192.168.0.1");
setCfgConfig("secondIp", "192.168.0.2");
setCfgConfig("publicIp", "192.168.0." + this.index);
setCfgConfig("internalIp", "192.168.0." + this.index);
setCfgConfig("privateIp", "192.168.0." + this.index);
}
setCfgConfig("dataDir", this.dataDir);
setCfgConfig("logDir", this.logDir);
setCfgConfig("numOfLogLines", "1000000/00");
setCfgConfig("mnodeEqualVnodeNum", "0");
setCfgConfig("walLevel", "1");
setCfgConfig("statusInterval", "1");
setCfgConfig("numOfMnodes", "3");
setCfgConfig("numOfThreadsPerCore", "2.0");
setCfgConfig("monitor", "0");
setCfgConfig("maxVnodeConnections", "30000");
setCfgConfig("maxMgmtConnections", "30000");
setCfgConfig("maxMeterConnections", "30000");
setCfgConfig("maxShellConns", "30000");
setCfgConfig("locale", "en_US.UTF-8");
setCfgConfig("charset", "UTF-8");
setCfgConfig("asyncLog", "0");
setCfgConfig("anyIp", "0");
setCfgConfig("dDebugFlag", "135");
setCfgConfig("mDebugFlag", "135");
setCfgConfig("sdbDebugFlag", "135");
setCfgConfig("rpcDebugFlag", "135");
setCfgConfig("tmrDebugFlag", "131");
setCfgConfig("cDebugFlag", "135");
setCfgConfig("httpDebugFlag", "135");
setCfgConfig("monitorDebugFlag", "135");
setCfgConfig("udebugFlag", "135");
setCfgConfig("jnidebugFlag", "135");
setCfgConfig("qdebugFlag", "135");
this.deployed = 1;
}
}
\ No newline at end of file
src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/TDNodes.java 已删除 100644 → 0
浏览文件 @ dee48f94
package com.taosdata.jdbc.utils;
import java.io.File;
import java.util.*;
public class TDNodes {
private ArrayList<TDNode> tdNodes;
private boolean testCluster;
public TDNodes () {
tdNodes = new ArrayList<>();
for(int i = 1; i < 11; i ++) {
tdNodes.add(new TDNode(i));
}
}
public void setTestCluster(boolean testCluster) {
this.testCluster = testCluster;
}
public void check(int index) {
if(index < 1 || index > 10) {
System.out.println("index: " + index + " should on a scale of [1, 10]");
return;
}
}
public void deploy(int index) {
try {
File file = new File(System.getProperty("user.dir") + "/../../../");
String projectRealPath = file.getCanonicalPath();
check(index);
tdNodes.get(index - 1).setTestCluster(this.testCluster);
tdNodes.get(index - 1).setPath(projectRealPath);
tdNodes.get(index - 1).deploy();
} catch (Exception e) {
e.printStackTrace();
System.out.println("deploy Test Exception");
}
}
public void cfg(int index, String option, String value) {
check(index);
tdNodes.get(index - 1).setCfgConfig(option, value);
}
public TDNode getTDNode(int index) {
check(index);
return tdNodes.get(index - 1);
}
public void start(int index) {
check(index);
tdNodes.get(index - 1).start();
}
public void stop(int index) {
check(index);
tdNodes.get(index - 1).stop();
}
public void startIP(int index) {
check(index);
tdNodes.get(index - 1).startIP();
}
public void stopIP(int index) {
check(index);
tdNodes.get(index - 1).stopIP();
}
}
\ No newline at end of file
src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/TaosInfo.java 0 → 100644
浏览文件 @ 219e0994
package com.taosdata.jdbc.utils;
import javax.management.*;
import java.lang.management.ManagementFactory;
import java.util.concurrent.atomic.AtomicLong;
public class TaosInfo implements TaosInfoMBean {
private static volatile TaosInfo instance;
private AtomicLong connect_open = new AtomicLong();
private AtomicLong connect_close = new AtomicLong();
private AtomicLong statement_count = new AtomicLong();
static {
try {
MBeanServer server = ManagementFactory.getPlatformMBeanServer();
ObjectName name = new ObjectName("TaosInfoMBean:name=TaosInfo");
server.registerMBean(TaosInfo.getInstance(), name);
} catch (MalformedObjectNameException | InstanceAlreadyExistsException | MBeanRegistrationException | NotCompliantMBeanException e) {
e.printStackTrace();
}
}
@Override
public long getConnect_open() {
return connect_open.get();
}
@Override
public long getConnect_close() {
return connect_close.get();
}
@Override
public long getConnect_active() {
return connect_open.get() - connect_close.get();
}
@Override
public long getStatement_count() {
return statement_count.get();
}
/*******************************************************/
public void conn_open_increment() {
connect_open.incrementAndGet();
}
public void connect_close_increment() {
connect_close.incrementAndGet();
}
public void stmt_count_increment() {
statement_count.incrementAndGet();
}
/********************************************************************************/
private TaosInfo() {
}
public static TaosInfo getInstance() {
if (instance == null) {
synchronized (TaosInfo.class) {
if (instance == null) {
instance = new TaosInfo();
}
}
}
return instance;
}
}
src/connector/jdbc/src/main/java/com/taosdata/jdbc/utils/TaosInfoMBean.java 0 → 100644
浏览文件 @ 219e0994
package com.taosdata.jdbc.utils;
public interface TaosInfoMBean {
long getConnect_open();
long getConnect_close();
long getConnect_active();
long getStatement_count();
}
src/connector/jdbc/src/test/java/com/taosdata/jdbc/cases/TaosInfoMonitorTest.java 0 → 100644
浏览文件 @ 219e0994
package com.taosdata.jdbc.cases;
import org.junit.Test;
import java.sql.*;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class TaosInfoMonitorTest {
@Test
public void testCreateTooManyConnection() throws ClassNotFoundException {
Class.forName("com.taosdata.jdbc.TSDBDriver");
final String url = "jdbc:TAOS://127.0.0.1:6030/?user=root&password=taosdata";
List<Connection> connectionList = IntStream.range(0, 100).mapToObj(i -> {
try {
TimeUnit.MILLISECONDS.sleep(100);
return DriverManager.getConnection(url);
} catch (SQLException | InterruptedException e) {
e.printStackTrace();
}
return null;
}).collect(Collectors.toList());
connectionList.stream().forEach(conn -> {
try (Statement stmt = conn.createStatement()) {
ResultSet rs = stmt.executeQuery("show databases");
while (rs.next()) {
}
TimeUnit.MILLISECONDS.sleep(100);
} catch (SQLException | InterruptedException e) {
e.printStackTrace();
}
});
connectionList.stream().forEach(conn -> {
try {
conn.close();
TimeUnit.MILLISECONDS.sleep(100);
} catch (SQLException | InterruptedException e) {
e.printStackTrace();
}
});
}
}
src/cq/src/cqMain.c
浏览文件 @ 219e0994
......@@ -125,8 +125,6 @@ void cqFree(void *handle) {
pthread_mutex_unlock(&pContext->mutex);
if (delete) {
pthread_mutex_unlock(&pContext->mutex);
pthread_mutex_destroy(&pContext->mutex);
taosTmrCleanUp(pContext->tmrCtrl);
......@@ -186,6 +184,18 @@ void *cqOpen(void *ahandle, const SCqCfg *pCfg) {
return pContext;
}
static void freeSCqContext(void *handle) {
if (handle == NULL) {
return;
}
SCqContext *pContext = handle;
pthread_mutex_destroy(&pContext->mutex);
taosTmrCleanUp(pContext->tmrCtrl);
pContext->tmrCtrl = NULL;
cDebug("vgId:%d, CQ is closed", pContext->vgId);
free(pContext);
}
void cqClose(void *handle) {
if (tsEnableStream == 0) {
return;
......@@ -217,6 +227,8 @@ void cqClose(void *handle) {
taosRemoveRef(cqObjRef, rid);
}
freeSCqContext(pContext);
}
void cqStart(void *handle) {
......
src/query/inc/queryLog.h
浏览文件 @ 219e0994
......@@ -30,6 +30,7 @@ extern uint32_t qDebugFlag;
#define qInfo(...) do { if (qDebugFlag & DEBUG_INFO) { taosPrintLog("QRY ", 255, __VA_ARGS__); }} while(0)
#define qDebug(...) do { if (qDebugFlag & DEBUG_DEBUG) { taosPrintLog("QRY ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qTrace(...) do { if (qDebugFlag & DEBUG_TRACE) { taosPrintLog("QRY ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qDump(a, l) do { if (qDebugFlag & DEBUG_DUMP) { taosDumpData((unsigned char *)a, l); }} while(0)
#ifdef __cplusplus
}
......
src/query/src/qExecutor.c
浏览文件 @ 219e0994
......@@ -7160,14 +7160,23 @@ static int32_t doDumpQueryResult(SQInfo *pQInfo, char *data) {
if (f) {
off_t s = lseek(fileno(f), 0, SEEK_END);
qDebug("QInfo:%p ts comp data return, file:%p, size:%"PRId64, pQInfo, f, s);
qDebug("QInfo:%p ts comp data return, file:%p, size:%"PRId64, pQInfo, f, (uint64_t)s);
if (fseek(f, 0, SEEK_SET) >= 0) {
size_t sz = fread(data, 1, s, f);
if(sz < s) { // todo handle error
qError("fread(f:%p,%d) failed, rsize:%" PRId64 ", expect size:%" PRId64, f, fileno(f), (uint64_t)sz, (uint64_t)s);
assert(0);
}
} else {
UNUSED(s);
qError("fseek(f:%p,%d) failed, error:%s", f, fileno(f), strerror(errno));
assert(0);
}
if (s <= (sizeof(STSBufFileHeader) + sizeof(STSGroupBlockInfo) + 6 * sizeof(int32_t))) {
qDump(data, s);
assert(0);
}
fclose(f);
......
tests/pytest/pytest_3.sh
浏览文件 @ 219e0994
......@@ -99,7 +99,7 @@ python3 test.py -f query/queryFillTest.py
python3 test.py -f tools/taosdemoTest.py
python3 test.py -f tools/taosdumpTest.py
python3 test.py -f tools/lowaTest.py
python3 test.py -f tools/taosdemoTest2.py
#python3 test.py -f tools/taosdemoTest2.py
# subscribe
python3 test.py -f subscribe/singlemeter.py
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或