This is an automated email from the ASF dual-hosted git repository. qiaojialin pushed a commit to branch refine_exception_code in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit abc6212905b0c2cc59613ccbf821f1d746651bf1 Author: qiaojialin <[email protected]> AuthorDate: Mon Mar 9 20:26:53 2020 +0800 modify return code --- .../apache/iotdb/rocketmq/RocketMQConsumer.java | 4 +- .../org/apache/iotdb/jdbc/IoTDBSQLException.java | 2 +- .../java/org/apache/iotdb/jdbc/IoTDBStatement.java | 48 ++- .../test/java/org/apache/iotdb/jdbc/BatchTest.java | 54 ++-- .../org/apache/iotdb/jdbc/IoTDBConnectionTest.java | 2 +- .../iotdb/jdbc/IoTDBPreparedStatementTest.java | 2 +- .../apache/iotdb/jdbc/IoTDBQueryResultSetTest.java | 4 +- .../org/apache/iotdb/jdbc/IoTDBStatementTest.java | 4 +- .../test/java/org/apache/iotdb/jdbc/UtilsTest.java | 7 +- .../org/apache/iotdb/db/engine/StorageEngine.java | 3 +- .../engine/storagegroup/StorageGroupProcessor.java | 20 +- .../db/engine/storagegroup/TsFileProcessor.java | 6 +- .../apache/iotdb/db/metrics/ui/MetricsPage.java | 4 +- .../apache/iotdb/db/qp/executor/IPlanExecutor.java | 3 +- .../apache/iotdb/db/qp/executor/PlanExecutor.java | 3 +- .../org/apache/iotdb/db/service/StaticResps.java | 3 +- .../org/apache/iotdb/db/service/TSServiceImpl.java | 325 ++++++++------------- .../apache/iotdb/rpc/BatchExecutionException.java | 27 ++ .../main/java/org/apache/iotdb/rpc/RpcUtils.java | 92 +++++- .../iotdb/rpc/StatementExecutionException.java | 13 + .../java/org/apache/iotdb/rpc/TSStatusCode.java | 2 +- service-rpc/src/main/thrift/rpc.thrift | 21 +- .../java/org/apache/iotdb/session/Session.java | 110 +++---- .../org/apache/iotdb/session/IoTDBSessionIT.java | 65 +++-- 24 files changed, 419 insertions(+), 405 deletions(-) diff --git a/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java index 59f7651..2b8594b 100644 --- a/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java +++ b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java @@ -84,7 +84,7 @@ public class RocketMQConsumer { session.createTimeseries(timeseries, dataType, encoding, compressionType); } - private void insert(String data) throws IoTDBConnectionException { + private void insert(String data) throws IoTDBConnectionException, StatementExecutionException { String[] dataArray = data.split(","); String device = dataArray[0]; long time = Long.parseLong(dataArray[1]); @@ -120,7 +120,7 @@ public class RocketMQConsumer { new String(msg.getBody()))); try { insert(new String(msg.getBody())); - } catch (IoTDBConnectionException e) { + } catch (Exception e) { logger.error(e.getMessage()); } } diff --git a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBSQLException.java b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBSQLException.java index f924078..9403b36 100644 --- a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBSQLException.java +++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBSQLException.java @@ -31,7 +31,7 @@ public class IoTDBSQLException extends SQLException { } public IoTDBSQLException(String reason, TSStatus status) { - super(reason, status.sqlState, status.statusType.code); + super(reason, status.message, status.code); } public IoTDBSQLException(Throwable cause) { diff --git a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java index 9ea181e..6ad35ba 100644 --- a/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java +++ b/jdbc/src/main/java/org/apache/iotdb/jdbc/IoTDBStatement.java @@ -258,41 +258,27 @@ public class IoTDBStatement implements Statement { } } - private int[] executeBatchSQL() throws TException, SQLException { + private int[] executeBatchSQL() throws TException, BatchUpdateException { isCancelled = false; - TSExecuteBatchStatementReq execReq = new TSExecuteBatchStatementReq(sessionId, - batchSQLList); + TSExecuteBatchStatementReq execReq = new TSExecuteBatchStatementReq(sessionId, batchSQLList); TSExecuteBatchStatementResp execResp = client.executeBatchStatement(execReq); - if (execResp.getStatus().getStatusType().getCode() == TSStatusCode.SUCCESS_STATUS - .getStatusCode()) { - if (execResp.getResult() == null) { - return new int[0]; - } else { - List<Integer> result = execResp.getResult(); - int len = result.size(); - int[] updateArray = new int[len]; - for (int i = 0; i < len; i++) { - updateArray[i] = result.get(i); - } - return updateArray; - } - } else { - BatchUpdateException exception; - if (execResp.getResult() == null) { - exception = new BatchUpdateException(execResp.getStatus().getStatusType().getMessage(), - new int[0]); - } else { - List<Integer> result = execResp.getResult(); - int len = result.size(); - int[] updateArray = new int[len]; - for (int i = 0; i < len; i++) { - updateArray[i] = result.get(i); - } - exception = new BatchUpdateException(execResp.getStatus().getStatusType().getMessage(), - updateArray); + if (execResp.statusList == null) { + + } + int[] result = new int[execResp.statusList.size()]; + boolean allSuccess = true; + String message = ""; + for (int i = 0; i < result.length; i++) { + result[i] = execResp.statusList.get(i).code; + if (result[i] != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + allSuccess = false; + message = execResp.statusList.get(i).message; } - throw exception; } + if (!allSuccess) { + throw new BatchUpdateException(message, result); + } + return result; } @Override diff --git a/jdbc/src/test/java/org/apache/iotdb/jdbc/BatchTest.java b/jdbc/src/test/java/org/apache/iotdb/jdbc/BatchTest.java index f19edd9..953f5a8 100644 --- a/jdbc/src/test/java/org/apache/iotdb/jdbc/BatchTest.java +++ b/jdbc/src/test/java/org/apache/iotdb/jdbc/BatchTest.java @@ -29,12 +29,12 @@ import java.sql.Statement; import java.time.ZoneId; import java.util.ArrayList; import java.util.List; +import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementReq; import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp; import org.apache.iotdb.service.rpc.thrift.TSIService; import org.apache.iotdb.service.rpc.thrift.TSStatus; -import org.apache.iotdb.service.rpc.thrift.TSStatusType; import org.apache.thrift.TException; import org.junit.After; import org.junit.Before; @@ -51,8 +51,8 @@ public class BatchTest { private long sessionId; @Mock private IoTDBStatement statement; - private TSStatusType successStatus = new TSStatusType(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); - private TSStatusType errorStatus = new TSStatusType(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(), ""); + private TSStatus successStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); + private TSStatus errorStatus = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(), ""); private TSStatus Status_SUCCESS = new TSStatus(successStatus); private TSStatus Status_ERROR = new TSStatus(errorStatus); private TSExecuteBatchStatementResp resp; @@ -74,25 +74,26 @@ public class BatchTest { @Test public void testExecuteBatchSQL1() throws SQLException, TException { Statement statement = connection.createStatement(); - resp = new TSExecuteBatchStatementResp(Status_SUCCESS); + resp = new TSExecuteBatchStatementResp(); + resp = RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.SUCCESS_STATUS); when(client.executeBatchStatement(any(TSExecuteBatchStatementReq.class))).thenReturn(resp); int[] result = statement.executeBatch(); - assertEquals(result.length, 0); + assertEquals(result.length, 1); - List<Integer> resExpected = new ArrayList<Integer>() { + List<TSStatus> resExpected = new ArrayList<TSStatus>() { { - add(Statement.SUCCESS_NO_INFO); - add(Statement.SUCCESS_NO_INFO); - add(Statement.SUCCESS_NO_INFO); - add(Statement.SUCCESS_NO_INFO); - add(Statement.SUCCESS_NO_INFO); - add(Statement.EXECUTE_FAILED); - add(Statement.SUCCESS_NO_INFO); - add(Statement.SUCCESS_NO_INFO); - add(Statement.EXECUTE_FAILED); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); } }; - resp.setResult(resExpected); + resp.setStatusList(resExpected); statement.addBatch("SET STORAGE GROUP TO root.ln.wf01.wt01"); statement.addBatch( @@ -112,9 +113,9 @@ public class BatchTest { statement.addBatch( "insert into root.ln.wf01.wt01(timestamp,temperature) vvvvvv(1509465720000,20.092794)"); result = statement.executeBatch(); - assertEquals(result.length, resExpected.size()); - for (int i = 0; i < resExpected.size(); i++) { - assertEquals(result[i], (int) resExpected.get(i)); + assertEquals(resp.statusList.size(), result.length); + for (int i = 0; i < resp.statusList.size(); i++) { + assertEquals(resExpected.get(i).code, result[i]); } statement.clearBatch(); } @@ -122,7 +123,8 @@ public class BatchTest { @Test(expected = BatchUpdateException.class) public void testExecuteBatchSQL2() throws SQLException, TException { Statement statement = connection.createStatement(); - resp = new TSExecuteBatchStatementResp(Status_ERROR); + resp = RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.SQL_PARSE_ERROR); + when(client.executeBatchStatement(any(TSExecuteBatchStatementReq.class))).thenReturn(resp); statement.executeBatch(); } @@ -131,21 +133,21 @@ public class BatchTest { @Test public void testExecuteBatchSQL3() throws SQLException, TException { Statement statement = connection.createStatement(); - resp = new TSExecuteBatchStatementResp(Status_ERROR); - List<Integer> resExpected = new ArrayList<Integer>() { + resp = RpcUtils.getTSBatchExecuteStatementResp(Status_ERROR); + List<TSStatus> resExpected = new ArrayList<TSStatus>() { { - add(Statement.SUCCESS_NO_INFO); - add(Statement.EXECUTE_FAILED); + add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + add(RpcUtils.getStatus(TSStatusCode.SQL_PARSE_ERROR)); } }; - resp.setResult(resExpected); + resp.setStatusList(resExpected); when(client.executeBatchStatement(any(TSExecuteBatchStatementReq.class))).thenReturn(resp); try { statement.executeBatch(); } catch (BatchUpdateException e) { int[] result = e.getUpdateCounts(); for (int i = 0; i < resExpected.size(); i++) { - assertEquals(result[i], (int) resExpected.get(i)); + assertEquals(resExpected.get(i).code, result[i]); } return; } diff --git a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBConnectionTest.java b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBConnectionTest.java index 787c48b..c11d8e7 100644 --- a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBConnectionTest.java +++ b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBConnectionTest.java @@ -40,7 +40,7 @@ public class IoTDBConnectionTest { private TSIService.Iface client; private IoTDBConnection connection = new IoTDBConnection(); - private TSStatusType successStatus = new TSStatusType(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); + private TSStatus successStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); private TSStatus Status_SUCCESS = new TSStatus(successStatus); private long sessionId; diff --git a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBPreparedStatementTest.java b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBPreparedStatementTest.java index b052252..2ee1b1f 100644 --- a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBPreparedStatementTest.java +++ b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBPreparedStatementTest.java @@ -48,7 +48,7 @@ public class IoTDBPreparedStatementTest { @Mock private Iface client; @Mock - private TSStatusType successStatus = new TSStatusType(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); + private TSStatus successStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); private TSStatus Status_SUCCESS = new TSStatus(successStatus); private long queryId; private long sessionId; diff --git a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBQueryResultSetTest.java b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBQueryResultSetTest.java index 80ce9e4..7f1fa24 100644 --- a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBQueryResultSetTest.java +++ b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBQueryResultSetTest.java @@ -47,7 +47,6 @@ import org.apache.iotdb.service.rpc.thrift.TSFetchResultsResp; import org.apache.iotdb.service.rpc.thrift.TSIService; import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet; import org.apache.iotdb.service.rpc.thrift.TSStatus; -import org.apache.iotdb.service.rpc.thrift.TSStatusType; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.junit.Assert; import org.junit.Before; @@ -118,8 +117,7 @@ public class IoTDBQueryResultSetTest { @Mock private TSFetchResultsResp fetchResultsResp; - private TSStatusType successStatus = new TSStatusType(TSStatusCode.SUCCESS_STATUS.getStatusCode(), - ""); + private TSStatus successStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); private TSStatus Status_SUCCESS = new TSStatus(successStatus); private ZoneId zoneID = ZoneId.systemDefault(); diff --git a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBStatementTest.java b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBStatementTest.java index ebb1741..31ccd99 100644 --- a/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBStatementTest.java +++ b/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBStatementTest.java @@ -29,7 +29,7 @@ import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataReq; import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataResp; import org.apache.iotdb.service.rpc.thrift.TSIService.Iface; import org.apache.iotdb.service.rpc.thrift.TSStatus; -import org.apache.iotdb.service.rpc.thrift.TSStatusType; +import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -49,7 +49,7 @@ public class IoTDBStatementTest { @Mock private TSFetchMetadataResp fetchMetadataResp; - private TSStatusType successStatus = new TSStatusType(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); + private TSStatus successStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); private TSStatus Status_SUCCESS = new TSStatus(successStatus); private ZoneId zoneID = ZoneId.systemDefault(); diff --git a/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java b/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java index c5fa182..0b2b3a7 100644 --- a/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java +++ b/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java @@ -21,7 +21,7 @@ package org.apache.iotdb.jdbc; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSStatus; -import org.apache.iotdb.service.rpc.thrift.TSStatusType; +import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -103,15 +103,14 @@ public class UtilsTest { @Test public void testVerifySuccess() { try { - TSStatusType successStatus = new TSStatusType(TSStatusCode.SUCCESS_STATUS.getStatusCode(), - ""); + TSStatus successStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode(), ""); RpcUtils.verifySuccess(new TSStatus(successStatus)); } catch (Exception e) { fail(); } try { - TSStatusType errorStatus = new TSStatusType( + TSStatus errorStatus = new TSStatus( TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(), ""); RpcUtils.verifySuccess(new TSStatus(errorStatus)); } catch (Exception e) { diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java index 531c06c..fcc0a37 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java @@ -65,6 +65,7 @@ import org.apache.iotdb.db.service.IService; import org.apache.iotdb.db.service.ServiceType; import org.apache.iotdb.db.utils.FilePathUtils; import org.apache.iotdb.db.utils.UpgradeUtils; +import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; @@ -276,7 +277,7 @@ public class StorageEngine implements IService { * * @return result of each row */ - public Integer[] insertBatch(BatchInsertPlan batchInsertPlan) throws StorageEngineException { + public TSStatus[] insertBatch(BatchInsertPlan batchInsertPlan) throws StorageEngineException { StorageGroupProcessor storageGroupProcessor; try { storageGroupProcessor = getProcessor(batchInsertPlan.getDeviceId()); diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java index 7791966..0afbb82 100755 --- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java @@ -77,11 +77,14 @@ import org.apache.iotdb.db.qp.physical.crud.DeletePlan; import org.apache.iotdb.db.qp.physical.crud.InsertPlan; import org.apache.iotdb.db.query.context.QueryContext; import org.apache.iotdb.db.query.control.QueryFileManager; +import org.apache.iotdb.db.service.TSServiceImpl; import org.apache.iotdb.db.utils.CopyOnReadLinkedList; import org.apache.iotdb.db.utils.SchemaUtils; import org.apache.iotdb.db.utils.UpgradeUtils; import org.apache.iotdb.db.writelog.recover.TsFileRecoverPerformer; +import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; @@ -491,10 +494,10 @@ public class StorageGroupProcessor { } } - public Integer[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException { + public TSStatus[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException { writeLock(); try { - Integer[] results = new Integer[batchInsertPlan.getRowCount()]; + TSStatus[] results = new TSStatus[batchInsertPlan.getRowCount()]; /* * assume that batch has been sorted by client @@ -504,7 +507,8 @@ public class StorageGroupProcessor { long currTime = batchInsertPlan.getTimes()[loc]; // skip points that do not satisfy TTL if (!checkTTL(currTime)) { - results[loc] = TSStatusCode.OUT_OF_TTL_ERROR.getStatusCode(); + results[loc] = RpcUtils.getStatus(TSStatusCode.OUT_OF_TTL_ERROR, + "time " + currTime + " in current line is out of TTL: " + dataTTL); loc++; } else { break; @@ -527,7 +531,7 @@ public class StorageGroupProcessor { while (loc < batchInsertPlan.getRowCount()) { long time = batchInsertPlan.getTimes()[loc]; long curTimePartition = StorageEngine.fromTimeToTimePartition(time); - results[loc] = TSStatusCode.SUCCESS_STATUS.getStatusCode(); + results[loc] = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); // start next partition if (curTimePartition != beforeTimePartition) { // insert last time partition @@ -583,18 +587,18 @@ public class StorageGroupProcessor { * @param timePartitionId time partition id */ private void insertBatchToTsFileProcessor(BatchInsertPlan batchInsertPlan, - int start, int end, boolean sequence, Integer[] results, long timePartitionId) + int start, int end, boolean sequence, TSStatus[] results, long timePartitionId) throws QueryProcessException { // return when start <= end if (start >= end) { return; } - TsFileProcessor tsFileProcessor = getOrCreateTsFileProcessor(timePartitionId, - sequence); + TsFileProcessor tsFileProcessor = getOrCreateTsFileProcessor(timePartitionId, sequence); if (tsFileProcessor == null) { for (int i = start; i < end; i++) { - results[i] = TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(); + results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, + "can not create TsFileProcessor, timePartitionId: " + timePartitionId); } return; } diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java index f11c629..75c605e 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java @@ -54,7 +54,9 @@ import org.apache.iotdb.db.rescon.MemTablePool; import org.apache.iotdb.db.utils.QueryUtils; import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager; import org.apache.iotdb.db.writelog.node.WriteLogNode; +import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; @@ -183,7 +185,7 @@ public class TsFileProcessor { } public boolean insertBatch(BatchInsertPlan batchInsertPlan, int start, int end, - Integer[] results) throws QueryProcessException { + TSStatus[] results) throws QueryProcessException { if (workMemTable == null) { workMemTable = MemTablePool.getInstance().getAvailableMemTable(this); @@ -201,7 +203,7 @@ public class TsFileProcessor { logger.error("{}: {} write WAL failed", storageGroupName, tsFileResource.getFile().getName(), e); for (int i = start; i < end; i++) { - results[i] = TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode(); + results[i] = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage()); } return false; } diff --git a/server/src/main/java/org/apache/iotdb/db/metrics/ui/MetricsPage.java b/server/src/main/java/org/apache/iotdb/db/metrics/ui/MetricsPage.java index 6380150..f325a25 100644 --- a/server/src/main/java/org/apache/iotdb/db/metrics/ui/MetricsPage.java +++ b/server/src/main/java/org/apache/iotdb/db/metrics/ui/MetricsPage.java @@ -100,8 +100,8 @@ public class MetricsPage { for (int i = (list.size() - 1); i >= 0; i--) { sqlArgument = list.get(i); resp = sqlArgument.getTSExecuteStatementResp(); - errMsg = resp.getStatus().getStatusType().getMessage(); - statusCode = resp.getStatus().getStatusType().getCode(); + errMsg = resp.getStatus().message; + statusCode = resp.getStatus().code; String status; if (statusCode == 200) { status = "FINISHED"; diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java index b2466db..cf09d6a 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java @@ -28,6 +28,7 @@ import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan; import org.apache.iotdb.db.qp.physical.crud.DeletePlan; import org.apache.iotdb.db.qp.physical.crud.InsertPlan; import org.apache.iotdb.db.query.context.QueryContext; +import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException; import org.apache.iotdb.tsfile.read.common.Path; import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet; @@ -90,5 +91,5 @@ public interface IPlanExecutor { * * @return result of each row */ - Integer[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException; + TSStatus[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException; } diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java index fd34fe6..6c1b0e8 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java @@ -108,6 +108,7 @@ import org.apache.iotdb.db.utils.AuthUtils; import org.apache.iotdb.db.utils.FileLoaderUtils; import org.apache.iotdb.db.utils.TypeInferenceUtils; import org.apache.iotdb.db.utils.UpgradeUtils; +import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException; import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException; @@ -757,7 +758,7 @@ public class PlanExecutor implements IPlanExecutor { } @Override - public Integer[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException { + public TSStatus[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException { try { String[] measurementList = batchInsertPlan.getMeasurements(); String deviceId = batchInsertPlan.getDeviceId(); diff --git a/server/src/main/java/org/apache/iotdb/db/service/StaticResps.java b/server/src/main/java/org/apache/iotdb/db/service/StaticResps.java index 441a73d..14cc64d 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/StaticResps.java +++ b/server/src/main/java/org/apache/iotdb/db/service/StaticResps.java @@ -40,6 +40,7 @@ import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_VERSION; import java.util.Arrays; import java.util.Collections; import java.util.List; +import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; @@ -131,7 +132,7 @@ class StaticResps { private static TSExecuteStatementResp getExecuteResp(List<String> columns, List<String> dataTypes, boolean ignoreTimeStamp) { TSExecuteStatementResp resp = - TSServiceImpl.getTSExecuteStatementResp(TSServiceImpl.getStatus(TSStatusCode.SUCCESS_STATUS)); + RpcUtils.getTSExecuteStatementResp(TSStatusCode.SUCCESS_STATUS); resp.setIgnoreTimeStamp(ignoreTimeStamp); resp.setColumns(columns); resp.setDataTypeList(dataTypes); diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java index 409112e..91abd16 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java @@ -18,13 +18,9 @@ */ package org.apache.iotdb.db.service; -import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES; -import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_VALUE; - import java.io.IOException; import java.nio.ByteBuffer; import java.sql.SQLException; -import java.sql.Statement; import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; @@ -84,6 +80,7 @@ import org.apache.iotdb.db.tools.watermark.GroupedLSBWatermarkEncoder; import org.apache.iotdb.db.tools.watermark.WatermarkEncoder; import org.apache.iotdb.db.utils.QueryDataSetUtils; import org.apache.iotdb.db.utils.SchemaUtils; +import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.ServerProperties; import org.apache.iotdb.service.rpc.thrift.TSBatchInsertionReq; @@ -94,7 +91,6 @@ import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq; import org.apache.iotdb.service.rpc.thrift.TSDeleteDataReq; import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementReq; import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp; -import org.apache.iotdb.service.rpc.thrift.TSExecuteInsertRowInBatchResp; import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq; import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp; import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataReq; @@ -112,7 +108,6 @@ import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet; import org.apache.iotdb.service.rpc.thrift.TSQueryNonAlignDataSet; import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq; import org.apache.iotdb.service.rpc.thrift.TSStatus; -import org.apache.iotdb.service.rpc.thrift.TSStatusType; import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException; import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; @@ -197,7 +192,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { //check the version compatibility boolean compatible = checkCompatibility(req.getClient_protocol()); if (!compatible) { - tsStatus = getStatus(TSStatusCode.INCOMPATIBLE_VERSION, + tsStatus = RpcUtils.getStatus(TSStatusCode.INCOMPATIBLE_VERSION, "The version is incompatible, please upgrade to " + IoTDBConstant.VERSION); TSOpenSessionResp resp = new TSOpenSessionResp(tsStatus, TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V2); @@ -205,21 +200,19 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { return resp; } - tsStatus = getStatus(TSStatusCode.SUCCESS_STATUS, "Login successfully"); + tsStatus = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "Login successfully"); sessionId = sessionIdGenerator.incrementAndGet(); sessionIdUsernameMap.put(sessionId, req.getUsername()); sessionIdZoneIdMap.put(sessionId, config.getZoneID()); currSessionId.set(sessionId); } else { - tsStatus = getStatus(TSStatusCode.WRONG_LOGIN_PASSWORD_ERROR); + tsStatus = RpcUtils.getStatus(TSStatusCode.WRONG_LOGIN_PASSWORD_ERROR); } TSOpenSessionResp resp = new TSOpenSessionResp(tsStatus, TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V2); resp.setSessionId(sessionId); logger.info( - "{}: Login status: {}. User : {}", - IoTDBConstant.GLOBAL_DB_NAME, - tsStatus.getStatusType().getMessage(), + "{}: Login status: {}. User : {}", IoTDBConstant.GLOBAL_DB_NAME, tsStatus.message, req.getUsername()); return resp; @@ -235,9 +228,9 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { long sessionId = req.getSessionId(); TSStatus tsStatus; if (sessionIdUsernameMap.remove(sessionId) == null) { - tsStatus = getStatus(TSStatusCode.NOT_LOGIN_ERROR); + tsStatus = RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); } else { - tsStatus = getStatus(TSStatusCode.SUCCESS_STATUS); + tsStatus = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); } sessionIdZoneIdMap.remove(sessionId); @@ -260,7 +253,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } if (!exceptions.isEmpty()) { return new TSStatus( - getStatus( + RpcUtils.getStatus( TSStatusCode.CLOSE_OPERATION_ERROR, String.format( "%d errors in closeOperation, see server logs for detail", exceptions.size()))); @@ -272,7 +265,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { @Override public TSStatus cancelOperation(TSCancelOperationReq req) { // TODO implement - return getStatus(TSStatusCode.QUERY_NOT_ALLOWED, "Cancellation is not implemented"); + return RpcUtils.getStatus(TSStatusCode.QUERY_NOT_ALLOWED, "Cancellation is not implemented"); } @Override @@ -280,7 +273,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { logger.info("{}: receive close operation", IoTDBConstant.GLOBAL_DB_NAME); if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getStatus(TSStatusCode.NOT_LOGIN_ERROR); + return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); } try { // statement close @@ -299,42 +292,21 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } catch (Exception e) { logger.error("Error in closeOperation : ", e); - return getStatus(TSStatusCode.CLOSE_OPERATION_ERROR, "Error in closeOperation"); + return RpcUtils.getStatus(TSStatusCode.CLOSE_OPERATION_ERROR, "Error in closeOperation"); } - return getStatus(TSStatusCode.SUCCESS_STATUS); + return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); } /** * release single operation resource */ - protected void releaseQueryResource(long queryId) throws StorageEngineException { + private void releaseQueryResource(long queryId) throws StorageEngineException { // remove the corresponding Physical Plan queryId2DataSet.remove(queryId); QueryResourceManager.getInstance().endQuery(queryId); } - /** - * convert from TSStatusCode to TSStatus according to status code and status message - * - * @param statusType status type - */ - static TSStatus getStatus(TSStatusCode statusType) { - TSStatusType statusCodeAndMessage = new TSStatusType(statusType.getStatusCode(), ""); - return new TSStatus(statusCodeAndMessage); - } - - /** - * convert from TSStatusCode to TSStatus, which has message appending with existed status message - * - * @param statusType status type - * @param appendMessage appending message - */ - private TSStatus getStatus(TSStatusCode statusType, String appendMessage) { - TSStatusType statusCodeAndMessage = new TSStatusType(statusType.getStatusCode(), appendMessage); - return new TSStatus(statusCodeAndMessage); - } - - protected TSDataType getSeriesType(String path) throws QueryProcessException, MetadataException { + private TSDataType getSeriesType(String path) throws QueryProcessException { try { return SchemaUtils.getSeriesType(path); } catch (MetadataException e) { @@ -347,7 +319,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { TSStatus status; if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - status = getStatus(TSStatusCode.NOT_LOGIN_ERROR); + status = RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); return new TSFetchMetadataResp(status); } @@ -357,24 +329,24 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { case "METADATA_IN_JSON": String metadataInJson = getMetadataInString(); resp.setMetadataInJson(metadataInJson); - status = getStatus(TSStatusCode.SUCCESS_STATUS); + status = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); break; case "COLUMN": resp.setDataType(getSeriesType(req.getColumnPath()).toString()); - status = getStatus(TSStatusCode.SUCCESS_STATUS); + status = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); break; case "ALL_COLUMNS": resp.setColumnsList(getPaths(req.getColumnPath())); - status = getStatus(TSStatusCode.SUCCESS_STATUS); + status = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); break; default: - status = getStatus(TSStatusCode.METADATA_ERROR, req.getType()); + status = RpcUtils.getStatus(TSStatusCode.METADATA_ERROR, req.getType()); break; } } catch (MetadataException | OutOfMemoryError | QueryProcessException e) { logger.error( String.format("Failed to fetch timeseries %s's metadata", req.getColumnPath()), e); - status = getStatus(TSStatusCode.METADATA_ERROR, e.getMessage()); + status = RpcUtils.getStatus(TSStatusCode.METADATA_ERROR, e.getMessage()); resp.setStatus(status); return resp; } @@ -447,31 +419,28 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { @Override public TSExecuteBatchStatementResp executeBatchStatement(TSExecuteBatchStatementReq req) { long t1 = System.currentTimeMillis(); - List<Integer> result = new ArrayList<>(); + List<TSStatus> result = new ArrayList<>(); try { if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getTSBatchExecuteStatementResp(getStatus(TSStatusCode.NOT_LOGIN_ERROR), null); + return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.NOT_LOGIN_ERROR); } List<String> statements = req.getStatements(); boolean isAllSuccessful = true; - StringBuilder batchErrorMessage = new StringBuilder(); for (String statement : statements) { long t2 = System.currentTimeMillis(); isAllSuccessful = - executeStatementInBatch(statement, batchErrorMessage, result, req.getSessionId()) + executeStatementInBatch(statement, result, req.getSessionId()) && isAllSuccessful; Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_ONE_SQL_IN_BATCH, t2); } if (isAllSuccessful) { - return getTSBatchExecuteStatementResp( - getStatus(TSStatusCode.SUCCESS_STATUS, "Execute batch statements successfully"), - result); + return RpcUtils.getTSBatchExecuteStatementResp( + TSStatusCode.SUCCESS_STATUS, "Execute batch statements successfully"); } else { - return getTSBatchExecuteStatementResp( - getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, batchErrorMessage.toString()), result); + return RpcUtils.getTSBatchExecuteStatementResp(result); } } finally { Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_JDBC_BATCH, t1); @@ -480,8 +449,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { // execute one statement of a batch. Currently, query is not allowed in a batch statement and // on finding queries in a batch, such query will be ignored and an error will be generated - private boolean executeStatementInBatch( - String statement, StringBuilder batchErrorMessage, List<Integer> result, long sessionId) { + private boolean executeStatementInBatch(String statement, List<TSStatus> result, long sessionId) { try { PhysicalPlan physicalPlan = processor.parseSQLToPhysicalPlan(statement, sessionIdZoneIdMap.get(sessionId)); @@ -489,36 +457,33 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { throw new QueryInBatchStatementException(statement); } TSExecuteStatementResp resp = executeUpdateStatement(physicalPlan, sessionId); - if (resp.getStatus().getStatusType().getCode() - == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - result.add(Statement.SUCCESS_NO_INFO); + if (resp.getStatus().code == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + result.add(resp.status); } else { - result.add(Statement.EXECUTE_FAILED); - batchErrorMessage.append(resp.getStatus().getStatusType().getCode()).append("\n"); + result.add(resp.status); return false; } } catch (ParseCancellationException e) { logger.debug(e.getMessage()); - result.add(Statement.EXECUTE_FAILED); - batchErrorMessage.append(TSStatusCode.SQL_PARSE_ERROR.getStatusCode()).append("\n"); + result.add(RpcUtils.getStatus(TSStatusCode.SQL_PARSE_ERROR, + "Statement format is not right: " + e.getMessage())); return false; } catch (SQLParserException e) { logger.error("Error occurred when executing {}, check metadata error: ", statement, e); - result.add(Statement.EXECUTE_FAILED); - batchErrorMessage.append(TSStatusCode.METADATA_ERROR.getStatusCode()).append("\n"); + result.add(RpcUtils.getStatus( + TSStatusCode.SQL_PARSE_ERROR, "Statement format is not right: " + e.getMessage())); return false; } catch (QueryProcessException e) { logger.info( "Error occurred when executing {}, meet error while parsing SQL to physical plan: {}", - statement, - e.getMessage()); - result.add(Statement.EXECUTE_FAILED); - batchErrorMessage.append(TSStatusCode.SQL_PARSE_ERROR.getStatusCode()).append("\n"); + statement, e.getMessage()); + result.add(RpcUtils.getStatus( + TSStatusCode.QUERY_PROCESS_ERROR, "Meet error in query process: " + e.getMessage())); return false; } catch (QueryInBatchStatementException e) { logger.info("Error occurred when executing {}, query statement not allowed: ", statement, e); - result.add(Statement.EXECUTE_FAILED); - batchErrorMessage.append(TSStatusCode.QUERY_NOT_ALLOWED.getStatusCode()).append("\n"); + result.add( + RpcUtils.getStatus(TSStatusCode.QUERY_NOT_ALLOWED, "query statement not allowed: " + statement)); return false; } return true; @@ -532,13 +497,13 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { try { if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getTSExecuteStatementResp(getStatus(TSStatusCode.NOT_LOGIN_ERROR)); + return RpcUtils.getTSExecuteStatementResp(TSStatusCode.NOT_LOGIN_ERROR); } String statement = req.getStatement(); if (execAdminCommand(statement, req.getSessionId())) { - return getTSExecuteStatementResp( - getStatus(TSStatusCode.SUCCESS_STATUS, "ADMIN_COMMAND_SUCCESS")); + return RpcUtils.getTSExecuteStatementResp( + RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "ADMIN_COMMAND_SUCCESS")); } PhysicalPlan physicalPlan = processor.parseSQLToPhysicalPlan(statement, sessionIdZoneIdMap.get(req.getSessionId())); @@ -561,20 +526,20 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } } catch (ParseCancellationException e) { logger.debug(e.getMessage()); - return getTSExecuteStatementResp(getStatus(TSStatusCode.SQL_PARSE_ERROR, e.getMessage())); + return RpcUtils.getTSExecuteStatementResp(TSStatusCode.SQL_PARSE_ERROR, e.getMessage()); } catch (SQLParserException e) { logger.error("check metadata error: ", e); - return getTSExecuteStatementResp( - getStatus(TSStatusCode.METADATA_ERROR, "Check metadata error: " + e.getMessage())); + return RpcUtils.getTSExecuteStatementResp( + TSStatusCode.METADATA_ERROR, "Check metadata error: " + e.getMessage()); } catch (QueryProcessException e) { logger.info(ERROR_PARSING_SQL, e.getMessage()); - return getTSExecuteStatementResp( - getStatus( - TSStatusCode.SQL_PARSE_ERROR, "Statement format is not right: " + e.getMessage())); + return RpcUtils.getTSExecuteStatementResp( + RpcUtils.getStatus(TSStatusCode.QUERY_PROCESS_ERROR, + "Meet error in query process: " + e.getMessage())); } catch (StorageEngineException e) { logger.info(ERROR_PARSING_SQL, e.getMessage()); - return getTSExecuteStatementResp( - getStatus(TSStatusCode.READ_ONLY_SYSTEM_ERROR, e.getMessage())); + return RpcUtils.getTSExecuteStatementResp( + RpcUtils.getStatus(TSStatusCode.READ_ONLY_SYSTEM_ERROR, e.getMessage())); } } @@ -587,7 +552,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getTSExecuteStatementResp(getStatus(TSStatusCode.NOT_LOGIN_ERROR)); + return RpcUtils.getTSExecuteStatementResp(TSStatusCode.NOT_LOGIN_ERROR); } String statement = req.getStatement(); @@ -597,12 +562,12 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { processor.parseSQLToPhysicalPlan(statement, sessionIdZoneIdMap.get(req.getSessionId())); } catch (QueryProcessException | SQLParserException e) { logger.info(ERROR_PARSING_SQL, e.getMessage()); - return getTSExecuteStatementResp(getStatus(TSStatusCode.SQL_PARSE_ERROR, e.getMessage())); + return RpcUtils.getTSExecuteStatementResp(TSStatusCode.SQL_PARSE_ERROR, e.getMessage()); } if (!physicalPlan.isQuery()) { - return getTSExecuteStatementResp( - getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is not a query statement.")); + return RpcUtils.getTSExecuteStatementResp( + TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is not a query statement."); } resp = internalExecuteQueryStatement( @@ -617,12 +582,12 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { return resp; } catch (ParseCancellationException e) { logger.debug(e.getMessage()); - return getTSExecuteStatementResp(getStatus(TSStatusCode.SQL_PARSE_ERROR, - "Statement format is not right: " + e.getMessage())); + return RpcUtils.getTSExecuteStatementResp(TSStatusCode.SQL_PARSE_ERROR, + "Statement format is not right: " + e.getMessage()); } catch (SQLParserException e) { logger.error("check metadata error: ", e); - return getTSExecuteStatementResp( - getStatus(TSStatusCode.METADATA_ERROR, "Check metadata error: " + e.getMessage())); + return RpcUtils.getTSExecuteStatementResp( + TSStatusCode.METADATA_ERROR, "Check metadata error: " + e.getMessage()); } } @@ -670,8 +635,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { return resp; } catch (Exception e) { logger.error("{}: Internal server error: ", IoTDBConstant.GLOBAL_DB_NAME, e); - return getTSExecuteStatementResp( - getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage())); + return RpcUtils.getTSExecuteStatementResp(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage()); } finally { Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_QUERY, t1); } @@ -734,9 +698,8 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { case LIST_USER_PRIVILEGE: return StaticResps.LIST_USER_PRIVILEGE_RESP; default: - return getTSExecuteStatementResp( - getStatus( - TSStatusCode.SQL_PARSE_ERROR, + return RpcUtils.getTSExecuteStatementResp( + RpcUtils.getStatus(TSStatusCode.SQL_PARSE_ERROR, String.format("%s is not an auth query", authorPlan.getAuthorType()))); } } @@ -752,13 +715,13 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { // check permissions if (!checkAuthorization(physicalPlan.getPaths(), physicalPlan, username)) { - return getTSExecuteStatementResp( - getStatus( - TSStatusCode.NO_PERMISSION_ERROR, + return RpcUtils.getTSExecuteStatementResp( + RpcUtils.getStatus(TSStatusCode.NO_PERMISSION_ERROR, "No permissions for this operation " + physicalPlan.getOperatorType())); } - TSExecuteStatementResp resp = getTSExecuteStatementResp(getStatus(TSStatusCode.SUCCESS_STATUS)); + TSExecuteStatementResp resp = RpcUtils + .getTSExecuteStatementResp(TSStatusCode.SUCCESS_STATUS); // align by device query QueryPlan plan = (QueryPlan) physicalPlan; @@ -805,11 +768,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } for (String column : respColumns) { - try { - columnTypes.add(getSeriesType(column).toString()); - } catch (MetadataException e) { - throw new QueryProcessException(e); - } + columnTypes.add(getSeriesType(column).toString()); } } @@ -856,25 +815,17 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { plan.setPaths(null); } - private void getLastQueryHeaders( - QueryPlan plan, List<String> respColumns, List<String> columnTypes) - throws TException, QueryProcessException { - respColumns.add(COLUMN_TIMESERIES); - respColumns.add(COLUMN_VALUE); - columnTypes.add(TSDataType.TEXT.toString()); - columnTypes.add(TSDataType.TEXT.toString()); - } @Override public TSFetchResultsResp fetchResults(TSFetchResultsReq req) { try { if (!checkLogin(req.getSessionId())) { - return getTSFetchResultsResp(getStatus(TSStatusCode.NOT_LOGIN_ERROR)); + return RpcUtils.getTSFetchResultsResp(TSStatusCode.NOT_LOGIN_ERROR); } if (!queryId2DataSet.containsKey(req.queryId)) { - return getTSFetchResultsResp( - getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, "Has not executed query")); + return RpcUtils.getTSFetchResultsResp( + RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, "Has not executed query")); } QueryDataSet queryDataSet = queryId2DataSet.get(req.queryId); @@ -886,11 +837,8 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { QueryResourceManager.getInstance().endQuery(req.queryId); queryId2DataSet.remove(req.queryId); } - TSFetchResultsResp resp = - getTSFetchResultsResp( - getStatus( - TSStatusCode.SUCCESS_STATUS, - "FetchResult successfully. Has more result: " + hasResultSet)); + TSFetchResultsResp resp = RpcUtils.getTSFetchResultsResp( + TSStatusCode.SUCCESS_STATUS, "FetchResult successfully. Has more result: " + hasResultSet); resp.setHasResultSet(hasResultSet); resp.setQueryDataSet(result); resp.setIsAlign(true); @@ -910,10 +858,8 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { queryId2DataSet.remove(req.queryId); } TSFetchResultsResp resp = - getTSFetchResultsResp( - getStatus( - TSStatusCode.SUCCESS_STATUS, - "FetchResult successfully. Has more result: " + hasResultSet)); + RpcUtils.getTSFetchResultsResp(TSStatusCode.SUCCESS_STATUS, + "FetchResult successfully. Has more result: " + hasResultSet); resp.setHasResultSet(hasResultSet); resp.setNonAlignQueryDataSet(nonAlignResult); resp.setIsAlign(false); @@ -921,7 +867,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } } catch (Exception e) { logger.error("{}: Internal server error: ", IoTDBConstant.GLOBAL_DB_NAME, e); - return getTSFetchResultsResp(getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage())); + return RpcUtils.getTSFetchResultsResp(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage()); } } @@ -1002,7 +948,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { return queryDataSet; } - protected QueryContext genQueryContext(long queryId) { + private QueryContext genQueryContext(long queryId) { return new QueryContext(queryId); } @@ -1011,14 +957,14 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { try { if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getTSExecuteStatementResp(getStatus(TSStatusCode.NOT_LOGIN_ERROR)); + return RpcUtils.getTSExecuteStatementResp(TSStatusCode.NOT_LOGIN_ERROR); } String statement = req.getStatement(); return executeUpdateStatement(statement, req.getSessionId()); } catch (Exception e) { logger.error("{}: server Internal Error: ", IoTDBConstant.GLOBAL_DB_NAME, e); - return getTSExecuteStatementResp( - getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage())); + return RpcUtils.getTSExecuteStatementResp( + RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR, e.getMessage())); } } @@ -1029,7 +975,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } status = executePlan(plan); - TSExecuteStatementResp resp = getTSExecuteStatementResp(status); + TSExecuteStatementResp resp = RpcUtils.getTSExecuteStatementResp(status); long queryId = generateQueryId(false); resp.setQueryId(queryId); return resp; @@ -1050,12 +996,12 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { physicalPlan = processor.parseSQLToPhysicalPlan(statement, sessionIdZoneIdMap.get(sessionId)); } catch (QueryProcessException | SQLParserException e) { logger.info(ERROR_PARSING_SQL, e.getMessage()); - return getTSExecuteStatementResp(getStatus(TSStatusCode.SQL_PARSE_ERROR, e.getMessage())); + return RpcUtils.getTSExecuteStatementResp(TSStatusCode.SQL_PARSE_ERROR, e.getMessage()); } if (physicalPlan.isQuery()) { - return getTSExecuteStatementResp( - getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is a query statement.")); + return RpcUtils.getTSExecuteStatementResp( + TSStatusCode.EXECUTE_STATEMENT_ERROR, "Statement is a query statement."); } return executeUpdateStatement(physicalPlan, sessionId); @@ -1079,30 +1025,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { return AuthorityChecker.check(username, paths, plan.getOperatorType(), targetUser); } - static TSExecuteStatementResp getTSExecuteStatementResp(TSStatus status) { - TSExecuteStatementResp resp = new TSExecuteStatementResp(); - TSStatus tsStatus = new TSStatus(status); - resp.setStatus(tsStatus); - return resp; - } - - private TSExecuteBatchStatementResp getTSBatchExecuteStatementResp( - TSStatus status, List<Integer> result) { - TSExecuteBatchStatementResp resp = new TSExecuteBatchStatementResp(); - TSStatus tsStatus = new TSStatus(status); - resp.setStatus(tsStatus); - resp.setResult(result); - return resp; - } - - private TSFetchResultsResp getTSFetchResultsResp(TSStatus status) { - TSFetchResultsResp resp = new TSFetchResultsResp(); - TSStatus tsStatus = new TSStatus(status); - resp.setStatus(tsStatus); - return resp; - } - - protected void handleClientExit() { + void handleClientExit() { Long sessionId = currSessionId.get(); if (sessionId != null) { TSCloseSessionReq req = new TSCloseSessionReq(sessionId); @@ -1115,14 +1038,14 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { TSStatus tsStatus; TSGetTimeZoneResp resp = null; try { - tsStatus = getStatus(TSStatusCode.SUCCESS_STATUS); + tsStatus = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); ZoneId zoneId = sessionIdZoneIdMap.get(sessionId); if (zoneId != null) { resp = new TSGetTimeZoneResp(tsStatus, zoneId.toString()); } } catch (Exception e) { logger.error("meet error while generating time zone.", e); - tsStatus = getStatus(TSStatusCode.GENERATE_TIME_ZONE_ERROR); + tsStatus = RpcUtils.getStatus(TSStatusCode.GENERATE_TIME_ZONE_ERROR); resp = new TSGetTimeZoneResp(tsStatus, "Unknown time zone"); } return resp; @@ -1134,10 +1057,10 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { try { String timeZoneID = req.getTimeZone(); sessionIdZoneIdMap.put(req.getSessionId(), ZoneId.of(timeZoneID)); - tsStatus = getStatus(TSStatusCode.SUCCESS_STATUS); + tsStatus = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); } catch (Exception e) { logger.error("meet error while setting time zone.", e); - tsStatus = getStatus(TSStatusCode.SET_TIME_ZONE_ERROR); + tsStatus = RpcUtils.getStatus(TSStatusCode.SET_TIME_ZONE_ERROR); } return new TSStatus(tsStatus); } @@ -1155,11 +1078,11 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } @Override - public TSExecuteInsertRowInBatchResp insertRowInBatch(TSInsertInBatchReq req) { - TSExecuteInsertRowInBatchResp resp = new TSExecuteInsertRowInBatchResp(); + public TSExecuteBatchStatementResp insertRowInBatch(TSInsertInBatchReq req) { + TSExecuteBatchStatementResp resp = new TSExecuteBatchStatementResp(); if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - resp.addToStatusList(getStatus(TSStatusCode.NOT_LOGIN_ERROR)); + resp.addToStatusList(RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR)); return resp; } @@ -1171,7 +1094,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { plan.setValues(req.getValuesList().get(i).toArray(new String[0])); TSStatus status = checkAuthority(plan, req.getSessionId()); if (status != null) { - resp.addToStatusList(new TSStatus(status)); + resp.addToStatusList(status); } else { resp.addToStatusList(executePlan(plan)); } @@ -1183,32 +1106,26 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { @Override public TSExecuteBatchStatementResp testInsertBatch(TSBatchInsertionReq req) { logger.debug("Test insert batch request receive."); - TSExecuteBatchStatementResp resp = new TSExecuteBatchStatementResp(); - resp.setStatus(getStatus(TSStatusCode.SUCCESS_STATUS)); - resp.setResult(Collections.emptyList()); - return resp; + return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.SUCCESS_STATUS); } @Override public TSStatus testInsertRow(TSInsertReq req) { logger.debug("Test insert row request receive."); - return getStatus(TSStatusCode.SUCCESS_STATUS); + return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); } @Override - public TSExecuteInsertRowInBatchResp testInsertRowInBatch(TSInsertInBatchReq req) { + public TSExecuteBatchStatementResp testInsertRowInBatch(TSInsertInBatchReq req) { logger.debug("Test insert row in batch request receive."); - - TSExecuteInsertRowInBatchResp resp = new TSExecuteInsertRowInBatchResp(); - resp.addToStatusList(getStatus(TSStatusCode.SUCCESS_STATUS)); - return resp; + return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.SUCCESS_STATUS); } @Override public TSStatus insert(TSInsertReq req) { if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getStatus(TSStatusCode.NOT_LOGIN_ERROR); + return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); } InsertPlan plan = new InsertPlan(); @@ -1219,16 +1136,16 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { TSStatus status = checkAuthority(plan, req.getSessionId()); if (status != null) { - return new TSStatus(status); + return status; } - return new TSStatus(executePlan(plan)); + return executePlan(plan); } @Override public TSStatus deleteData(TSDeleteDataReq req) { if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getStatus(TSStatusCode.NOT_LOGIN_ERROR); + return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); } DeletePlan plan = new DeletePlan(); @@ -1252,7 +1169,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { try { if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getTSBatchExecuteStatementResp(getStatus(TSStatusCode.NOT_LOGIN_ERROR), null); + return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.NOT_LOGIN_ERROR); } BatchInsertPlan batchInsertPlan = new BatchInsertPlan(req.deviceId, req.measurements); @@ -1266,30 +1183,30 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { boolean isAllSuccessful = true; TSStatus status = checkAuthority(batchInsertPlan, req.getSessionId()); if (status != null) { - return new TSExecuteBatchStatementResp(status); + return RpcUtils.getTSBatchExecuteStatementResp(status); } - Integer[] results = executor.insertBatch(batchInsertPlan); + TSStatus[] tsStatusArray = executor.insertBatch(batchInsertPlan); - for (Integer result : results) { - if (result != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + for (TSStatus tsStatus : tsStatusArray) { + if (tsStatus.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { isAllSuccessful = false; break; } } if (isAllSuccessful) { - logger.debug("Insert one RowBatch successfully"); - return getTSBatchExecuteStatementResp( - getStatus(TSStatusCode.SUCCESS_STATUS), Arrays.asList(results)); + if (logger.isDebugEnabled()) { + logger.debug("Insert one RowBatch successfully"); + } + return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.SUCCESS_STATUS); } else { logger.debug("Insert one RowBatch failed!"); - return getTSBatchExecuteStatementResp( - getStatus(TSStatusCode.INTERNAL_SERVER_ERROR), Arrays.asList(results)); + return RpcUtils.getTSBatchExecuteStatementResp(Arrays.asList(tsStatusArray)); } } catch (Exception e) { logger.info("{}: error occurs when executing statements", IoTDBConstant.GLOBAL_DB_NAME, e); - return getTSBatchExecuteStatementResp( - getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage()), null); + return RpcUtils + .getTSBatchExecuteStatementResp(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage()); } finally { Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_RPC_BATCH_INSERT, t1); } @@ -1299,7 +1216,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { public TSStatus setStorageGroup(long sessionId, String storageGroup) { if (!checkLogin(sessionId)) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getStatus(TSStatusCode.NOT_LOGIN_ERROR); + return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); } SetStorageGroupPlan plan = new SetStorageGroupPlan(new Path(storageGroup)); @@ -1314,7 +1231,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { public TSStatus deleteStorageGroups(long sessionId, List<String> storageGroups) { if (!checkLogin(sessionId)) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getStatus(TSStatusCode.NOT_LOGIN_ERROR); + return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); } List<Path> storageGroupList = new ArrayList<>(); for (String storageGroup : storageGroups) { @@ -1332,7 +1249,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { public TSStatus createTimeseries(TSCreateTimeseriesReq req) { if (!checkLogin(req.getSessionId())) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getStatus(TSStatusCode.NOT_LOGIN_ERROR); + return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); } CreateTimeSeriesPlan plan = new CreateTimeSeriesPlan( @@ -1352,7 +1269,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { public TSStatus deleteTimeseries(long sessionId, List<String> paths) { if (!checkLogin(sessionId)) { logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME); - return getStatus(TSStatusCode.NOT_LOGIN_ERROR); + return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN_ERROR); } List<Path> pathList = new ArrayList<>(); for (String path : paths) { @@ -1377,29 +1294,29 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { List<Path> paths = plan.getPaths(); try { if (!checkAuthorization(paths, plan, sessionIdUsernameMap.get(sessionId))) { - return getStatus( + return RpcUtils.getStatus( TSStatusCode.NO_PERMISSION_ERROR, "No permissions for this operation " + plan.getOperatorType().toString()); } } catch (AuthException e) { logger.error("meet error while checking authorization.", e); - return getStatus(TSStatusCode.UNINITIALIZED_AUTH_ERROR, e.getMessage()); + return RpcUtils.getStatus(TSStatusCode.UNINITIALIZED_AUTH_ERROR, e.getMessage()); } return null; } - protected TSStatus executePlan(PhysicalPlan plan) { + private TSStatus executePlan(PhysicalPlan plan) { boolean execRet; try { execRet = executeNonQuery(plan); } catch (QueryProcessException e) { logger.debug("meet error while processing non-query. ", e); - return new TSStatus(new TSStatusType(e.getErrorCode(), e.getMessage())); + return new TSStatus(e.getErrorCode(), e.getMessage()); } return execRet - ? getStatus(TSStatusCode.SUCCESS_STATUS, "Execute successfully") - : getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR); + ? RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "Execute successfully") + : RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR); } private long generateQueryId(boolean isDataQuery) { diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/BatchExecutionException.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/BatchExecutionException.java new file mode 100644 index 0000000..ab768ff --- /dev/null +++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/BatchExecutionException.java @@ -0,0 +1,27 @@ +package org.apache.iotdb.rpc; + +import java.util.List; +import org.apache.iotdb.service.rpc.thrift.TSStatus; + +public class BatchExecutionException extends Exception{ + + private List<TSStatus> statusList; + + public BatchExecutionException(String message) { + super(message); + } + + public BatchExecutionException(List<TSStatus> statusList) { + this.statusList = statusList; + } + + public BatchExecutionException(List<TSStatus> statusList, String message) { + super(message); + this.statusList = statusList; + } + + public List<TSStatus> getStatusList() { + return statusList; + } + +} diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java index 8b2e956..1e909b3 100644 --- a/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java +++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcUtils.java @@ -19,6 +19,10 @@ package org.apache.iotdb.rpc; import java.lang.reflect.Proxy; +import java.util.List; +import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp; +import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp; +import org.apache.iotdb.service.rpc.thrift.TSFetchResultsResp; import org.apache.iotdb.service.rpc.thrift.TSIService; import org.apache.iotdb.service.rpc.thrift.TSStatus; @@ -35,11 +39,93 @@ public class RpcUtils { * @param status -status */ public static void verifySuccess(TSStatus status) throws StatementExecutionException { - if (status.getStatusType().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { throw new StatementExecutionException(String.format("%d: %s", - status.getStatusType().getCode(), - status.getStatusType().getMessage())); + status.code, status.message)); } } + public static void verifySuccess(List<TSStatus> statuses) throws BatchExecutionException { + for (TSStatus status : statuses) { + if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new BatchExecutionException(statuses, status.message); + } + } + } + + /** + * convert from TSStatusCode to TSStatus according to status code and status message + */ + public static TSStatus getStatus(TSStatusCode tsStatusCode) { + return new TSStatus(tsStatusCode.getStatusCode(), ""); + } + + /** + * convert from TSStatusCode to TSStatus, which has message appending with existed status message + * + * @param tsStatusCode status type + * @param message appending message + */ + public static TSStatus getStatus(TSStatusCode tsStatusCode, String message) { + return new TSStatus(tsStatusCode.getStatusCode(), message); + } + + public static TSExecuteStatementResp getTSExecuteStatementResp(TSStatusCode tsStatusCode) { + TSStatus status = getStatus(tsStatusCode); + return getTSExecuteStatementResp(status); + } + + public static TSExecuteStatementResp getTSExecuteStatementResp(TSStatusCode tsStatusCode, String message) { + TSStatus status = getStatus(tsStatusCode, message); + return getTSExecuteStatementResp(status); + } + + public static TSExecuteStatementResp getTSExecuteStatementResp(TSStatus status) { + TSExecuteStatementResp resp = new TSExecuteStatementResp(); + TSStatus tsStatus = new TSStatus(status); + resp.setStatus(tsStatus); + return resp; + } + + + public static TSExecuteBatchStatementResp getTSBatchExecuteStatementResp(TSStatusCode tsStatusCode) { + TSStatus status = getStatus(tsStatusCode); + return getTSBatchExecuteStatementResp(status); + } + + public static TSExecuteBatchStatementResp getTSBatchExecuteStatementResp(TSStatusCode tsStatusCode, String message) { + TSStatus status = getStatus(tsStatusCode, message); + return getTSBatchExecuteStatementResp(status); + } + + public static TSExecuteBatchStatementResp getTSBatchExecuteStatementResp(TSStatus status) { + TSExecuteBatchStatementResp resp = new TSExecuteBatchStatementResp(); + resp.addToStatusList(status); + return resp; + } + + public static TSExecuteBatchStatementResp getTSBatchExecuteStatementResp(List<TSStatus> statusList) { + TSExecuteBatchStatementResp resp = new TSExecuteBatchStatementResp(); + resp.setStatusList(statusList); + return resp; + } + + + public static TSFetchResultsResp getTSFetchResultsResp(TSStatusCode tsStatusCode) { + TSStatus status = getStatus(tsStatusCode); + return getTSFetchResultsResp(status); + } + + public static TSFetchResultsResp getTSFetchResultsResp(TSStatusCode tsStatusCode, String appendMessage) { + TSStatus status = getStatus(tsStatusCode, appendMessage); + return getTSFetchResultsResp(status); + } + + public static TSFetchResultsResp getTSFetchResultsResp(TSStatus status) { + TSFetchResultsResp resp = new TSFetchResultsResp(); + TSStatus tsStatus = new TSStatus(status); + resp.setStatus(tsStatus); + return resp; + } + } diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/StatementExecutionException.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/StatementExecutionException.java index d7e09fc..0ff2898 100644 --- a/service-rpc/src/main/java/org/apache/iotdb/rpc/StatementExecutionException.java +++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/StatementExecutionException.java @@ -18,8 +18,21 @@ */ package org.apache.iotdb.rpc; +import org.apache.iotdb.service.rpc.thrift.TSStatus; + public class StatementExecutionException extends Exception{ + private int code; + + public StatementExecutionException(TSStatus status) { + super(status.message); + code = status.code; + } + + public int getCode() { + return code; + } + public StatementExecutionException(String reason) { super(reason); } diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java index fb307df..4c62201 100644 --- a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java +++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java @@ -20,7 +20,7 @@ package org.apache.iotdb.rpc; public enum TSStatusCode { - SUCCESS_STATUS(200), + SUCCESS_STATUS(-2), // equal to java.sql.Statement.SUCCESS_NO_INFO STILL_EXECUTING_STATUS(201), INVALID_HANDLE_STATUS(202), diff --git a/service-rpc/src/main/thrift/rpc.thrift b/service-rpc/src/main/thrift/rpc.thrift index fb96f02..091316b 100644 --- a/service-rpc/src/main/thrift/rpc.thrift +++ b/service-rpc/src/main/thrift/rpc.thrift @@ -19,18 +19,11 @@ namespace java org.apache.iotdb.service.rpc.thrift // The return status code and message in each response. -struct TSStatusType { +struct TSStatus { 1: required i32 code 2: required string message } -// The return status of a remote request -struct TSStatus { - 1: required TSStatusType statusType - 2: optional list<string> infoMessages - 3: optional string sqlState // as defined in the ISO/IEF CLIENT specification -} - struct TSExecuteStatementResp { 1: required TSStatus status 2: optional i64 queryId @@ -97,17 +90,11 @@ struct TSExecuteStatementReq { 4: optional i32 fetchSize } -struct TSExecuteInsertRowInBatchResp{ +struct TSExecuteBatchStatementResp{ 1: required i64 sessionId 2: required list<TSStatus> statusList } -struct TSExecuteBatchStatementResp{ - 1: required TSStatus status - // For each value in result, Statement.SUCCESS_NO_INFO represents success, Statement.EXECUTE_FAILED represents fail otherwise. - 2: optional list<i32> result -} - struct TSExecuteBatchStatementReq{ // The session to execute the statement against 1: required i64 sessionId @@ -280,13 +267,13 @@ service TSIService { TSExecuteBatchStatementResp insertBatch(1:TSBatchInsertionReq req); - TSExecuteInsertRowInBatchResp insertRowInBatch(1:TSInsertInBatchReq req); + TSExecuteBatchStatementResp insertRowInBatch(1:TSInsertInBatchReq req); TSExecuteBatchStatementResp testInsertBatch(1:TSBatchInsertionReq req); TSStatus testInsertRow(1:TSInsertReq req); - TSExecuteInsertRowInBatchResp testInsertRowInBatch(1:TSInsertInBatchReq req); + TSExecuteBatchStatementResp testInsertRowInBatch(1:TSInsertInBatchReq req); TSStatus deleteData(1:TSDeleteDataReq req); diff --git a/session/src/main/java/org/apache/iotdb/session/Session.java b/session/src/main/java/org/apache/iotdb/session/Session.java index 3cc53f0..4c36f29 100644 --- a/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/session/src/main/java/org/apache/iotdb/session/Session.java @@ -23,18 +23,16 @@ import static org.apache.iotdb.session.Config.PATH_PATTERN; import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.Comparator; import java.util.List; +import org.apache.iotdb.rpc.BatchExecutionException; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.StatementExecutionException; -import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSBatchInsertionReq; import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq; import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq; import org.apache.iotdb.service.rpc.thrift.TSDeleteDataReq; -import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp; import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq; import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp; import org.apache.iotdb.service.rpc.thrift.TSGetTimeZoneResp; @@ -201,8 +199,8 @@ public class Session { * * @param rowBatch data batch */ - private TSExecuteBatchStatementResp insertSortedBatchIntern(RowBatch rowBatch) - throws IoTDBConnectionException { + private void insertSortedBatchIntern(RowBatch rowBatch) + throws IoTDBConnectionException, BatchExecutionException { TSBatchInsertionReq request = new TSBatchInsertionReq(); request.setSessionId(sessionId); request.deviceId = rowBatch.deviceId; @@ -215,7 +213,7 @@ public class Session { request.setSize(rowBatch.batchSize); try { - return checkAndReturn(client.insertBatch(request)); + RpcUtils.verifySuccess(client.insertBatch(request).statusList); } catch (TException e) { throw new IoTDBConnectionException(e); } @@ -227,12 +225,12 @@ public class Session { * * @param rowBatch data batch */ - public TSExecuteBatchStatementResp insertSortedBatch(RowBatch rowBatch) - throws StatementExecutionException, IoTDBConnectionException { + public void insertSortedBatch(RowBatch rowBatch) + throws BatchExecutionException, IoTDBConnectionException { if(!checkSorted(rowBatch)){ - throw new StatementExecutionException("Row batch has't been sorted when calling insertSortedBatch"); + throw new BatchExecutionException("Row batch has't been sorted when calling insertSortedBatch"); } - return insertSortedBatchIntern(rowBatch); + insertSortedBatchIntern(rowBatch); } /** @@ -240,11 +238,12 @@ public class Session { * * @param rowBatch data batch */ - public TSExecuteBatchStatementResp insertBatch(RowBatch rowBatch) throws IoTDBConnectionException { + public void insertBatch(RowBatch rowBatch) + throws IoTDBConnectionException, BatchExecutionException { sortRowBatch(rowBatch); - return insertSortedBatchIntern(rowBatch); + insertSortedBatchIntern(rowBatch); } private void sortRowBatch(RowBatch rowBatch){ @@ -329,9 +328,9 @@ public class Session { * * @see Session#insertBatch(RowBatch) */ - public List<TSStatus> insertInBatch(List<String> deviceIds, List<Long> times, + public void insertInBatch(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList) - throws IoTDBConnectionException { + throws IoTDBConnectionException, BatchExecutionException { // check params size int len = deviceIds.size(); if (len != times.size() || len != measurementsList.size() || len != valuesList.size()) { @@ -347,11 +346,7 @@ public class Session { request.setValuesList(valuesList); try { - List<TSStatus> result = new ArrayList<>(); - for (TSStatus cur : client.insertRowInBatch(request).getStatusList()) { - result.add(checkAndReturn(cur)); - } - return result; + RpcUtils.verifySuccess(client.insertRowInBatch(request).statusList); } catch (TException e) { throw new IoTDBConnectionException(e); } @@ -364,8 +359,8 @@ public class Session { * @see Session#insertInBatch(List, List, List, List) * @see Session#insertBatch(RowBatch) */ - public TSStatus insert(String deviceId, long time, List<String> measurements, - List<String> values) throws IoTDBConnectionException { + public void insert(String deviceId, long time, List<String> measurements, + List<String> values) throws IoTDBConnectionException, StatementExecutionException { TSInsertReq request = new TSInsertReq(); request.setSessionId(sessionId); request.setDeviceId(deviceId); @@ -374,7 +369,7 @@ public class Session { request.setValues(values); try { - return checkAndReturn(client.insert(request)); + RpcUtils.verifySuccess(client.insert(request)); } catch (TException e) { throw new IoTDBConnectionException(e); } @@ -384,8 +379,8 @@ public class Session { * This method NOT insert data into database and the server just return after accept the request, * this method should be used to test other time cost in client */ - public TSExecuteBatchStatementResp testInsertBatch(RowBatch rowBatch) - throws IoTDBConnectionException { + public void testInsertBatch(RowBatch rowBatch) + throws IoTDBConnectionException, BatchExecutionException { TSBatchInsertionReq request = new TSBatchInsertionReq(); request.setSessionId(sessionId); request.deviceId = rowBatch.deviceId; @@ -398,7 +393,7 @@ public class Session { request.setSize(rowBatch.batchSize); try { - return client.testInsertBatch(request); + RpcUtils.verifySuccess(client.testInsertBatch(request).statusList); } catch (TException e) { throw new IoTDBConnectionException(e); } @@ -408,9 +403,9 @@ public class Session { * This method NOT insert data into database and the server just return after accept the request, * this method should be used to test other time cost in client */ - public List<TSStatus> testInsertInBatch(List<String> deviceIds, List<Long> times, + public void testInsertInBatch(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList) - throws IoTDBConnectionException { + throws IoTDBConnectionException, BatchExecutionException { // check params size int len = deviceIds.size(); if (len != times.size() || len != measurementsList.size() || len != valuesList.size()) { @@ -426,8 +421,7 @@ public class Session { request.setValuesList(valuesList); try { - client.testInsertRowInBatch(request); - return Collections.emptyList(); + RpcUtils.verifySuccess(client.testInsertRowInBatch(request).statusList); } catch (TException e) { throw new IoTDBConnectionException(e); } @@ -437,8 +431,8 @@ public class Session { * This method NOT insert data into database and the server just return after accept the request, * this method should be used to test other time cost in client */ - public TSStatus testInsert(String deviceId, long time, List<String> measurements, - List<String> values) throws IoTDBConnectionException { + public void testInsert(String deviceId, long time, List<String> measurements, + List<String> values) throws IoTDBConnectionException, StatementExecutionException { TSInsertReq request = new TSInsertReq(); request.setSessionId(sessionId); request.setDeviceId(deviceId); @@ -447,7 +441,7 @@ public class Session { request.setValues(values); try { - return client.testInsertRow(request); + RpcUtils.verifySuccess(client.testInsertRow(request)); } catch (TException e) { throw new IoTDBConnectionException(e); } @@ -458,10 +452,11 @@ public class Session { * * @param path timeseries to delete, should be a whole path */ - public TSStatus deleteTimeseries(String path) throws IoTDBConnectionException { + public void deleteTimeseries(String path) + throws IoTDBConnectionException, StatementExecutionException { List<String> paths = new ArrayList<>(); paths.add(path); - return deleteTimeseries(paths); + deleteTimeseries(paths); } /** @@ -469,9 +464,10 @@ public class Session { * * @param paths timeseries to delete, should be a whole path */ - public TSStatus deleteTimeseries(List<String> paths) throws IoTDBConnectionException { + public void deleteTimeseries(List<String> paths) + throws IoTDBConnectionException, StatementExecutionException { try { - return checkAndReturn(client.deleteTimeseries(sessionId, paths)); + RpcUtils.verifySuccess(client.deleteTimeseries(sessionId, paths)); } catch (TException e) { throw new IoTDBConnectionException(e); } @@ -483,10 +479,11 @@ public class Session { * @param path data in which time series to delete * @param time data with time stamp less than or equal to time will be deleted */ - public TSStatus deleteData(String path, long time) throws IoTDBConnectionException { + public void deleteData(String path, long time) + throws IoTDBConnectionException, StatementExecutionException { List<String> paths = new ArrayList<>(); paths.add(path); - return deleteData(paths, time); + deleteData(paths, time); } /** @@ -495,45 +492,48 @@ public class Session { * @param paths data in which time series to delete * @param time data with time stamp less than or equal to time will be deleted */ - public TSStatus deleteData(List<String> paths, long time) throws IoTDBConnectionException { + public void deleteData(List<String> paths, long time) + throws IoTDBConnectionException, StatementExecutionException { TSDeleteDataReq request = new TSDeleteDataReq(); request.setSessionId(sessionId); request.setPaths(paths); request.setTimestamp(time); try { - return checkAndReturn(client.deleteData(request)); + RpcUtils.verifySuccess(client.deleteData(request)); } catch (TException e) { throw new IoTDBConnectionException(e); } } - public TSStatus setStorageGroup(String storageGroupId) + public void setStorageGroup(String storageGroupId) throws IoTDBConnectionException, StatementExecutionException { checkPathValidity(storageGroupId); try { - return checkAndReturn(client.setStorageGroup(sessionId, storageGroupId)); + RpcUtils.verifySuccess(client.setStorageGroup(sessionId, storageGroupId)); } catch (TException e) { throw new IoTDBConnectionException(e); } } - public TSStatus deleteStorageGroup(String storageGroup) throws IoTDBConnectionException { + public void deleteStorageGroup(String storageGroup) + throws IoTDBConnectionException, StatementExecutionException { List<String> groups = new ArrayList<>(); groups.add(storageGroup); - return deleteStorageGroups(groups); + deleteStorageGroups(groups); } - public TSStatus deleteStorageGroups(List<String> storageGroup) throws IoTDBConnectionException { + public void deleteStorageGroups(List<String> storageGroup) + throws IoTDBConnectionException, StatementExecutionException { try { - return checkAndReturn(client.deleteStorageGroups(sessionId, storageGroup)); + RpcUtils.verifySuccess(client.deleteStorageGroups(sessionId, storageGroup)); } catch (TException e) { throw new IoTDBConnectionException(e); } } - public TSStatus createTimeseries(String path, TSDataType dataType, + public void createTimeseries(String path, TSDataType dataType, TSEncoding encoding, CompressionType compressor) throws IoTDBConnectionException, StatementExecutionException { checkPathValidity(path); @@ -545,7 +545,7 @@ public class Session { request.setCompressor(compressor.ordinal()); try { - return checkAndReturn(client.createTimeseries(request)); + RpcUtils.verifySuccess(client.createTimeseries(request)); } catch (TException e) { throw new IoTDBConnectionException(e); } @@ -561,20 +561,6 @@ public class Session { } } - private TSStatus checkAndReturn(TSStatus resp) { - if (resp.statusType.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - logger.error(resp.statusType.getMessage()); - } - return resp; - } - - private TSExecuteBatchStatementResp checkAndReturn(TSExecuteBatchStatementResp resp) { - if (resp.status.statusType.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - logger.error(resp.status.statusType.getMessage()); - } - return resp; - } - private synchronized String getTimeZone() throws StatementExecutionException, IoTDBConnectionException { if (zoneId != null) { diff --git a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionIT.java b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionIT.java index c5f330b..980bf4f 100644 --- a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionIT.java +++ b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionIT.java @@ -19,6 +19,7 @@ package org.apache.iotdb.session; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import java.io.File; import java.sql.Connection; @@ -34,6 +35,7 @@ import org.apache.iotdb.db.conf.IoTDBConstant; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.jdbc.Config; +import org.apache.iotdb.rpc.BatchExecutionException; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; @@ -67,8 +69,8 @@ public class IoTDBSessionIT { } @Test - public void testAlignByDevice() - throws SQLException, IoTDBConnectionException, StatementExecutionException { + public void testAlignByDevice() throws SQLException, IoTDBConnectionException, + StatementExecutionException, BatchExecutionException { session = new Session("127.0.0.1", 6667, "root", "root"); session.open(); @@ -83,7 +85,8 @@ public class IoTDBSessionIT { } // it's will output too much to travis, so ignore it - public void testTime() throws IoTDBConnectionException, StatementExecutionException { + public void testTime() + throws IoTDBConnectionException, StatementExecutionException, BatchExecutionException { session = new Session("127.0.0.1", 6667, "root", "root"); session.open(); @@ -95,8 +98,8 @@ public class IoTDBSessionIT { } @Test - public void testBatchInsertSeqAndUnseq() - throws SQLException, ClassNotFoundException, IoTDBConnectionException, StatementExecutionException { + public void testBatchInsertSeqAndUnseq() throws SQLException, ClassNotFoundException, + IoTDBConnectionException, StatementExecutionException, BatchExecutionException { session = new Session("127.0.0.1", 6667, "root", "root"); session.open(); @@ -119,8 +122,8 @@ public class IoTDBSessionIT { } @Test - public void testBatchInsert() - throws StatementExecutionException, SQLException, ClassNotFoundException, IoTDBConnectionException { + public void testBatchInsert() throws StatementExecutionException, SQLException, + ClassNotFoundException, IoTDBConnectionException, BatchExecutionException { session = new Session("127.0.0.1", 6667, "root", "root"); session.open(); @@ -133,7 +136,8 @@ public class IoTDBSessionIT { queryForBatch(); } - public void testTestMethod() throws StatementExecutionException, IoTDBConnectionException { + public void testTestMethod() + throws StatementExecutionException, IoTDBConnectionException, BatchExecutionException { session = new Session("127.0.0.1", 6667, "root", "root"); session.open(); @@ -200,7 +204,7 @@ public class IoTDBSessionIT { @Test public void test() throws ClassNotFoundException, SQLException, - IoTDBConnectionException, StatementExecutionException { + IoTDBConnectionException, StatementExecutionException, BatchExecutionException { session = new Session("127.0.0.1", 6667, "root", "root"); try { session.open(); @@ -313,7 +317,7 @@ public class IoTDBSessionIT { CompressionType.SNAPPY); } - private void insertInBatch() throws IoTDBConnectionException { + private void insertInBatch() throws IoTDBConnectionException, BatchExecutionException { String deviceId = "root.sg1.d2"; List<String> measurements = new ArrayList<>(); measurements.add("s1"); @@ -346,7 +350,7 @@ public class IoTDBSessionIT { session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList); } - private void insert() throws IoTDBConnectionException { + private void insert() throws IoTDBConnectionException, StatementExecutionException { String deviceId = "root.sg1.d1"; List<String> measurements = new ArrayList<>(); measurements.add("s1"); @@ -361,7 +365,8 @@ public class IoTDBSessionIT { } } - private void insertRowBatchTest1(String deviceId) throws IoTDBConnectionException { + private void insertRowBatchTest1(String deviceId) + throws IoTDBConnectionException, BatchExecutionException { Schema schema = new Schema(); schema.registerTimeseries(new Path(deviceId, "s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); schema.registerTimeseries(new Path(deviceId, "s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); @@ -391,7 +396,7 @@ public class IoTDBSessionIT { } } - private void deleteData() throws IoTDBConnectionException { + private void deleteData() throws IoTDBConnectionException, StatementExecutionException { String path1 = "root.sg1.d1.s1"; String path2 = "root.sg1.d1.s2"; String path3 = "root.sg1.d1.s3"; @@ -404,7 +409,7 @@ public class IoTDBSessionIT { session.deleteData(paths, deleteTime); } - private void deleteTimeseries() throws IoTDBConnectionException { + private void deleteTimeseries() throws IoTDBConnectionException, StatementExecutionException { session.deleteTimeseries("root.sg1.d1.s1"); session.deleteTimeseries("root.laptop.d1.1_2"); session.deleteTimeseries("root.laptop.d1.\"1.2.3\""); @@ -425,7 +430,7 @@ public class IoTDBSessionIT { final int colCount = metaData.getColumnCount(); StringBuilder resultStr = new StringBuilder(); for (int i = 0; i < colCount; i++) { - resultStr.append(metaData.getColumnLabel(i + 1) + "\n"); + resultStr.append(metaData.getColumnLabel(i + 1)).append("\n"); } while (resultSet.next()) { for (int i = 1; i <= colCount; i++) { @@ -437,8 +442,7 @@ public class IoTDBSessionIT { } } - private void queryForAlignByDevice() - throws SQLException, StatementExecutionException, IoTDBConnectionException { + private void queryForAlignByDevice() throws StatementExecutionException, IoTDBConnectionException { SessionDataSet sessionDataSet = session .executeQueryStatement("select '11', s1, '11' from root.sg1.d1 align by device"); sessionDataSet.setBatchSize(1024); @@ -456,8 +460,7 @@ public class IoTDBSessionIT { sessionDataSet.closeOperationHandle(); } - private void queryForAlignByDevice2() - throws SQLException, IoTDBConnectionException, StatementExecutionException { + private void queryForAlignByDevice2() throws IoTDBConnectionException, StatementExecutionException { SessionDataSet sessionDataSet = session.executeQueryStatement( "select '11', s1, '11', s5, s1, s5 from root.sg1.d1 align by device"); sessionDataSet.setBatchSize(1024); @@ -489,7 +492,7 @@ public class IoTDBSessionIT { final int colCount = metaData.getColumnCount(); StringBuilder resultStr = new StringBuilder(); for (int i = 0; i < colCount; i++) { - resultStr.append(metaData.getColumnLabel(i + 1) + "\n"); + resultStr.append(metaData.getColumnLabel(i + 1)).append("\n"); } while (resultSet.next()) { for (int i = 1; i <= colCount; i++) { @@ -510,7 +513,7 @@ public class IoTDBSessionIT { } session.deleteStorageGroup("root.sg1"); File folder = new File("data/system/storage_groups/root.sg1/"); - assertEquals(folder.exists(), false); + assertFalse(folder.exists()); session.setStorageGroup("root.sg1.d1"); session.createTimeseries("root.sg1.d1.s1", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY); @@ -525,7 +528,7 @@ public class IoTDBSessionIT { final int colCount = metaData.getColumnCount(); StringBuilder resultStr = new StringBuilder(); for (int i = 0; i < colCount; i++) { - resultStr.append(metaData.getColumnLabel(i + 1) + "\n"); + resultStr.append(metaData.getColumnLabel(i + 1)).append("\n"); } while (resultSet.next()) { for (int i = 1; i <= colCount; i++) { @@ -636,7 +639,8 @@ public class IoTDBSessionIT { assertEquals(correctStatus, status); } - private void insertRowBatchTest2(String deviceId) throws IoTDBConnectionException { + private void insertRowBatchTest2(String deviceId) + throws IoTDBConnectionException, BatchExecutionException { Schema schema = new Schema(); schema.registerTimeseries(new Path(deviceId,"s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); @@ -669,7 +673,8 @@ public class IoTDBSessionIT { } } - private void insertRowBatchTest3(String deviceId) throws IoTDBConnectionException { + private void insertRowBatchTest3(String deviceId) + throws IoTDBConnectionException, BatchExecutionException { Schema schema = new Schema(); schema.registerTimeseries(new Path(deviceId,"s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); @@ -702,7 +707,8 @@ public class IoTDBSessionIT { } } - private void insertRowBatchTestForTime(String deviceId) throws IoTDBConnectionException { + private void insertRowBatchTestForTime(String deviceId) + throws IoTDBConnectionException, BatchExecutionException { Schema schema = new Schema(); schema.registerTimeseries(new Path(deviceId,"s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); @@ -716,7 +722,6 @@ public class IoTDBSessionIT { new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); schema.registerTimeseries(new Path(deviceId,"s6"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE)); - long countTime = 0; long count = 10000000; long begin = 0; //long begin = 1579414903000L; @@ -737,13 +742,11 @@ public class IoTDBSessionIT { long start = System.currentTimeMillis(); session.insertBatch(rowBatch); long val = System.currentTimeMillis() - start; - countTime += val; rowBatch.reset(); } } if (rowBatch.batchSize != 0) { - long start = System.currentTimeMillis(); session.insertBatch(rowBatch); rowBatch.reset(); } @@ -763,7 +766,7 @@ public class IoTDBSessionIT { final int colCount = metaData.getColumnCount(); StringBuilder resultStr = new StringBuilder(); for (int i = 0; i < colCount; i++) { - resultStr.append(metaData.getColumnLabel(i + 1) + "\n"); + resultStr.append(metaData.getColumnLabel(i + 1)).append("\n"); } int count = 0; @@ -791,7 +794,7 @@ public class IoTDBSessionIT { final int colCount = metaData.getColumnCount(); StringBuilder resultStr = new StringBuilder(); for (int i = 0; i < colCount; i++) { - resultStr.append(metaData.getColumnLabel(i + 1) + "\n"); + resultStr.append(metaData.getColumnLabel(i + 1)).append("\n"); } int count = 0; @@ -831,7 +834,7 @@ public class IoTDBSessionIT { final int colCount = metaData.getColumnCount(); StringBuilder resultStr = new StringBuilder(); for (int i = 0; i < colCount; i++) { - resultStr.append(metaData.getColumnLabel(i + 1) + "\n"); + resultStr.append(metaData.getColumnLabel(i + 1)).append("\n"); } int count = 0;
