This is an automated email from the ASF dual-hosted git repository. qiaojialin pushed a commit to branch test_req in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit ea1d96ceb213c9e42694100f7549615cf1375ef3 Author: qiaojialin <[email protected]> AuthorDate: Fri May 8 13:36:50 2020 +0800 test --- .../main/java/org/apache/iotdb/SessionExample.java | 371 +++++---------------- .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +- .../org/apache/iotdb/db/service/TSServiceImpl.java | 5 + 3 files changed, 88 insertions(+), 290 deletions(-) diff --git a/example/session/src/main/java/org/apache/iotdb/SessionExample.java b/example/session/src/main/java/org/apache/iotdb/SessionExample.java index 57d0a64..4f7ee5c 100644 --- a/example/session/src/main/java/org/apache/iotdb/SessionExample.java +++ b/example/session/src/main/java/org/apache/iotdb/SessionExample.java @@ -19,323 +19,116 @@ package org.apache.iotdb; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; -import java.util.Map; -import org.apache.iotdb.rpc.BatchExecutionException; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; -import org.apache.iotdb.session.Session; -import org.apache.iotdb.session.SessionDataSet; -import org.apache.iotdb.session.SessionDataSet.DataIterator; -import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; -import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; -import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; -import org.apache.iotdb.tsfile.write.record.Tablet; -import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.apache.iotdb.session.pool.SessionDataSetWrapper; +import org.apache.iotdb.session.pool.SessionPool; public class SessionExample { - private static Session session; + public static void main(String[] args) { - public static void main(String[] args) - throws IoTDBConnectionException, StatementExecutionException, BatchExecutionException { - session = new Session("127.0.0.1", 6667, "root", "root"); - session.open(false); + for (int i = 0; i < 6; i++) { + new Thread(new WriteThread(i)).start(); + } try { - session.setStorageGroup("root.sg1"); - } catch (StatementExecutionException e) { - if (!e.getMessage().contains("StorageGroupAlreadySetException")) { - throw e; - } + Thread.sleep(10000); + } catch (InterruptedException e) { + e.printStackTrace(); } - createTimeseries(); - createMultiTimeseries(); - insertRecord(); - insertTablet(); - insertTablets(); - insertRecords(); - nonQuery(); - query(); - queryByIterator(); - deleteData(); - deleteTimeseries(); - session.close(); - } - - private static void createTimeseries() - throws IoTDBConnectionException, StatementExecutionException { - - if (!session.checkTimeseriesExists("root.sg1.d1.s1")) { - session.createTimeseries("root.sg1.d1.s1", TSDataType.INT64, TSEncoding.RLE, - CompressionType.SNAPPY); - } - if (!session.checkTimeseriesExists("root.sg1.d1.s2")) { - session.createTimeseries("root.sg1.d1.s2", TSDataType.INT64, TSEncoding.RLE, - CompressionType.SNAPPY); - } - if (!session.checkTimeseriesExists("root.sg1.d1.s3")) { - session.createTimeseries("root.sg1.d1.s3", TSDataType.INT64, TSEncoding.RLE, - CompressionType.SNAPPY); + for (int i = 0; i < 6; i++) { + new Thread(new ReadThread(i)).start(); } - // create timeseries with tags and attributes - if (!session.checkTimeseriesExists("root.sg1.d1.s4")) { - Map<String, String> tags = new HashMap<>(); - tags.put("tag1", "v1"); - Map<String, String> attributes = new HashMap<>(); - tags.put("description", "v1"); - session.createTimeseries("root.sg1.d1.s4", TSDataType.INT64, TSEncoding.RLE, - CompressionType.SNAPPY, null, tags, attributes, "temperature"); - } } - private static void createMultiTimeseries() - throws IoTDBConnectionException, BatchExecutionException { - - if (!session.checkTimeseriesExists("root.sg1.d2.s1") && !session - .checkTimeseriesExists("root.sg1.d2.s2")) { - List<String> paths = new ArrayList<>(); - paths.add("root.sg1.d2.s1"); - paths.add("root.sg1.d2.s2"); - List<TSDataType> tsDataTypes = new ArrayList<>(); - tsDataTypes.add(TSDataType.INT64); - tsDataTypes.add(TSDataType.INT64); - List<TSEncoding> tsEncodings = new ArrayList<>(); - tsEncodings.add(TSEncoding.RLE); - tsEncodings.add(TSEncoding.RLE); - List<CompressionType> compressionTypes = new ArrayList<>(); - compressionTypes.add(CompressionType.SNAPPY); - compressionTypes.add(CompressionType.SNAPPY); - - List<Map<String, String>> tagsList = new ArrayList<>(); - Map<String, String> tags = new HashMap<>(); - tags.put("unit", "kg"); - tagsList.add(tags); - tagsList.add(tags); + static class WriteThread implements Runnable{ + int device; - List<Map<String, String>> attributesList = new ArrayList<>(); - Map<String, String> attributes = new HashMap<>(); - attributes.put("minValue", "1"); - attributes.put("maxValue", "100"); - attributesList.add(attributes); - attributesList.add(attributes); - - List<String> alias = new ArrayList<>(); - alias.add("weight1"); - alias.add("weight2"); - - session - .createMultiTimeseries(paths, tsDataTypes, tsEncodings, compressionTypes, null, tagsList, - attributesList, alias); + WriteThread(int device) { + this.device = device; } - } - private static void insertRecord() throws IoTDBConnectionException, StatementExecutionException { - String deviceId = "root.sg1.d1"; - List<String> measurements = new ArrayList<>(); - measurements.add("s1"); - measurements.add("s2"); - measurements.add("s3"); - for (long time = 0; time < 100; time++) { - List<String> values = new ArrayList<>(); - values.add("1"); - values.add("2"); - values.add("3"); - session.insertRecord(deviceId, time, measurements, values); - } - } - - private static void insertRecordInObject() - throws IoTDBConnectionException, StatementExecutionException { - String deviceId = "root.sg1.d1"; - List<String> measurements = new ArrayList<>(); - measurements.add("s1"); - measurements.add("s2"); - measurements.add("s3"); - for (long time = 0; time < 100; time++) { - session.insertRecord(deviceId, time, measurements, 1L, 1L, 1L); - } - } - - private static void insertRecords() throws IoTDBConnectionException, BatchExecutionException { - String deviceId = "root.sg1.d1"; - List<String> measurements = new ArrayList<>(); - measurements.add("s1"); - measurements.add("s2"); - measurements.add("s3"); - List<String> deviceIds = new ArrayList<>(); - List<List<String>> measurementsList = new ArrayList<>(); - List<List<String>> valuesList = new ArrayList<>(); - List<Long> timestamps = new ArrayList<>(); - - for (long time = 0; time < 500; time++) { - List<String> values = new ArrayList<>(); - values.add("1"); - values.add("2"); - values.add("3"); - - deviceIds.add(deviceId); - measurementsList.add(measurements); - valuesList.add(values); - timestamps.add(time); - if (time != 0 && time % 100 == 0) { - session.insertRecords(deviceIds, timestamps, measurementsList, valuesList); - deviceIds.clear(); - measurementsList.clear(); - valuesList.clear(); - timestamps.clear(); + @Override + public void run() { + SessionPool session = new SessionPool("127.0.0.1", 6667, "root", "root", 6); + + long time = 0; + while (true) { + try { + Thread.sleep(5000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + long start = System.currentTimeMillis(); + + time += 5000; + String deviceId = "root.sg1.d1"; + List<String> measurements = new ArrayList<>(); + for (int i = 0; i < 50000; i++) { + measurements.add("s" + (i + device * 50000)); + } + + List<String> values = new ArrayList<>(); + for (int i = 0; i < 50000; i++) { + values.add("1"); + } + + try { + session.insertRecord(deviceId, time, measurements, values); + } catch (IoTDBConnectionException | StatementExecutionException e) { + e.printStackTrace(); + } + System.out.println( + Thread.currentThread().getName() + " write: " + (System.currentTimeMillis() - start)); } } - - session.insertRecords(deviceIds, timestamps, measurementsList, valuesList); } - /** - * insert the data of a device. For each timestamp, the number of measurements is the same. - * - * a Tablet example: - * - * device1 - * time s1, s2, s3 - * 1, 1, 1, 1 - * 2, 2, 2, 2 - * 3, 3, 3, 3 - * - * Users need to control the count of Tablet and write a batch when it reaches the maxBatchSize - */ - private static void insertTablet() throws IoTDBConnectionException, BatchExecutionException { - // The schema of sensors of one device - List<MeasurementSchema> schemaList = new ArrayList<>(); - schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); - schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); - schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE)); + static class ReadThread implements Runnable { + int device; - Tablet tablet = new Tablet("root.sg1.d1", schemaList, 100); - - long[] timestamps = tablet.timestamps; - Object[] values = tablet.values; - - for (long time = 0; time < 100; time++) { - int row = tablet.rowSize++; - timestamps[row] = time; - for (int i = 0; i < 3; i++) { - long[] sensor = (long[]) values[i]; - sensor[row] = i; - } - if (tablet.rowSize == tablet.getMaxRowNumber()) { - session.insertTablet(tablet, true); - tablet.reset(); - } - } - - if (tablet.rowSize != 0) { - session.insertTablet(tablet); - tablet.reset(); + ReadThread(int device) { + this.device = device; } - } - - private static void insertTablets() throws IoTDBConnectionException, BatchExecutionException { - // The schema of sensors of one device - List<MeasurementSchema> schemaList = new ArrayList<>(); - schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); - schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); - schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE)); - - Tablet tablet1 = new Tablet("root.sg1.d1", schemaList, 100); - Tablet tablet2 = new Tablet("root.sg1.d2", schemaList, 100); - Tablet tablet3 = new Tablet("root.sg1.d3", schemaList, 100); - Map<String, Tablet> tabletMap = new HashMap<>(); - tabletMap.put("root.sg1.d1", tablet1); - tabletMap.put("root.sg1.d2", tablet2); - tabletMap.put("root.sg1.d3", tablet3); - - long[] timestamps1 = tablet1.timestamps; - Object[] values1 = tablet1.values; - long[] timestamps2 = tablet2.timestamps; - Object[] values2 = tablet2.values; - long[] timestamps3 = tablet3.timestamps; - Object[] values3 = tablet3.values; - - for (long time = 0; time < 100; time++) { - int row1 = tablet1.rowSize++; - int row2 = tablet2.rowSize++; - int row3 = tablet3.rowSize++; - timestamps1[row1] = time; - timestamps2[row2] = time; - timestamps3[row3] = time; - for (int i = 0; i < 3; i++) { - long[] sensor1 = (long[]) values1[i]; - sensor1[row1] = i; - long[] sensor2 = (long[]) values2[i]; - sensor2[row2] = i; - long[] sensor3 = (long[]) values3[i]; - sensor3[row3] = i; - } - if (tablet1.rowSize == tablet1.getMaxRowNumber()) { - session.insertTablets(tabletMap, true); - - tablet1.reset(); - tablet2.reset(); - tablet3.reset(); + @Override + public void run() { + SessionDataSetWrapper dataSet = null; + SessionPool session = new SessionPool("127.0.0.1", 6667, "root", "root", 2); + + try { +// while (true) { + Thread.sleep(5000); + long start = System.currentTimeMillis(); + + StringBuilder builder = new StringBuilder("select last "); + for (int c = 50000*device; c < 50000*device + 49999; c++) { + builder.append("s").append(c).append(","); + } + + builder.append("s" + ((device+1)*50000-1)); + builder.append(" from root.sg1.d1"); + + dataSet = session.executeQueryStatement(builder.toString()); + System.out.println(builder.toString()); + int a = 0; + while (dataSet.hasNext()) { + a++; + dataSet.next(); + } + System.out.print(Thread.currentThread().getName() + " read " + a + " "); + System.out.println(System.currentTimeMillis() - start); + session.closeResultSet(dataSet); +// } + } catch (Exception e) { + e.printStackTrace(); } - } - - if (tablet1.rowSize != 0) { - session.insertTablets(tabletMap, true); - tablet1.reset(); - tablet2.reset(); - tablet3.reset(); - } - } - - private static void deleteData() throws IoTDBConnectionException, StatementExecutionException { - String path = "root.sg1.d1.s1"; - long deleteTime = 99; - session.deleteData(path, deleteTime); - } - - private static void deleteTimeseries() - throws IoTDBConnectionException, StatementExecutionException { - List<String> paths = new ArrayList<>(); - paths.add("root.sg1.d1.s1"); - paths.add("root.sg1.d1.s2"); - paths.add("root.sg1.d1.s3"); - session.deleteTimeseries(paths); - } - private static void query() throws IoTDBConnectionException, StatementExecutionException { - SessionDataSet dataSet; - dataSet = session.executeQueryStatement("select * from root.sg1.d1"); - System.out.println(dataSet.getColumnNames()); - dataSet.setFetchSize(1024); // default is 512 - while (dataSet.hasNext()) { - System.out.println(dataSet.next()); } - - dataSet.closeOperationHandle(); - } - - private static void queryByIterator() - throws IoTDBConnectionException, StatementExecutionException { - SessionDataSet dataSet; - dataSet = session.executeQueryStatement("select * from root.sg1.d1"); - DataIterator iterator = dataSet.iterator(); - System.out.println(dataSet.getColumnNames()); - dataSet.setFetchSize(1024); // default is 512 - while (iterator.next()) { - System.out.println(String.format("%s,%s,%s,%s,%s", iterator.getLong(1), iterator.getLong(2), - iterator.getLong("root.sg1.d1.s2"), iterator.getLong(4), - iterator.getObject("root.sg1.d1.s4"))); - } - - dataSet.closeOperationHandle(); } - private static void nonQuery() throws IoTDBConnectionException, StatementExecutionException { - session.executeNonQueryStatement("insert into root.sg1.d1(timestamp,s1) values(200, 1);"); - } } \ No newline at end of file diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index 51bb3fc..4758727 100644 --- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -131,7 +131,7 @@ public class IoTDBConfig { /** * Is dynamic parameter adapter enable. */ - private boolean enableParameterAdapter = true; + private boolean enableParameterAdapter = false; /** * Is the write ahead log enable. 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 bfce48f..9b46e7a 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 @@ -265,6 +265,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { protected void releaseQueryResource(long queryId) throws StorageEngineException { // remove the corresponding Physical Plan queryId2DataSet.remove(queryId); + logger.warn("remove queryid {} from keyset {}", queryId, queryId2DataSet.keySet()); QueryResourceManager.getInstance().endQuery(queryId); } @@ -741,10 +742,12 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } if (!queryId2DataSet.containsKey(req.queryId)) { + logger.warn("@++++<<<< queryid {} does not exist in queryId2DataSet keyset {}", req.queryId, queryId2DataSet.keySet()); return RpcUtils.getTSFetchResultsResp( RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, "Has not executed query")); } + logger.warn("@++++<<<< queryid {} exist in queryId2DataSet keyset {}", req.queryId, queryId2DataSet.keySet()); QueryDataSet queryDataSet = queryId2DataSet.get(req.queryId); if (req.isAlign) { TSQueryDataSet result = @@ -771,6 +774,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { } if (!hasResultSet) { queryId2DataSet.remove(req.queryId); + logger.warn("remove queryid {} from keyset {}", req.queryId, queryId2DataSet.keySet()); } TSFetchResultsResp resp = RpcUtils.getTSFetchResultsResp(TSStatusCode.SUCCESS_STATUS); resp.setHasResultSet(hasResultSet); @@ -863,6 +867,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { QueryContext context = genQueryContext(queryId); QueryDataSet queryDataSet = executor.processQuery(physicalPlan, context); queryId2DataSet.put(queryId, queryDataSet); + logger.warn("@+++++<<<<: put queryId {} to queryId2DataSet", queryId); return queryDataSet; }
