提交 11e68d6e 编写于 作者: H Hongze Cheng

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

......@@ -86,6 +86,7 @@ void qDestroyQueryInfo(qinfo_t qHandle);
void* qOpenQueryMgmt(int32_t vgId);
void qQueryMgmtNotifyClosed(void* pExecutor);
void qQueryMgmtReOpen(void *pExecutor);
void qCleanupQueryMgmt(void* pExecutor);
void** qRegisterQInfo(void* pMgmt, uint64_t qInfo);
void** qAcquireQInfo(void* pMgmt, uint64_t key);
......
......@@ -1051,7 +1051,10 @@ static int32_t sdbWriteFwdToQueue(int32_t vgId, void *wparam, int32_t qtype, voi
memcpy(pRow->pHead, pHead, sizeof(SWalHead) + pHead->len);
pRow->rowData = pRow->pHead->cont;
return sdbWriteToQueue(pRow, qtype);
int32_t code = sdbWriteToQueue(pRow, qtype);
if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) code = 0;
return code;
}
static int32_t sdbWriteRowToQueue(SSdbRow *pInputRow, int32_t action) {
......
......@@ -89,12 +89,17 @@ int taosInitTimer(void (*callback)(int), int ms) {
if (code != 0) {
uError("failed to create timer thread");
return -1;
} else {
uDebug("timer thread:0x%08" PRIx64 " is created", taosGetPthreadId(timerThread));
}
return 0;
}
void taosUninitTimer() {
stopTimer = true;
uDebug("join timer thread:0x%08" PRIx64, taosGetPthreadId(timerThread));
pthread_join(timerThread, NULL);
}
......
......@@ -1667,7 +1667,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
_end:
assert(offset >= 0 && tsCols != NULL);
if (prevTs != INT64_MIN) {
if (prevTs != INT64_MIN && prevTs != *(int64_t*)pRuntimeEnv->prevRow[0]) {
assert(prevRowIndex >= 0);
item->lastKey = prevTs + step;
}
......@@ -7635,6 +7635,19 @@ void qQueryMgmtNotifyClosed(void* pQMgmt) {
taosCacheRefresh(pQueryMgmt->qinfoPool, queryMgmtKillQueryFn);
}
void qQueryMgmtReOpen(void *pQMgmt) {
if (pQMgmt == NULL) {
return;
}
SQueryMgmt *pQueryMgmt = pQMgmt;
qDebug("vgId:%d, set querymgmt reopen", pQueryMgmt->vgId);
pthread_mutex_lock(&pQueryMgmt->lock);
pQueryMgmt->closed = false;
pthread_mutex_unlock(&pQueryMgmt->lock);
}
void qCleanupQueryMgmt(void* pQMgmt) {
if (pQMgmt == NULL) {
return;
......
......@@ -90,7 +90,7 @@ static int32_t syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
break;
}
sDebug("%s, file:%s info is received from master, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%d", pPeer->id,
sDebug("%s, file:%s info is received from master, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%u", pPeer->id,
minfo.name, minfo.index, minfo.size, minfo.fversion, minfo.magic);
// remove extra files on slave between the current and last index
......@@ -100,13 +100,13 @@ static int32_t syncRestoreFile(SSyncPeer *pPeer, uint64_t *fversion) {
// check the file info
sinfo = minfo;
sinfo.magic = (*pNode->getFileInfo)(pNode->vgId, sinfo.name, &sinfo.index, TAOS_SYNC_MAX_INDEX, &sinfo.size, &sinfo.fversion);
sDebug("%s, local file:%s info, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%d", pPeer->id, sinfo.name,
sDebug("%s, local file:%s info, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%u", pPeer->id, sinfo.name,
sinfo.index, sinfo.size, sinfo.fversion, sinfo.magic);
// if file not there or magic is not the same, file shall be synced
memset(&fileAck, 0, sizeof(SFileAck));
syncBuildFileAck(&fileAck, pNode->vgId);
fileAck.sync = (sinfo.magic != minfo.magic || sinfo.name[0] == 0) ? 1 : 0;
fileAck.sync = (sinfo.magic != minfo.magic || sinfo.size != minfo.size || sinfo.name[0] == 0) ? 1 : 0;
// send file ack
ret = taosWriteMsg(pPeer->syncFd, &fileAck, sizeof(SFileAck));
......@@ -195,7 +195,11 @@ static int32_t syncRestoreWal(SSyncPeer *pPeer, uint64_t *wver) {
}
lastVer = pHead->version;
(*pNode->writeToCache)(pNode->vgId, pHead, TAOS_QTYPE_WAL, NULL);
ret = (*pNode->writeToCache)(pNode->vgId, pHead, TAOS_QTYPE_WAL, NULL);
if (ret != 0) {
sError("%s, failed to restore record since %s, hver:%" PRIu64, pPeer->id, tstrerror(ret), pHead->version);
break;
}
}
if (code < 0) {
......
......@@ -104,7 +104,7 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
fileInfo.magic = (*pNode->getFileInfo)(pNode->vgId, fileInfo.name, &fileInfo.index, TAOS_SYNC_MAX_INDEX,
&fileInfo.size, &fileInfo.fversion);
syncBuildFileInfo(&fileInfo, pNode->vgId);
sDebug("%s, file:%s info is sent, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%d", pPeer->id, fileInfo.name,
sDebug("%s, file:%s info is sent, index:%d size:%" PRId64 " fver:%" PRIu64 " magic:%u", pPeer->id, fileInfo.name,
fileInfo.index, fileInfo.size, fileInfo.fversion, fileInfo.magic);
// send the file info
......@@ -143,10 +143,10 @@ static int32_t syncRetrieveFile(SSyncPeer *pPeer) {
// if sync is not required, continue
if (fileAck.sync == 0) {
fileInfo.index++;
sDebug("%s, %s is the same", pPeer->id, fileInfo.name);
sDebug("%s, %s is the same, fver:%" PRIu64, pPeer->id, fileInfo.name, fileInfo.fversion);
continue;
} else {
sDebug("%s, %s will be sent", pPeer->id, fileInfo.name);
sDebug("%s, %s will be sent, fver:%" PRIu64, pPeer->id, fileInfo.name, fileInfo.fversion);
}
// get the full path to file
......@@ -328,7 +328,8 @@ static int32_t syncProcessLastWal(SSyncPeer *pPeer, char *wname, int64_t index)
// if bytes > 0, file is updated, or fversion is not reached but file still open, read again
once = 1;
offset += bytes;
sDebug("%s, continue retrieve last wal, bytes:%d offset:%" PRId64, pPeer->id, bytes, offset);
sDebug("%s, continue retrieve last wal, bytes:%d offset:%" PRId64 " sver:%" PRIu64 " fver:%" PRIu64, pPeer->id,
bytes, offset, pPeer->sversion, fversion);
}
return -1;
......
......@@ -27,6 +27,7 @@ void vnodeCleanupRead(void);
int32_t vnodeWriteToRQueue(void *pVnode, void *pCont, int32_t contLen, int8_t qtype, void *rparam);
void vnodeFreeFromRQueue(void *pVnode, SVReadMsg *pRead);
int32_t vnodeProcessRead(void *pVnode, SVReadMsg *pRead);
void vnodeWaitReadCompleted(void *pVnode);
#ifdef __cplusplus
}
......
......@@ -37,6 +37,7 @@ bool vnodeSetResetStatus(SVnodeObj* pVnode);
bool vnodeInInitStatus(SVnodeObj* pVnode);
bool vnodeInReadyStatus(SVnodeObj* pVnode);
bool vnodeInReadyOrUpdatingStatus(SVnodeObj* pVnode);
bool vnodeInClosingStatus(SVnodeObj* pVnode);
bool vnodeInResetStatus(SVnodeObj* pVnode);
......
......@@ -27,6 +27,7 @@ void vnodeCleanupWrite(void);
int32_t vnodeWriteToWQueue(void *pVnode, void *pHead, int32_t qtype, void *pRpcMsg);
void vnodeFreeFromWQueue(void *pVnode, SVWriteMsg *pWrite);
int32_t vnodeProcessWrite(void *pVnode, void *pHead, int32_t qtype, void *pRspRet);
void vnodeWaitWriteCompleted(void *pVnode);
#ifdef __cplusplus
}
......
......@@ -142,6 +142,11 @@ static int32_t vnodeAlterImp(SVnodeObj *pVnode, SCreateVnodeMsg *pVnodeCfg) {
int32_t vnodeAlter(void *vparam, SCreateVnodeMsg *pVnodeCfg) {
SVnodeObj *pVnode = vparam;
if (pVnode->dbCfgVersion == pVnodeCfg->cfg.dbCfgVersion && pVnode->vgCfgVersion == pVnodeCfg->cfg.vgCfgVersion) {
vDebug("vgId:%d, dbCfgVersion:%d and vgCfgVersion:%d not change", pVnode->vgId, pVnode->dbCfgVersion,
pVnode->vgCfgVersion);
return TSDB_CODE_SUCCESS;
}
// vnode in non-ready state and still needs to return success instead of TSDB_CODE_VND_INVALID_STATUS
// dbCfgVersion can be corrected by status msg
......@@ -400,15 +405,12 @@ void vnodeDestroy(SVnodeObj *pVnode) {
}
void vnodeCleanUp(SVnodeObj *pVnode) {
if (!vnodeInInitStatus(pVnode)) {
// it may be in updateing or reset state, then it shall wait
int32_t i = 0;
while (!vnodeSetClosingStatus(pVnode)) {
if (++i % 1000 == 0) {
sched_yield();
}
}
}
vDebug("vgId:%d, vnode will cleanup, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
vnodeSetClosingStatus(pVnode);
// release local resources only after cutting off outside connections
qQueryMgmtNotifyClosed(pVnode->qMgmt);
// stop replication module
if (pVnode->sync > 0) {
......@@ -417,10 +419,7 @@ void vnodeCleanUp(SVnodeObj *pVnode) {
syncStop(sync);
}
vDebug("vgId:%d, vnode will cleanup, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
// release local resources only after cutting off outside connections
qQueryMgmtNotifyClosed(pVnode->qMgmt);
vDebug("vgId:%d, vnode is cleaned, refCount:%d pVnode:%p", pVnode->vgId, pVnode->refCount, pVnode);
vnodeRelease(pVnode);
}
......
......@@ -88,22 +88,15 @@ void vnodeFreeFromRQueue(void *vparam, SVReadMsg *pRead) {
vnodeRelease(pVnode);
}
int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qtype, void *rparam) {
SVnodeObj *pVnode = vparam;
if (qtype == TAOS_QTYPE_RPC || qtype == TAOS_QTYPE_QUERY) {
int32_t code = vnodeCheckRead(pVnode);
if (code != TSDB_CODE_SUCCESS) return code;
}
static SVReadMsg *vnodeBuildVReadMsg(SVnodeObj *pVnode, void *pCont, int32_t contLen, int8_t qtype, SRpcMsg *pRpcMsg) {
int32_t size = sizeof(SVReadMsg) + contLen;
SVReadMsg *pRead = taosAllocateQitem(size);
if (pRead == NULL) {
return TSDB_CODE_VND_OUT_OF_MEMORY;
terrno = TSDB_CODE_VND_OUT_OF_MEMORY;
return NULL;
}
if (rparam != NULL) {
SRpcMsg *pRpcMsg = rparam;
if (pRpcMsg != NULL) {
pRead->rpcHandle = pRpcMsg->handle;
pRead->rpcAhandle = pRpcMsg->ahandle;
pRead->msgType = pRpcMsg->msgType;
......@@ -119,13 +112,35 @@ int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qt
pRead->qtype = qtype;
atomic_add_fetch_32(&pVnode->refCount, 1);
return pRead;
}
int32_t vnodeWriteToRQueue(void *vparam, void *pCont, int32_t contLen, int8_t qtype, void *rparam) {
SVReadMsg *pRead = vnodeBuildVReadMsg(vparam, pCont, contLen, qtype, rparam);
if (pRead == NULL) {
assert(terrno != 0);
return terrno;
}
SVnodeObj *pVnode = vparam;
int32_t code = vnodeCheckRead(pVnode);
if (code != TSDB_CODE_SUCCESS) {
taosFreeQitem(pRead);
vnodeRelease(pVnode);
return code;
}
atomic_add_fetch_32(&pVnode->queuedRMsg, 1);
if (pRead->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pRead->msgType == TSDB_MSG_TYPE_FETCH) {
vTrace("vgId:%d, write into vfetch queue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedRMsg);
vTrace("vgId:%d, write into vfetch queue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount,
pVnode->queuedRMsg);
return taosWriteQitem(pVnode->fqueue, qtype, pRead);
} else {
vTrace("vgId:%d, write into vquery queue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount, pVnode->queuedRMsg);
vTrace("vgId:%d, write into vquery queue, refCount:%d queued:%d", pVnode->vgId, pVnode->refCount,
pVnode->queuedRMsg);
return taosWriteQitem(pVnode->qqueue, qtype, pRead);
}
}
......@@ -420,3 +435,5 @@ int32_t vnodeNotifyCurrentQhandle(void *handle, void *qhandle, int32_t vgId) {
vTrace("QInfo:%p register qhandle to connect:%p", qhandle, handle);
return rpcReportProgress(handle, (char *)pMsg, sizeof(SRetrieveTableMsg));
}
void vnodeWaitReadCompleted(void *pVnode) {}
\ No newline at end of file
......@@ -15,6 +15,8 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "taosmsg.h"
#include "query.h"
#include "vnodeStatus.h"
char* vnodeStatus[] = {
......@@ -44,11 +46,13 @@ bool vnodeSetReadyStatus(SVnodeObj* pVnode) {
vDebug("vgId:%d, cannot set status:ready, old:%s", pVnode->vgId, vnodeStatus[pVnode->status]);
}
qQueryMgmtReOpen(pVnode->qMgmt);
pthread_mutex_unlock(&pVnode->statusMutex);
return set;
}
bool vnodeSetClosingStatus(SVnodeObj* pVnode) {
static bool vnodeSetClosingStatusImp(SVnodeObj* pVnode) {
bool set = false;
pthread_mutex_lock(&pVnode->statusMutex);
......@@ -63,6 +67,20 @@ bool vnodeSetClosingStatus(SVnodeObj* pVnode) {
return set;
}
bool vnodeSetClosingStatus(SVnodeObj* pVnode) {
if (!vnodeInInitStatus(pVnode)) {
// it may be in updating or reset state, then it shall wait
int32_t i = 0;
while (!vnodeSetClosingStatusImp(pVnode)) {
if (++i % 1000 == 0) {
sched_yield();
}
}
}
return true;
}
bool vnodeSetUpdatingStatus(SVnodeObj* pVnode) {
bool set = false;
pthread_mutex_lock(&pVnode->statusMutex);
......@@ -117,6 +135,18 @@ bool vnodeInReadyStatus(SVnodeObj* pVnode) {
return in;
}
bool vnodeInReadyOrUpdatingStatus(SVnodeObj* pVnode) {
bool in = false;
pthread_mutex_lock(&pVnode->statusMutex);
if (pVnode->status == TAOS_VN_STATUS_READY || pVnode->status == TAOS_VN_STATUS_UPDATING) {
in = true;
}
pthread_mutex_unlock(&pVnode->statusMutex);
return in;
}
bool vnodeInClosingStatus(SVnodeObj* pVnode) {
bool in = false;
pthread_mutex_lock(&pVnode->statusMutex);
......
......@@ -90,7 +90,10 @@ int32_t vnodeProcessWrite(void *vparam, void *wparam, int32_t qtype, void *rpara
// write into WAL
code = walWrite(pVnode->wal, pHead);
if (code < 0) return code;
if (code < 0) {
vError("vgId:%d, hver:%" PRIu64 " vver:%" PRIu64 " code:0x%x", pVnode->vgId, pHead->version, pVnode->version, code);
return code;
}
pVnode->version = pHead->version;
......@@ -120,12 +123,6 @@ static int32_t vnodeCheckWrite(SVnodeObj *pVnode) {
return TSDB_CODE_APP_NOT_READY;
}
if (vnodeInClosingStatus(pVnode)) {
vDebug("vgId:%d, vnode status is %s, refCount:%d pVnode:%p", pVnode->vgId, vnodeStatus[pVnode->status],
pVnode->refCount, pVnode);
return TSDB_CODE_APP_NOT_READY;
}
if (pVnode->isFull) {
vDebug("vgId:%d, vnode is full, refCount:%d", pVnode->vgId, pVnode->refCount);
return TSDB_CODE_VND_IS_FULL;
......@@ -248,12 +245,21 @@ static int32_t vnodeWriteToWQueueImp(SVWriteMsg *pWrite) {
if (pWrite->qtype == TAOS_QTYPE_RPC) {
int32_t code = vnodeCheckWrite(pVnode);
if (code != TSDB_CODE_SUCCESS) {
vError("vgId:%d, failed to write into vwqueue since %s", pVnode->vgId, tstrerror(code));
taosFreeQitem(pWrite);
vnodeRelease(pVnode);
return code;
}
}
if (!vnodeInReadyOrUpdatingStatus(pVnode)) {
vError("vgId:%d, failed to write into vwqueue, vstatus is %s, refCount:%d pVnode:%p", pVnode->vgId,
vnodeStatus[pVnode->status], pVnode->refCount, pVnode);
taosFreeQitem(pWrite);
vnodeRelease(pVnode);
return TSDB_CODE_APP_NOT_READY;
}
int32_t queued = atomic_add_fetch_32(&pVnode->queuedWMsg, 1);
if (queued > MAX_QUEUED_MSG_NUM) {
int32_t ms = (queued / MAX_QUEUED_MSG_NUM) * 10 + 3;
......@@ -338,3 +344,5 @@ static int32_t vnodePerformFlowCtrl(SVWriteMsg *pWrite) {
return TSDB_CODE_VND_ACTION_IN_PROGRESS;
}
}
void vnodeWaitWriteCompleted(void *pVnode) {}
\ No newline at end of file
......@@ -5,9 +5,10 @@
<modelVersion>4.0.0</modelVersion>
<groupId>com.taosdata.jdbc</groupId>
<artifactId>jdbcChecker</artifactId>
<artifactId>JDBCDemo</artifactId>
<version>SNAPSHOT</version>
<packaging>jar</packaging>
<build>
<plugins>
<plugin>
......@@ -17,7 +18,7 @@
<configuration>
<archive>
<manifest>
<mainClass>com.taosdata.example.JdbcChecker</mainClass>
<mainClass>com.taosdata.example.JDBCDemo</mainClass>
</manifest>
</archive>
<descriptorRefs>
......@@ -43,25 +44,8 @@
<target>8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>com.taosdata.jdbc</groupId>
<artifactId>taos-jdbcdriver</artifactId>
<version>2.0.12</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.13.1</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
# How to Run the JDBC Demo Code On A Linux OS
# How to Run the JDBC Demo Code On Linux OS
TDengine's JDBC demo project is organized in a Maven way so that users can easily compile, package and run the project. If you don't have Maven on your server, you may install it using
<pre>sudo apt-get install maven</pre>
```
sudo apt-get install maven
```
## Install TDengine Client
Make sure you have already installed a tdengine client on your current develop environment.
Download the tdengine package on our website: ``https://www.taosdata.com/cn/all-downloads/`` and install the client.
## How to run jdbcChecker
<pre>mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.JdbcChecker" -Dexec.args="-host localhost"</pre>
## How to run jdbcTaosDemo
## Run jdbcDemo using mvn plugin
run command:
<pre> mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.jdbcTaosdemo.JdbcTaosdemo"</pre>
```
mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.JdbcDemo"
```
and run with your customed args
<pre>mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.jdbcTaosdemo.JdbcTaosdemo" -Dexec.args="-host localhost"</pre>
```
mvn clean compile exec:java -Dexec.mainClass="com.taosdata.example.JdbcDemo" -Dexec.args="-host [HOSTNAME]"
```
## Compile the Demo Code and Run It
To compile taos-jdbcdriver, go to the source directory ``TDengine/src/connector/jdbc`` and execute
```
mvn clean package -Dmaven.test.skip=true
```
To compile the demo project, go to the source directory ``TDengine/tests/examples/JDBC/JDBCDemo`` and execute
<pre>
```
mvn clean package assembly:single
</pre>
```
The ``pom.xml`` is configured to package all the dependencies into one executable jar file.
To run JDBCDemo.jar, go to ``TDengine/tests/examples/JDBC/JDBCDemo`` and execute
```
java -Djava.ext.dirs=../../../../src/connector/jdbc/target:$JAVA_HOME/jre/lib/ext -jar target/JDBCDemo-SNAPSHOT-jar-with-dependencies.jar -host [HOSTNAME]
```
To run it, go to ``examples/JDBC/JDBCDemo/target`` and execute
<pre>java -jar jdbcChecker-SNAPSHOT-jar-with-dependencies.jar -host localhost</pre>
package com.taosdata.example;
import com.taosdata.jdbc.TSDBDriver;
import java.sql.*;
import java.util.Properties;
public class JdbcChecker {
public class JDBCDemo {
private static String host;
private static String dbName = "test";
private static String tbName = "weather";
private static String driverType;
private static final String dbName = "test";
private static final String tbName = "weather";
private Connection connection;
/**
* get connection
**/
public static void main(String[] args) {
for (int i = 0; i < args.length; i++) {
if ("-host".equalsIgnoreCase(args[i]) && i < args.length - 1)
host = args[++i];
if ("-driverType".equalsIgnoreCase(args[i]) && i < args.length - 1) {
driverType = args[++i];
if (!"jni".equalsIgnoreCase(driverType) && !"restful".equalsIgnoreCase(driverType))
printHelp();
}
}
if (host == null || driverType == null) {
printHelp();
}
JDBCDemo demo = new JDBCDemo();
demo.init();
demo.createDatabase();
demo.useDatabase();
demo.dropTable();
demo.createTable();
demo.insert();
demo.select();
demo.dropTable();
demo.close();
}
private void init() {
// get connection
try {
String url = "jdbc:TAOS://" + host + ":6030/?user=root&password=taosdata";
if (driverType.equals("restful")) {
Class.forName("com.taosdata.jdbc.rs.RestfulDriver");
url = "jdbc:TAOS-RS://" + host + ":6041/?user=root&password=taosdata";
} else {
Class.forName("com.taosdata.jdbc.TSDBDriver");
}
Properties properties = new Properties();
properties.setProperty(TSDBDriver.PROPERTY_KEY_HOST, host);
properties.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
properties.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "en_US.UTF-8");
properties.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
properties.setProperty("host", host);
properties.setProperty("charset", "UTF-8");
properties.setProperty("locale", "en_US.UTF-8");
properties.setProperty("timezone", "UTC-8");
System.out.println("get connection starting...");
connection = DriverManager.getConnection("jdbc:TAOS://" + host + ":0/", properties);
connection = DriverManager.getConnection(url, properties);
if (connection != null)
System.out.println("[ OK ] Connection established.");
} catch (ClassNotFoundException | SQLException e) {
throw new RuntimeException("connection failed: " + host);
e.printStackTrace();
}
}
/**
* create database
*/
private void createDatabase() {
String sql = "create database if not exists " + dbName;
exuete(sql);
}
/**
* use database
*/
private void useDatabase() {
String sql = "use " + dbName;
exuete(sql);
}
/**
* select
*/
private void checkSelect() {
private void select() {
final String sql = "select * from test.weather";
executeQuery(sql);
}
......@@ -79,40 +100,21 @@ public class JdbcChecker {
}
}
private String formatString(String str) {
StringBuilder sb = new StringBuilder();
int blankCnt = (26 - str.length()) / 2;
for (int j = 0; j < blankCnt; j++)
sb.append(" ");
sb.append(str);
for (int j = 0; j < blankCnt; j++)
sb.append(" ");
sb.append("|");
return sb.toString();
}
/**
* insert
*/
private void checkInsert() {
private void insert() {
final String sql = "insert into test.weather (ts, temperature, humidity) values(now, 20.5, 34)";
exuete(sql);
}
/**
* create table
*/
private void createTable() {
final String sql = "create table if not exists " + dbName + "." + tbName + " (ts timestamp, temperature float, humidity int)";
exuete(sql);
}
private final void printSql(String sql, boolean succeed, long cost) {
private void printSql(String sql, boolean succeed, long cost) {
System.out.println("[ " + (succeed ? "OK" : "ERROR!") + " ] time cost: " + cost + " ms, execute statement ====> " + sql);
}
private final void exuete(String sql) {
private void exuete(String sql) {
try (Statement statement = connection.createStatement()) {
long start = System.currentTimeMillis();
boolean execute = statement.execute(sql);
......@@ -135,39 +137,15 @@ public class JdbcChecker {
}
}
private void checkDropTable() {
private void dropTable() {
final String sql = "drop table if exists " + dbName + "." + tbName + "";
exuete(sql);
}
public static void main(String[] args) {
for (int i = 0; i < args.length; i++) {
if ("-host".equalsIgnoreCase(args[i]) && i < args.length - 1) {
host = args[++i];
}
if ("-db".equalsIgnoreCase(args[i]) && i < args.length - 1) {
dbName = args[++i];
}
if ("-t".equalsIgnoreCase(args[i]) && i < args.length - 1) {
tbName = args[++i];
}
}
if (host == null) {
System.out.println("Usage: java -jar JDBCConnectorChecker.jar -host <hostname>");
return;
private static void printHelp() {
System.out.println("Usage: java -jar JdbcDemo.jar -host <hostname> -driverType <jni|restful>");
System.exit(0);
}
JdbcChecker checker = new JdbcChecker();
checker.init();
checker.createDatabase();
checker.useDatabase();
checker.checkDropTable();
checker.createTable();
checker.checkInsert();
checker.checkSelect();
checker.checkDropTable();
checker.close();
}
}
package com.taosdata.example.jdbcTaosdemo;
import com.taosdata.example.jdbcTaosdemo.domain.JdbcTaosdemoConfig;
import com.taosdata.example.jdbcTaosdemo.task.CreateTableTask;
import com.taosdata.example.jdbcTaosdemo.task.InsertTableDatetimeTask;
import com.taosdata.example.jdbcTaosdemo.task.InsertTableTask;
import com.taosdata.example.jdbcTaosdemo.utils.ConnectionFactory;
import com.taosdata.example.jdbcTaosdemo.utils.SqlSpeller;
import com.taosdata.example.jdbcTaosdemo.utils.TimeStampUtil;
import org.apache.log4j.Logger;
import java.sql.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class JdbcTaosdemo {
private static Logger logger = Logger.getLogger(JdbcTaosdemo.class);
private final JdbcTaosdemoConfig config;
private Connection connection;
public JdbcTaosdemo(JdbcTaosdemoConfig config) {
this.config = config;
}
public static void main(String[] args) {
// parse config from args
JdbcTaosdemoConfig config = new JdbcTaosdemoConfig(args);
boolean isHelp = Arrays.asList(args).contains("--help");
if (isHelp || config.host == null || config.host.isEmpty()) {
JdbcTaosdemoConfig.printHelp();
return;
}
JdbcTaosdemo taosdemo = new JdbcTaosdemo(config);
// establish connection
taosdemo.init();
// drop database
taosdemo.dropDatabase();
// create database
taosdemo.createDatabase();
// use db
taosdemo.useDatabase();
// create super table
taosdemo.createSuperTable();
// create sub tables
taosdemo.createTableMultiThreads();
boolean infinite = Arrays.asList(args).contains("--infinite");
if (infinite) {
logger.info("!!! Infinite Insert Mode Started. !!!");
taosdemo.insertInfinite();
} else {
// insert into table
taosdemo.insertMultiThreads();
// select from sub table
taosdemo.selectFromTableLimit();
taosdemo.selectCountFromTable();
taosdemo.selectAvgMinMaxFromTable();
// select last from
taosdemo.selectLastFromTable();
// select from super table
taosdemo.selectFromSuperTableLimit();
taosdemo.selectCountFromSuperTable();
taosdemo.selectAvgMinMaxFromSuperTable();
//select avg ,max from stb where tag
taosdemo.selectAvgMinMaxFromSuperTableWhereTag();
//select last from stb where location = ''
taosdemo.selectLastFromSuperTableWhere();
// select group by
taosdemo.selectGroupBy();
// select like
taosdemo.selectLike();
// select where ts >= ts<=
taosdemo.selectLastOneHour();
taosdemo.selectLastOneDay();
taosdemo.selectLastOneWeek();
taosdemo.selectLastOneMonth();
taosdemo.selectLastOneYear();
// drop super table
if (config.dropTable)
taosdemo.dropSuperTable();
taosdemo.close();
}
}
/**
* establish the connection
*/
private void init() {
try {
Class.forName("com.taosdata.jdbc.TSDBDriver");
connection = ConnectionFactory.build(config);
if (connection != null)
logger.info("[ OK ] Connection established.");
} catch (ClassNotFoundException | SQLException e) {
logger.error(e.getMessage());
throw new RuntimeException("connection failed: " + config.host);
}
}
/**
* create database
*/
private void createDatabase() {
String sql = SqlSpeller.createDatabaseSQL(config.database, config.keep, config.days);
execute(sql);
}
/**
* drop database
*/
private void dropDatabase() {
String sql = SqlSpeller.dropDatabaseSQL(config.database);
execute(sql);
}
/**
* use database
*/
private void useDatabase() {
String sql = SqlSpeller.useDatabaseSQL(config.database);
execute(sql);
}
/**
* create super table
*/
private void createSuperTable() {
String sql = SqlSpeller.createSuperTableSQL(config.superTable);
execute(sql);
}
/**
* create table use super table with multi threads
*/
private void createTableMultiThreads() {
try {
final int tableSize = (int) (config.numOfTables / config.numOfThreadsForCreate);
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < config.numOfThreadsForCreate; i++) {
Thread thread = new Thread(new CreateTableTask(config, i * tableSize, tableSize), "Thread-" + i);
threads.add(thread);
thread.start();
}
for (Thread thread : threads) {
thread.join();
}
logger.info("<<< Multi Threads create table finished.");
} catch (InterruptedException e) {
logger.error(e.getMessage());
e.printStackTrace();
}
}
/**
* insert data infinitely
*/
private void insertInfinite() {
try {
final long startDatetime = TimeStampUtil.datetimeToLong("2005-01-01 00:00:00.000");
final long finishDatetime = TimeStampUtil.datetimeToLong("2030-01-01 00:00:00.000");
final int tableSize = (int) (config.numOfTables / config.numOfThreadsForInsert);
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < config.numOfThreadsForInsert; i++) {
Thread thread = new Thread(new InsertTableDatetimeTask(config, i * tableSize, tableSize, startDatetime, finishDatetime), "Thread-" + i);
threads.add(thread);
thread.start();
}
for (Thread thread : threads) {
thread.join();
}
logger.info("<<< Multi Threads insert table finished.");
} catch (InterruptedException e) {
logger.error(e.getMessage());
e.printStackTrace();
}
}
private void insertMultiThreads() {
try {
final int tableSize = (int) (config.numOfTables / config.numOfThreadsForInsert);
final int numberOfRecordsPerTable = (int) config.numOfRowsPerTable;
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < config.numOfThreadsForInsert; i++) {
Thread thread = new Thread(new InsertTableTask(config, i * tableSize, tableSize, numberOfRecordsPerTable), "Thread-" + i);
threads.add(thread);
thread.start();
}
for (Thread thread : threads) {
thread.join();
}
logger.info("<<< Multi Threads insert table finished.");
} catch (InterruptedException e) {
logger.error(e.getMessage());
e.printStackTrace();
}
}
private void selectFromTableLimit() {
String sql = SqlSpeller.selectFromTableLimitSQL(config.database, config.prefixOfTable, 1, 10, 0);
executeQuery(sql);
}
private void selectCountFromTable() {
String sql = SqlSpeller.selectCountFromTableSQL(config.database, config.prefixOfTable, 1);
executeQuery(sql);
}
private void selectAvgMinMaxFromTable() {
String sql = SqlSpeller.selectAvgMinMaxFromTableSQL("current", config.database, config.prefixOfTable, 1);
executeQuery(sql);
}
private void selectLastFromTable() {
String sql = SqlSpeller.selectLastFromTableSQL(config.database, config.prefixOfTable, 1);
executeQuery(sql);
}
private void selectFromSuperTableLimit() {
String sql = SqlSpeller.selectFromSuperTableLimitSQL(config.database, config.superTable, 10, 0);
executeQuery(sql);
}
private void selectCountFromSuperTable() {
String sql = SqlSpeller.selectCountFromSuperTableSQL(config.database, config.superTable);
executeQuery(sql);
}
private void selectAvgMinMaxFromSuperTable() {
String sql = SqlSpeller.selectAvgMinMaxFromSuperTableSQL("current", config.database, config.superTable);
executeQuery(sql);
}
private void selectAvgMinMaxFromSuperTableWhereTag() {
String sql = SqlSpeller.selectAvgMinMaxFromSuperTableWhere("current", config.database, config.superTable);
executeQuery(sql);
}
private void selectLastFromSuperTableWhere() {
String sql = SqlSpeller.selectLastFromSuperTableWhere("current", config.database, config.superTable);
executeQuery(sql);
}
private void selectGroupBy() {
String sql = SqlSpeller.selectGroupBy("current", config.database, config.superTable);
executeQuery(sql);
}
private void selectLike() {
String sql = SqlSpeller.selectLike(config.database, config.superTable);
executeQuery(sql);
}
private void selectLastOneHour() {
String sql = SqlSpeller.selectLastOneHour(config.database, config.superTable);
executeQuery(sql);
}
private void selectLastOneDay() {
String sql = SqlSpeller.selectLastOneDay(config.database, config.superTable);
executeQuery(sql);
}
private void selectLastOneWeek() {
String sql = SqlSpeller.selectLastOneWeek(config.database, config.superTable);
executeQuery(sql);
}
private void selectLastOneMonth() {
String sql = SqlSpeller.selectLastOneMonth(config.database, config.superTable);
executeQuery(sql);
}
private void selectLastOneYear() {
String sql = SqlSpeller.selectLastOneYear(config.database, config.superTable);
executeQuery(sql);
}
private void close() {
try {
if (connection != null) {
this.connection.close();
logger.info("connection closed.");
}
} catch (SQLException e) {
logger.error(e.getMessage());
e.printStackTrace();
}
}
/**
* drop super table
*/
private void dropSuperTable() {
String sql = SqlSpeller.dropSuperTableSQL(config.database, config.superTable);
execute(sql);
}
/**
* execute sql, use this method when sql is create, alter, drop..
*/
private void execute(String sql) {
try (Statement statement = connection.createStatement()) {
long start = System.currentTimeMillis();
boolean execute = statement.execute(sql);
long end = System.currentTimeMillis();
printSql(sql, execute, (end - start));
} catch (SQLException e) {
logger.error("ERROR execute SQL ===> " + sql);
logger.error(e.getMessage());
e.printStackTrace();
}
}
private static void printSql(String sql, boolean succeed, long cost) {
System.out.println("[ " + (succeed ? "OK" : "ERROR!") + " ] time cost: " + cost + " ms, execute statement ====> " + sql);
}
private void executeQuery(String sql) {
try (Statement statement = connection.createStatement()) {
long start = System.currentTimeMillis();
ResultSet resultSet = statement.executeQuery(sql);
long end = System.currentTimeMillis();
printSql(sql, true, (end - start));
printResult(resultSet);
} catch (SQLException e) {
logger.error("ERROR execute SQL ===> " + sql);
logger.error(e.getMessage());
e.printStackTrace();
}
}
private static void printResult(ResultSet resultSet) throws SQLException {
ResultSetMetaData metaData = resultSet.getMetaData();
while (resultSet.next()) {
StringBuilder sb = new StringBuilder();
for (int i = 1; i <= metaData.getColumnCount(); i++) {
String columnLabel = metaData.getColumnLabel(i);
String value = resultSet.getString(i);
sb.append(columnLabel + ": " + value + "\t");
}
System.out.println(sb.toString());
}
}
}
package com.taosdata.example.jdbcTaosdemo.domain;
import com.taosdata.example.jdbcTaosdemo.utils.TimeStampUtil;
public final class JdbcTaosdemoConfig {
// instance
public String host; //host
public int port = 6030; //port
public String user = "root"; //user
public String password = "taosdata"; //password
// database
public String database = "test"; //database
public int keep = 3650; //keep
public int days = 30; //days
public int replica = 1; //replica
//super table
public boolean doCreateTable = true;
public String superTable = "weather"; //super table name
public String prefixOfFields = "col";
public int numOfFields;
public String prefixOfTags = "tag";
public int numOfTags;
public String superTableSQL;
//sub table
public String prefixOfTable = "t";
// insert task
public boolean autoCreateTable = true;
public long numOfTables = 100;
public long numOfRowsPerTable = 100;
public int numOfTablesPerSQL = 10;
public int numOfValuesPerSQL = 10;
public int numOfThreadsForCreate = 1;
public int numOfThreadsForInsert = 1;
public long startTime;
public long timeGap = 1;
public int frequency;
public int order;
public int rate = 10;
public long range = 1000l;
// select task
// drop task
public boolean dropTable = false;
public static void printHelp() {
System.out.println("Usage: java -jar jdbc-taosdemo-2.0.jar [OPTION...]");
// instance
System.out.println("-host The host to connect to TDengine which you must specify");
System.out.println("-port The TCP/IP port number to use for the connection. Default is 6030");
System.out.println("-user The TDengine user name to use when connecting to the server. Default is 'root'");
System.out.println("-password The password to use when connecting to the server.Default is 'taosdata'");
// database
System.out.println("-database Destination database. Default is 'test'");
System.out.println("-keep database keep parameter. Default is 3650");
System.out.println("-days database days parameter. Default is 30");
System.out.println("-replica database replica parameter. Default 1, min: 1, max: 3");
// super table
System.out.println("-doCreateTable do create super table and sub table, true or false, Default true");
System.out.println("-superTable super table name. Default 'weather'");
System.out.println("-prefixOfFields The prefix of field in super table. Default is 'col'");
System.out.println("-numOfFields The number of field in super table. Default is (ts timestamp, temperature float, humidity int).");
System.out.println("-prefixOfTags The prefix of tag in super table. Default is 'tag'");
System.out.println("-numOfTags The number of tag in super table. Default is (location nchar(64), groupId int).");
System.out.println("-superTableSQL specify a sql statement for the super table.\n" +
" Default is 'create table weather(ts timestamp, temperature float, humidity int) tags(location nchar(64), groupId int). \n" +
" if you use this parameter, the numOfFields and numOfTags will be invalid'");
// sub table
System.out.println("-prefixOfTable The prefix of sub tables. Default is 't'");
System.out.println("-numOfTables The number of tables. Default is 1");
System.out.println("-numOfThreadsForCreate The number of thread during create sub table. Default is 1");
// insert task
System.out.println("-autoCreateTable Use auto Create sub tables SQL. Default is false");
System.out.println("-numOfRowsPerTable The number of records per table. Default is 1");
System.out.println("-numOfThreadsForInsert The number of threads during insert row. Default is 1");
System.out.println("-numOfTablesPerSQL The number of table per SQL. Default is 1");
System.out.println("-numOfValuesPerSQL The number of value per SQL. Default is 1");
System.out.println("-startTime start time for insert task, The format is \"yyyy-MM-dd HH:mm:ss.SSS\".");
System.out.println("-timeGap the number of time gap. Default is 1000 ms");
System.out.println("-frequency the number of records per second inserted into one table. default is 0, do not control frequency");
System.out.println("-order Insert mode--0: In order, 1: Out of order. Default is in order");
System.out.println("-rate The proportion of data out of order. effective only if order is 1. min 0, max 100, default is 10");
System.out.println("-range The range of data out of order. effective only if order is 1. default is 1000 ms");
// query task
// System.out.println("-sqlFile The select sql file");
// drop task
System.out.println("-dropTable Drop data before quit. Default is false");
System.out.println("--help Give this help list");
}
/**
* parse args from command line
*
* @param args command line args
* @return JdbcTaosdemoConfig
*/
public JdbcTaosdemoConfig(String[] args) {
for (int i = 0; i < args.length; i++) {
// instance
if ("-host".equals(args[i]) && i < args.length - 1) {
host = args[++i];
}
if ("-port".equals(args[i]) && i < args.length - 1) {
port = Integer.parseInt(args[++i]);
}
if ("-user".equals(args[i]) && i < args.length - 1) {
user = args[++i];
}
if ("-password".equals(args[i]) && i < args.length - 1) {
password = args[++i];
}
// database
if ("-database".equals(args[i]) && i < args.length - 1) {
database = args[++i];
}
if ("-keep".equals(args[i]) && i < args.length - 1) {
keep = Integer.parseInt(args[++i]);
}
if ("-days".equals(args[i]) && i < args.length - 1) {
days = Integer.parseInt(args[++i]);
}
if ("-replica".equals(args[i]) && i < args.length - 1) {
replica = Integer.parseInt(args[++i]);
}
// super table
if ("-doCreateTable".equals(args[i]) && i < args.length - 1) {
doCreateTable = Boolean.parseBoolean(args[++i]);
}
if ("-superTable".equals(args[i]) && i < args.length - 1) {
superTable = args[++i];
}
if ("-prefixOfFields".equals(args[i]) && i < args.length - 1) {
prefixOfFields = args[++i];
}
if ("-numOfFields".equals(args[i]) && i < args.length - 1) {
numOfFields = Integer.parseInt(args[++i]);
}
if ("-prefixOfTags".equals(args[i]) && i < args.length - 1) {
prefixOfTags = args[++i];
}
if ("-numOfTags".equals(args[i]) && i < args.length - 1) {
numOfTags = Integer.parseInt(args[++i]);
}
if ("-superTableSQL".equals(args[i]) && i < args.length - 1) {
superTableSQL = args[++i];
}
// sub table
if ("-prefixOfTable".equals(args[i]) && i < args.length - 1) {
prefixOfTable = args[++i];
}
if ("-numOfTables".equals(args[i]) && i < args.length - 1) {
numOfTables = Long.parseLong(args[++i]);
}
if ("-autoCreateTable".equals(args[i]) && i < args.length - 1) {
autoCreateTable = Boolean.parseBoolean(args[++i]);
}
if ("-numOfThreadsForCreate".equals(args[i]) && i < args.length - 1) {
numOfThreadsForCreate = Integer.parseInt(args[++i]);
}
// insert task
if ("-numOfRowsPerTable".equals(args[i]) && i < args.length - 1) {
numOfRowsPerTable = Long.parseLong(args[++i]);
}
if ("-numOfThreadsForInsert".equals(args[i]) && i < args.length - 1) {
numOfThreadsForInsert = Integer.parseInt(args[++i]);
}
if ("-numOfTablesPerSQL".equals(args[i]) && i < args.length - 1) {
numOfTablesPerSQL = Integer.parseInt(args[++i]);
}
if ("-numOfValuesPerSQL".equals(args[i]) && i < args.length - 1) {
numOfValuesPerSQL = Integer.parseInt(args[++i]);
}
if ("-startTime".equals(args[i]) && i < args.length - 1) {
startTime = TimeStampUtil.datetimeToLong(args[++i]);
}
if ("-timeGap".equals(args[i]) && i < args.length - 1) {
timeGap = Long.parseLong(args[++i]);
}
if ("-frequency".equals(args[i]) && i < args.length - 1) {
frequency = Integer.parseInt(args[++i]);
}
if ("-order".equals(args[i]) && i < args.length - 1) {
order = Integer.parseInt(args[++i]);
}
if ("-rate".equals(args[i]) && i < args.length - 1) {
rate = Integer.parseInt(args[++i]);
if (rate < 0 || rate > 100)
throw new IllegalArgumentException("rate must between 0 and 100");
}
if ("-range".equals(args[i]) && i < args.length - 1) {
range = Integer.parseInt(args[++i]);
}
// select task
// drop task
if ("-dropTable".equals(args[i]) && i < args.length - 1) {
dropTable = Boolean.parseBoolean(args[++i]);
}
}
}
public static void main(String[] args) {
JdbcTaosdemoConfig config = new JdbcTaosdemoConfig(args);
}
}
package com.taosdata.example.jdbcTaosdemo.task;
import com.taosdata.example.jdbcTaosdemo.domain.JdbcTaosdemoConfig;
import com.taosdata.example.jdbcTaosdemo.utils.ConnectionFactory;
import com.taosdata.example.jdbcTaosdemo.utils.SqlSpeller;
import org.apache.log4j.Logger;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
public class CreateTableTask implements Runnable {
private static Logger logger = Logger.getLogger(CreateTableTask.class);
private final JdbcTaosdemoConfig config;
private final int startIndex;
private final int tableNumber;
public CreateTableTask(JdbcTaosdemoConfig config, int startIndex, int tableNumber) {
this.config = config;
this.startIndex = startIndex;
this.tableNumber = tableNumber;
}
@Override
public void run() {
try {
Connection connection = ConnectionFactory.build(config);
for (int i = startIndex; i < startIndex + tableNumber; i++) {
Statement statement = connection.createStatement();
String sql = SqlSpeller.createTableSQL(i + 1, config.database, config.superTable);
statement.execute(sql);
statement.close();
logger.info(">>> " + sql);
}
connection.close();
} catch (SQLException e) {
logger.error(e.getMessage());
e.printStackTrace();
}
}
}
package com.taosdata.example.jdbcTaosdemo.task;
import com.taosdata.example.jdbcTaosdemo.domain.JdbcTaosdemoConfig;
import com.taosdata.example.jdbcTaosdemo.utils.ConnectionFactory;
import com.taosdata.example.jdbcTaosdemo.utils.SqlSpeller;
import org.apache.log4j.Logger;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
public class InsertTableDatetimeTask implements Runnable {
private static Logger logger = Logger.getLogger(InsertTableDatetimeTask.class);
private final JdbcTaosdemoConfig config;
private final int startTableIndex;
private final int tableNumber;
private final long startDatetime;
private final long finishedDatetime;
public InsertTableDatetimeTask(JdbcTaosdemoConfig config, int startTableIndex, int tableNumber, long startDatetime, long finishedDatetime) {
this.config = config;
this.startTableIndex = startTableIndex;
this.tableNumber = tableNumber;
this.startDatetime = startDatetime;
this.finishedDatetime = finishedDatetime;
}
@Override
public void run() {
try {
Connection connection = ConnectionFactory.build(config);
int valuesCount = config.numOfValuesPerSQL;
for (long ts = startDatetime; ts < finishedDatetime; ts += valuesCount) {
for (int i = startTableIndex; i < startTableIndex + tableNumber; i++) {
String sql = SqlSpeller.insertBatchSizeRowsSQL(config.database, config.prefixOfTable, i + 1, ts, valuesCount);
Statement statement = connection.createStatement();
statement.execute(sql);
statement.close();
logger.info(Thread.currentThread().getName() + ">>> " + sql);
}
}
connection.close();
} catch (SQLException e) {
logger.error(e.getMessage());
e.printStackTrace();
}
}
}
package com.taosdata.example.jdbcTaosdemo.task;
import com.taosdata.example.jdbcTaosdemo.domain.JdbcTaosdemoConfig;
import com.taosdata.example.jdbcTaosdemo.utils.ConnectionFactory;
import com.taosdata.example.jdbcTaosdemo.utils.SqlSpeller;
import org.apache.log4j.Logger;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
public class InsertTableTask implements Runnable {
private static final Logger logger = Logger.getLogger(InsertTableTask.class);
private final JdbcTaosdemoConfig config;
private final int startTbIndex;
private final int tableNumber;
private final int recordsNumberPerTable;
public InsertTableTask(JdbcTaosdemoConfig config, int startTbIndex, int tableNumber, int recordsNumberPerTable) {
this.config = config;
this.startTbIndex = startTbIndex;
this.tableNumber = tableNumber;
this.recordsNumberPerTable = recordsNumberPerTable;
}
@Override
public void run() {
try {
Connection connection = ConnectionFactory.build(config);
int keep = config.keep;
Instant end = Instant.now();
Instant start = end.minus(Duration.ofDays(keep - 1));
long timeGap = ChronoUnit.MILLIS.between(start, end) / (recordsNumberPerTable - 1);
// iterate insert
for (int j = 0; j < recordsNumberPerTable; j++) {
long ts = start.toEpochMilli() + (j * timeGap);
// insert data into echo table
for (int i = startTbIndex; i < startTbIndex + tableNumber; i++) {
String sql = SqlSpeller.insertBatchSizeRowsSQL(config.database, config.prefixOfTable, i + 1, ts, config.numOfValuesPerSQL);
logger.info(Thread.currentThread().getName() + ">>> " + sql);
Statement statement = connection.createStatement();
statement.execute(sql);
statement.close();
}
}
connection.close();
} catch (SQLException e) {
logger.error(e.getMessage());
e.printStackTrace();
}
}
}
package com.taosdata.example.jdbcTaosdemo.utils;
import com.taosdata.example.jdbcTaosdemo.domain.JdbcTaosdemoConfig;
import com.taosdata.jdbc.TSDBDriver;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Properties;
public class ConnectionFactory {
public static Connection build(JdbcTaosdemoConfig config) throws SQLException {
return build(config.host, config.port, config.database, config.user, config.password);
}
public static Connection build(String host, int port, String dbName) throws SQLException {
return build(host, port, dbName, "root", "taosdata");
}
private static Connection build(String host, int port, String dbName, String user, String password) throws SQLException {
Properties properties = new Properties();
properties.setProperty(TSDBDriver.PROPERTY_KEY_USER, user);
properties.setProperty(TSDBDriver.PROPERTY_KEY_PASSWORD, password);
properties.setProperty(TSDBDriver.PROPERTY_KEY_CHARSET, "UTF-8");
properties.setProperty(TSDBDriver.PROPERTY_KEY_LOCALE, "en_US.UTF-8");
properties.setProperty(TSDBDriver.PROPERTY_KEY_TIME_ZONE, "UTC-8");
return DriverManager.getConnection("jdbc:TAOS://" + host + ":" + port + "/" + dbName + "", properties);
}
}
package com.taosdata.example.jdbcTaosdemo.utils;
import java.util.Random;
public class SqlSpeller {
private static final Random random = new Random(System.currentTimeMillis());
private static final String[] locations = {
"Beijing", "Shanghai", "Guangzhou", "Shenzhen",
"HangZhou", "Tianjin", "Wuhan", "Changsha", "Nanjing", "Xian"
};
public static String createDatabaseSQL(String dbName, int keep, int days) {
return "create database if not exists " + dbName + " keep " + keep + " days " + days;
}
public static String dropDatabaseSQL(String dbName) {
return "drop database if exists " + dbName;
}
public static String useDatabaseSQL(String dbName) {
return "use " + dbName;
}
public static String createSuperTableSQL(String superTableName) {
return "create table if not exists " + superTableName + "(ts timestamp, current float, voltage int, phase float) tags(location binary(64), groupId int)";
}
public static String dropSuperTableSQL(String dbName, String superTableName) {
return "drop table if exists " + dbName + "." + superTableName;
}
public static String createTableSQL(int tableIndex, String dbName, String superTableName) {
String location = locations[random.nextInt(locations.length)];
return "create table d" + tableIndex + " using " + dbName + "." + superTableName + " tags('" + location + "'," + tableIndex + ")";
}
public static String insertOneRowSQL(String dbName, String tbPrefix, int tableIndex, long ts) {
float current = 10 + random.nextFloat();
int voltage = 200 + random.nextInt(20);
float phase = random.nextFloat();
String sql = "insert into " + dbName + "." + tbPrefix + "" + tableIndex + " " + "values(" + ts + ", " + current + ", " + voltage + ", " + phase + ")";
return sql;
}
public static String insertBatchSizeRowsSQL(String dbName, String tbPrefix, int tbIndex, long ts, int valuesCount) {
float current = 10 + random.nextFloat();
int voltage = 200 + random.nextInt(20);
float phase = random.nextFloat();
StringBuilder sb = new StringBuilder();
sb.append("insert into " + dbName + "." + tbPrefix + "" + tbIndex + " " + "values");
for (int i = 0; i < valuesCount; i++) {
sb.append("(" + (ts + i) + ", " + current + ", " + voltage + ", " + phase + ") ");
}
return sb.toString();
}
public static String selectFromTableLimitSQL(String dbName, String tbPrefix, int tbIndex, int limit, int offset) {
return "select * from " + dbName + "." + tbPrefix + "" + tbIndex + " limit " + limit + " offset " + offset;
}
public static String selectCountFromTableSQL(String dbName, String tbPrefix, int tbIndex) {
return "select count(*) from " + dbName + "." + tbPrefix + "" + tbIndex;
}
public static String selectAvgMinMaxFromTableSQL(String field, String dbName, String tbPrefix, int tbIndex) {
return "select avg(" + field + "),min(" + field + "),max(" + field + ") from " + dbName + "." + tbPrefix + "" + tbIndex;
}
public static String selectFromSuperTableLimitSQL(String dbName, String stbName, int limit, int offset) {
return "select * from " + dbName + "." + stbName + " limit " + limit + " offset " + offset;
}
public static String selectCountFromSuperTableSQL(String dbName, String stableName) {
return "select count(*) from " + dbName + "." + stableName;
}
public static String selectAvgMinMaxFromSuperTableSQL(String field, String dbName, String stbName) {
return "select avg(" + field + "),min(" + field + "),max(" + field + ") from " + dbName + "." + stbName + "";
}
public static String selectLastFromTableSQL(String dbName, String tbPrefix, int tbIndex) {
return "select last(*) from " + dbName + "." + tbPrefix + "" + tbIndex;
}
//select avg ,max from stb where tag
public static String selectAvgMinMaxFromSuperTableWhere(String field, String dbName, String stbName) {
return "select avg(" + field + "),min(" + field + "),max(" + field + ") from " + dbName + "." + stbName + " where location = '" + locations[random.nextInt(locations.length)] + "'";
}
//select last from stb where
public static String selectLastFromSuperTableWhere(String field, String dbName, String stbName) {
return "select last(" + field + ") from " + dbName + "." + stbName + " where location = '" + locations[random.nextInt(locations.length)] + "'";
}
public static String selectGroupBy(String field, String dbName, String stbName) {
return "select avg(" + field + ") from " + dbName + "." + stbName + " group by location";
}
public static String selectLike(String dbName, String stbName) {
return "select * from " + dbName + "." + stbName + " where location like 'S%'";
}
public static String selectLastOneHour(String dbName, String stbName) {
return "select * from " + dbName + "." + stbName + " where ts >= now - 1h";
}
public static String selectLastOneDay(String dbName, String stbName) {
return "select * from " + dbName + "." + stbName + " where ts >= now - 1d";
}
public static String selectLastOneWeek(String dbName, String stbName) {
return "select * from " + dbName + "." + stbName + " where ts >= now - 1w";
}
public static String selectLastOneMonth(String dbName, String stbName) {
return "select * from " + dbName + "." + stbName + " where ts >= now - 1n";
}
public static String selectLastOneYear(String dbName, String stbName) {
return "select * from " + dbName + "." + stbName + " where ts >= now - 1y";
}
// select group by
// select like
// select ts >= ts<=
}
\ No newline at end of file
package com.taosdata.example.jdbcTaosdemo.utils;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.time.Instant;
import java.util.Date;
public class TimeStampUtil {
private static final String datetimeFormat = "yyyy-MM-dd HH:mm:ss.SSS";
public static long datetimeToLong(String dateTime) {
SimpleDateFormat sdf = new SimpleDateFormat(datetimeFormat);
try {
return sdf.parse(dateTime).getTime();
} catch (ParseException e) {
throw new RuntimeException(e);
}
}
public static String longToDatetime(long time) {
SimpleDateFormat sdf = new SimpleDateFormat(datetimeFormat);
return sdf.format(new Date(time));
}
public static void main(String[] args) throws ParseException {
// Instant now = Instant.now();
// System.out.println(now);
// Instant years20Ago = now.minus(Duration.ofDays(365));
// System.out.println(years20Ago);
}
}
### 设置###
log4j.rootLogger=debug,stdout,DebugLog,ErrorLog
### 输出信息到控制抬 ###
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n
### 输出DEBUG 级别以上的日志到=logs/error.log ###
log4j.appender.DebugLog=org.apache.log4j.DailyRollingFileAppender
log4j.appender.DebugLog.File=logs/debug.log
log4j.appender.DebugLog.Append=true
log4j.appender.DebugLog.Threshold=DEBUG
log4j.appender.DebugLog.layout=org.apache.log4j.PatternLayout
log4j.appender.DebugLog.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss} [ %t:%r ] - [ %p ] %m%n
### 输出ERROR 级别以上的日志到=logs/error.log ###
log4j.appender.ErrorLog=org.apache.log4j.DailyRollingFileAppender
log4j.appender.ErrorLog.File=logs/error.log
log4j.appender.ErrorLog.Append=true
log4j.appender.ErrorLog.Threshold=ERROR
log4j.appender.ErrorLog.layout=org.apache.log4j.PatternLayout
log4j.appender.ErrorLog.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss} [ %t:%r ] - [ %p ] %m%n
\ No newline at end of file
package com.taosdata.example.jdbcTaosdemo.utils;
import org.junit.Test;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.time.Instant;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.Date;
import static org.junit.Assert.*;
public class TimeStampUtilTest {
@Test
public void datetimeToLong() {
final String startTime = "2005-01-01 00:00:00.000";
long start = TimeStampUtil.datetimeToLong(startTime);
assertEquals(1104508800000l, start);
}
@Test
public void longToDatetime() {
String datetime = TimeStampUtil.longToDatetime(1510000000000L);
assertEquals("2017-11-07 04:26:40.000", datetime);
}
@Test
public void getStartDateTime() {
int keep = 365;
Instant end = Instant.now();
System.out.println(end.toString());
System.out.println(end.toEpochMilli());
Instant start = end.minus(Duration.ofDays(keep));
System.out.println(start.toString());
System.out.println(start.toEpochMilli());
int numberOfRecordsPerTable = 10;
long timeGap = ChronoUnit.MILLIS.between(start, end) / (numberOfRecordsPerTable - 1);
System.out.println(timeGap);
System.out.println("===========================");
for (int i = 0; i < numberOfRecordsPerTable; i++) {
long ts = start.toEpochMilli() + (i * timeGap);
System.out.println(i + " : " + ts);
}
}
}
\ No newline at end of file
......@@ -292,6 +292,9 @@ bool simExecuteRunBackCmd(SScript *script, char *option) {
if (pthread_create(&newScript->bgPid, NULL, simExecuteScript, (void *)newScript) != 0) {
sprintf(script->error, "lineNum:%d. create background thread failed", script->lines[script->linePos].lineNum);
return false;
} else {
simDebug("script:%s, background thread:0x%08" PRIx64 " is created", newScript->fileName,
taosGetPthreadId(newScript->bgPid));
}
script->linePos++;
......@@ -448,7 +451,6 @@ void simCloseNativeConnect(SScript *script) {
simDebug("script:%s, taos:%p closed", script->fileName, script->taos);
taos_close(script->taos);
taosMsleep(1200);
script->taos = NULL;
}
......
......@@ -40,14 +40,14 @@ int32_t main(int32_t argc, char *argv[]) {
printf("usage: %s [options] \n", argv[0]);
printf(" [-c config]: config directory, default is: %s\n", configDir);
printf(" [-f script]: script filename\n");
exit(0);
return 0;
}
}
if (!simSystemInit()) {
simError("failed to initialize the system");
simSystemCleanUp();
exit(1);
return -1;
}
simInfo("simulator is running ...");
......@@ -56,7 +56,7 @@ int32_t main(int32_t argc, char *argv[]) {
SScript *script = simParseScript(scriptFile);
if (script == NULL) {
simError("parse script file:%s failed", scriptFile);
exit(-1);
return -1;
}
simScriptList[++simScriptPos] = script;
......
......@@ -93,27 +93,34 @@ void simFreeScript(SScript *script) {
for (int32_t i = 0; i < script->bgScriptLen; ++i) {
SScript *bgScript = script->bgScripts[i];
simInfo("script:%s, set stop flag", script->fileName);
simDebug("script:%s, is background script, set stop flag", bgScript->fileName);
bgScript->killed = true;
if (taosCheckPthreadValid(bgScript->bgPid)) {
pthread_join(bgScript->bgPid, NULL);
}
}
simDebug("script:%s, background thread joined", bgScript->fileName);
taos_close(bgScript->taos);
tfree(bgScript->lines);
tfree(bgScript->optionBuffer);
tfree(bgScript);
}
simDebug("script:%s, is freed", script->fileName);
simDebug("script:%s, is cleaned", script->fileName);
taos_close(script->taos);
tfree(script->lines);
tfree(script->optionBuffer);
tfree(script);
}
}
SScript *simProcessCallOver(SScript *script) {
if (script->type == SIM_SCRIPT_TYPE_MAIN) {
simDebug("script:%s, is main script, set stop flag", script->fileName);
if (script->killed) {
simInfo("script:" FAILED_PREFIX "%s" FAILED_POSTFIX ", " FAILED_PREFIX "failed" FAILED_POSTFIX ", error:%s",
script->fileName, script->error);
exit(-1);
return NULL;
} else {
simInfo("script:" SUCCESS_PREFIX "%s" SUCCESS_POSTFIX ", " SUCCESS_PREFIX "success" SUCCESS_POSTFIX,
script->fileName);
......@@ -125,13 +132,13 @@ SScript *simProcessCallOver(SScript *script) {
if (simScriptPos == -1) {
simInfo("----------------------------------------------------------------------");
simInfo("Simulation Test Done, " SUCCESS_PREFIX "%d" SUCCESS_POSTFIX " Passed:\n", simScriptSucced);
exit(0);
return NULL;
}
return simScriptList[simScriptPos];
}
} else {
simInfo("script:%s, is stopped by main script", script->fileName);
simDebug("script:%s, is stopped", script->fileName);
simFreeScript(script);
return NULL;
}
......@@ -161,5 +168,6 @@ void *simExecuteScript(void *inputScript) {
}
}
simInfo("thread is stopped");
return NULL;
}
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册