You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2021/02/03 02:20:24 UTC
[iotdb] 01/03: session
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch pipeline_test
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 6899c63fa7a2d510d0a87f7637d2bc0d3acf0fc8
Author: JackieTien97 <Ja...@foxmail.com>
AuthorDate: Sat Jan 30 09:47:17 2021 +0800
session
---
example/session/pom.xml | 30 ++
.../org/apache/iotdb/DataMigrationExample.java | 185 ---------
.../main/java/org/apache/iotdb/SessionExample.java | 449 +--------------------
.../java/org/apache/iotdb/SessionPoolExample.java | 119 ------
4 files changed, 42 insertions(+), 741 deletions(-)
diff --git a/example/session/pom.xml b/example/session/pom.xml
index 6fd056b..2c4d4d1 100644
--- a/example/session/pom.xml
+++ b/example/session/pom.xml
@@ -40,4 +40,34 @@
<version>${project.version}</version>
</dependency>
</dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <version>${maven.assembly.version}</version>
+ <configuration>
+ <archive>
+ <manifest>
+ <addClasspath>true</addClasspath>
+ <mainClass>org.apache.iotdb.SessionExample</mainClass>
+ </manifest>
+ </archive>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id>
+ <!-- this is used for inheritance merges -->
+ <phase>package</phase>
+ <!-- bind to the packaging phase -->
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
</project>
diff --git a/example/session/src/main/java/org/apache/iotdb/DataMigrationExample.java b/example/session/src/main/java/org/apache/iotdb/DataMigrationExample.java
deleted file mode 100644
index ccb878d..0000000
--- a/example/session/src/main/java/org/apache/iotdb/DataMigrationExample.java
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import org.apache.iotdb.rpc.IoTDBConnectionException;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.session.SessionDataSet.DataIterator;
-import org.apache.iotdb.session.pool.SessionDataSetWrapper;
-import org.apache.iotdb.session.pool.SessionPool;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.read.common.Path;
-import org.apache.iotdb.tsfile.utils.Binary;
-import org.apache.iotdb.tsfile.write.record.Tablet;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-
-
-/**
- * Migrate all data belongs to a path from one IoTDB to another IoTDB Each thread migrate one
- * series, the concurrent thread can be configured by concurrency
- *
- * This example is migrating all timeseries from a local IoTDB with 6667 port to a local IoTDB with
- * 6668 port
- */
-public class DataMigrationExample {
-
- // used to read data from the source IoTDB
- private static SessionPool readerPool;
- // used to write data into the destination IoTDB
- private static SessionPool writerPool;
- // concurrent thread of loading timeseries data
- private static int concurrency = 5;
-
- public static void main(String[] args)
- throws IoTDBConnectionException, StatementExecutionException, ExecutionException, InterruptedException {
-
- ExecutorService executorService = Executors.newFixedThreadPool(2 * concurrency + 1);
-
- String path = "root";
-
- if (args.length != 0) {
- path = args[0];
- }
-
- readerPool = new SessionPool("127.0.0.1", 6667, "root", "root", concurrency);
- writerPool = new SessionPool("127.0.0.1", 6668, "root", "root", concurrency);
-
- SessionDataSetWrapper schemaDataSet = readerPool
- .executeQueryStatement("count timeseries " + path);
- DataIterator schemaIter = schemaDataSet.iterator();
- int total;
- if (schemaIter.next()) {
- total = schemaIter.getInt(1);
- System.out.println("Total timeseries: " + total);
- } else {
- System.out.println("Can not get timeseries schema");
- System.exit(1);
- }
- readerPool.closeResultSet(schemaDataSet);
-
- schemaDataSet = readerPool
- .executeQueryStatement("show timeseries " + path);
- schemaIter = schemaDataSet.iterator();
-
- List<Future> futureList = new ArrayList<>();
- int count = 0;
- while (schemaIter.next()) {
- count ++;
- Path currentPath = new Path(schemaIter.getString("timeseries"));
- Future future = executorService.submit(
- new LoadThread(count, currentPath, TSDataType.valueOf(schemaIter.getString("dataType"))));
- futureList.add(future);
- }
- readerPool.closeResultSet(schemaDataSet);
-
- for (Future future : futureList) {
- future.get();
- }
- executorService.shutdown();
-
- readerPool.close();
- writerPool.close();
- }
-
-
- static class LoadThread implements Callable<Void> {
-
- String device;
- String measurement;
- Path series;
- TSDataType dataType;
- Tablet tablet;
- int i;
-
- public LoadThread(int i, Path series, TSDataType dataType) {
- this.i = i;
- this.device = series.getDevice();
- this.measurement = series.getMeasurement();
- this.dataType = dataType;
- this.series = series;
- List<MeasurementSchema> schemaList = new ArrayList<>();
- schemaList.add(new MeasurementSchema(measurement, dataType));
- tablet = new Tablet(device, schemaList, 300000);
- }
-
- @Override
- public Void call() {
-
- SessionDataSetWrapper dataSet = null;
-
- try {
-
- dataSet = readerPool
- .executeQueryStatement(String.format("select %s from %s", measurement, device));
-
- DataIterator dataIter = dataSet.iterator();
- while (dataIter.next()) {
- int row = tablet.rowSize++;
- tablet.timestamps[row] = dataIter.getLong(1);
- switch (dataType) {
- case BOOLEAN:
- ((boolean[]) tablet.values[0])[row] = dataIter.getBoolean(2);
- break;
- case INT32:
- ((int[]) tablet.values[0])[row] = dataIter.getInt(2);
- break;
- case INT64:
- ((long[]) tablet.values[0])[row] = dataIter.getLong(2);
- break;
- case FLOAT:
- ((float[]) tablet.values[0])[row] = dataIter.getFloat(2);
- break;
- case DOUBLE:
- ((double[]) tablet.values[0])[row] = dataIter.getDouble(2);
- break;
- case TEXT:
- ((Binary[]) tablet.values[0])[row] = new Binary(dataIter.getString(2));
- break;
- }
- if (tablet.rowSize == tablet.getMaxRowNumber()) {
- writerPool.insertTablet(tablet, true);
- tablet.reset();
- }
- }
- if (tablet.rowSize != 0) {
- writerPool.insertTablet(tablet);
- tablet.reset();
- }
-
- } catch (Exception e) {
- System.out.println(
- "Loading the " + i + "-th timeseries: " + series + " failed " + e.getMessage());
- return null;
- } finally {
- readerPool.closeResultSet(dataSet);
- }
-
- System.out.println("Loading the " + i + "-th timeseries: " + series + " success");
- return null;
- }
-
- }
-}
\ No newline at end of file
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 e8ca2fa..b6d390c 100644
--- a/example/session/src/main/java/org/apache/iotdb/SessionExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
@@ -19,31 +19,18 @@
package org.apache.iotdb;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.Random;
-import org.apache.iotdb.rpc.BatchExecutionException;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.rpc.TSStatusCode;
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;
public class SessionExample {
private static Session session;
- private static final String ROOT_SG1_D1_S1 = "root.sg1.d1.s1";
- private static final String ROOT_SG1_D1_S2 = "root.sg1.d1.s2";
- private static final String ROOT_SG1_D1_S3 = "root.sg1.d1.s3";
- private static final String ROOT_SG1_D1_S4 = "root.sg1.d1.s4";
- private static final String ROOT_SG1_D1_S5 = "root.sg1.d1.s5";
private static final String ROOT_SG1_D1 = "root.sg1.d1";
@@ -55,205 +42,11 @@ public class SessionExample {
//set session fetchSize
session.setFetchSize(10000);
- try {
- session.setStorageGroup("root.sg1");
- } catch (StatementExecutionException e) {
- if (e.getStatusCode() != TSStatusCode.PATH_ALREADY_EXIST_ERROR.getStatusCode()) {
- throw e;
- }
- }
-
- createTimeseries();
- createMultiTimeseries();
- insertRecord();
- insertTablet();
- insertTablets();
- insertRecords();
- nonQuery();
- query();
- queryWithTimeout();
- rawDataQuery();
- queryByIterator();
- deleteData();
- deleteTimeseries();
- setTimeout();
+ long MAX_ROW_NUM = Long.parseLong(args[0]);
+ System.out.println("MAX_ROW_NUM: " + MAX_ROW_NUM);
+ insertTablet(MAX_ROW_NUM);
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);
- }
-
- // 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");
- }
-
- // create timeseries with SDT property, SDT will take place when flushing
- if (!session.checkTimeseriesExists(ROOT_SG1_D1_S5)) {
- // COMPDEV is required
- // COMPMAXTIME and COMPMINTIME are optional and their unit is ms
- Map<String, String> props = new HashMap<>();
- props.put("LOSS", "sdt");
- props.put("COMPDEV", "0.01");
- props.put("COMPMINTIME", "2");
- props.put("COMPMAXTIME", "10");
- session.createTimeseries(ROOT_SG1_D1_S5, TSDataType.INT64, TSEncoding.RLE,
- CompressionType.SNAPPY, props, null, null, null);
- }
- }
-
- private static void createMultiTimeseries()
- throws IoTDBConnectionException, BatchExecutionException, StatementExecutionException {
-
- 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);
-
- 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);
- }
- }
-
- private static void insertRecord() throws IoTDBConnectionException, StatementExecutionException {
- String deviceId = ROOT_SG1_D1;
- List<String> measurements = new ArrayList<>();
- List<TSDataType> types = new ArrayList<>();
- measurements.add("s1");
- measurements.add("s2");
- measurements.add("s3");
- types.add(TSDataType.INT64);
- types.add(TSDataType.INT64);
- types.add(TSDataType.INT64);
-
- for (long time = 0; time < 100; time++) {
- List<Object> values = new ArrayList<>();
- values.add(1L);
- values.add(2L);
- values.add(3L);
- session.insertRecord(deviceId, time, measurements, types, values);
- }
- }
-
- private static void insertStrRecord()
- 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 < 10; 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<>();
- List<TSDataType> types = new ArrayList<>();
- measurements.add("s1");
- measurements.add("s2");
- measurements.add("s3");
- types.add(TSDataType.INT64);
- types.add(TSDataType.INT64);
- types.add(TSDataType.INT64);
-
- for (long time = 0; time < 100; time++) {
- session.insertRecord(deviceId, time, measurements, types, 1L, 1L, 1L);
- }
- }
-
- private static void insertRecords() throws IoTDBConnectionException, StatementExecutionException {
- 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<Object>> valuesList = new ArrayList<>();
- List<Long> timestamps = new ArrayList<>();
- List<List<TSDataType>> typesList = new ArrayList<>();
-
- for (long time = 0; time < 500; time++) {
- List<Object> values = new ArrayList<>();
- List<TSDataType> types = new ArrayList<>();
- values.add(1L);
- values.add(2L);
- values.add(3L);
- types.add(TSDataType.INT64);
- types.add(TSDataType.INT64);
- types.add(TSDataType.INT64);
-
- deviceIds.add(deviceId);
- measurementsList.add(measurements);
- valuesList.add(values);
- typesList.add(types);
- timestamps.add(time);
- if (time != 0 && time % 100 == 0) {
- session.insertRecords(deviceIds, timestamps, measurementsList, typesList, valuesList);
- deviceIds.clear();
- measurementsList.clear();
- valuesList.clear();
- timestamps.clear();
- }
- }
-
- session.insertRecords(deviceIds, timestamps, measurementsList, typesList, valuesList);
- }
-
/**
* insert the data of a device. For each timestamp, the number of measurements is the same.
*
@@ -267,24 +60,24 @@ public class SessionExample {
*
* Users need to control the count of Tablet and write a batch when it reaches the maxBatchSize
*/
- private static void insertTablet() throws IoTDBConnectionException, StatementExecutionException {
+ private static void insertTablet(long MAX_ROW_NUM) throws IoTDBConnectionException, StatementExecutionException {
// The schema of measurements of one device
// only measurementId and data type in MeasurementSchema take effects in Tablet
List<MeasurementSchema> schemaList = new ArrayList<>();
- schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
- schemaList.add(new MeasurementSchema("s2", TSDataType.INT64));
- schemaList.add(new MeasurementSchema("s3", TSDataType.INT64));
+ for (int i = 0; i < 1000; i++) {
+ schemaList.add(new MeasurementSchema("s" + i, TSDataType.DOUBLE));
+ }
- Tablet tablet = new Tablet(ROOT_SG1_D1, schemaList, 100);
+ Tablet tablet = new Tablet(ROOT_SG1_D1, schemaList, 1000);
//Method 1 to add tablet data
long timestamp = System.currentTimeMillis();
- for (long row = 0; row < 100; row++) {
+ for (long row = 0; row < MAX_ROW_NUM; row++) {
int rowIndex = tablet.rowSize++;
tablet.addTimestamp(rowIndex, timestamp);
- for (int s = 0; s < 3; s++) {
- long value = new Random().nextLong();
+ for (int s = 0; s < 1000; s++) {
+ double value = new Random().nextDouble();
tablet.addValue(schemaList.get(s).getMeasurementId(), rowIndex, value);
}
if (tablet.rowSize == tablet.getMaxRowNumber()) {
@@ -298,223 +91,5 @@ public class SessionExample {
session.insertTablet(tablet);
tablet.reset();
}
-
- //Method 2 to add tablet data
- 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();
- }
- }
-
- private static void insertTablets() throws IoTDBConnectionException, StatementExecutionException {
- // The schema of measurements of one device
- // only measurementId and data type in MeasurementSchema take effects in Tablet
- List<MeasurementSchema> schemaList = new ArrayList<>();
- schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
- schemaList.add(new MeasurementSchema("s2", TSDataType.INT64));
- schemaList.add(new MeasurementSchema("s3", TSDataType.INT64));
-
- 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);
-
- //Method 1 to add tablet data
- long timestamp = System.currentTimeMillis();
- for (long row = 0; row < 100; row++) {
- int row1 = tablet1.rowSize++;
- int row2 = tablet2.rowSize++;
- int row3 = tablet3.rowSize++;
- tablet1.addTimestamp(row1, timestamp);
- tablet2.addTimestamp(row2, timestamp);
- tablet3.addTimestamp(row3, timestamp);
- for (int i = 0; i < 3; i++) {
- long value = new Random().nextLong();
- tablet1.addValue(schemaList.get(i).getMeasurementId(), row1, value);
- tablet2.addValue(schemaList.get(i).getMeasurementId(), row2, value);
- tablet3.addValue(schemaList.get(i).getMeasurementId(), row3, value);
- }
- if (tablet1.rowSize == tablet1.getMaxRowNumber()) {
- session.insertTablets(tabletMap, true);
- tablet1.reset();
- tablet2.reset();
- tablet3.reset();
- }
- timestamp++;
- }
-
- if (tablet1.rowSize != 0) {
- session.insertTablets(tabletMap, true);
- tablet1.reset();
- tablet2.reset();
- tablet3.reset();
- }
-
- //Method 2 to add tablet data
- 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();
- }
- }
-
- 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 = session.executeQueryStatement("select * from root.sg1.d1");
- System.out.println(dataSet.getColumnNames());
- dataSet.setFetchSize(1024); // default is 10000
- while (dataSet.hasNext()) {
- System.out.println(dataSet.next());
- }
-
- dataSet.closeOperationHandle();
- }
-
- private static void queryWithTimeout()
- throws IoTDBConnectionException, StatementExecutionException {
- SessionDataSet dataSet = session.executeQueryStatement("select * from root.sg1.d1", 2000);
- System.out.println(dataSet.getColumnNames());
- dataSet.setFetchSize(1024); // default is 10000
- while (dataSet.hasNext()) {
- System.out.println(dataSet.next());
- }
-
- dataSet.closeOperationHandle();
- }
-
- private static void rawDataQuery() 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);
- long startTime = 10L;
- long endTime = 200L;
-
- SessionDataSet dataSet = session.executeRawDataQuery(paths, startTime, endTime);
- System.out.println(dataSet.getColumnNames());
- dataSet.setFetchSize(1024);
- while (dataSet.hasNext()) {
- System.out.println(dataSet.next());
- }
- dataSet.closeOperationHandle();
- }
-
- private static void queryByIterator()
- throws IoTDBConnectionException, StatementExecutionException {
- SessionDataSet dataSet = session.executeQueryStatement("select * from root.sg1.d1");
- DataIterator iterator = dataSet.iterator();
- System.out.println(dataSet.getColumnNames());
- dataSet.setFetchSize(1024); // default is 10000
- while (iterator.next()) {
- StringBuilder builder = new StringBuilder();
- // get time
- builder.append(iterator.getLong(1)).append(",");
- // get second column
- if (!iterator.isNull(2)) {
- builder.append(iterator.getLong(2)).append(",");
- } else {
- builder.append("null").append(",");
- }
-
- // get third column
- if (!iterator.isNull(ROOT_SG1_D1_S2)) {
- builder.append(iterator.getLong(ROOT_SG1_D1_S2)).append(",");
- } else {
- builder.append("null").append(",");
- }
-
- // get forth column
- if (!iterator.isNull(4)) {
- builder.append(iterator.getLong(4)).append(",");
- } else {
- builder.append("null").append(",");
- }
-
- // get fifth column
- if (!iterator.isNull(ROOT_SG1_D1_S4)) {
- builder.append(iterator.getObject(ROOT_SG1_D1_S4));
- } else {
- builder.append("null");
- }
-
- System.out.println(builder.toString());
- }
-
- dataSet.closeOperationHandle();
- }
-
- private static void nonQuery() throws IoTDBConnectionException, StatementExecutionException {
- session.executeNonQueryStatement("insert into root.sg1.d1(timestamp,s1) values(200, 1)");
- }
-
- private static void setTimeout() throws StatementExecutionException {
- Session tempSession = new Session("127.0.0.1", 6667, "root", "root", 10000, 20000);
- tempSession.setTimeout(60000);
}
-}
+}
\ No newline at end of file
diff --git a/example/session/src/main/java/org/apache/iotdb/SessionPoolExample.java b/example/session/src/main/java/org/apache/iotdb/SessionPoolExample.java
deleted file mode 100644
index 395312e..0000000
--- a/example/session/src/main/java/org/apache/iotdb/SessionPoolExample.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import org.apache.iotdb.rpc.IoTDBConnectionException;
-import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.session.SessionDataSet.DataIterator;
-import org.apache.iotdb.session.pool.SessionDataSetWrapper;
-import org.apache.iotdb.session.pool.SessionPool;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-
-public class SessionPoolExample {
-
- private static SessionPool pool;
- private static ExecutorService service;
-
- public static void main(String[] args)
- throws StatementExecutionException, IoTDBConnectionException, InterruptedException {
- pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3);
- service = Executors.newFixedThreadPool(10);
-
- insertRecord();
- queryByRowRecord();
- Thread.sleep(1000);
- queryByIterator();
- pool.close();
- service.shutdown();
- }
-
- // more insert example, see SessionExample.java
- private static void insertRecord() throws StatementExecutionException, IoTDBConnectionException {
- String deviceId = "root.sg1.d1";
- List<String> measurements = new ArrayList<>();
- List<TSDataType> types = new ArrayList<>();
- measurements.add("s1");
- measurements.add("s2");
- measurements.add("s3");
- types.add(TSDataType.INT64);
- types.add(TSDataType.INT64);
- types.add(TSDataType.INT64);
-
- for (long time = 0; time < 10; time++) {
- List<Object> values = new ArrayList<>();
- values.add(1L);
- values.add(2L);
- values.add(3L);
- pool.insertRecord(deviceId, time, measurements, types, values);
- }
- }
-
- private static void queryByRowRecord() {
- for (int i = 0; i < 1; i++) {
- service.submit(() -> {
- SessionDataSetWrapper wrapper = null;
- try {
- wrapper = pool.executeQueryStatement("select * from root.sg1.d1");
- System.out.println(wrapper.getColumnNames());
- System.out.println(wrapper.getColumnTypes());
- while (wrapper.hasNext()) {
- System.out.println(wrapper.next());
- }
- } catch (IoTDBConnectionException | StatementExecutionException e) {
- e.printStackTrace();
- } finally {
- // remember to close data set finally!
- pool.closeResultSet(wrapper);
- }
- });
- }
- }
-
- private static void queryByIterator() {
- for (int i = 0; i < 1; i++) {
- service.submit(() -> {
- SessionDataSetWrapper wrapper = null;
- try {
- wrapper = pool.executeQueryStatement("select * from root.sg1.d1");
- // get DataIterator like JDBC
- DataIterator dataIterator = wrapper.iterator();
- System.out.println(wrapper.getColumnNames());
- System.out.println(wrapper.getColumnTypes());
- while (dataIterator.next()) {
- StringBuilder builder = new StringBuilder();
- for (String columnName: wrapper.getColumnNames()) {
- builder.append(dataIterator.getString(columnName) + " ");
- }
- System.out.println(builder.toString());
- }
- } catch (IoTDBConnectionException | StatementExecutionException e) {
- e.printStackTrace();
- } finally {
- // remember to close data set finally!
- pool.closeResultSet(wrapper);
- }
- });
- }
- }
-
-}