You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2024/09/04 02:44:59 UTC
(iotdb) 01/03: yes
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch iotdb-influxdb-proto-example
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0adadac9796cfb89fda3a894c448ee9d1816638b
Author: Steve Yurong Su <ro...@apache.org>
AuthorDate: Tue Sep 3 18:17:37 2024 +0800
yes
---
example/influxdb-proto-example/pom.xml | 43 +
.../iotdb/AlignedTimeseriesSessionExample.java | 637 ++++++++++++++
.../org/apache/iotdb/DataMigrationExample.java | 211 +++++
.../iotdb/HybridTimeseriesSessionExample.java | 131 +++
.../org/apache/iotdb/SessionConcurrentExample.java | 197 +++++
.../main/java/org/apache/iotdb/SessionExample.java | 914 +++++++++++++++++++++
.../java/org/apache/iotdb/SessionPoolExample.java | 154 ++++
.../apache/iotdb/SubscriptionSessionExample.java | 345 ++++++++
.../iotdb/SyntaxConventionRelatedExample.java | 156 ++++
.../org/apache/iotdb/TableModelSessionExample.java | 136 +++
.../apache/iotdb/TableModelSessionPoolExample.java | 147 ++++
.../main/java/org/apache/iotdb/TabletExample.java | 204 +++++
example/pom.xml | 1 +
iotdb-client/session/pom.xml | 5 +
.../iotdb/session/influxdb/IoTDBInfluxDB.java | 393 +++++++++
.../influxdb/IoTDBInfluxDBProtoFactory.java | 74 ++
.../influxdb/example/InfluxProtoExample.java | 72 ++
.../protocol/constant/InfluxDBConstant.java | 25 +
.../influxdb/protocol/dto/SessionPoint.java | 88 ++
.../protocol/impl/IoTDBInfluxDBService.java | 77 ++
.../influxdb/protocol/util/ParameterUtils.java | 36 +
.../session/influxdb/session/InfluxDBSession.java | 219 +++++
22 files changed, 4265 insertions(+)
diff --git a/example/influxdb-proto-example/pom.xml b/example/influxdb-proto-example/pom.xml
new file mode 100644
index 00000000000..92d8e9b5b4d
--- /dev/null
+++ b/example/influxdb-proto-example/pom.xml
@@ -0,0 +1,43 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-examples</artifactId>
+ <version>1.4.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>influxdb-proto-example</artifactId>
+ <name>IoTDB: Example: InfluxDB Proto</name>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-session</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>node-commons</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java
new file mode 100644
index 00000000000..1c412482800
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java
@@ -0,0 +1,637 @@
+/*
+ * 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 org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.template.Template;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.template.InternalNode;
+import org.apache.iotdb.session.template.MeasurementNode;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.io.IOException;
+import java.security.SecureRandom;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@SuppressWarnings({"squid:S106", "squid:S1144", "squid:S125"})
+public class AlignedTimeseriesSessionExample {
+
+ private static Session session;
+ private static final String ROOT_SG1_D1 = "root.sg_1.d1";
+ private static final String ROOT_SG1_D1_VECTOR2 = "root.sg_1.d1.vector2";
+ private static final String ROOT_SG1_D1_VECTOR3 = "root.sg_1.d1.vector3";
+ private static final String ROOT_SG2_D1_VECTOR4 = "root.sg_2.d1.vector4";
+ private static final String ROOT_SG2_D1_VECTOR5 = "root.sg_2.d1.vector5";
+ private static final String ROOT_SG2_D1_VECTOR6 = "root.sg_2.d1.vector6";
+ private static final String ROOT_SG2_D1_VECTOR7 = "root.sg_2.d1.vector7";
+ private static final String ROOT_SG2_D1_VECTOR8 = "root.sg_2.d1.vector8";
+ public static final String FLUSH = "flush";
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException, StatementExecutionException {
+ session = new Session("127.0.0.1", 6667, "root", "root");
+ session.open(false);
+
+ // set session fetchSize
+ session.setFetchSize(10000);
+
+ // createTemplate();
+ createAlignedTimeseries();
+
+ insertAlignedRecord();
+ // insertAlignedRecords();
+ // insertAlignedRecordsOfOneDevice();
+
+ // insertAlignedStringRecord();
+ // insertAlignedStringRecords();
+
+ // insertTabletWithAlignedTimeseriesMethod1();
+ // insertTabletWithAlignedTimeseriesMethod2();
+ // insertNullableTabletWithAlignedTimeseries();
+ // insertTabletsWithAlignedTimeseries();
+ session.executeNonQueryStatement(FLUSH);
+ selectTest();
+ selectWithValueFilterTest();
+ selectWithLastTest();
+ selectWithLastTestWithoutValueFilter();
+ session.executeNonQueryStatement("delete from root.sg_1.d1.s1 where time <= 5");
+ System.out.println("execute sql delete from root.sg_1.d1.s1 where time <= 5");
+ selectTest();
+ selectWithValueFilterTest();
+ selectWithLastTest();
+ selectWithLastTestWithoutValueFilter();
+ session.executeNonQueryStatement("delete from root.sg_1.d1.s2 where time <= 3");
+ System.out.println("execute sql delete from root.sg_1.d1.s2 where time <= 3");
+
+ selectTest();
+ selectWithValueFilterTest();
+ selectWithLastTest();
+ selectWithLastTestWithoutValueFilter();
+ session.executeNonQueryStatement("delete from root.sg_1.d1.s1 where time <= 10");
+ System.out.println("execute sql delete from root.sg_1.d1.s1 where time <= 10");
+ selectTest();
+ selectWithValueFilterTest();
+ selectWithLastTest();
+ selectWithLastTestWithoutValueFilter();
+
+ // selectWithValueFilterTest();
+ // selectWithGroupByTest();
+ // selectWithLastTest();
+
+ // selectWithAggregationTest();
+
+ // selectWithAlignByDeviceTest();
+
+ session.close();
+ }
+
+ private static void selectTest() throws StatementExecutionException, IoTDBConnectionException {
+ SessionDataSet dataSet = session.executeQueryStatement("select s1 from root.sg_1.d1");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ dataSet = session.executeQueryStatement("select * from root.sg_1.d1");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ }
+
+ private static void selectWithAlignByDeviceTest()
+ throws StatementExecutionException, IoTDBConnectionException {
+ SessionDataSet dataSet =
+ session.executeQueryStatement("select * from root.sg_1 align by device");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ }
+
+ private static void selectWithValueFilterTest()
+ throws StatementExecutionException, IoTDBConnectionException {
+ SessionDataSet dataSet =
+ session.executeQueryStatement("select s1 from root.sg_1.d1 where s1 > 3 and time < 9");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ dataSet =
+ session.executeQueryStatement(
+ "select * from root.sg_1.d1 where time < 8 and s1 > 3 and s2 > 5");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ }
+
+ private static void selectWithAggregationTest()
+ throws StatementExecutionException, IoTDBConnectionException {
+ SessionDataSet dataSet = session.executeQueryStatement("select count(s1) from root.sg_1.d1");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ dataSet = session.executeQueryStatement("select count(*) from root.sg_1.d1");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ dataSet =
+ session.executeQueryStatement(
+ "select sum(*) from root.sg_1.d1.vector where time > 50 and s1 > 0 and s2 > 10000");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ }
+
+ private static void selectWithGroupByTest()
+ throws StatementExecutionException, IoTDBConnectionException {
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select count(s1) from root.sg_1.d1.vector GROUP BY ([1, 100), 20ms)");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from root.sg_1.d1.vector where time > 50 and s1 > 0 and s2 > 10000"
+ + " GROUP BY ([50, 100), 10ms)");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ }
+
+ private static void selectWithLastTest()
+ throws StatementExecutionException, IoTDBConnectionException {
+ SessionDataSet dataSet = session.executeQueryStatement("select last s1 from root.sg_1.d1");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ dataSet = session.executeQueryStatement("select last * from root.sg_1.d1");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ }
+
+ private static void selectWithLastTestWithoutValueFilter()
+ throws StatementExecutionException, IoTDBConnectionException {
+ SessionDataSet dataSet =
+ session.executeQueryStatement("select last s1 from root.sg_1.d1 where time >= 5");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+
+ dataSet = session.executeQueryStatement("select last * from root.sg_1.d1 where time >= 5");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ dataSet.closeOperationHandle();
+
+ dataSet = session.executeQueryStatement("select last * from root.sg_1.d1 where time >= 20");
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ dataSet.closeOperationHandle();
+ }
+
+ private static void createAlignedTimeseries()
+ throws StatementExecutionException, IoTDBConnectionException {
+ List<String> measurements = new ArrayList<>();
+ for (int i = 1; i <= 2; i++) {
+ measurements.add("s" + i);
+ }
+ List<TSDataType> dataTypes = new ArrayList<>();
+ dataTypes.add(TSDataType.INT64);
+ dataTypes.add(TSDataType.INT32);
+ List<TSEncoding> encodings = new ArrayList<>();
+ List<CompressionType> compressors = new ArrayList<>();
+ for (int i = 1; i <= 2; i++) {
+ encodings.add(TSEncoding.RLE);
+ compressors.add(CompressionType.SNAPPY);
+ }
+ session.createAlignedTimeseries(
+ ROOT_SG1_D1, measurements, dataTypes, encodings, compressors, null, null, null);
+ }
+
+ // be sure template is coordinate with tablet
+ private static void createTemplate()
+ throws StatementExecutionException, IoTDBConnectionException, IOException {
+ Template template = new Template("template1");
+ InternalNode iNodeVector = new InternalNode("vector", true);
+ MeasurementNode mNodeS1 =
+ new MeasurementNode("s1", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
+ MeasurementNode mNodeS2 =
+ new MeasurementNode("s2", TSDataType.INT32, TSEncoding.RLE, CompressionType.SNAPPY);
+
+ iNodeVector.addChild(mNodeS1);
+ iNodeVector.addChild(mNodeS2);
+
+ template.addToTemplate(iNodeVector);
+
+ session.createSchemaTemplate(template);
+ session.setSchemaTemplate("template1", "root.sg_1");
+ }
+
+ /** Method 1 for insert tablet with aligned timeseries */
+ private static void insertTabletWithAlignedTimeseriesMethod1()
+ throws IoTDBConnectionException, StatementExecutionException {
+ // The schema of measurements of one device
+ // only measurementId and data type in MeasurementSchema take effects in Tablet
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT32));
+
+ Tablet tablet = new Tablet(ROOT_SG1_D1, schemaList);
+ long timestamp = 1;
+
+ for (long row = 1; row < 100; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp);
+ tablet.addValue(
+ schemaList.get(0).getMeasurementId(), rowIndex, new SecureRandom().nextLong());
+ tablet.addValue(schemaList.get(1).getMeasurementId(), rowIndex, new SecureRandom().nextInt());
+
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertAlignedTablet(tablet, true);
+ tablet.reset();
+ }
+ timestamp++;
+ }
+
+ if (tablet.rowSize != 0) {
+ session.insertAlignedTablet(tablet);
+ tablet.reset();
+ }
+
+ session.executeNonQueryStatement(FLUSH);
+ }
+
+ /** Method 2 for insert tablet with aligned timeseries */
+ private static void insertTabletWithAlignedTimeseriesMethod2()
+ throws IoTDBConnectionException, StatementExecutionException {
+ // The schema of measurements of one device
+ // only measurementId and data type in MeasurementSchema take effects in Tablet
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT32));
+
+ Tablet tablet = new Tablet(ROOT_SG1_D1_VECTOR2, schemaList);
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
+
+ for (long time = 100; time < 200; time++) {
+ int row = tablet.rowSize++;
+ timestamps[row] = time;
+
+ long[] sensor1 = (long[]) values[0];
+ sensor1[row] = new SecureRandom().nextLong();
+
+ int[] sensor2 = (int[]) values[1];
+ sensor2[row] = new SecureRandom().nextInt();
+
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertAlignedTablet(tablet, true);
+ tablet.reset();
+ }
+ }
+
+ if (tablet.rowSize != 0) {
+ session.insertAlignedTablet(tablet, true);
+ tablet.reset();
+ }
+
+ session.executeNonQueryStatement(FLUSH);
+ }
+
+ private static void insertNullableTabletWithAlignedTimeseries()
+ throws IoTDBConnectionException, StatementExecutionException {
+ // The schema of measurements of one device
+ // only measurementId and data type in MeasurementSchema take effects in Tablet
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT32));
+
+ Tablet tablet = new Tablet(ROOT_SG1_D1_VECTOR3, schemaList);
+
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
+ // Use the bitMap to mark the null value point
+ BitMap[] bitMaps = new BitMap[values.length];
+ tablet.bitMaps = bitMaps;
+
+ bitMaps[1] = new BitMap(tablet.getMaxRowNumber());
+ for (long time = 200; time < 300; time++) {
+ int row = tablet.rowSize++;
+ timestamps[row] = time;
+
+ long[] sensor1 = (long[]) values[0];
+ sensor1[row] = new SecureRandom().nextLong();
+
+ int[] sensor2 = (int[]) values[1];
+ sensor2[row] = new SecureRandom().nextInt();
+
+ // mark this point as null value
+ if (time % 5 == 0) {
+ bitMaps[1].mark(row);
+ }
+
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertAlignedTablet(tablet, true);
+ tablet.reset();
+ bitMaps[1].reset();
+ }
+ }
+
+ if (tablet.rowSize != 0) {
+ session.insertAlignedTablet(tablet, true);
+ tablet.reset();
+ }
+
+ session.executeNonQueryStatement(FLUSH);
+ }
+
+ private static void insertAlignedRecord()
+ throws IoTDBConnectionException, StatementExecutionException {
+ // first file we have both sensots' data
+ List<String> measurements = new ArrayList<>();
+ List<TSDataType> types = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+ types.add(TSDataType.INT64);
+ types.add(TSDataType.INT32);
+
+ for (long time = 0; time < 10; time++) {
+ List<Object> values = new ArrayList<>();
+ values.add(time);
+ values.add((int) time);
+ session.insertAlignedRecord(ROOT_SG1_D1, time, measurements, types, values);
+ }
+ session.executeNonQueryStatement(FLUSH);
+ // second file we only have s1's data
+ measurements.clear();
+ types.clear();
+ measurements.add("s1");
+ types.add(TSDataType.INT64);
+ for (long time = 10; time < 20; time++) {
+ List<Object> values = new ArrayList<>();
+ values.add(time);
+ session.insertAlignedRecord(ROOT_SG1_D1, time, measurements, types, values);
+ }
+ }
+
+ private static void insertAlignedStringRecord()
+ throws IoTDBConnectionException, StatementExecutionException {
+ List<String> measurements = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+
+ for (long time = 0; time < 1; time++) {
+ List<String> values = new ArrayList<>();
+ values.add("3");
+ values.add("4");
+ session.insertAlignedRecord(ROOT_SG2_D1_VECTOR5, time, measurements, values);
+ }
+ }
+
+ private static void insertAlignedRecords()
+ throws IoTDBConnectionException, StatementExecutionException {
+ List<String> deviceIds = new ArrayList<>();
+ List<List<String>> measurementsList = new ArrayList<>();
+ List<List<TSDataType>> typeList = new ArrayList<>();
+ List<Long> times = new ArrayList<>();
+ List<List<Object>> valueList = new ArrayList<>();
+
+ for (long time = 1; time < 5; time++) {
+ List<String> measurements = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+
+ List<TSDataType> types = new ArrayList<>();
+ types.add(TSDataType.INT64);
+ types.add(TSDataType.INT32);
+
+ List<Object> values = new ArrayList<>();
+ values.add(1L);
+ values.add(2);
+
+ deviceIds.add(ROOT_SG2_D1_VECTOR4);
+ times.add(time);
+ measurementsList.add(measurements);
+ typeList.add(types);
+ valueList.add(values);
+ }
+ session.insertAlignedRecords(deviceIds, times, measurementsList, typeList, valueList);
+ }
+
+ private static void insertAlignedStringRecords()
+ throws IoTDBConnectionException, StatementExecutionException {
+ List<String> deviceIds = new ArrayList<>();
+ List<List<String>> measurementsList = new ArrayList<>();
+ List<Long> times = new ArrayList<>();
+ List<List<String>> valueList = new ArrayList<>();
+
+ for (long time = 1; time < 5; time++) {
+ List<String> measurements = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+
+ List<String> values = new ArrayList<>();
+ values.add("3");
+ values.add("4");
+
+ deviceIds.add(ROOT_SG2_D1_VECTOR5);
+ times.add(time);
+ measurementsList.add(measurements);
+ valueList.add(values);
+ }
+ session.insertAlignedRecords(deviceIds, times, measurementsList, valueList);
+ }
+
+ private static void insertAlignedRecordsOfOneDevice()
+ throws IoTDBConnectionException, StatementExecutionException {
+ List<List<String>> measurementsList = new ArrayList<>();
+ List<List<TSDataType>> typeList = new ArrayList<>();
+ List<Long> times = new ArrayList<>();
+ List<List<Object>> valueList = new ArrayList<>();
+
+ for (long time = 10; time < 15; time++) {
+ List<String> measurements = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+
+ List<TSDataType> types = new ArrayList<>();
+ types.add(TSDataType.INT64);
+ types.add(TSDataType.INT32);
+
+ List<Object> values = new ArrayList<>();
+ values.add(1L);
+ values.add(2);
+
+ times.add(time);
+ measurementsList.add(measurements);
+ typeList.add(types);
+ valueList.add(values);
+ }
+ session.insertAlignedRecordsOfOneDevice(
+ ROOT_SG2_D1_VECTOR4, times, measurementsList, typeList, valueList);
+ }
+
+ private static void insertTabletsWithAlignedTimeseries()
+ throws IoTDBConnectionException, StatementExecutionException {
+
+ List<IMeasurementSchema> schemaList1 = new ArrayList<>();
+ schemaList1.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList1.add(new MeasurementSchema("s2", TSDataType.INT64));
+
+ List<IMeasurementSchema> schemaList2 = new ArrayList<>();
+ schemaList2.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList2.add(new MeasurementSchema("s2", TSDataType.INT64));
+
+ List<IMeasurementSchema> schemaList3 = new ArrayList<>();
+ schemaList3.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList3.add(new MeasurementSchema("s2", TSDataType.INT64));
+
+ Tablet tablet1 = new Tablet(ROOT_SG2_D1_VECTOR6, schemaList1, 100);
+ Tablet tablet2 = new Tablet(ROOT_SG2_D1_VECTOR7, schemaList2, 100);
+ Tablet tablet3 = new Tablet(ROOT_SG2_D1_VECTOR8, schemaList3, 100);
+
+ Map<String, Tablet> tabletMap = new HashMap<>();
+ tabletMap.put(ROOT_SG2_D1_VECTOR6, tablet1);
+ tabletMap.put(ROOT_SG2_D1_VECTOR7, tablet2);
+ tabletMap.put(ROOT_SG2_D1_VECTOR8, 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 < 2; i++) {
+ long value = new SecureRandom().nextLong();
+ tablet1.addValue(schemaList1.get(i).getMeasurementId(), row1, value);
+ tablet2.addValue(schemaList2.get(i).getMeasurementId(), row2, value);
+ tablet3.addValue(schemaList3.get(i).getMeasurementId(), row3, value);
+ }
+ if (tablet1.rowSize == tablet1.getMaxRowNumber()) {
+ session.insertAlignedTablets(tabletMap, true);
+ tablet1.reset();
+ tablet2.reset();
+ tablet3.reset();
+ }
+ timestamp++;
+ }
+
+ if (tablet1.rowSize != 0) {
+ session.insertAlignedTablets(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 < 2; 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.insertAlignedTablets(tabletMap, true);
+
+ tablet1.reset();
+ tablet2.reset();
+ tablet3.reset();
+ }
+ }
+
+ if (tablet1.rowSize != 0) {
+ session.insertAlignedTablets(tabletMap, true);
+ tablet1.reset();
+ tablet2.reset();
+ tablet3.reset();
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/DataMigrationExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/DataMigrationExample.java
new file mode 100644
index 00000000000..288ed7eaa03
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/DataMigrationExample.java
@@ -0,0 +1,211 @@
+/*
+ * 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 org.apache.iotdb.commons.path.MeasurementPath;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.isession.SessionDataSet.DataIterator;
+import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.pool.SessionPool;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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;
+
+/**
+ * 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
+ *
+ * <p>This example is migrating all timeseries from a local IoTDB with 6667 port to a local IoTDB
+ * with 6668 port
+ */
+public class DataMigrationExample {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(DataMigrationExample.class);
+
+ // 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 final int CONCURRENCY = 5;
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException,
+ StatementExecutionException,
+ ExecutionException,
+ InterruptedException {
+
+ // the thread used for dataMigration must be smaller than session pool size
+ ExecutorService executorService = Executors.newFixedThreadPool(CONCURRENCY);
+
+ 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 deviceDataSet = readerPool.executeQueryStatement("count devices " + path);
+ DataIterator deviceIter = deviceDataSet.iterator();
+ int total;
+ if (deviceIter.next()) {
+ total = deviceIter.getInt(1);
+ LOGGER.info("Total devices: {}", total);
+ } else {
+ LOGGER.error("Can not get devices schema");
+ System.exit(1);
+ }
+ readerPool.closeResultSet(deviceDataSet);
+
+ deviceDataSet = readerPool.executeQueryStatement("show devices " + path);
+ deviceIter = deviceDataSet.iterator();
+
+ List<Future<Void>> futureList = new ArrayList<>();
+ int count = 0;
+ while (deviceIter.next()) {
+ count++;
+ Future<Void> future =
+ executorService.submit(new LoadThread(count, deviceIter.getString("Device")));
+ futureList.add(future);
+ }
+ readerPool.closeResultSet(deviceDataSet);
+
+ for (Future<Void> future : futureList) {
+ future.get();
+ }
+ executorService.shutdown();
+
+ readerPool.close();
+ writerPool.close();
+ }
+
+ static class LoadThread implements Callable<Void> {
+
+ String device;
+ Tablet tablet;
+ int i;
+
+ public LoadThread(int i, String device) {
+ this.i = i;
+ this.device = device;
+ }
+
+ @Override
+ public Void call() {
+ SessionDataSetWrapper dataSet = null;
+ long startTime = System.currentTimeMillis();
+ try {
+ dataSet = readerPool.executeQueryStatement(String.format("select * from %s", device));
+ DataIterator dataIter = dataSet.iterator();
+ List<String> columnNameList = dataIter.getColumnNameList();
+ List<String> columnTypeList = dataIter.getColumnTypeList();
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ for (int j = 1; j < columnNameList.size(); j++) {
+ PartialPath currentPath = new MeasurementPath(columnNameList.get(j));
+ schemaList.add(
+ new MeasurementSchema(
+ currentPath.getMeasurement(), TSDataType.valueOf(columnTypeList.get(j))));
+ }
+ tablet = new Tablet(device, schemaList, 300000);
+ while (dataIter.next()) {
+ int row = tablet.rowSize++;
+ tablet.timestamps[row] = dataIter.getLong(1);
+ for (int j = 0; j < schemaList.size(); ++j) {
+ if (dataIter.isNull(j + 2)) {
+ tablet.addValue(schemaList.get(j).getMeasurementId(), row, null);
+ continue;
+ }
+ switch (schemaList.get(j).getType()) {
+ case BOOLEAN:
+ tablet.addValue(
+ schemaList.get(j).getMeasurementId(), row, dataIter.getBoolean(j + 2));
+ break;
+ case INT32:
+ tablet.addValue(schemaList.get(j).getMeasurementId(), row, dataIter.getInt(j + 2));
+ break;
+ case INT64:
+ case TIMESTAMP:
+ tablet.addValue(schemaList.get(j).getMeasurementId(), row, dataIter.getLong(j + 2));
+ break;
+ case FLOAT:
+ tablet.addValue(
+ schemaList.get(j).getMeasurementId(), row, dataIter.getFloat(j + 2));
+ break;
+ case DOUBLE:
+ tablet.addValue(
+ schemaList.get(j).getMeasurementId(), row, dataIter.getDouble(j + 2));
+ break;
+ case TEXT:
+ case STRING:
+ tablet.addValue(
+ schemaList.get(j).getMeasurementId(), row, dataIter.getString(j + 2));
+ break;
+ case DATE:
+ case BLOB:
+ tablet.addValue(
+ schemaList.get(j).getMeasurementId(), row, dataIter.getObject(j + 2));
+ break;
+ default:
+ LOGGER.info("Migration of this type of data is not supported");
+ }
+ }
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ writerPool.insertTablet(tablet, true);
+ tablet.reset();
+ }
+ }
+ if (tablet.rowSize != 0) {
+ writerPool.insertTablet(tablet);
+ tablet.reset();
+ }
+
+ } catch (Exception e) {
+ LOGGER.error("Loading the {}-th device: {} failed {}", i, device, e.getMessage());
+ return null;
+ } finally {
+ if (dataSet != null) {
+ readerPool.closeResultSet(dataSet);
+ }
+ long endTime = System.currentTimeMillis();
+ long totalTime = endTime - startTime;
+ LOGGER.info("migrate device ：{} using {} ms", device, totalTime);
+ }
+
+ LOGGER.info("Loading the {}-th device: {} success", i, device);
+ return null;
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/HybridTimeseriesSessionExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/HybridTimeseriesSessionExample.java
new file mode 100644
index 00000000000..63f91072c39
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/HybridTimeseriesSessionExample.java
@@ -0,0 +1,131 @@
+/*
+ * 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 org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This example shows how to insert and select Hybrid Timeseries by session Hybrid Timeseries
+ * includes Aligned Timeseries and Normal Timeseries
+ */
+public class HybridTimeseriesSessionExample {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(HybridTimeseriesSessionExample.class);
+
+ private static Session session;
+ private static final String ROOT_SG1_ALIGNEDDEVICE = "root.sg_1.aligned_device";
+ private static final String ROOT_SG1_D1 = "root.sg_1.d1";
+ private static final String ROOT_SG1_D2 = "root.sg_1.d2";
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException, StatementExecutionException {
+ session = new Session("127.0.0.1", 6667, "root", "root");
+ session.open(false);
+
+ // set session fetchSize
+ session.setFetchSize(10000);
+
+ insertRecord(ROOT_SG1_D2, 0, 100);
+ insertTabletWithAlignedTimeseriesMethod(0, 100);
+ insertRecord(ROOT_SG1_D1, 0, 100);
+ session.executeNonQueryStatement("flush");
+ selectTest();
+
+ session.close();
+ }
+
+ private static void selectTest() throws StatementExecutionException, IoTDBConnectionException {
+ SessionDataSet dataSet = session.executeQueryStatement("select ** from root.sg_1");
+ LOGGER.info("columnNames = {}", dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ LOGGER.info("data = {}", dataSet.next());
+ }
+
+ dataSet.closeOperationHandle();
+ }
+
+ /** Method 1 for insert tablet with aligned timeseries */
+ private static void insertTabletWithAlignedTimeseriesMethod(int minTime, int maxTime)
+ throws IoTDBConnectionException, StatementExecutionException {
+ // The schema of measurements of one device
+ // only measurementId and data type in MeasurementSchema take effects in Tablet
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT32));
+
+ Tablet tablet = new Tablet(ROOT_SG1_ALIGNEDDEVICE, schemaList);
+ long timestamp = minTime;
+
+ for (long row = minTime; row < maxTime; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp);
+ tablet.addValue(schemaList.get(0).getMeasurementId(), rowIndex, row * 10 + 1L);
+ tablet.addValue(schemaList.get(1).getMeasurementId(), rowIndex, (int) (row * 10 + 2));
+
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertAlignedTablet(tablet, true);
+ tablet.reset();
+ }
+ timestamp++;
+ }
+
+ if (tablet.rowSize != 0) {
+ session.insertAlignedTablet(tablet);
+ tablet.reset();
+ }
+ }
+
+ private static void insertRecord(String deviceId, int minTime, int maxTime)
+ throws IoTDBConnectionException, StatementExecutionException {
+ List<String> measurements = new ArrayList<>();
+ List<TSDataType> types = new ArrayList<>();
+ measurements.add("s2");
+ measurements.add("s4");
+ measurements.add("s5");
+ measurements.add("s6");
+ types.add(TSDataType.INT64);
+ types.add(TSDataType.INT64);
+ types.add(TSDataType.INT64);
+ types.add(TSDataType.INT64);
+
+ for (long time = minTime; time < maxTime; time++) {
+ List<Object> values = new ArrayList<>();
+ values.add(time * 10 + 3L);
+ values.add(time * 10 + 4L);
+ values.add(time * 10 + 5L);
+ values.add(time * 10 + 6L);
+ session.insertRecord(deviceId, time, measurements, types, values);
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionConcurrentExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionConcurrentExample.java
new file mode 100644
index 00000000000..68e2696724d
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionConcurrentExample.java
@@ -0,0 +1,197 @@
+/*
+ * 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 org.apache.iotdb.isession.template.Template;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.template.MeasurementNode;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class SessionConcurrentExample {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SessionConcurrentExample.class);
+
+ private static final int SG_NUM = 20;
+ private static final int DEVICE_NUM = 100;
+ private static final int PARALLEL_DEGREE_FOR_ONE_SG = 3;
+ private static Random random = new Random();
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException, StatementExecutionException, IOException {
+
+ try (Session session = new Session("127.0.0.1", 6667, "root", "root")) {
+ session.open(false);
+ createTemplate(session);
+ } catch (Exception e) {
+ LOGGER.error("create template with Session error", e);
+ }
+
+ CountDownLatch latch = new CountDownLatch(SG_NUM * PARALLEL_DEGREE_FOR_ONE_SG);
+ ExecutorService es = Executors.newFixedThreadPool(SG_NUM * PARALLEL_DEGREE_FOR_ONE_SG);
+
+ for (int i = 0; i < SG_NUM * PARALLEL_DEGREE_FOR_ONE_SG; i++) {
+ int currentIndex = i;
+ es.execute(() -> concurrentOperation(latch, currentIndex));
+ }
+
+ es.shutdown();
+
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ LOGGER.warn("CountDownLatch interrupted", e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private static void concurrentOperation(CountDownLatch latch, int currentIndex) {
+
+ Session session = new Session("127.0.0.1", 6667, "root", "root");
+ try {
+ session.open(false);
+ } catch (IoTDBConnectionException e) {
+ LOGGER.error("Open Session error", e);
+ }
+
+ for (int j = 0; j < DEVICE_NUM; j++) {
+ try {
+ insertTablet(
+ session,
+ String.format("root.sg_%d.d_%d", currentIndex / PARALLEL_DEGREE_FOR_ONE_SG, j));
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ LOGGER.error("Insert tablet error", e);
+ }
+ }
+
+ try {
+ session.close();
+ } catch (IoTDBConnectionException e) {
+ LOGGER.error("Close session error", e);
+ }
+
+ latch.countDown();
+ }
+
+ private static void createTemplate(Session session)
+ throws IoTDBConnectionException, StatementExecutionException, IOException {
+
+ Template template = new Template("template1", false);
+ MeasurementNode mNodeS1 =
+ new MeasurementNode("s1", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
+ MeasurementNode mNodeS2 =
+ new MeasurementNode("s2", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
+ MeasurementNode mNodeS3 =
+ new MeasurementNode("s3", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
+
+ template.addToTemplate(mNodeS1);
+ template.addToTemplate(mNodeS2);
+ template.addToTemplate(mNodeS3);
+
+ session.createSchemaTemplate(template);
+ for (int i = 0; i < SG_NUM; i++) {
+ session.setSchemaTemplate("template1", "root.sg_" + i);
+ }
+ }
+
+ /**
+ * insert the data of a device. For each timestamp, the number of measurements is the same.
+ *
+ * <p>Users need to control the count of Tablet and write a batch when it reaches the maxBatchSize
+ */
+ private static void insertTablet(Session session, String deviceId)
+ throws IoTDBConnectionException, StatementExecutionException {
+ /*
+ * A Tablet example:
+ * device1
+ * time s1, s2, s3
+ * 1, 1, 1, 1
+ * 2, 2, 2, 2
+ * 3, 3, 3, 3
+ */
+ // The schema of measurements of one device
+ // only measurementId and data type in MeasurementSchema take effects in Tablet
+ List<IMeasurementSchema> 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 tablet = new Tablet(deviceId, schemaList, 100);
+
+ // Method 1 to add tablet data
+ long timestamp = System.currentTimeMillis();
+ for (long row = 0; row < 100; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp);
+ for (int s = 0; s < 3; s++) {
+ long value = random.nextLong();
+ tablet.addValue(schemaList.get(s).getMeasurementId(), rowIndex, value);
+ }
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet, true);
+ tablet.reset();
+ }
+ timestamp++;
+ }
+
+ if (tablet.rowSize != 0) {
+ 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();
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionExample.java
new file mode 100644
index 00000000000..47254abfae6
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionExample.java
@@ -0,0 +1,914 @@
+/*
+ * 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 org.apache.iotdb.common.rpc.thrift.TAggregationType;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.SessionDataSet.DataIterator;
+import org.apache.iotdb.isession.template.Template;
+import org.apache.iotdb.isession.util.Version;
+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.template.MeasurementNode;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+@SuppressWarnings({"squid:S106", "squid:S1144", "squid:S125"})
+public class SessionExample {
+
+ private static Session session;
+ private static Session sessionEnableRedirect;
+ 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";
+ private static final String ROOT_SG1 = "root.sg1";
+ private static final String LOCAL_HOST = "127.0.0.1";
+ public static final String SELECT_D1 = "select * from root.sg1.d1";
+
+ private static Random random = new Random();
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException, StatementExecutionException {
+ session =
+ new Session.Builder()
+ .host(LOCAL_HOST)
+ .port(6667)
+ .username("root")
+ .password("root")
+ .version(Version.V_1_0)
+ .build();
+ session.open(false);
+
+ // set session fetchSize
+ session.setFetchSize(10000);
+
+ try {
+ session.createDatabase("root.sg1");
+ } catch (StatementExecutionException e) {
+ if (e.getStatusCode() != TSStatusCode.DATABASE_ALREADY_EXISTS.getStatusCode()) {
+ throw e;
+ }
+ }
+
+ // createTemplate();
+ createTimeseries();
+ createMultiTimeseries();
+ insertRecord();
+ insertTablet();
+ // insertTabletWithNullValues();
+ // insertTablets();
+ // insertRecords();
+ // insertText();
+ // selectInto();
+ // createAndDropContinuousQueries();
+ // nonQuery();
+ query();
+ // queryWithTimeout();
+ rawDataQuery();
+ lastDataQuery();
+ aggregationQuery();
+ groupByQuery();
+ // queryByIterator();
+ // deleteData();
+ // deleteTimeseries();
+ // setTimeout();
+
+ sessionEnableRedirect = new Session(LOCAL_HOST, 6667, "root", "root");
+ sessionEnableRedirect.setEnableQueryRedirection(true);
+ sessionEnableRedirect.open(false);
+
+ // set session fetchSize
+ sessionEnableRedirect.setFetchSize(10000);
+
+ fastLastDataQueryForOneDevice();
+ insertRecord4Redirect();
+ query4Redirect();
+ sessionEnableRedirect.close();
+ session.close();
+ }
+
+ private static void createAndDropContinuousQueries()
+ throws StatementExecutionException, IoTDBConnectionException {
+ session.executeNonQueryStatement(
+ "CREATE CONTINUOUS QUERY cq1 "
+ + "BEGIN SELECT max_value(s1) INTO temperature_max FROM root.sg1.* "
+ + "GROUP BY time(10s) END");
+ session.executeNonQueryStatement(
+ "CREATE CONTINUOUS QUERY cq2 "
+ + "BEGIN SELECT count(s2) INTO temperature_cnt FROM root.sg1.* "
+ + "GROUP BY time(10s), level=1 END");
+ session.executeNonQueryStatement(
+ "CREATE CONTINUOUS QUERY cq3 "
+ + "RESAMPLE EVERY 20s FOR 20s "
+ + "BEGIN SELECT avg(s3) INTO temperature_avg FROM root.sg1.* "
+ + "GROUP BY time(10s), level=1 END");
+ session.executeNonQueryStatement("DROP CONTINUOUS QUERY cq1");
+ session.executeNonQueryStatement("DROP CONTINUOUS QUERY cq2");
+ session.executeNonQueryStatement("DROP CONTINUOUS QUERY cq3");
+ }
+
+ 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<>();
+ attributes.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, 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 createTemplate()
+ throws IoTDBConnectionException, StatementExecutionException, IOException {
+
+ Template template = new Template("template1", false);
+ MeasurementNode mNodeS1 =
+ new MeasurementNode("s1", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
+ MeasurementNode mNodeS2 =
+ new MeasurementNode("s2", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
+ MeasurementNode mNodeS3 =
+ new MeasurementNode("s3", TSDataType.INT64, TSEncoding.RLE, CompressionType.SNAPPY);
+
+ template.addToTemplate(mNodeS1);
+ template.addToTemplate(mNodeS2);
+ template.addToTemplate(mNodeS3);
+
+ session.createSchemaTemplate(template);
+ session.setSchemaTemplate("template1", "root.sg1");
+ }
+
+ 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 insertRecord4Redirect()
+ throws IoTDBConnectionException, StatementExecutionException {
+ for (int i = 0; i < 6; i++) {
+ for (int j = 0; j < 2; j++) {
+ String deviceId = "root.redirect" + i + ".d" + j;
+ List<String> measurements = new ArrayList<>();
+ measurements.add("s1");
+ measurements.add("s2");
+ measurements.add("s3");
+ List<TSDataType> types = new ArrayList<>();
+ types.add(TSDataType.INT64);
+ types.add(TSDataType.INT64);
+ types.add(TSDataType.INT64);
+
+ for (long time = 0; time < 5; time++) {
+ List<Object> values = new ArrayList<>();
+ values.add(1L + time);
+ values.add(2L + time);
+ values.add(3L + time);
+ 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();
+ typesList.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.
+ *
+ * <p>Users need to control the count of Tablet and write a batch when it reaches the maxBatchSize
+ */
+ private static void insertTablet() throws IoTDBConnectionException, StatementExecutionException {
+ /*
+ * A Tablet example:
+ * device1
+ * time s1, s2, s3
+ * 1, 1, 1, 1
+ * 2, 2, 2, 2
+ * 3, 3, 3, 3
+ */
+ // The schema of measurements of one device
+ // only measurementId and data type in MeasurementSchema take effects in Tablet
+ List<IMeasurementSchema> 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 tablet = new Tablet(ROOT_SG1_D1, schemaList, 100);
+
+ // Method 1 to add tablet data
+ long timestamp = System.currentTimeMillis();
+
+ for (long row = 0; row < 100; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp);
+ for (int s = 0; s < 3; s++) {
+ long value = random.nextLong();
+ tablet.addValue(schemaList.get(s).getMeasurementId(), rowIndex, value);
+ }
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet, true);
+ tablet.reset();
+ }
+ timestamp++;
+ }
+
+ if (tablet.rowSize != 0) {
+ 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 insertTabletWithNullValues()
+ throws IoTDBConnectionException, StatementExecutionException {
+ /*
+ * A Tablet example:
+ * device1
+ * time s1, s2, s3
+ * 1, null, 1, 1
+ * 2, 2, null, 2
+ * 3, 3, 3, null
+ */
+ // The schema of measurements of one device
+ // only measurementId and data type in MeasurementSchema take effects in Tablet
+ List<IMeasurementSchema> 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 tablet = new Tablet(ROOT_SG1_D1, schemaList, 100);
+
+ // Method 1 to add tablet data
+ insertTablet1(schemaList, tablet);
+
+ // Method 2 to add tablet data
+ insertTablet2(schemaList, tablet);
+ }
+
+ private static void insertTablet1(List<IMeasurementSchema> schemaList, Tablet tablet)
+ throws IoTDBConnectionException, StatementExecutionException {
+ tablet.initBitMaps();
+
+ long timestamp = System.currentTimeMillis();
+ for (long row = 0; row < 100; row++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, timestamp);
+ for (int s = 0; s < 3; s++) {
+ long value = random.nextLong();
+ // mark null value
+ if (row % 3 == s) {
+ tablet.bitMaps[s].mark((int) row);
+ }
+ tablet.addValue(schemaList.get(s).getMeasurementId(), rowIndex, value);
+ }
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet, true);
+ tablet.reset();
+ }
+ timestamp++;
+ }
+
+ if (tablet.rowSize != 0) {
+ session.insertTablet(tablet);
+ tablet.reset();
+ }
+ }
+
+ private static void insertTablet2(List<IMeasurementSchema> schemaList, Tablet tablet)
+ throws IoTDBConnectionException, StatementExecutionException {
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
+ BitMap[] bitMaps = new BitMap[schemaList.size()];
+ for (int s = 0; s < 3; s++) {
+ bitMaps[s] = new BitMap(tablet.getMaxRowNumber());
+ }
+ tablet.bitMaps = bitMaps;
+
+ 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];
+ // mark null value
+ if (row % 3 == i) {
+ bitMaps[i].mark(row);
+ }
+ 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<IMeasurementSchema> 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 = 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();
+ }
+ }
+
+ /**
+ * This example shows how to insert data of TSDataType.TEXT. You can use the session interface to
+ * write data of String type or Binary type.
+ */
+ private static void insertText() throws IoTDBConnectionException, StatementExecutionException {
+ String device = "root.sg1.text";
+ // the first data is String type and the second data is Binary type
+ List<Object> datas = Arrays.asList("String", new Binary("Binary", TSFileConfig.STRING_CHARSET));
+ // insertRecord example
+ for (int i = 0; i < datas.size(); i++) {
+ // write data of String type or Binary type
+ session.insertRecord(
+ device,
+ i,
+ Collections.singletonList("s1"),
+ Collections.singletonList(TSDataType.TEXT),
+ datas.get(i));
+ }
+
+ // insertTablet example
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s2", TSDataType.TEXT));
+ Tablet tablet = new Tablet(device, schemaList, 100);
+ for (int i = 0; i < datas.size(); i++) {
+ int rowIndex = tablet.rowSize++;
+ tablet.addTimestamp(rowIndex, i);
+ // write data of String type or Binary type
+ tablet.addValue(schemaList.get(0).getMeasurementId(), rowIndex, datas.get(i));
+ }
+ session.insertTablet(tablet);
+ try (SessionDataSet dataSet = session.executeQueryStatement("select s1, s2 from " + device)) {
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+
+ private static void selectInto() throws IoTDBConnectionException, StatementExecutionException {
+ session.executeNonQueryStatement(
+ "select s1, s2, s3 into into_s1, into_s2, into_s3 from root.sg1.d1");
+
+ try (SessionDataSet dataSet =
+ session.executeQueryStatement("select into_s1, into_s2, into_s3 from root.sg1.d1")) {
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+
+ 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 {
+ try (SessionDataSet dataSet = session.executeQueryStatement(SELECT_D1)) {
+ System.out.println(dataSet.getColumnNames());
+ dataSet.setFetchSize(1024); // default is 10000
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+
+ private static void query4Redirect()
+ throws IoTDBConnectionException, StatementExecutionException {
+ String selectPrefix = "select * from root.redirect";
+ for (int i = 0; i < 6; i++) {
+ try (SessionDataSet dataSet =
+ sessionEnableRedirect.executeQueryStatement(selectPrefix + i + ".d1")) {
+
+ System.out.println(dataSet.getColumnNames());
+ dataSet.setFetchSize(1024); // default is 10000
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+
+ for (int i = 0; i < 6; i++) {
+ try (SessionDataSet dataSet =
+ sessionEnableRedirect.executeQueryStatement(
+ selectPrefix + i + ".d1 where time >= 1 and time < 10")) {
+
+ System.out.println(dataSet.getColumnNames());
+ dataSet.setFetchSize(1024); // default is 10000
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+
+ for (int i = 0; i < 6; i++) {
+ try (SessionDataSet dataSet =
+ sessionEnableRedirect.executeQueryStatement(
+ selectPrefix + i + ".d1 where time >= 1 and time < 10 align by device")) {
+
+ System.out.println(dataSet.getColumnNames());
+ dataSet.setFetchSize(1024); // default is 10000
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+
+ for (int i = 0; i < 6; i++) {
+ try (SessionDataSet dataSet =
+ sessionEnableRedirect.executeQueryStatement(
+ selectPrefix
+ + i
+ + ".d1 where time >= 1 and time < 10 and root.redirect"
+ + i
+ + ".d1.s1 > 1")) {
+ System.out.println(dataSet.getColumnNames());
+ dataSet.setFetchSize(1024); // default is 10000
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+ }
+
+ private static void queryWithTimeout()
+ throws IoTDBConnectionException, StatementExecutionException {
+ try (SessionDataSet dataSet = session.executeQueryStatement(SELECT_D1, 2000)) {
+ System.out.println(dataSet.getColumnNames());
+ dataSet.setFetchSize(1024); // default is 10000
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+
+ 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;
+ long timeOut = 60000;
+
+ try (SessionDataSet dataSet = session.executeRawDataQuery(paths, startTime, endTime, timeOut)) {
+
+ System.out.println(dataSet.getColumnNames());
+ dataSet.setFetchSize(1024);
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+
+ private static void lastDataQuery() 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);
+ try (SessionDataSet sessionDataSet = session.executeLastDataQuery(paths, 3, 60000)) {
+ System.out.println(sessionDataSet.getColumnNames());
+ sessionDataSet.setFetchSize(1024);
+ while (sessionDataSet.hasNext()) {
+ System.out.println(sessionDataSet.next());
+ }
+ }
+ }
+
+ private static void fastLastDataQueryForOneDevice()
+ throws IoTDBConnectionException, StatementExecutionException {
+ System.out.println("-------fastLastQuery------");
+ List<String> paths = new ArrayList<>();
+ paths.add("s1");
+ paths.add("s2");
+ paths.add("s3");
+ try (SessionDataSet sessionDataSet =
+ sessionEnableRedirect.executeLastDataQueryForOneDevice(
+ ROOT_SG1, ROOT_SG1_D1, paths, true)) {
+ System.out.println(sessionDataSet.getColumnNames());
+ sessionDataSet.setFetchSize(1024);
+ while (sessionDataSet.hasNext()) {
+ System.out.println(sessionDataSet.next());
+ }
+ }
+ }
+
+ private static void aggregationQuery()
+ 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);
+
+ List<TAggregationType> aggregations = new ArrayList<>();
+ aggregations.add(TAggregationType.COUNT);
+ aggregations.add(TAggregationType.SUM);
+ aggregations.add(TAggregationType.MAX_VALUE);
+ try (SessionDataSet sessionDataSet = session.executeAggregationQuery(paths, aggregations)) {
+ System.out.println(sessionDataSet.getColumnNames());
+ sessionDataSet.setFetchSize(1024);
+ while (sessionDataSet.hasNext()) {
+ System.out.println(sessionDataSet.next());
+ }
+ }
+ }
+
+ private static void groupByQuery() 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);
+
+ List<TAggregationType> aggregations = new ArrayList<>();
+ aggregations.add(TAggregationType.COUNT);
+ aggregations.add(TAggregationType.SUM);
+ aggregations.add(TAggregationType.MAX_VALUE);
+ try (SessionDataSet sessionDataSet =
+ session.executeAggregationQuery(paths, aggregations, 0, 100, 10, 20)) {
+ System.out.println(sessionDataSet.getColumnNames());
+ sessionDataSet.setFetchSize(1024);
+ while (sessionDataSet.hasNext()) {
+ System.out.println(sessionDataSet.next());
+ }
+ }
+ }
+
+ private static void queryByIterator()
+ throws IoTDBConnectionException, StatementExecutionException {
+ try (SessionDataSet dataSet = session.executeQueryStatement(SELECT_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);
+ }
+ }
+ }
+
+ private static void nonQuery() throws IoTDBConnectionException, StatementExecutionException {
+ session.executeNonQueryStatement("insert into root.sg1.d1(timestamp,s1) values(200, 1)");
+ }
+
+ private static void setTimeout() throws IoTDBConnectionException {
+ try (Session tempSession = new Session(LOCAL_HOST, 6667, "root", "root", 10000, 20000)) {
+ tempSession.setQueryTimeout(60000);
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionPoolExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionPoolExample.java
new file mode 100644
index 00000000000..578a82d4b84
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SessionPoolExample.java
@@ -0,0 +1,154 @@
+/*
+ * 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 org.apache.iotdb.isession.SessionDataSet.DataIterator;
+import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.pool.SessionPool;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@SuppressWarnings({"squid:S106", "squid:S1144"})
+public class SessionPoolExample {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SessionPoolExample.class);
+
+ private static SessionPool sessionPool;
+ private static ExecutorService service;
+
+ /** Build a custom SessionPool for this example */
+ private static void constructCustomSessionPool() {
+ sessionPool =
+ new SessionPool.Builder()
+ .host("127.0.0.1")
+ .port(6667)
+ .user("root")
+ .password("root")
+ .maxSize(3)
+ .build();
+ }
+
+ /** Build a redirect-able SessionPool for this example */
+ private static void constructRedirectSessionPool() {
+ List<String> nodeUrls = new ArrayList<>();
+ nodeUrls.add("127.0.0.1:6667");
+ nodeUrls.add("127.0.0.1:6668");
+ sessionPool =
+ new SessionPool.Builder()
+ .nodeUrls(nodeUrls)
+ .user("root")
+ .password("root")
+ .maxSize(3)
+ .build();
+ }
+
+ public static void main(String[] args)
+ throws StatementExecutionException, IoTDBConnectionException, InterruptedException {
+ // Choose the SessionPool you going to use
+ constructRedirectSessionPool();
+
+ service = Executors.newFixedThreadPool(10);
+ insertRecord();
+ queryByRowRecord();
+ Thread.sleep(1000);
+ queryByIterator();
+ sessionPool.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);
+ sessionPool.insertRecord(deviceId, time, measurements, types, values);
+ }
+ }
+
+ private static void queryByRowRecord() {
+ for (int i = 0; i < 1; i++) {
+ service.submit(
+ () -> {
+ SessionDataSetWrapper wrapper = null;
+ try {
+ wrapper = sessionPool.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) {
+ LOGGER.error("Query by row record error", e);
+ } finally {
+ // remember to close data set finally!
+ sessionPool.closeResultSet(wrapper);
+ }
+ });
+ }
+ }
+
+ private static void queryByIterator() {
+ for (int i = 0; i < 1; i++) {
+ service.submit(
+ () -> {
+ SessionDataSetWrapper wrapper = null;
+ try {
+ wrapper = sessionPool.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);
+ }
+ } catch (IoTDBConnectionException | StatementExecutionException e) {
+ LOGGER.error("Query by Iterator error", e);
+ } finally {
+ // remember to close data set finally!
+ sessionPool.closeResultSet(wrapper);
+ }
+ });
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java
new file mode 100644
index 00000000000..1ec74a34a29
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SubscriptionSessionExample.java
@@ -0,0 +1,345 @@
+/*
+ * 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 org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.util.Version;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.subscription.SubscriptionSession;
+import org.apache.iotdb.session.subscription.consumer.AckStrategy;
+import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
+import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer;
+import org.apache.iotdb.session.subscription.consumer.SubscriptionPushConsumer;
+import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
+import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
+import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
+
+import org.apache.tsfile.read.TsFileReader;
+import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.expression.QueryExpression;
+import org.apache.tsfile.read.query.dataset.QueryDataSet;
+
+import java.io.IOException;
+import java.net.URLEncoder;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.LockSupport;
+
+public class SubscriptionSessionExample {
+
+ private static Session session;
+
+ private static final String HOST = "127.0.0.1";
+ private static final int PORT = 6667;
+
+ private static final String TOPIC_1 = "topic1";
+ private static final String TOPIC_2 = "`'topic2'`";
+ private static final String TOPIC_3 = "`\"topic3\"`";
+ private static final String TOPIC_4 = "`\"top \\.i.c4\"`";
+
+ private static final long SLEEP_NS = 1_000_000_000L;
+ private static final long POLL_TIMEOUT_MS = 10_000L;
+ private static final int MAX_RETRY_TIMES = 3;
+ private static final int PARALLELISM = 8;
+ private static final long CURRENT_TIME = System.currentTimeMillis();
+
+ private static void prepareData() throws Exception {
+ // Open session
+ session =
+ new Session.Builder()
+ .host(HOST)
+ .port(PORT)
+ .username("root")
+ .password("root")
+ .version(Version.V_1_0)
+ .build();
+ session.open(false);
+
+ // Insert some historical data
+ for (int i = 0; i < 100; ++i) {
+ session.executeNonQueryStatement(
+ String.format("insert into root.db.d1(time, s1, s2) values (%s, 1, 2)", i));
+ session.executeNonQueryStatement(
+ String.format(
+ "insert into root.db.d2(time, s1, s2) values (%s, 3, 4)", CURRENT_TIME + i));
+ session.executeNonQueryStatement(
+ String.format("insert into root.sg.d3(time, s1) values (%s, 5)", CURRENT_TIME + 2 * i));
+ }
+ session.executeNonQueryStatement("flush");
+
+ // Close session
+ session.close();
+ session = null;
+ }
+
+ private static void dataQuery() throws Exception {
+ // Open session
+ session =
+ new Session.Builder()
+ .host(HOST)
+ .port(PORT)
+ .username("root")
+ .password("root")
+ .version(Version.V_1_0)
+ .build();
+ session.open(false);
+
+ // Query
+ final SessionDataSet dataSet = session.executeQueryStatement("select ** from root.**");
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+
+ // Close session
+ session.close();
+ session = null;
+ }
+
+ /** single pull consumer subscribe topic with path and time range */
+ private static void dataSubscription1() throws Exception {
+ // Create topics
+ try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
+ subscriptionSession.open();
+ final Properties config = new Properties();
+ config.put(TopicConstant.PATH_KEY, "root.db.d1.s1");
+ config.put(TopicConstant.START_TIME_KEY, 25);
+ config.put(TopicConstant.END_TIME_KEY, 75);
+ subscriptionSession.createTopic(TOPIC_1, config);
+ }
+
+ int retryCount = 0;
+ // Subscription: property-style ctor
+ final Properties config = new Properties();
+ config.put(ConsumerConstant.CONSUMER_ID_KEY, "c1");
+ config.put(ConsumerConstant.CONSUMER_GROUP_ID_KEY, "cg1");
+ try (SubscriptionPullConsumer consumer1 = new SubscriptionPullConsumer(config)) {
+ consumer1.open();
+ consumer1.subscribe(TOPIC_1);
+ while (true) {
+ final List<SubscriptionMessage> messages = consumer1.poll(POLL_TIMEOUT_MS);
+ if (messages.isEmpty()) {
+ retryCount++;
+ if (retryCount >= MAX_RETRY_TIMES) {
+ break;
+ }
+ }
+ for (final SubscriptionMessage message : messages) {
+ for (final SubscriptionSessionDataSet dataSet : message.getSessionDataSetsHandler()) {
+ System.out.println(dataSet.getColumnNames());
+ System.out.println(dataSet.getColumnTypes());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+ // Auto commit
+ }
+
+ // Show topics and subscriptions
+ try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
+ subscriptionSession.open();
+ subscriptionSession.getTopics().forEach((System.out::println));
+ subscriptionSession.getSubscriptions().forEach((System.out::println));
+ }
+
+ consumer1.unsubscribe(TOPIC_1);
+ }
+ }
+
+ /** multi pull consumer subscribe topic with tsfile format */
+ private static void dataSubscription2() throws Exception {
+ try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
+ subscriptionSession.open();
+ final Properties config = new Properties();
+ config.put(TopicConstant.START_TIME_KEY, CURRENT_TIME + 33);
+ config.put(TopicConstant.END_TIME_KEY, CURRENT_TIME + 66);
+ config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
+ subscriptionSession.createTopic(TOPIC_2, config);
+ }
+
+ final List<Thread> threads = new ArrayList<>();
+ for (int i = 0; i < PARALLELISM; ++i) {
+ final int idx = i;
+ final Thread thread =
+ new Thread(
+ () -> {
+ int retryCount = 0;
+ // Subscription: builder-style ctor
+ try (final SubscriptionPullConsumer consumer2 =
+ new SubscriptionPullConsumer.Builder()
+ .consumerId("c" + idx)
+ .consumerGroupId("cg2")
+ .autoCommit(false)
+ .buildPullConsumer()) {
+ consumer2.open();
+ consumer2.subscribe(TOPIC_2);
+ while (true) {
+ final List<SubscriptionMessage> messages =
+ consumer2.poll(Collections.singleton(TOPIC_2), POLL_TIMEOUT_MS);
+ if (messages.isEmpty()) {
+ retryCount++;
+ if (retryCount >= MAX_RETRY_TIMES) {
+ break;
+ }
+ }
+ for (final SubscriptionMessage message : messages) {
+ try (final TsFileReader reader = message.getTsFileHandler().openReader()) {
+ final QueryDataSet dataSet =
+ reader.query(
+ QueryExpression.create(
+ Arrays.asList(
+ new Path("root.db.d2", "s2", true),
+ new Path("root.sg.d3", "s1", true)),
+ null));
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+ }
+ consumer2.commitSync(messages);
+ }
+ consumer2.unsubscribe(TOPIC_2);
+ } catch (final IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ thread.start();
+ threads.add(thread);
+ }
+
+ for (final Thread thread : threads) {
+ thread.join();
+ }
+ }
+
+ /** multi push consumer subscribe topic with tsfile format and snapshot mode */
+ private static void dataSubscription3() throws Exception {
+ try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
+ subscriptionSession.open();
+ final Properties config = new Properties();
+ config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
+ config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_SNAPSHOT_VALUE);
+ subscriptionSession.createTopic(TOPIC_3, config);
+ }
+
+ final List<Thread> threads = new ArrayList<>();
+ for (int i = 0; i < PARALLELISM; ++i) {
+ final int idx = i;
+ final Thread thread =
+ new Thread(
+ () -> {
+ // Subscription: builder-style ctor
+ try (final SubscriptionPushConsumer consumer3 =
+ new SubscriptionPushConsumer.Builder()
+ .consumerId("c" + idx)
+ .consumerGroupId("cg3")
+ .ackStrategy(AckStrategy.AFTER_CONSUME)
+ .consumeListener(
+ message -> {
+ // do something for SubscriptionTsFileHandler
+ System.out.println(
+ message.getTsFileHandler().getFile().getAbsolutePath());
+ return ConsumeResult.SUCCESS;
+ })
+ .buildPushConsumer()) {
+ consumer3.open();
+ consumer3.subscribe(TOPIC_3);
+ while (!consumer3.allSnapshotTopicMessagesHaveBeenConsumed()) {
+ LockSupport.parkNanos(SLEEP_NS); // wait some time
+ }
+ }
+ });
+ thread.start();
+ threads.add(thread);
+ }
+
+ for (final Thread thread : threads) {
+ thread.join();
+ }
+ }
+
+ /** multi pull consumer subscribe topic with tsfile format and snapshot mode */
+ private static void dataSubscription4() throws Exception {
+ try (final SubscriptionSession subscriptionSession = new SubscriptionSession(HOST, PORT)) {
+ subscriptionSession.open();
+ final Properties config = new Properties();
+ config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_HANDLER_VALUE);
+ config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_SNAPSHOT_VALUE);
+ subscriptionSession.createTopic(TOPIC_4, config);
+ }
+
+ final AtomicLong counter = new AtomicLong();
+ final List<Thread> threads = new ArrayList<>();
+ for (int i = 0; i < PARALLELISM; ++i) {
+ final int idx = i;
+ final Thread thread =
+ new Thread(
+ () -> {
+ // Subscription: builder-style ctor
+ try (final SubscriptionPullConsumer consumer4 =
+ new SubscriptionPullConsumer.Builder()
+ .consumerId("c" + idx)
+ .consumerGroupId("cg4")
+ .autoCommit(true)
+ .fileSaveFsync(true)
+ .buildPullConsumer()) {
+ consumer4.open();
+ consumer4.subscribe(TOPIC_4);
+ while (!consumer4.allSnapshotTopicMessagesHaveBeenConsumed()) {
+ for (final SubscriptionMessage message : consumer4.poll(POLL_TIMEOUT_MS)) {
+ final SubscriptionTsFileHandler handler = message.getTsFileHandler();
+ handler.moveFile(
+ Paths.get(System.getProperty("user.dir"), "exported-tsfiles")
+ .resolve(
+ URLEncoder.encode(TOPIC_4)
+ + "-"
+ + counter.getAndIncrement()
+ + ".tsfile"));
+ }
+ }
+ } catch (final IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ thread.start();
+ threads.add(thread);
+ }
+
+ for (final Thread thread : threads) {
+ thread.join();
+ }
+ }
+
+ public static void main(final String[] args) throws Exception {
+ prepareData();
+ // dataQuery();
+ // dataSubscription1();
+ // dataSubscription2();
+ // dataSubscription3();
+ dataSubscription4();
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SyntaxConventionRelatedExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SyntaxConventionRelatedExample.java
new file mode 100644
index 00000000000..ada3646fcce
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/SyntaxConventionRelatedExample.java
@@ -0,0 +1,156 @@
+/*
+ * 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 org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.util.Version;
+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.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * When using session API, measurement, device, database and path are represented by String. The
+ * content of the String should be the same as what you would write in a SQL statement. This class
+ * is an example to help you understand better.
+ */
+public class SyntaxConventionRelatedExample {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(SyntaxConventionRelatedExample.class);
+
+ private static Session session;
+ private static final String LOCAL_HOST = "127.0.0.1";
+
+ /**
+ * if you want to create a time series named root.sg1.select, a possible SQL statement would be
+ * like: create timeseries root.sg1.select with datatype=FLOAT, encoding=RLE As described before,
+ * when using session API, path is represented using String. The path should be written as
+ * "root.sg1.select".
+ */
+ private static final String ROOT_SG1_KEYWORD_EXAMPLE = "root.sg1.select";
+
+ /**
+ * if you want to create a time series named root.sg1.111, a possible SQL statement would be like:
+ * create timeseries root.sg1.`111` with datatype=FLOAT, encoding=RLE The path should be written
+ * as "root.sg1.`111`".
+ */
+ private static final String ROOT_SG1_DIGITS_EXAMPLE = "root.sg1.`111`";
+
+ /**
+ * if you want to create a time series named root.sg1.`a"b'c``, a possible SQL statement would be
+ * like: create timeseries root.sg1.`a"b'c``` with datatype=FLOAT, encoding=RLE The path should be
+ * written as "root.sg1.`a"b`c```".
+ */
+ private static final String ROOT_SG1_SPECIAL_CHARACTER_EXAMPLE = "root.sg1.`a\"b'c```";
+
+ /**
+ * if you want to create a time series named root.sg1.a, a possible SQL statement would be like:
+ * create timeseries root.sg1.a with datatype=FLOAT, encoding=RLE The path should be written as
+ * "root.sg1.a".
+ */
+ private static final String ROOT_SG1_NORMAL_NODE_EXAMPLE = "root.sg1.a";
+
+ public static void main(String[] args)
+ throws IoTDBConnectionException, StatementExecutionException {
+ session =
+ new Session.Builder()
+ .host(LOCAL_HOST)
+ .port(6667)
+ .username("root")
+ .password("root")
+ .version(Version.V_1_0)
+ .build();
+ session.open(false);
+
+ // set session fetchSize
+ session.setFetchSize(10000);
+
+ try {
+ session.setStorageGroup("root.sg1");
+ } catch (StatementExecutionException e) {
+ if (e.getStatusCode() != TSStatusCode.PATH_ALREADY_EXIST.getStatusCode()) {
+ throw e;
+ }
+ }
+
+ // createTimeSeries
+ createTimeSeries();
+ SessionDataSet dataSet = session.executeQueryStatement("show timeseries root.sg1.*");
+ // the expected paths would be:
+ // [root.sg1.select, root.sg1.`111`, root.sg1.`a"b'c```, root.sg1.a]
+ // You could see that time series in dataSet are exactly the same as
+ // the initial String you used as path. Node names consist of digits or contain special
+ // characters are quoted with ``, both in SQL statement and in header of result dataset.
+ // It's convenient that you can use the result of show timeseries as input parameter directly
+ // for other
+ // session APIs such as insertRecord or executeRawDataQuery.
+ List<String> paths = new ArrayList<>();
+ while (dataSet.hasNext()) {
+ paths.add(dataSet.next().getFields().get(0).toString());
+ }
+
+ long startTime = 1L;
+ long endTime = 100L;
+ long timeOut = 60000;
+
+ try (SessionDataSet dataSet1 =
+ session.executeRawDataQuery(paths, startTime, endTime, timeOut)) {
+
+ LOGGER.info("columnNames = {}", dataSet1.getColumnNames());
+ dataSet1.setFetchSize(1024);
+ while (dataSet1.hasNext()) {
+ LOGGER.info("data = {}", dataSet1.next());
+ }
+ }
+ }
+
+ private static void createTimeSeries()
+ throws IoTDBConnectionException, StatementExecutionException {
+ if (!session.checkTimeseriesExists(ROOT_SG1_KEYWORD_EXAMPLE)) {
+ session.createTimeseries(
+ ROOT_SG1_KEYWORD_EXAMPLE, TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY);
+ }
+ if (!session.checkTimeseriesExists(ROOT_SG1_DIGITS_EXAMPLE)) {
+ session.createTimeseries(
+ ROOT_SG1_DIGITS_EXAMPLE, TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY);
+ }
+ if (!session.checkTimeseriesExists(ROOT_SG1_SPECIAL_CHARACTER_EXAMPLE)) {
+ session.createTimeseries(
+ ROOT_SG1_SPECIAL_CHARACTER_EXAMPLE,
+ TSDataType.FLOAT,
+ TSEncoding.RLE,
+ CompressionType.SNAPPY);
+ }
+ if (!session.checkTimeseriesExists(ROOT_SG1_NORMAL_NODE_EXAMPLE)) {
+ session.createTimeseries(
+ ROOT_SG1_NORMAL_NODE_EXAMPLE, TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY);
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TableModelSessionExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TableModelSessionExample.java
new file mode 100644
index 00000000000..b57b5903e07
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TableModelSessionExample.java
@@ -0,0 +1,136 @@
+/*
+ * 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 org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.util.Version;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+
+public class TableModelSessionExample {
+
+ private static final String LOCAL_HOST = "127.0.0.1";
+
+ public static void main(String[] args) {
+
+ // don't specify database in constructor
+ Session session =
+ new Session.Builder()
+ .host(LOCAL_HOST)
+ .port(6667)
+ .username("root")
+ .password("root")
+ .version(Version.V_1_0)
+ .sqlDialect("table")
+ .build();
+
+ try {
+ session.open(false);
+
+ session.executeNonQueryStatement("CREATE DATABASE test1");
+ session.executeNonQueryStatement("CREATE DATABASE test2");
+
+ session.executeNonQueryStatement("use test2");
+
+ // or use full qualified table name
+ session.executeNonQueryStatement(
+ "create table test1.table1(region_id STRING ID, plant_id STRING ID, device_id STRING ID, model STRING ATTRIBUTE, temperature FLOAT MEASUREMENT, humidity DOUBLE MEASUREMENT) with (TTL=3600000)");
+
+ session.executeNonQueryStatement(
+ "create table table2(region_id STRING ID, plant_id STRING ID, color STRING ATTRIBUTE, temperature FLOAT MEASUREMENT, speed DOUBLE MEASUREMENT) with (TTL=6600000)");
+
+ // show tables from current database
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES")) {
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ // show tables by specifying another database
+ // using SHOW tables FROM
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES FROM test1")) {
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ } catch (StatementExecutionException e) {
+ e.printStackTrace();
+ } finally {
+ try {
+ session.close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ }
+
+ // specify database in constructor
+ session =
+ new Session.Builder()
+ .host(LOCAL_HOST)
+ .port(6667)
+ .username("root")
+ .password("root")
+ .version(Version.V_1_0)
+ .sqlDialect("table")
+ .database("test1")
+ .build();
+
+ try {
+ session.open(false);
+
+ // show tables from current database
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES")) {
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ // change database to test2
+ session.executeNonQueryStatement("use test2");
+
+ // show tables by specifying another database
+ // using SHOW tables FROM
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES")) {
+ System.out.println(dataSet.getColumnNames());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ } catch (StatementExecutionException e) {
+ e.printStackTrace();
+ } finally {
+ try {
+ session.close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TableModelSessionPoolExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TableModelSessionPoolExample.java
new file mode 100644
index 00000000000..78af9efdd6d
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TableModelSessionPoolExample.java
@@ -0,0 +1,147 @@
+/*
+ * 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 org.apache.iotdb.isession.IPooledSession;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.pool.SessionPool;
+
+public class TableModelSessionPoolExample {
+
+ private static final String LOCAL_HOST = "127.0.0.1";
+
+ public static void main(String[] args) {
+
+ // don't specify database in constructor
+ SessionPool sessionPool =
+ new SessionPool.Builder()
+ .host(LOCAL_HOST)
+ .port(6667)
+ .user("root")
+ .password("root")
+ .maxSize(1)
+ .sqlDialect("table")
+ .build();
+
+ try (IPooledSession session = sessionPool.getPooledSession()) {
+
+ session.executeNonQueryStatement("CREATE DATABASE test1");
+ session.executeNonQueryStatement("CREATE DATABASE test2");
+
+ session.executeNonQueryStatement("use test2");
+
+ // or use full qualified table name
+ session.executeNonQueryStatement(
+ "create table test1.table1(region_id STRING ID, plant_id STRING ID, device_id STRING ID, model STRING ATTRIBUTE, temperature FLOAT MEASUREMENT, humidity DOUBLE MEASUREMENT) with (TTL=3600000)");
+
+ session.executeNonQueryStatement(
+ "create table table2(region_id STRING ID, plant_id STRING ID, color STRING ATTRIBUTE, temperature FLOAT MEASUREMENT, speed DOUBLE MEASUREMENT) with (TTL=6600000)");
+
+ // show tables from current database
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES")) {
+ System.out.println(dataSet.getColumnNames());
+ System.out.println(dataSet.getColumnTypes());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ // show tables by specifying another database
+ // using SHOW tables FROM
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES FROM test1")) {
+ System.out.println(dataSet.getColumnNames());
+ System.out.println(dataSet.getColumnTypes());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ } catch (StatementExecutionException e) {
+ e.printStackTrace();
+ } finally {
+ sessionPool.close();
+ }
+
+ // specify database in constructor
+ sessionPool =
+ new SessionPool.Builder()
+ .host(LOCAL_HOST)
+ .port(6667)
+ .user("root")
+ .password("root")
+ .maxSize(1)
+ .sqlDialect("table")
+ .database("test1")
+ .build();
+
+ try (IPooledSession session = sessionPool.getPooledSession()) {
+
+ // show tables from current database
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES")) {
+ System.out.println(dataSet.getColumnNames());
+ System.out.println(dataSet.getColumnTypes());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ // change database to test2
+ session.executeNonQueryStatement("use test2");
+
+ // show tables by specifying another database
+ // using SHOW tables FROM
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES")) {
+ System.out.println(dataSet.getColumnNames());
+ System.out.println(dataSet.getColumnTypes());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ } catch (StatementExecutionException e) {
+ e.printStackTrace();
+ }
+
+ try (IPooledSession session = sessionPool.getPooledSession()) {
+
+ // show tables from default database test1
+ try (SessionDataSet dataSet = session.executeQueryStatement("SHOW TABLES")) {
+ System.out.println(dataSet.getColumnNames());
+ System.out.println(dataSet.getColumnTypes());
+ while (dataSet.hasNext()) {
+ System.out.println(dataSet.next());
+ }
+ }
+
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ } catch (StatementExecutionException e) {
+ e.printStackTrace();
+ } finally {
+ sessionPool.close();
+ }
+ }
+}
diff --git a/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TabletExample.java b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TabletExample.java
new file mode 100644
index 00000000000..39f1223efca
--- /dev/null
+++ b/example/influxdb-proto-example/src/main/java/org/apache/iotdb/TabletExample.java
@@ -0,0 +1,204 @@
+/*
+ * 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 org.apache.iotdb.session.Session;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.BytesUtils;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+@SuppressWarnings({"squid:S106"})
+public class TabletExample {
+
+ private static final String TIME_STR = "time";
+
+ /**
+ * load csv data.
+ *
+ * @param measureTSTypeInfos key: measurement name, value: measurement data type
+ * @param dataFileName the csv file name to load
+ * @return key: measurement name, value: series in format of {@link ArrayList}
+ * @throws IOException if the csv format is incorrect
+ */
+ private static Map<String, ArrayList<Object>> loadCSVData(
+ Map<String, TSDataType> measureTSTypeInfos, String dataFileName) throws IOException {
+ measureTSTypeInfos.put(TIME_STR, TSDataType.INT64);
+ try (BufferedReader reader = new BufferedReader(new FileReader(dataFileName))) {
+ String headline = reader.readLine();
+ if (headline == null) {
+ throw new IOException("Given csv data file has not headers");
+ }
+ // check the csv file format
+ String[] fileColumns = headline.split(",");
+ Map<String, Integer> columnToIdMap = new HashMap<>();
+ for (int col = 0; col < fileColumns.length; col++) {
+ String columnName = fileColumns[col];
+ if (columnToIdMap.containsKey(columnName)) {
+ throw new IOException(
+ String.format("csv file contains duplicate columns: %s", columnName));
+ }
+ columnToIdMap.put(columnName, col);
+ }
+ Map<String, ArrayList<Object>> ret = new HashMap<>();
+ // make sure that all measurements can be found from the data file
+ for (Entry<String, TSDataType> entry : measureTSTypeInfos.entrySet()) {
+ String measurement = entry.getKey();
+ if (!columnToIdMap.containsKey(entry.getKey())) {
+ throw new IOException(String.format("measurement %s's is not in csv file.", measurement));
+ } else {
+ ret.put(measurement, new ArrayList<>());
+ }
+ }
+
+ String line;
+ while ((line = reader.readLine()) != null) {
+ String[] items = line.split(",");
+ for (Entry<String, TSDataType> entry : measureTSTypeInfos.entrySet()) {
+ String measurement = entry.getKey();
+ TSDataType dataType = entry.getValue();
+ int idx = columnToIdMap.get(measurement);
+ switch (dataType) {
+ case BOOLEAN:
+ ret.get(measurement).add(Boolean.parseBoolean(items[idx]));
+ break;
+ case DATE:
+ case INT32:
+ ret.get(measurement).add(Integer.parseInt(items[idx]));
+ break;
+ case TIMESTAMP:
+ case INT64:
+ ret.get(measurement).add(Long.parseLong(items[idx]));
+ break;
+ case FLOAT:
+ ret.get(measurement).add(Float.parseFloat(items[idx]));
+ break;
+ case DOUBLE:
+ ret.get(measurement).add(Double.parseDouble(items[idx]));
+ break;
+ case STRING:
+ case BLOB:
+ case TEXT:
+ ret.get(measurement).add(BytesUtils.valueOf(items[idx]));
+ break;
+ case VECTOR:
+ throw new IOException(String.format("data type %s is not yet.", TSDataType.VECTOR));
+ default:
+ throw new IOException("no type");
+ }
+ }
+ }
+ return ret;
+ } finally {
+ measureTSTypeInfos.remove(TIME_STR);
+ }
+ }
+
+ /**
+ * Read csv file and insert tablet to IoTDB
+ *
+ * @param args: arg(with default value): arg0: dataFileName(sample.csv), arg1: rowSize(10000),
+ * arg2: colSize(5000).
+ */
+ public static void main(String[] args) throws Exception {
+
+ try (Session session = new Session("127.0.0.1", 6667, "root", "root")) {
+ session.open();
+ String dataFileName = "sample.csv";
+ int rowSize = 10000;
+ int colSize = 5000;
+ if (args.length > 1) {
+ dataFileName = args[0];
+ }
+ if (args.length > 2) {
+ rowSize = Integer.parseInt(args[1]);
+ }
+ if (args.length > 3) {
+ colSize = Integer.parseInt(args[2]);
+ }
+
+ // construct the tablet's measurements.
+ Map<String, TSDataType> measureTSTypeInfos = new HashMap<>();
+ measureTSTypeInfos.put("s0", TSDataType.BOOLEAN);
+ measureTSTypeInfos.put("s1", TSDataType.FLOAT);
+ measureTSTypeInfos.put("s2", TSDataType.INT32);
+ measureTSTypeInfos.put("s3", TSDataType.DOUBLE);
+ measureTSTypeInfos.put("s4", TSDataType.INT64);
+ measureTSTypeInfos.put("s5", TSDataType.TEXT);
+ List<IMeasurementSchema> schemas = new ArrayList<>();
+ measureTSTypeInfos.forEach((mea, type) -> schemas.add(new MeasurementSchema(mea, type)));
+
+ System.out.println(
+ String.format(
+ "Test Java: csv file name: %s, row: %d, col: %d", dataFileName, rowSize, colSize));
+ System.out.println(String.format("Total points: %d", rowSize * colSize * schemas.size()));
+
+ // test start
+ long allStart = System.nanoTime();
+
+ Map<String, ArrayList<Object>> data = loadCSVData(measureTSTypeInfos, dataFileName);
+ long loadCost = System.nanoTime() - allStart;
+
+ long insertCost = 0;
+ for (int i = 0; i < colSize; i++) {
+ String deviceId = "root.sg" + i % 8 + "." + i;
+
+ Tablet ta = new Tablet(deviceId, schemas, rowSize);
+ ta.rowSize = rowSize;
+ for (int t = 0; t < ta.rowSize; t++) {
+ ta.addTimestamp(t, (Long) data.get(TIME_STR).get(t));
+ for (Entry<String, TSDataType> entry : measureTSTypeInfos.entrySet()) {
+ String mea = entry.getKey();
+ ta.addValue(mea, t, data.get(mea).get(t));
+ }
+ }
+ long insertSt = System.nanoTime();
+ session.insertTablet(ta, false);
+ insertCost += (System.nanoTime() - insertSt);
+ }
+ // test end
+ long allEnd = System.nanoTime();
+
+ session.executeNonQueryStatement("delete timeseries root.*");
+
+ System.out.println(String.format("load cost: %.3f", ((float) loadCost / 1000_000_000)));
+ System.out.println(
+ String.format(
+ "construct tablet cost: %.3f",
+ ((float) (allEnd - allStart - insertCost - loadCost) / 1000_000_000)));
+ System.out.println(
+ String.format("insert tablet cost: %.3f", ((float) insertCost / 1000_000_000)));
+ System.out.println(
+ String.format("total cost: %.3f", ((float) (allEnd - allStart) / 1000_000_000)));
+ System.out.println(String.format("%.3f", ((float) loadCost / 1000_000_000)));
+ }
+ }
+}
diff --git a/example/pom.xml b/example/pom.xml
index 094ca541af4..78adf4430cf 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -40,6 +40,7 @@
<module>session</module>
<module>trigger</module>
<module>udf</module>
+ <module>influxdb-proto-example</module>
</modules>
<build>
<pluginManagement>
diff --git a/iotdb-client/session/pom.xml b/iotdb-client/session/pom.xml
index b465de97ed0..80c6ce30a62 100644
--- a/iotdb-client/session/pom.xml
+++ b/iotdb-client/session/pom.xml
@@ -87,6 +87,11 @@
<artifactId>powermock-reflect</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.influxdb</groupId>
+ <artifactId>influxdb-java</artifactId>
+ <version>2.21</version>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/IoTDBInfluxDB.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/IoTDBInfluxDB.java
new file mode 100644
index 00000000000..946521723c2
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/IoTDBInfluxDB.java
@@ -0,0 +1,393 @@
+/*
+ * 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.session.influxdb;
+
+import org.influxdb.InfluxDB;
+
+import java.util.*;
+
+public abstract class IoTDBInfluxDB implements InfluxDB {
+
+ // private final IoTDBInfluxDBService influxDBService;
+ //
+ // public IoTDBInfluxDB(String url, String userName, String password) {
+ // URI uri;
+ // try {
+ // uri = new URI(url);
+ // } catch (URISyntaxException e) {
+ // throw new IllegalArgumentException("Unable to parse url: " + url, e);
+ // }
+ // influxDBService = new IoTDBInfluxDBService(uri.getHost(), uri.getPort(), userName,
+ // password);
+ // }
+ //
+ // public IoTDBInfluxDB(String host, int rpcPort, String userName, String password) {
+ // influxDBService = new IoTDBInfluxDBService(host, rpcPort, userName, password);
+ // }
+ //
+ // public IoTDBInfluxDB(Session.Builder builder) {
+ // this(builder.build());
+ // }
+
+ // public IoTDBInfluxDB(Session session) {
+ // SessionPoint sessionPoint = new SessionPoint(session);
+ // influxDBService =
+ // new IoTDBInfluxDBService(
+ // sessionPoint.getHost(),
+ // sessionPoint.getRpcPort(),
+ // sessionPoint.getUsername(),
+ // sessionPoint.getPassword());
+ // }
+ //
+ // @Override
+ // public void write(final Point point) {
+ // write(null, null, point);
+ // }
+ //
+ // @Override
+ // public void write(final String database, final String retentionPolicy, final Point point) {
+ // TimeUnit precision = TimeUnit.NANOSECONDS;
+ // // Get the precision of point in influxdb by reflection
+ // for (java.lang.reflect.Field reflectField : point.getClass().getDeclaredFields()) {
+ // reflectField.setAccessible(true);
+ // try {
+ // if (reflectField.getType().getName().equalsIgnoreCase("java.util.concurrent.TimeUnit")
+ // && reflectField.getName().equalsIgnoreCase("precision")) {
+ // precision = (TimeUnit) reflectField.get(point);
+ // }
+ // } catch (IllegalAccessException e) {
+ // throw new IllegalArgumentException(e.getMessage());
+ // }
+ // }
+ // BatchPoints batchPoints =
+ // BatchPoints.database(database)
+ // .retentionPolicy(retentionPolicy)
+ // .precision(precision)
+ // .build();
+ // batchPoints.point(point);
+ // write(batchPoints);
+ // }
+ //
+ // @Override
+ // public void write(final int udpPort, final Point point) {
+ // write(null, null, point);
+ // }
+ //
+ // @Override
+ // public void write(final BatchPoints batchPoints) {
+ // influxDBService.writePoints(
+ // batchPoints.getDatabase(),
+ // batchPoints.getRetentionPolicy(),
+ // TimeUtil.toTimePrecision(batchPoints.getPrecision()),
+ // batchPoints.getConsistency().value(),
+ // batchPoints.lineProtocol());
+ // }
+ //
+ // @Override
+ // public void writeWithRetry(final BatchPoints batchPoints) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void write(final String records) {
+ // write(null, null, null, null, records);
+ // }
+ //
+ // @Override
+ // public void write(final List<String> records) {
+ // write(String.join("\n", records));
+ // }
+ //
+ // @Override
+ // public void write(
+ // final String database,
+ // final String retentionPolicy,
+ // final ConsistencyLevel consistency,
+ // final String records) {
+ // write(database, retentionPolicy, consistency, null, records);
+ // }
+ //
+ // @Override
+ // public void write(
+ // final String database,
+ // final String retentionPolicy,
+ // final ConsistencyLevel consistency,
+ // final TimeUnit precision,
+ // final String records) {
+ // influxDBService.writePoints(
+ // database,
+ // retentionPolicy,
+ // consistency.value(),
+ // precision == null ? "" : TimeUtil.toTimePrecision(precision),
+ // records);
+ // }
+ //
+ // @Override
+ // public void write(
+ // final String database,
+ // final String retentionPolicy,
+ // final ConsistencyLevel consistency,
+ // final List<String> records) {
+ // write(database, retentionPolicy, consistency, null, String.join("\n", records));
+ // }
+ //
+ // @Override
+ // public void write(
+ // final String database,
+ // final String retentionPolicy,
+ // final ConsistencyLevel consistency,
+ // final TimeUnit precision,
+ // final List<String> records) {
+ // write(database, retentionPolicy, consistency, precision, String.join("\n", records));
+ // }
+ //
+ // @Override
+ // public void write(final int udpPort, final String records) {
+ // write(records);
+ // }
+ //
+ // @Override
+ // public void write(final int udpPort, final List<String> records) {
+ // write(String.join("\n", records));
+ // }
+ //
+ // @Override
+ // public QueryResult query(final Query query) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void createDatabase(final String name) {
+ // ParameterUtils.checkNonEmptyString(name, "database name");
+ // influxDBService.createDatabase(name);
+ // }
+ //
+ // @Override
+ // public void deleteDatabase(final String name) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB setDatabase(final String database) {
+ // ParameterUtils.checkNonEmptyString(database, "database name");
+ // influxDBService.setDatabase(database);
+ // return this;
+ // }
+ //
+ // @Override
+ // public void query(
+ // final Query query,
+ // final Consumer<QueryResult> onSuccess,
+ // final Consumer<Throwable> onFailure) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void query(Query query, int chunkSize, Consumer<QueryResult> onNext) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void query(Query query, int chunkSize, BiConsumer<Cancellable, QueryResult> onNext) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void query(Query query, int chunkSize, Consumer<QueryResult> onNext, Runnable
+ // onComplete) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void query(
+ // Query query,
+ // int chunkSize,
+ // BiConsumer<Cancellable, QueryResult> onNext,
+ // Runnable onComplete) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void query(
+ // Query query,
+ // int chunkSize,
+ // BiConsumer<Cancellable, QueryResult> onNext,
+ // Runnable onComplete,
+ // Consumer<Throwable> onFailure) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public QueryResult query(Query query, TimeUnit timeUnit) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public List<String> describeDatabases() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public boolean databaseExists(final String name) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void flush() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void close() {
+ // influxDBService.close();
+ // }
+ //
+ // @Override
+ // public InfluxDB setConsistency(final ConsistencyLevel consistencyLevel) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB setRetentionPolicy(final String retentionPolicy) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void createRetentionPolicy(
+ // final String rpName,
+ // final String database,
+ // final String duration,
+ // final String shardDuration,
+ // final int replicationFactor,
+ // final boolean isDefault) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void createRetentionPolicy(
+ // final String rpName,
+ // final String database,
+ // final String duration,
+ // final int replicationFactor,
+ // final boolean isDefault) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void createRetentionPolicy(
+ // final String rpName,
+ // final String database,
+ // final String duration,
+ // final String shardDuration,
+ // final int replicationFactor) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void dropRetentionPolicy(final String rpName, final String database) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB setLogLevel(LogLevel logLevel) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB enableGzip() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB disableGzip() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public boolean isGzipEnabled() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB enableBatch() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB enableBatch(BatchOptions batchOptions) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB enableBatch(
+ // final int actions, final int flushDuration, final TimeUnit flushDurationTimeUnit) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB enableBatch(
+ // final int actions,
+ // final int flushDuration,
+ // final TimeUnit flushDurationTimeUnit,
+ // final ThreadFactory threadFactory) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB enableBatch(
+ // int actions,
+ // int flushDuration,
+ // TimeUnit flushDurationTimeUnit,
+ // ThreadFactory threadFactory,
+ // BiConsumer<Iterable<Point>, Throwable> exceptionHandler,
+ // ConsistencyLevel consistency) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public InfluxDB enableBatch(
+ // final int actions,
+ // final int flushDuration,
+ // final TimeUnit flushDurationTimeUnit,
+ // final ThreadFactory threadFactory,
+ // final BiConsumer<Iterable<Point>, Throwable> exceptionHandler) {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public void disableBatch() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public boolean isBatchEnabled() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public Pong ping() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+ //
+ // @Override
+ // public String version() {
+ // throw new UnsupportedOperationException(InfluxDBConstant.METHOD_NOT_SUPPORTED);
+ // }
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/IoTDBInfluxDBProtoFactory.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/IoTDBInfluxDBProtoFactory.java
new file mode 100644
index 00000000000..8ebc0f761f9
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/IoTDBInfluxDBProtoFactory.java
@@ -0,0 +1,74 @@
+/*
+ * 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.session.influxdb;
+
+import org.apache.iotdb.session.pool.SessionPool;
+
+import okhttp3.OkHttpClient;
+import org.influxdb.InfluxDB;
+import org.influxdb.impl.InfluxDBImpl;
+
+public enum IoTDBInfluxDBProtoFactory {
+ INSTANCE;
+
+ private static final SessionPool sessionPool =
+ new SessionPool("127.0.0.1", 6667, "root", "root", 2);
+
+ IoTDBInfluxDBProtoFactory() {}
+
+ //
+
+ public static InfluxDB connect(
+ String url, String username, String password, final OkHttpClient.Builder client) {
+ return new InfluxDBImpl(url, username, password, client) {};
+ }
+ //
+ // public static InfluxDB connect(String host, int rpcPort, String userName, String password) {
+ // ParameterUtils.checkNonEmptyString(host, "host");
+ // ParameterUtils.checkNonEmptyString(userName, "username");
+ // return new IoTDBInfluxDB(host, rpcPort, userName, password);
+ // }
+ //
+ // public static InfluxDB connect(
+ // String url, String username, String password, OkHttpClient.Builder client) {
+ // ParameterUtils.checkNonEmptyString(url, "url");
+ // ParameterUtils.checkNonEmptyString(username, "username");
+ // return connect(url, username, password);
+ // }
+ //
+ // public static InfluxDB connect(
+ // String url,
+ // String username,
+ // String password,
+ // OkHttpClient.Builder client,
+ // InfluxDB.ResponseFormat responseFormat) {
+ // ParameterUtils.checkNonEmptyString(url, "url");
+ // ParameterUtils.checkNonEmptyString(username, "username");
+ // return connect(url, username, password);
+ // }
+ //
+ // public static InfluxDB connect(Session.Builder builder) {
+ // return new IoTDBInfluxDB(builder);
+ // }
+ //
+ // public static InfluxDB connect(Session session) {
+ // return new IoTDBInfluxDB(session);
+ // }
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/example/InfluxProtoExample.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/example/InfluxProtoExample.java
new file mode 100644
index 00000000000..d13594bb59b
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/example/InfluxProtoExample.java
@@ -0,0 +1,72 @@
+/*
+ * 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.session.influxdb.example;
+
+import org.apache.iotdb.session.influxdb.IoTDBInfluxDBProtoFactory;
+
+import okhttp3.OkHttpClient;
+import org.influxdb.InfluxDB;
+import org.influxdb.dto.Point;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class InfluxProtoExample {
+
+ private static InfluxDB influxDB;
+
+ public static void main(String[] args) {
+ influxDB =
+ IoTDBInfluxDBProtoFactory.connect(
+ "http://127.0.0.1:8883/", "root", "root", new OkHttpClient.Builder());
+ influxDB.deleteDatabase("db_test");
+ influxDB.createDatabase("db_test");
+ influxDB.setDatabase("db_test");
+
+ insertData();
+ queryData();
+
+ influxDB.close();
+ }
+
+ private static void insertData() {
+ Point.Builder builder = Point.measurement("student");
+ Map<String, String> tags = new HashMap<>();
+ Map<String, Object> fields = new HashMap<>();
+ tags.put("name", "xie");
+ tags.put("country", "china");
+ fields.put("score", 86.0);
+ builder.tag(tags);
+ builder.fields(fields);
+ builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+ Point point = builder.build();
+ influxDB.write(point);
+ }
+
+ private static void queryData() {
+ QueryResult queryResult = influxDB.query(new Query("select * from student", "database"));
+ for (QueryResult.Result result : queryResult.getResults()) {
+ System.out.println(result);
+ }
+ }
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/constant/InfluxDBConstant.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/constant/InfluxDBConstant.java
new file mode 100644
index 00000000000..7a0705f1114
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/constant/InfluxDBConstant.java
@@ -0,0 +1,25 @@
+/*
+ * 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.session.influxdb.protocol.constant;
+
+public class InfluxDBConstant {
+
+ public static final String METHOD_NOT_SUPPORTED = "Method not supported.";
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/dto/SessionPoint.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/dto/SessionPoint.java
new file mode 100644
index 00000000000..5fee3f40fb2
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/dto/SessionPoint.java
@@ -0,0 +1,88 @@
+/*
+ * 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.session.influxdb.protocol.dto;
+
+public class SessionPoint {
+ // private final String host;
+ // private final int rpcPort;
+ // private final String username;
+ // private final String password;
+ //
+ // public SessionPoint(String host, int rpcPort, String username, String password) {
+ // this.host = host;
+ // this.rpcPort = rpcPort;
+ // this.username = username;
+ // this.password = password;
+ // }
+ //
+ // public SessionPoint(Session session) {
+ //
+ // EndPoint endPoint = null;
+ // String username = null;
+ // String password = null;
+ //
+ // // Get the property of session by reflection
+ // for (java.lang.reflect.Field reflectField : session.getClass().getDeclaredFields()) {
+ // reflectField.setAccessible(true);
+ // try {
+ // if (reflectField
+ // .getType()
+ // .getName()
+ // .equalsIgnoreCase("org.apache.iotdb.service.rpc.thrift.EndPoint")
+ // && reflectField.getName().equalsIgnoreCase("defaultEndPoint")) {
+ // endPoint = (EndPoint) reflectField.get(session);
+ // }
+ // if (reflectField.getType().getName().equalsIgnoreCase("java.lang.String")
+ // && reflectField.getName().equalsIgnoreCase("username")) {
+ // username = (String) reflectField.get(session);
+ // }
+ // if (reflectField.getType().getName().equalsIgnoreCase("java.lang.String")
+ // && reflectField.getName().equalsIgnoreCase("password")) {
+ // password = (String) reflectField.get(session);
+ // }
+ // } catch (IllegalAccessException e) {
+ // throw new IllegalArgumentException(e.getMessage());
+ // }
+ // }
+ // if (endPoint == null) {
+ // throw new InfluxDBException("session's ip and port is null");
+ // }
+ //
+ // this.host = endPoint.ip;
+ // this.rpcPort = endPoint.port;
+ // this.username = username;
+ // this.password = password;
+ // }
+ //
+ // public String getHost() {
+ // return host;
+ // }
+ //
+ // public int getRpcPort() {
+ // return rpcPort;
+ // }
+ //
+ // public String getUsername() {
+ // return username;
+ // }
+ //
+ // public String getPassword() {
+ // return password;
+ // }
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/impl/IoTDBInfluxDBService.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/impl/IoTDBInfluxDBService.java
new file mode 100644
index 00000000000..30ed4dfefbe
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/impl/IoTDBInfluxDBService.java
@@ -0,0 +1,77 @@
+/*
+ * 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.session.influxdb.protocol.impl;
+
+public class IoTDBInfluxDBService {
+ //
+ // private String currentDatabase;
+ //
+ // private final InfluxDBSession influxDBSession;
+ //
+ // public IoTDBInfluxDBService(String host, int rpcPort, String username, String password) {
+ // influxDBSession = new InfluxDBSession(host, rpcPort, username, password);
+ // try {
+ // influxDBSession.open();
+ // } catch (IoTDBConnectionException e) {
+ // throw new InfluxDBException(e.getMessage());
+ // }
+ // }
+ //
+ // public void writePoints(
+ // String database,
+ // String retentionPolicy,
+ // String precision,
+ // String consistency,
+ // String lineProtocol) {
+ // TSWritePointsReq tsWritePointsReq = new TSWritePointsReq();
+ // if (database == null) {
+ // tsWritePointsReq.setDatabase(currentDatabase);
+ // } else {
+ // tsWritePointsReq.setDatabase(database);
+ // }
+ // tsWritePointsReq.setRetentionPolicy(retentionPolicy);
+ // tsWritePointsReq.setPrecision(precision);
+ // tsWritePointsReq.setConsistency(consistency);
+ // tsWritePointsReq.setLineProtocol(lineProtocol);
+ // try {
+ // influxDBSession.writePoints(tsWritePointsReq);
+ // } catch (StatementExecutionException | IoTDBConnectionException e) {
+ // throw new InfluxDBException(e.getMessage());
+ // }
+ // }
+ //
+ // public void setDatabase(String database) {
+ // currentDatabase = database;
+ // }
+ //
+ // public void createDatabase(String database) {
+ // TSCreateDatabaseReq tsCreateDatabaseReq = new TSCreateDatabaseReq();
+ // tsCreateDatabaseReq.setDatabase(database);
+ // try {
+ // influxDBSession.createDatabase(tsCreateDatabaseReq);
+ // } catch (StatementExecutionException | IoTDBConnectionException e) {
+ // throw new InfluxDBException(e.getMessage());
+ // }
+ // }
+ //
+ // public void close() {
+ // influxDBSession.close();
+ // }
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/util/ParameterUtils.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/util/ParameterUtils.java
new file mode 100644
index 00000000000..9b2c76ada7b
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/protocol/util/ParameterUtils.java
@@ -0,0 +1,36 @@
+/*
+ * 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.session.influxdb.protocol.util;
+
+public class ParameterUtils {
+
+ /**
+ * check whether the field is empty. If it is empty, an error will be thrown
+ *
+ * @param string string to check
+ * @param name prompt information in error throwing
+ */
+ public static void checkNonEmptyString(String string, String name)
+ throws IllegalArgumentException {
+ if (string == null || string.isEmpty()) {
+ throw new IllegalArgumentException("Expecting a non-empty string for " + name);
+ }
+ }
+}
diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/session/InfluxDBSession.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/session/InfluxDBSession.java
new file mode 100644
index 00000000000..91ed05e3d9b
--- /dev/null
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/influxdb/session/InfluxDBSession.java
@@ -0,0 +1,219 @@
+/*
+ * 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.session.influxdb.session;
+
+public class InfluxDBSession {
+ // private static final Logger logger = LoggerFactory.getLogger(InfluxDBSession.class);
+ //
+ // private TTransport transport;
+ // private InfluxDBService.Iface client;
+ // private List<EndPoint> endPointList = new ArrayList<>();
+ // private long sessionId;
+ //
+ // protected String username;
+ // protected String password;
+ // protected int fetchSize;
+ // protected ZoneId zoneId;
+ // protected EndPoint defaultEndPoint;
+ // protected int thriftDefaultBufferSize;
+ // protected int thriftMaxFrameSize;
+ //
+ // private boolean isClosed = true;
+ // protected boolean enableRPCCompression;
+ // protected int connectionTimeoutInMs;
+ //
+ // public InfluxDBSession(String host, int rpcPort, String username, String password) {
+ // this(
+ // host,
+ // rpcPort,
+ // username,
+ // password,
+ // Config.DEFAULT_FETCH_SIZE,
+ // ZoneId.systemDefault(),
+ // Config.DEFAULT_INITIAL_BUFFER_CAPACITY,
+ // Config.DEFAULT_MAX_FRAME_SIZE);
+ // }
+ //
+ // public InfluxDBSession(
+ // String host,
+ // int rpcPort,
+ // String username,
+ // String password,
+ // int fetchSize,
+ // ZoneId zoneId,
+ // int thriftDefaultBufferSize,
+ // int thriftMaxFrameSize) {
+ // this.defaultEndPoint = new EndPoint(host, rpcPort);
+ // this.username = username;
+ // this.password = password;
+ // this.fetchSize = fetchSize;
+ // this.zoneId = zoneId;
+ // this.thriftDefaultBufferSize = thriftDefaultBufferSize;
+ // this.thriftMaxFrameSize = thriftMaxFrameSize;
+ // }
+ //
+ // public synchronized void open() throws IoTDBConnectionException {
+ // open(false, Config.DEFAULT_CONNECTION_TIMEOUT_MS);
+ // }
+ //
+ // public synchronized void open(boolean enableRPCCompression, int connectionTimeoutInMs)
+ // throws IoTDBConnectionException {
+ // if (!isClosed) {
+ // return;
+ // }
+ // this.enableRPCCompression = enableRPCCompression;
+ // this.connectionTimeoutInMs = connectionTimeoutInMs;
+ // this.endPointList.add(defaultEndPoint);
+ //
+ // init();
+ // isClosed = false;
+ // }
+ //
+ // public void init() throws IoTDBConnectionException {
+ // RpcTransportFactory.setDefaultBufferCapacity(thriftDefaultBufferSize);
+ // RpcTransportFactory.setThriftMaxFrameSize(thriftMaxFrameSize);
+ // try {
+ // transport =
+ // RpcTransportFactory.INSTANCE.getTransport(
+ // // as there is a try-catch already, we do not need to use TSocket.wrap
+ // defaultEndPoint.getIp(), defaultEndPoint.getPort(), connectionTimeoutInMs);
+ // transport.open();
+ // } catch (TTransportException e) {
+ // throw new IoTDBConnectionException(e);
+ // }
+ //
+ // if (IoTDBDescriptor.getInstance().getConfig().isRpcThriftCompressionEnable()) {
+ // client = new InfluxDBService.Client(new TCompactProtocol(transport));
+ // } else {
+ // client = new InfluxDBService.Client(new TBinaryProtocol(transport));
+ // }
+ // client = RpcUtils.newSynchronizedClient(client);
+ //
+ // TSOpenSessionReq openReq = new TSOpenSessionReq();
+ // openReq.setUsername(username);
+ // openReq.setPassword(password);
+ // openReq.setZoneId(zoneId.toString());
+ //
+ // try {
+ // TSOpenSessionResp openResp = client.openSession(openReq);
+ // RpcUtils.verifySuccess(openResp.getStatus());
+ // sessionId = openResp.getSessionId();
+ //
+ // } catch (Exception e) {
+ // transport.close();
+ // throw new IoTDBConnectionException(e);
+ // }
+ // }
+ //
+ // public void writePoints(TSWritePointsReq request)
+ // throws StatementExecutionException, IoTDBConnectionException {
+ // request.setSessionId(sessionId);
+ // try {
+ // RpcUtils.verifySuccess(client.writePoints(request));
+ // } catch (TException e) {
+ // logger.error(e.getMessage());
+ // if (reconnect()) {
+ // try {
+ // request.setSessionId(sessionId);
+ // RpcUtils.verifySuccess(client.writePoints(request));
+ // } catch (TException tException) {
+ // throw new IoTDBConnectionException(tException);
+ // }
+ // } else {
+ // throw new IoTDBConnectionException(MSG_RECONNECTION_FAIL);
+ // }
+ // }
+ // }
+ //
+ // public void createDatabase(TSCreateDatabaseReq request)
+ // throws StatementExecutionException, IoTDBConnectionException {
+ // request.setSessionId(sessionId);
+ // try {
+ // RpcUtils.verifySuccess(client.createDatabase(request));
+ // } catch (TException e) {
+ // e.printStackTrace();
+ // logger.error(e.getMessage());
+ // if (reconnect()) {
+ // try {
+ // request.setSessionId(sessionId);
+ // RpcUtils.verifySuccess(client.createDatabase(request));
+ // } catch (TException tException) {
+ // throw new IoTDBConnectionException(tException);
+ // }
+ // } else {
+ // throw new IoTDBConnectionException(MSG_RECONNECTION_FAIL);
+ // }
+ // }
+ // }
+ //
+ // public synchronized void close() {
+ // if (isClosed) {
+ // return;
+ // }
+ // try {
+ // client.closeSession(new TSCloseSessionReq(sessionId));
+ // } catch (TException e) {
+ // throw new InfluxDBException(
+ // "Error occurs when closing session at server. Maybe server is down.", e);
+ // } finally {
+ // if (transport != null) {
+ // transport.close();
+ // }
+ // if (!isClosed) {
+ // isClosed = true;
+ // }
+ // }
+ // }
+ //
+ // private boolean reconnect() {
+ // boolean connectedSuccess = false;
+ // Random random = new Random();
+ // for (int i = 1; i <= Config.RETRY_NUM; i++) {
+ // if (transport != null) {
+ // transport.close();
+ // int currHostIndex = random.nextInt(endPointList.size());
+ // int tryHostNum = 0;
+ // for (int j = currHostIndex; j < endPointList.size(); j++) {
+ // if (tryHostNum == endPointList.size()) {
+ // break;
+ // }
+ // defaultEndPoint = endPointList.get(j);
+ // if (j == endPointList.size() - 1) {
+ // j = -1;
+ // }
+ // tryHostNum++;
+ // try {
+ // init();
+ // connectedSuccess = true;
+ // } catch (IoTDBConnectionException e) {
+ // logger.error("The current node may have been down {},try next node",
+ // defaultEndPoint);
+ // continue;
+ // }
+ // break;
+ // }
+ // }
+ // if (connectedSuccess) {
+ // break;
+ // }
+ // }
+ // return connectedSuccess;
+ // }
+}