Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
apache
Iotdb
提交
ca45800e
I
Iotdb
项目概览
apache
/
Iotdb
12 个月 前同步成功
通知
25
Star
3344
Fork
916
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
I
Iotdb
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
ca45800e
编写于
8月 24, 2019
作者:
Q
qiaojialin
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
make session module depend on service-rpc instead of jdbc
上级
07775693
变更
16
隐藏空白更改
内联
并排
Showing
16 changed file
with
243 addition
and
163 deletion
+243
-163
jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
+4
-4
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
.../src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
+17
-44
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBDatabaseMetadata.java
...ain/java/org/apache/iotdb/jdbc/IoTDBDatabaseMetadata.java
+15
-9
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBQueryResultSet.java
.../main/java/org/apache/iotdb/jdbc/IoTDBQueryResultSet.java
+6
-4
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBSQLException.java
...rc/main/java/org/apache/iotdb/jdbc/IoTDBSQLException.java
+4
-0
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
+19
-5
jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
+0
-73
jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
+3
-2
service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRPCException.java
...src/main/java/org/apache/iotdb/rpc/IoTDBRPCException.java
+11
-0
service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java
service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java
+26
-0
service-rpc/src/main/java/org/apache/iotdb/rpc/SynchronizedHandler.java
...c/main/java/org/apache/iotdb/rpc/SynchronizedHandler.java
+36
-0
session/pom.xml
session/pom.xml
+6
-1
session/src/main/java/org/apache/iotdb/session/Config.java
session/src/main/java/org/apache/iotdb/session/Config.java
+9
-0
session/src/main/java/org/apache/iotdb/session/IoTDBSessionException.java
.../java/org/apache/iotdb/session/IoTDBSessionException.java
+1
-1
session/src/main/java/org/apache/iotdb/session/Session.java
session/src/main/java/org/apache/iotdb/session/Session.java
+14
-20
session/src/main/java/org/apache/iotdb/session/SessionUtils.java
.../src/main/java/org/apache/iotdb/session/SessionUtils.java
+72
-0
未找到文件。
jdbc/src/main/java/org/apache/iotdb/jdbc/Config.java
浏览文件 @
ca45800e
...
...
@@ -38,11 +38,11 @@ public class Config {
*/
public
static
final
String
DEFAULT_SERIES_NAME
=
"default"
;
public
static
final
String
AUTH_USER
=
"
user
"
;
public
static
final
String
DEFAULT_USER
=
"
user
"
;
public
static
final
String
AUTH_USER
=
"
root
"
;
public
static
final
String
DEFAULT_USER
=
"
root
"
;
public
static
final
String
AUTH_PASSWORD
=
"
password
"
;
public
static
final
String
DEFALUT_PASSWORD
=
"
password
"
;
public
static
final
String
AUTH_PASSWORD
=
"
root
"
;
public
static
final
String
DEFALUT_PASSWORD
=
"
root
"
;
public
static
final
int
RETRY_NUM
=
3
;
public
static
final
long
RETRY_INTERVAL
=
1000
;
...
...
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBConnection.java
浏览文件 @
ca45800e
...
...
@@ -18,10 +18,6 @@
*/
package
org.apache.iotdb.jdbc
;
import
java.lang.reflect.InvocationHandler
;
import
java.lang.reflect.InvocationTargetException
;
import
java.lang.reflect.Method
;
import
java.lang.reflect.Proxy
;
import
java.sql.Array
;
import
java.sql.Blob
;
import
java.sql.CallableStatement
;
...
...
@@ -42,6 +38,8 @@ import java.time.ZoneId;
import
java.util.Map
;
import
java.util.Properties
;
import
java.util.concurrent.Executor
;
import
org.apache.iotdb.rpc.IoTDBRPCException
;
import
org.apache.iotdb.rpc.RpcUtils
;
import
org.apache.iotdb.service.rpc.thrift.ServerProperties
;
import
org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq
;
import
org.apache.iotdb.service.rpc.thrift.TSGetTimeZoneResp
;
...
...
@@ -94,15 +92,10 @@ public class IoTDBConnection implements Connection {
// open client session
openSession
();
// Wrap the client with a thread-safe proxy to serialize the RPC calls
client
=
newSynchronizedClient
(
client
);
client
=
RpcUtils
.
newSynchronizedClient
(
client
);
autoCommit
=
false
;
}
public
static
TSIService
.
Iface
newSynchronizedClient
(
TSIService
.
Iface
client
)
{
return
(
TSIService
.
Iface
)
Proxy
.
newProxyInstance
(
IoTDBConnection
.
class
.
getClassLoader
(),
new
Class
[]{
TSIService
.
Iface
.
class
},
new
SynchronizedHandler
(
client
));
}
@Override
public
boolean
isWrapperFor
(
Class
<?>
arg0
)
throws
SQLException
{
throw
new
SQLException
(
"Method not supported"
);
...
...
@@ -427,11 +420,11 @@ public class IoTDBConnection implements Connection {
// validate connection
try
{
Utils
.
verifySuccess
(
openResp
.
getStatus
());
}
catch
(
IoTDB
SQL
Exception
e
)
{
Rpc
Utils
.
verifySuccess
(
openResp
.
getStatus
());
}
catch
(
IoTDB
RPC
Exception
e
)
{
// failed to connect, disconnect from the server
transport
.
close
();
throw
e
;
throw
new
SQLException
(
e
)
;
}
if
(
protocolVersion
.
getValue
()
!=
openResp
.
getServerProtocolVersion
().
getValue
())
{
throw
new
TException
(
String
...
...
@@ -468,7 +461,7 @@ public class IoTDBConnection implements Connection {
client
=
new
TSIService
.
Client
(
new
TBinaryProtocol
(
transport
));
}
openSession
();
client
=
newSynchronizedClient
(
client
);
client
=
RpcUtils
.
newSynchronizedClient
(
client
);
flag
=
true
;
break
;
}
...
...
@@ -489,14 +482,22 @@ public class IoTDBConnection implements Connection {
}
TSGetTimeZoneResp
resp
=
client
.
getTimeZone
();
Utils
.
verifySuccess
(
resp
.
getStatus
());
try
{
RpcUtils
.
verifySuccess
(
resp
.
getStatus
());
}
catch
(
IoTDBRPCException
e
)
{
throw
new
IoTDBSQLException
(
e
);
}
return
resp
.
getTimeZone
();
}
public
void
setTimeZone
(
String
zoneId
)
throws
TException
,
IoTDBSQLException
{
TSSetTimeZoneReq
req
=
new
TSSetTimeZoneReq
(
zoneId
);
TSSetTimeZoneResp
resp
=
client
.
setTimeZone
(
req
);
Utils
.
verifySuccess
(
resp
.
getStatus
());
try
{
RpcUtils
.
verifySuccess
(
resp
.
getStatus
());
}
catch
(
IoTDBRPCException
e
)
{
throw
new
IoTDBSQLException
(
e
);
}
this
.
zoneId
=
ZoneId
.
of
(
zoneId
);
}
...
...
@@ -512,32 +513,4 @@ public class IoTDBConnection implements Connection {
this
.
protocol
=
protocol
;
}
public
static
class
SynchronizedHandler
implements
InvocationHandler
{
private
final
TSIService
.
Iface
client
;
public
SynchronizedHandler
(
TSIService
.
Iface
client
)
{
this
.
client
=
client
;
}
@Override
public
Object
invoke
(
Object
proxy
,
Method
method
,
Object
[]
args
)
throws
Throwable
{
try
{
synchronized
(
client
)
{
return
method
.
invoke
(
client
,
args
);
}
}
catch
(
InvocationTargetException
e
)
{
// all IFace APIs throw TException
if
(
e
.
getTargetException
()
instanceof
TException
)
{
throw
e
.
getTargetException
();
}
else
{
// should not happen
throw
new
TException
(
"Error in calling method "
+
method
.
getName
(),
e
.
getTargetException
());
}
}
catch
(
Exception
e
)
{
throw
new
TException
(
"Error in calling method "
+
method
.
getName
(),
e
);
}
}
}
}
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBDatabaseMetadata.java
浏览文件 @
ca45800e
...
...
@@ -25,6 +25,8 @@ import java.sql.RowIdLifetime;
import
java.sql.SQLException
;
import
java.util.List
;
import
java.util.Set
;
import
org.apache.iotdb.rpc.IoTDBRPCException
;
import
org.apache.iotdb.rpc.RpcUtils
;
import
org.apache.iotdb.service.rpc.thrift.TSFetchMetadataReq
;
import
org.apache.iotdb.service.rpc.thrift.TSFetchMetadataResp
;
import
org.apache.iotdb.service.rpc.thrift.TSIService
;
...
...
@@ -82,9 +84,9 @@ public class IoTDBDatabaseMetadata implements DatabaseMetaData {
req
.
setColumnPath
(
schemaPattern
);
try
{
TSFetchMetadataResp
resp
=
client
.
fetchMetadata
(
req
);
Utils
.
verifySuccess
(
resp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
resp
.
getStatus
());
return
new
IoTDBMetadataResultSet
(
resp
.
getColumnsList
(),
null
,
null
);
}
catch
(
TException
e
)
{
}
catch
(
TException
|
IoTDBRPCException
e
)
{
throw
new
TException
(
"Conncetion error when fetching column metadata"
,
e
);
}
case
Constant
.
CATALOG_DEVICE
:
...
...
@@ -92,19 +94,19 @@ public class IoTDBDatabaseMetadata implements DatabaseMetaData {
req
.
setColumnPath
(
schemaPattern
);
try
{
TSFetchMetadataResp
resp
=
client
.
fetchMetadata
(
req
);
Utils
.
verifySuccess
(
resp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
resp
.
getStatus
());
return
new
IoTDBMetadataResultSet
(
resp
.
getColumnsList
(),
null
,
null
);
}
catch
(
TException
e
)
{
}
catch
(
TException
|
IoTDBRPCException
e
)
{
throw
new
TException
(
"Conncetion error when fetching delta object metadata"
,
e
);
}
case
Constant
.
CATALOG_STORAGE_GROUP
:
req
=
new
TSFetchMetadataReq
(
Constant
.
GLOBAL_SHOW_STORAGE_GROUP_REQ
);
try
{
TSFetchMetadataResp
resp
=
client
.
fetchMetadata
(
req
);
Utils
.
verifySuccess
(
resp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
resp
.
getStatus
());
Set
<
String
>
showStorageGroup
=
resp
.
getShowStorageGroups
();
return
new
IoTDBMetadataResultSet
(
null
,
showStorageGroup
,
null
);
}
catch
(
TException
e
)
{
}
catch
(
TException
|
IoTDBRPCException
e
)
{
throw
new
TException
(
"Conncetion error when fetching storage group metadata"
,
e
);
}
case
Constant
.
CATALOG_TIMESERIES
:
...
...
@@ -112,10 +114,10 @@ public class IoTDBDatabaseMetadata implements DatabaseMetaData {
req
.
setColumnPath
(
schemaPattern
);
try
{
TSFetchMetadataResp
resp
=
client
.
fetchMetadata
(
req
);
Utils
.
verifySuccess
(
resp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
resp
.
getStatus
());
List
<
List
<
String
>>
showTimeseriesList
=
resp
.
getShowTimeseriesList
();
return
new
IoTDBMetadataResultSet
(
null
,
null
,
showTimeseriesList
);
}
catch
(
TException
e
)
{
}
catch
(
TException
|
IoTDBRPCException
e
)
{
throw
new
TException
(
"Conncetion error when fetching timeseries metadata"
,
e
);
}
default
:
...
...
@@ -1247,7 +1249,11 @@ public class IoTDBDatabaseMetadata implements DatabaseMetaData {
TSFetchMetadataReq
req
=
new
TSFetchMetadataReq
(
"METADATA_IN_JSON"
);
TSFetchMetadataResp
resp
;
resp
=
client
.
fetchMetadata
(
req
);
Utils
.
verifySuccess
(
resp
.
getStatus
());
try
{
RpcUtils
.
verifySuccess
(
resp
.
getStatus
());
}
catch
(
IoTDBRPCException
e
)
{
throw
new
IoTDBSQLException
(
e
);
}
return
resp
.
getMetadataInJson
();
}
}
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBQueryResultSet.java
浏览文件 @
ca45800e
...
...
@@ -46,6 +46,8 @@ import java.util.HashMap;
import
java.util.Iterator
;
import
java.util.List
;
import
java.util.Map
;
import
org.apache.iotdb.rpc.IoTDBRPCException
;
import
org.apache.iotdb.rpc.RpcUtils
;
import
org.apache.iotdb.service.rpc.thrift.TSCloseOperationReq
;
import
org.apache.iotdb.service.rpc.thrift.TSCloseOperationResp
;
import
org.apache.iotdb.service.rpc.thrift.TSFetchResultsReq
;
...
...
@@ -214,9 +216,9 @@ public class IoTDBQueryResultSet implements ResultSet {
if
(
operationHandle
!=
null
)
{
TSCloseOperationReq
closeReq
=
new
TSCloseOperationReq
(
operationHandle
,
queryId
);
TSCloseOperationResp
closeResp
=
client
.
closeOperation
(
closeReq
);
Utils
.
verifySuccess
(
closeResp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
closeResp
.
getStatus
());
}
}
catch
(
SQL
Exception
e
)
{
}
catch
(
IoTDBRPC
Exception
e
)
{
throw
new
SQLException
(
"Error occurs for close opeation in server side becasuse "
+
e
);
}
catch
(
TException
e
)
{
throw
new
SQLException
(
...
...
@@ -708,7 +710,7 @@ public class IoTDBQueryResultSet implements ResultSet {
try
{
TSFetchResultsResp
resp
=
client
.
fetchResults
(
req
);
Utils
.
verifySuccess
(
resp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
resp
.
getStatus
());
if
(!
resp
.
hasResultSet
)
{
emptyResultSet
=
true
;
}
else
{
...
...
@@ -716,7 +718,7 @@ public class IoTDBQueryResultSet implements ResultSet {
List
<
RowRecord
>
records
=
Utils
.
convertRowRecords
(
tsQueryDataSet
);
recordItr
=
records
.
iterator
();
}
}
catch
(
TException
e
)
{
}
catch
(
TException
|
IoTDBRPCException
e
)
{
throw
new
SQLException
(
"Cannot fetch result from server, because of network connection: {} "
,
e
);
}
...
...
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBSQLException.java
浏览文件 @
ca45800e
...
...
@@ -29,4 +29,8 @@ public class IoTDBSQLException extends SQLException {
super
(
reason
);
}
public
IoTDBSQLException
(
Throwable
cause
)
{
super
(
cause
);
}
}
jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java
浏览文件 @
ca45800e
...
...
@@ -30,6 +30,8 @@ import java.time.ZoneId;
import
java.util.ArrayList
;
import
java.util.List
;
import
java.util.concurrent.atomic.AtomicLong
;
import
org.apache.iotdb.rpc.IoTDBRPCException
;
import
org.apache.iotdb.rpc.RpcUtils
;
import
org.apache.iotdb.service.rpc.thrift.TSCancelOperationReq
;
import
org.apache.iotdb.service.rpc.thrift.TSCancelOperationResp
;
import
org.apache.iotdb.service.rpc.thrift.TSCloseOperationReq
;
...
...
@@ -131,7 +133,7 @@ public class IoTDBStatement implements Statement {
if
(
operationHandle
!=
null
)
{
TSCancelOperationReq
closeReq
=
new
TSCancelOperationReq
(
operationHandle
);
TSCancelOperationResp
closeResp
=
client
.
cancelOperation
(
closeReq
);
Utils
.
verifySuccess
(
closeResp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
closeResp
.
getStatus
());
}
}
catch
(
Exception
e
)
{
throw
new
SQLException
(
"Error occurs when canceling statement."
,
e
);
...
...
@@ -158,7 +160,7 @@ public class IoTDBStatement implements Statement {
TSCloseOperationReq
closeReq
=
new
TSCloseOperationReq
(
operationHandle
,
-
1
);
closeReq
.
setStmtId
(
stmtId
);
TSCloseOperationResp
closeResp
=
client
.
closeOperation
(
closeReq
);
Utils
.
verifySuccess
(
closeResp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
closeResp
.
getStatus
());
}
}
catch
(
Exception
e
)
{
throw
new
SQLException
(
"Error occurs when closing statement."
,
e
);
...
...
@@ -246,7 +248,11 @@ public class IoTDBStatement implements Statement {
TSExecuteStatementReq
execReq
=
new
TSExecuteStatementReq
(
sessionHandle
,
sql
);
TSExecuteStatementResp
execResp
=
client
.
executeStatement
(
execReq
);
operationHandle
=
execResp
.
getOperationHandle
();
Utils
.
verifySuccess
(
execResp
.
getStatus
());
try
{
RpcUtils
.
verifySuccess
(
execResp
.
getStatus
());
}
catch
(
IoTDBRPCException
e
)
{
throw
new
SQLException
(
e
);
}
if
(
execResp
.
getOperationHandle
().
hasResultSet
)
{
IoTDBQueryResultSet
resSet
=
new
IoTDBQueryResultSet
(
this
,
execResp
.
getColumns
(),
client
,
...
...
@@ -346,7 +352,11 @@ public class IoTDBStatement implements Statement {
TSExecuteStatementReq
execReq
=
new
TSExecuteStatementReq
(
sessionHandle
,
sql
);
TSExecuteStatementResp
execResp
=
client
.
executeQueryStatement
(
execReq
);
operationHandle
=
execResp
.
getOperationHandle
();
Utils
.
verifySuccess
(
execResp
.
getStatus
());
try
{
RpcUtils
.
verifySuccess
(
execResp
.
getStatus
());
}
catch
(
IoTDBRPCException
e
)
{
throw
new
SQLException
(
e
);
}
IoTDBQueryResultSet
resSet
=
new
IoTDBQueryResultSet
(
this
,
execResp
.
getColumns
(),
client
,
operationHandle
,
sql
,
execResp
.
getOperationType
(),
execResp
.
getDataTypeList
(),
queryId
.
getAndIncrement
());
...
...
@@ -399,7 +409,11 @@ public class IoTDBStatement implements Statement {
TSExecuteStatementReq
execReq
=
new
TSExecuteStatementReq
(
sessionHandle
,
sql
);
TSExecuteStatementResp
execResp
=
client
.
executeUpdateStatement
(
execReq
);
operationHandle
=
execResp
.
getOperationHandle
();
Utils
.
verifySuccess
(
execResp
.
getStatus
());
try
{
RpcUtils
.
verifySuccess
(
execResp
.
getStatus
());
}
catch
(
IoTDBRPCException
e
)
{
throw
new
IoTDBSQLException
(
e
);
}
return
0
;
}
...
...
jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java
浏览文件 @
ca45800e
...
...
@@ -27,8 +27,6 @@ import java.util.regex.Pattern;
import
org.apache.iotdb.service.rpc.thrift.TSDataValue
;
import
org.apache.iotdb.service.rpc.thrift.TSQueryDataSet
;
import
org.apache.iotdb.service.rpc.thrift.TSRowRecord
;
import
org.apache.iotdb.service.rpc.thrift.TS_Status
;
import
org.apache.iotdb.service.rpc.thrift.TS_StatusCode
;
import
org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException
;
import
org.apache.iotdb.tsfile.file.metadata.enums.TSDataType
;
import
org.apache.iotdb.tsfile.read.common.Field
;
...
...
@@ -75,17 +73,6 @@ public class Utils {
return
params
;
}
/**
* verify success.
*
* @param status -status
*/
public
static
void
verifySuccess
(
TS_Status
status
)
throws
IoTDBSQLException
{
if
(
status
.
getStatusCode
()
!=
TS_StatusCode
.
SUCCESS_STATUS
)
{
throw
new
IoTDBSQLException
(
status
.
errorMessage
);
}
}
/**
* convert row records.
*
...
...
@@ -148,64 +135,4 @@ public class Utils {
}
}
public
static
ByteBuffer
getTimeBuffer
(
RowBatch
rowBatch
)
{
ByteBuffer
timeBuffer
=
ByteBuffer
.
allocate
(
rowBatch
.
getTimeBytesSize
());
for
(
int
i
=
0
;
i
<
rowBatch
.
batchSize
;
i
++)
{
timeBuffer
.
putLong
(
rowBatch
.
timestamps
[
i
]);
}
timeBuffer
.
flip
();
return
timeBuffer
;
}
public
static
ByteBuffer
getValueBuffer
(
RowBatch
rowBatch
)
{
ByteBuffer
valueBuffer
=
ByteBuffer
.
allocate
(
rowBatch
.
getValueBytesSize
());
for
(
int
i
=
0
;
i
<
rowBatch
.
measurements
.
size
();
i
++)
{
TSDataType
dataType
=
rowBatch
.
measurements
.
get
(
i
).
getType
();
switch
(
dataType
)
{
case
INT32:
int
[]
intValues
=
(
int
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putInt
(
intValues
[
index
]);
}
break
;
case
INT64:
long
[]
longValues
=
(
long
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putLong
(
longValues
[
index
]);
}
break
;
case
FLOAT:
float
[]
floatValues
=
(
float
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putFloat
(
floatValues
[
index
]);
}
break
;
case
DOUBLE:
double
[]
doubleValues
=
(
double
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putDouble
(
doubleValues
[
index
]);
}
break
;
case
BOOLEAN:
boolean
[]
boolValues
=
(
boolean
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
put
(
BytesUtils
.
boolToByte
(
boolValues
[
index
]));
}
break
;
case
TEXT:
Binary
[]
binaryValues
=
(
Binary
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putInt
(
binaryValues
[
index
].
getLength
());
valueBuffer
.
put
(
binaryValues
[
index
].
getValues
());
}
break
;
default
:
throw
new
UnSupportedDataTypeException
(
String
.
format
(
"Data type %s is not supported."
,
dataType
));
}
}
valueBuffer
.
flip
();
return
valueBuffer
;
}
}
jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java
浏览文件 @
ca45800e
...
...
@@ -26,6 +26,7 @@ import java.nio.ByteBuffer;
import
java.util.ArrayList
;
import
java.util.List
;
import
java.util.Properties
;
import
org.apache.iotdb.rpc.RpcUtils
;
import
org.apache.iotdb.service.rpc.thrift.TSDataValue
;
import
org.apache.iotdb.service.rpc.thrift.TSQueryDataSet
;
import
org.apache.iotdb.service.rpc.thrift.TSRowRecord
;
...
...
@@ -69,13 +70,13 @@ public class UtilsTest {
@Test
public
void
testVerifySuccess
()
{
try
{
Utils
.
verifySuccess
(
new
TS_Status
(
TS_StatusCode
.
SUCCESS_STATUS
));
Rpc
Utils
.
verifySuccess
(
new
TS_Status
(
TS_StatusCode
.
SUCCESS_STATUS
));
}
catch
(
Exception
e
)
{
fail
();
}
try
{
Utils
.
verifySuccess
(
new
TS_Status
(
TS_StatusCode
.
ERROR_STATUS
));
Rpc
Utils
.
verifySuccess
(
new
TS_Status
(
TS_StatusCode
.
ERROR_STATUS
));
}
catch
(
Exception
e
)
{
return
;
}
...
...
service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRPCException.java
0 → 100644
浏览文件 @
ca45800e
package
org.apache.iotdb.rpc
;
public
class
IoTDBRPCException
extends
Exception
{
private
static
final
long
serialVersionUID
=
-
1268775292265203036L
;
public
IoTDBRPCException
(
String
reason
)
{
super
(
reason
);
}
}
service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java
0 → 100644
浏览文件 @
ca45800e
package
org.apache.iotdb.rpc
;
import
java.lang.reflect.Proxy
;
import
org.apache.iotdb.service.rpc.thrift.TSIService
;
import
org.apache.iotdb.service.rpc.thrift.TS_Status
;
import
org.apache.iotdb.service.rpc.thrift.TS_StatusCode
;
public
class
RpcUtils
{
public
static
TSIService
.
Iface
newSynchronizedClient
(
TSIService
.
Iface
client
)
{
return
(
TSIService
.
Iface
)
Proxy
.
newProxyInstance
(
RpcUtils
.
class
.
getClassLoader
(),
new
Class
[]{
TSIService
.
Iface
.
class
},
new
SynchronizedHandler
(
client
));
}
/**
* verify success.
*
* @param status -status
*/
public
static
void
verifySuccess
(
TS_Status
status
)
throws
IoTDBRPCException
{
if
(
status
.
getStatusCode
()
!=
TS_StatusCode
.
SUCCESS_STATUS
)
{
throw
new
IoTDBRPCException
(
status
.
errorMessage
);
}
}
}
service-rpc/src/main/java/org/apache/iotdb/rpc/SynchronizedHandler.java
0 → 100644
浏览文件 @
ca45800e
package
org.apache.iotdb.rpc
;
import
java.lang.reflect.InvocationHandler
;
import
java.lang.reflect.InvocationTargetException
;
import
java.lang.reflect.Method
;
import
org.apache.iotdb.service.rpc.thrift.TSIService
;
import
org.apache.thrift.TException
;
public
class
SynchronizedHandler
implements
InvocationHandler
{
private
final
TSIService
.
Iface
client
;
public
SynchronizedHandler
(
TSIService
.
Iface
client
)
{
this
.
client
=
client
;
}
@Override
public
Object
invoke
(
Object
proxy
,
Method
method
,
Object
[]
args
)
throws
Throwable
{
try
{
synchronized
(
client
)
{
return
method
.
invoke
(
client
,
args
);
}
}
catch
(
InvocationTargetException
e
)
{
// all IFace APIs throw TException
if
(
e
.
getTargetException
()
instanceof
TException
)
{
throw
e
.
getTargetException
();
}
else
{
// should not happen
throw
new
TException
(
"Error in calling method "
+
method
.
getName
(),
e
.
getTargetException
());
}
}
catch
(
Exception
e
)
{
throw
new
TException
(
"Error in calling method "
+
method
.
getName
(),
e
);
}
}
}
session/pom.xml
浏览文件 @
ca45800e
...
...
@@ -54,7 +54,12 @@
<dependencies>
<dependency>
<groupId>
org.apache.iotdb
</groupId>
<artifactId>
iotdb-jdbc
</artifactId>
<artifactId>
service-rpc
</artifactId>
<version>
0.9.0-SNAPSHOT
</version>
</dependency>
<dependency>
<groupId>
org.apache.iotdb
</groupId>
<artifactId>
tsfile
</artifactId>
<version>
0.9.0-SNAPSHOT
</version>
</dependency>
</dependencies>
...
...
session/src/main/java/org/apache/iotdb/session/Config.java
0 → 100644
浏览文件 @
ca45800e
package
org.apache.iotdb.session
;
public
class
Config
{
public
static
final
String
DEFAULT_USER
=
"root"
;
public
static
final
String
DEFALUT_PASSWORD
=
"root"
;
}
session/src/main/java/org/apache/iotdb/
except
ion/IoTDBSessionException.java
→
session/src/main/java/org/apache/iotdb/
sess
ion/IoTDBSessionException.java
浏览文件 @
ca45800e
...
...
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
package
org.apache.iotdb.
except
ion
;
package
org.apache.iotdb.
sess
ion
;
public
class
IoTDBSessionException
extends
RuntimeException
{
...
...
session/src/main/java/org/apache/iotdb/session/Session.java
浏览文件 @
ca45800e
...
...
@@ -19,11 +19,8 @@
package
org.apache.iotdb.session
;
import
java.time.ZoneId
;
import
org.apache.iotdb.exception.IoTDBSessionException
;
import
org.apache.iotdb.jdbc.Config
;
import
org.apache.iotdb.jdbc.IoTDBConnection
;
import
org.apache.iotdb.jdbc.IoTDBSQLException
;
import
org.apache.iotdb.jdbc.Utils
;
import
org.apache.iotdb.rpc.IoTDBRPCException
;
import
org.apache.iotdb.rpc.RpcUtils
;
import
org.apache.iotdb.service.rpc.thrift.TSBatchInsertionReq
;
import
org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq
;
import
org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp
;
...
...
@@ -35,6 +32,7 @@ import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
import
org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq
;
import
org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneResp
;
import
org.apache.iotdb.service.rpc.thrift.TS_SessionHandle
;
import
org.apache.iotdb.service.rpc.thrift.TS_StatusCode
;
import
org.apache.iotdb.tsfile.write.record.RowBatch
;
import
org.apache.iotdb.tsfile.write.schema.MeasurementSchema
;
import
org.apache.thrift.TException
;
...
...
@@ -102,13 +100,8 @@ public class Session {
try
{
TSOpenSessionResp
openResp
=
client
.
openSession
(
openReq
);
// validate connection
try
{
Utils
.
verifySuccess
(
openResp
.
getStatus
());
}
catch
(
IoTDBSQLException
e
)
{
transport
.
close
();
throw
e
;
}
// validate connectiontry {
RpcUtils
.
verifySuccess
(
openResp
.
getStatus
());
if
(
protocolVersion
.
getValue
()
!=
openResp
.
getServerProtocolVersion
().
getValue
())
{
throw
new
TException
(
String
...
...
@@ -124,13 +117,14 @@ public class Session {
zoneId
=
ZoneId
.
of
(
getTimeZone
());
}
}
catch
(
TException
|
IoTDBSQLException
e
)
{
}
catch
(
TException
|
IoTDBRPCException
e
)
{
transport
.
close
();
throw
new
IoTDBSessionException
(
String
.
format
(
"Can not open session to %s:%s with user: %s."
,
host
,
port
,
username
),
e
);
}
isClosed
=
false
;
client
=
IoTDBConnection
.
newSynchronizedClient
(
client
);
client
=
RpcUtils
.
newSynchronizedClient
(
client
);
}
...
...
@@ -158,8 +152,8 @@ public class Session {
request
.
addToMeasurements
(
measurementSchema
.
getMeasurementId
());
request
.
addToTypes
(
measurementSchema
.
getType
().
ordinal
());
}
request
.
setTimestamps
(
Utils
.
getTimeBuffer
(
rowBatch
));
request
.
setValues
(
Utils
.
getValueBuffer
(
rowBatch
));
request
.
setTimestamps
(
Session
Utils
.
getTimeBuffer
(
rowBatch
));
request
.
setValues
(
Session
Utils
.
getValueBuffer
(
rowBatch
));
request
.
setSize
(
rowBatch
.
batchSize
);
try
{
...
...
@@ -169,20 +163,20 @@ public class Session {
}
}
public
String
getTimeZone
()
throws
TException
,
IoTDB
SQL
Exception
{
public
String
getTimeZone
()
throws
TException
,
IoTDB
RPC
Exception
{
if
(
zoneId
!=
null
)
{
return
zoneId
.
toString
();
}
TSGetTimeZoneResp
resp
=
client
.
getTimeZone
();
Utils
.
verifySuccess
(
resp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
resp
.
getStatus
());
return
resp
.
getTimeZone
();
}
public
void
setTimeZone
(
String
zoneId
)
throws
TException
,
IoTDB
SQL
Exception
{
public
void
setTimeZone
(
String
zoneId
)
throws
TException
,
IoTDB
RPC
Exception
{
TSSetTimeZoneReq
req
=
new
TSSetTimeZoneReq
(
zoneId
);
TSSetTimeZoneResp
resp
=
client
.
setTimeZone
(
req
);
Utils
.
verifySuccess
(
resp
.
getStatus
());
Rpc
Utils
.
verifySuccess
(
resp
.
getStatus
());
this
.
zoneId
=
ZoneId
.
of
(
zoneId
);
}
...
...
session/src/main/java/org/apache/iotdb/session/SessionUtils.java
0 → 100644
浏览文件 @
ca45800e
package
org.apache.iotdb.session
;
import
java.nio.ByteBuffer
;
import
org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException
;
import
org.apache.iotdb.tsfile.file.metadata.enums.TSDataType
;
import
org.apache.iotdb.tsfile.utils.Binary
;
import
org.apache.iotdb.tsfile.utils.BytesUtils
;
import
org.apache.iotdb.tsfile.write.record.RowBatch
;
public
class
SessionUtils
{
public
static
ByteBuffer
getTimeBuffer
(
RowBatch
rowBatch
)
{
ByteBuffer
timeBuffer
=
ByteBuffer
.
allocate
(
rowBatch
.
getTimeBytesSize
());
for
(
int
i
=
0
;
i
<
rowBatch
.
batchSize
;
i
++)
{
timeBuffer
.
putLong
(
rowBatch
.
timestamps
[
i
]);
}
timeBuffer
.
flip
();
return
timeBuffer
;
}
public
static
ByteBuffer
getValueBuffer
(
RowBatch
rowBatch
)
{
ByteBuffer
valueBuffer
=
ByteBuffer
.
allocate
(
rowBatch
.
getValueBytesSize
());
for
(
int
i
=
0
;
i
<
rowBatch
.
measurements
.
size
();
i
++)
{
TSDataType
dataType
=
rowBatch
.
measurements
.
get
(
i
).
getType
();
switch
(
dataType
)
{
case
INT32:
int
[]
intValues
=
(
int
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putInt
(
intValues
[
index
]);
}
break
;
case
INT64:
long
[]
longValues
=
(
long
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putLong
(
longValues
[
index
]);
}
break
;
case
FLOAT:
float
[]
floatValues
=
(
float
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putFloat
(
floatValues
[
index
]);
}
break
;
case
DOUBLE:
double
[]
doubleValues
=
(
double
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putDouble
(
doubleValues
[
index
]);
}
break
;
case
BOOLEAN:
boolean
[]
boolValues
=
(
boolean
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
put
(
BytesUtils
.
boolToByte
(
boolValues
[
index
]));
}
break
;
case
TEXT:
Binary
[]
binaryValues
=
(
Binary
[])
rowBatch
.
values
[
i
];
for
(
int
index
=
0
;
index
<
rowBatch
.
batchSize
;
index
++)
{
valueBuffer
.
putInt
(
binaryValues
[
index
].
getLength
());
valueBuffer
.
put
(
binaryValues
[
index
].
getValues
());
}
break
;
default
:
throw
new
UnSupportedDataTypeException
(
String
.
format
(
"Data type %s is not supported."
,
dataType
));
}
}
valueBuffer
.
flip
();
return
valueBuffer
;
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录