You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2020/04/25 02:22:45 UTC
[incubator-iotdb] branch master updated: [IOTDB-597] Rename methods
in Session (#1060)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new c116b65 [IOTDB-597] Rename methods in Session (#1060)
c116b65 is described below
commit c116b65e3bd20bb497b2b37f762ac1b9268cbb8d
Author: Jialin Qiao <qj...@mails.tsinghua.edu.cn>
AuthorDate: Sat Apr 25 10:22:34 2020 +0800
[IOTDB-597] Rename methods in Session (#1060)
* rename insertInBatch to insertRecords in Session
* rename insertBatch to insertTablet in Session
* rename insert to insertRecord in Session
* rename RowBatch to Tablet
* add insertTablets rpc interface
---
client-py/src/client_example.py | 8 +-
docs/Development/ContributeGuide.md | 3 +-
docs/SystemDesign/1-TsFile/3-Write.md | 2 +-
.../4-StorageEngine/6-DataManipulation.md | 2 +-
.../4-Client/2-Programming - Native API.md | 92 +++--
.../4-Client/5-Programming - TsFile API.md | 141 +------
docs/zh/Development/ContributeGuide.md | 2 +-
docs/zh/SystemDesign/1-TsFile/3-Write.md | 2 +-
.../4-StorageEngine/6-DataManipulation.md | 2 +-
.../4-Client/2-Programming - Native API.md | 96 +++--
docs/zh/UserGuide/4-Client/3-Programming - JDBC.md | 2 +-
.../4-Client/5-Programming - TsFile API.md | 139 +------
.../apache/iotdb/hadoop/tsfile/TsFileHelper.java | 50 ++-
.../apache/iotdb/rocketmq/RocketMQConsumer.java | 2 +-
.../main/java/org/apache/iotdb/SessionExample.java | 211 +++++------
.../java/org/apache/iotdb/tsfile/TsFileRead.java | 4 +-
...ithRowBatch.java => TsFileWriteWithTablet.java} | 76 ++--
.../java/org/apache/iotdb/flink/IoTDBSink.java | 4 +-
.../iotdb/flink/IoTDBSinkBatchInsertTest.java | 4 +-
.../iotdb/flink/IoTDBSinkBatchTimerTest.java | 2 +-
.../apache/iotdb/flink/IoTDBSinkInsertTest.java | 2 +-
.../iotdb/hadoop/tsfile/TsFileTestHelper.java | 40 +-
.../org/apache/iotdb/hive/TsFileTestHelper.java | 41 +-
.../org/apache/iotdb/db/engine/StorageEngine.java | 10 +-
.../iotdb/db/engine/memtable/AbstractMemTable.java | 20 +-
.../apache/iotdb/db/engine/memtable/IMemTable.java | 6 +-
.../engine/storagegroup/StorageGroupProcessor.java | 46 +--
.../db/engine/storagegroup/TsFileProcessor.java | 17 +-
.../apache/iotdb/db/qp/executor/IPlanExecutor.java | 4 +-
.../apache/iotdb/db/qp/executor/PlanExecutor.java | 16 +-
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 4 +-
...{BatchInsertPlan.java => InsertTabletPlan.java} | 10 +-
.../org/apache/iotdb/db/service/TSServiceImpl.java | 84 ++++-
.../java/org/apache/iotdb/db/utils/MemUtils.java | 10 +-
.../iotdb/db/writelog/recover/LogReplayer.java | 31 +-
.../storagegroup/StorageGroupProcessorTest.java | 29 +-
service-rpc/rpc-changelist.md | 68 ++--
service-rpc/src/main/thrift/rpc.thrift | 30 +-
.../java/org/apache/iotdb/session/Session.java | 419 +++++++++++----------
.../org/apache/iotdb/session/SessionUtils.java | 42 +--
.../org/apache/iotdb/session/pool/SessionPool.java | 119 +++---
.../org/apache/iotdb/session/IoTDBSessionIT.java | 201 +++++-----
.../apache/iotdb/session/pool/SessionPoolTest.java | 17 +-
.../apache/iotdb/tsfile/write/TsFileWriter.java | 40 +-
.../tsfile/write/chunk/ChunkGroupWriterImpl.java | 24 +-
.../tsfile/write/chunk/IChunkGroupWriter.java | 14 +-
.../write/record/{RowBatch.java => Tablet.java} | 100 ++---
.../apache/iotdb/tsfile/write/schema/Schema.java | 22 --
.../iotdb/tsfile/write/TsFileWriterTest.java | 35 +-
49 files changed, 1092 insertions(+), 1253 deletions(-)
diff --git a/client-py/src/client_example.py b/client-py/src/client_example.py
index aad3f9b..42119ee 100644
--- a/client-py/src/client_example.py
+++ b/client-py/src/client_example.py
@@ -24,8 +24,8 @@ sys.path.append("../target")
from thrift.protocol import TBinaryProtocol
from thrift.transport import TSocket, TTransport
-from iotdb.rpc.TSIService import Client, TSCreateTimeseriesReq, TSInsertionReq, \
- TSBatchInsertionReq, TSExecuteStatementReq, TSOpenSessionReq, TSQueryDataSet, \
+from iotdb.rpc.TSIService import Client, TSCreateTimeseriesReq, TSInsertRecordReq, \
+ TSInsertTabletReq, TSExecuteStatementReq, TSOpenSessionReq, TSQueryDataSet, \
TSFetchResultsReq, TSCloseOperationReq, \
TSCloseSessionReq
from iotdb.rpc.ttypes import TSProtocolVersion, TSFetchMetadataReq
@@ -226,7 +226,7 @@ if __name__ == '__main__':
# insert a single row
values = ["1", "11", "1.1", "11.1", "TRUE", "\'text0\'"]
timestamp = 1
- status = client.insert(TSInsertionReq(sessionId, deviceId, measurements,
+ status = client.insert(TSInsertRecordReq(sessionId, deviceId, measurements,
values, timestamp))
print(status.status)
@@ -255,7 +255,7 @@ if __name__ == '__main__':
bytes('\'text3\'', encoding)))
# warning: the data in batch must be sorted by time
times.extend(struct.pack(time_pack_str, 2, 3, 4))
- resp = client.insertBatch(TSBatchInsertionReq(sessionId,deviceId,
+ resp = client.insertTablet(TSInsertTabletReq(sessionId,deviceId,
measurements, values,
times, dataTypes, rowCnt))
status = resp.status
diff --git a/docs/Development/ContributeGuide.md b/docs/Development/ContributeGuide.md
index 8a48d38..fec705a 100644
--- a/docs/Development/ContributeGuide.md
+++ b/docs/Development/ContributeGuide.md
@@ -103,7 +103,8 @@ Recommended Use Intellij idea。```mvn clean package -DskipTests``` After puttin
* Client：```client/src/main/java/org/apache/iotdb/client/```，Use Clinet for linux and WinClint for windows, you can start directly, need the parameter "-h 127.0.0.1 -p 6667 -u root -pw root"
* Server rpc implementation (mainly used for client and server communication, generally start interruption point here):```server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl```
* all jdbc statements：executeStatement(TSExecuteStatementReq req)
- * jdbc query：executeQueryStatement(TSExecuteStatementReq req) * native Write interface：insert(TSInsertReq req)
+ * jdbc query：executeQueryStatement(TSExecuteStatementReq req)
+ * native Write interface：insertRecord(TSInsertRecordReq req)
* Storage engine org.apache.iotdb.db.engine.StorageEngine
* Query engine org.apache.iotdb.db.qp.QueryProcessor
diff --git a/docs/SystemDesign/1-TsFile/3-Write.md b/docs/SystemDesign/1-TsFile/3-Write.md
index a292e4a..cf62c0b 100644
--- a/docs/SystemDesign/1-TsFile/3-Write.md
+++ b/docs/SystemDesign/1-TsFile/3-Write.md
@@ -43,7 +43,7 @@ TsFile file layer has two write interfaces
Write a device with a timestamp and multiple measurement points.
-- TsFileWriter.write(RowBatch rowBatch)
+- TsFileWriter.write(Tablet tablet)
Write multiple timestamps and multiple measurement points on one device.
diff --git a/docs/SystemDesign/4-StorageEngine/6-DataManipulation.md b/docs/SystemDesign/4-StorageEngine/6-DataManipulation.md
index 78dd676..d16a8a0 100644
--- a/docs/SystemDesign/4-StorageEngine/6-DataManipulation.md
+++ b/docs/SystemDesign/4-StorageEngine/6-DataManipulation.md
@@ -29,7 +29,7 @@ The following describes four common data manipulation operations, which are inse
* Corresponding interface
* JDBC's execute and executeBatch interfaces
- * Session's insert and insertInBatch
+ * Session's insertRecord and insertRecords
* Main entrance: ```public void insert(InsertPlan insertPlan)``` StorageEngine.java
* Find the corresponding StorageGroupProcessor
* Find the corresponding TsFileProcessor according to the time of writing the data and the last time stamp of the current device order
diff --git a/docs/UserGuide/4-Client/2-Programming - Native API.md b/docs/UserGuide/4-Client/2-Programming - Native API.md
index 79ea6f7..0338560 100644
--- a/docs/UserGuide/4-Client/2-Programming - Native API.md
+++ b/docs/UserGuide/4-Client/2-Programming - Native API.md
@@ -20,19 +20,18 @@
-->
# Programming - Native API
-## Usage
-### Dependencies
+## Dependencies
* JDK >= 1.8
* Maven >= 3.1
-### How to install in local maven repository
+## How to install in local maven repository
In root directory:
> mvn clean install -pl session -am -DskipTests
-### Using IoTDB Native API with Maven
+## Using IoTDB Native API with Maven
```
<dependencies>
@@ -45,12 +44,10 @@ In root directory:
```
-### Examples with Native API
+## Native APIs
Here we show the commonly used interfaces and their parameters in the Native API:
-#### Run the Native API
-
* Initialize a Session
Session(String host, int port)
@@ -67,84 +64,101 @@ Here we show the commonly used interfaces and their parameters in the Native API
Session.close()
-#### Operations
-
* Set storage group
- TSStatus setStorageGroup(String storageGroupId)
+ ```
+ TSStatus setStorageGroup(String storageGroupId)
+ ```
+
* Delete one or several storage groups
- TSStatus deleteStorageGroup(String storageGroup)
-
```
- TSStatus deleteStorageGroups(List<String> storageGroups)
+ void deleteStorageGroup(String storageGroup)
+ void deleteStorageGroups(List<String> storageGroups)
```
* Create one timeseries under a existing storage group
```
- TSStatus createTimeseries(String path, TSDataType dataType, TSEncoding encoding, CompressionType compressor)
+ void createTimeseries(String path, TSDataType dataType,
+ TSEncoding encoding, CompressionType compressor, Map<String, String> props,
+ Map<String, String> tags, Map<String, String> attributes, String measurementAlias)
+
+ void createMultiTimeseries(List<String> paths, List<TSDataType> dataTypes,
+ List<TSEncoding> encodings, List<CompressionType> compressors,
+ List<Map<String, String>> propsList, List<Map<String, String>> tagsList,
+ List<Map<String, String>> attributesList, List<String> measurementAliasList)
```
* Delete one or several timeseries
- TSStatus deleteTimeseries(String path)
-
- ```java
- TSStatus deleteTimeseries(List<String> paths)
+ ```
+ void deleteTimeseries(String path)
+ void deleteTimeseries(List<String> paths)
```
-* Delete one or several timeseries before a certain timestamp
-
- TSStatus deleteData(String path, long time)
+* Delete data before or equal to a timestap of one or several timeseries
```
- TSStatus deleteData(List<String> paths, long time)
+ void deleteData(String path, long time)
+ void deleteData(List<String> paths, long time)
```
-* Insert data into existing timeseries in batch
+* Insert a Record，which contains multiple measurement value of a device at a timestamp
- ```java
- TSStatus insertInBatch(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList)
- ```
+ ```
+ void insertRecord(String deviceId, long time, List<String> measurements, List<String> values)
+ ```
-* Insert data into existing timeseries
+* Insert a Tablet，which is multiple rows of a device, each row has the same measurements
```
- TSStatus insert(String deviceId, long time, List<String> measurements, List<String> values)
+ void insertTablet(Tablet tablet)
```
-* Batch insertion into timeseries
+* Insert multiple Tablets
+
+ ```
+ void insertTablets(Map<String, Tablet> tablet)
+ ```
+
+* Insert multiple Records
```
- TSExecuteBatchStatementResp insertBatch(RowBatch rowBatch)
+ void insertRecords(List<String> deviceIds, List<Long> times,
+ List<List<String>> measurementsList, List<List<String>> valuesList)
```
-* Test Insert data into existing timeseries in batch. This method NOT insert data into database and server just return after accept the request, this method should be used to test other time cost in client
+## Native APIs for profiling network cost
+
+* Test the network and client cost of insertRecords. This method NOT insert data into database and server just return after accept the request, this method should be used to test other time cost in client
```
- TSStatus testInsertInBatch(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList)
+ void testInsertRecords(List<String> deviceIds, List<Long> times,
+ List<List<String>> measurementsList, List<List<String>> valuesList)
```
-* Insert data into existing timeseries. This method NOT insert data into database and server just return after accept the request, this method should be used to test other time cost in client
+* Test the network and client cost of insertRecord. This method NOT insert data into database and server just return after accept the request, this method should be used to test other time cost in client
```
- TSStatus testInsert(String deviceId, long time, List<String> measurements, List<String> values)
+ void testInsertRecord(String deviceId, long time, List<String> measurements, List<String> values)
```
-* Batch insertion into timeseries. This method NOT insert data into database and server just return after accept the request, this method should be used to test other time cost in client
-
- TSExecuteBatchStatementResp testInsertBatch(RowBatch rowBatch)
+* Test the network and client cost of insertTablet. This method NOT insert data into database and server just return after accept the request, this method should be used to test other time cost in client
-#### Sample code
+ ```
+ void testInsertTablet(Tablet tablet)
+ ```
+
+## Sample code
To get more information of the following interfaces, please view session/src/main/java/org/apache/iotdb/session/Session.java
The sample code of using these interfaces is in example/session/src/main/java/org/apache/iotdb/SessionExample.java，which provides an example of how to open an IoTDB session, execute a batch insertion.
-# Session Pool for Native API
+## Session Pool for Native API
We provided a connection pool (`SessionPool) for Native API.
Using the interface, you need to define the pool size.
diff --git a/docs/UserGuide/4-Client/5-Programming - TsFile API.md b/docs/UserGuide/4-Client/5-Programming - TsFile API.md
index 9aaad0f..a45d9c1 100644
--- a/docs/UserGuide/4-Client/5-Programming - TsFile API.md
+++ b/docs/UserGuide/4-Client/5-Programming - TsFile API.md
@@ -276,146 +276,9 @@ You could write a TsFile by constructing **TSRecord** if you have the **non-alig
A more thorough example can be found at `/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithTSRecord.java`
-```java
-package org.apache.iotdb.tsfile;
-
-import java.io.File;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.write.TsFileWriter;
-import org.apache.iotdb.tsfile.write.record.TSRecord;
-import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
-import org.apache.iotdb.tsfile.write.record.datapoint.FloatDataPoint;
-import org.apache.iotdb.tsfile.write.record.datapoint.IntDataPoint;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-/**
- * An example of writing data to TsFile
- * It uses the interface:
- * public void addMeasurement(MeasurementSchema MeasurementSchema) throws WriteProcessException
- */
-public class TsFileWriteWithTSRecord {
-
- public static void main(String args[]) {
- try {
- String path = "test.tsfile";
- File f = new File(path);
- if (f.exists()) {
- f.delete();
- }
- TsFileWriter tsFileWriter = new TsFileWriter(f);
-
- // add measurements into file schema
- tsFileWriter
- .addMeasurement(new MeasurementSchema("sensor_1", TSDataType.INT64, TSEncoding.RLE));
- tsFileWriter
- .addMeasurement(new MeasurementSchema("sensor_2", TSDataType.INT64, TSEncoding.RLE));
- tsFileWriter
- .addMeasurement(new MeasurementSchema("sensor_3", TSDataType.INT64, TSEncoding.RLE));
-
- // construct TSRecord
- TSRecord tsRecord = new TSRecord(1, "device_1");
- DataPoint dPoint1 = new LongDataPoint("sensor_1", 1);
- DataPoint dPoint2 = new LongDataPoint("sensor_2", 2);
- DataPoint dPoint3 = new LongDataPoint("sensor_3", 3);
- tsRecord.addTuple(dPoint1);
- tsRecord.addTuple(dPoint2);
- tsRecord.addTuple(dPoint3);
-
- // write TSRecord
- tsFileWriter.write(tsRecord);
-
- // close TsFile
- tsFileWriter.close();
- } catch (Throwable e) {
- e.printStackTrace();
- System.out.println(e.getMessage());
- }
- }
-}
-
-```
-
-You could write a TsFile by constructing **RowBatch** if you have the **aligned** time series data.
+You could write a TsFile by constructing **Tablet** if you have the **aligned** time series data.
-A more thorough example can be found at `/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithRowBatch.java`
-
-```java
-package org.apache.iotdb.tsfile;
-
-import java.io.File;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.write.TsFileWriter;
-import org.apache.iotdb.tsfile.write.schema.Schema;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
-/**
- * An example of writing data with RowBatch to TsFile
- */
-public class TsFileWriteWithRowBatch {
-
- public static void main(String[] args) {
- try {
- String path = "test.tsfile";
- File f = new File(path);
- if (f.exists()) {
- f.delete();
- }
-
- Schema schema = new Schema();
-
- // the number of rows to include in the row batch
- int rowNum = 1000000;
- // the number of values to include in the row batch
- int sensorNum = 10;
-
- // add measurements into file schema (all with INT64 data type)
- for (int i = 0; i < sensorNum; i++) {
- schema.registerMeasurement(
- new MeasurementSchema("sensor_" + (i + 1), TSDataType.INT64, TSEncoding.TS_2DIFF));
- }
-
- // add measurements into TSFileWriter
- TsFileWriter tsFileWriter = new TsFileWriter(f, schema);
-
- // construct the row batch
- RowBatch rowBatch = schema.createRowBatch("device_1");
-
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
-
- long timestamp = 1;
- long value = 1000000L;
-
- for (int r = 0; r < rowNum; r++, value++) {
- int row = rowBatch.batchSize++;
- timestamps[row] = timestamp++;
- for (int i = 0; i < sensorNum; i++) {
- long[] sensor = (long[]) values[i];
- sensor[row] = value;
- }
- // write RowBatch to TsFile
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
- }
- }
- // write RowBatch to TsFile
- if (rowBatch.batchSize != 0) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
- }
-
- // close TsFile
- tsFileWriter.close();
- } catch (Throwable e) {
- e.printStackTrace();
- System.out.println(e.getMessage());
- }
- }
-}
-
-```
+A more thorough example can be found at `/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithTablet.java`
### Interface for Reading TsFile
diff --git a/docs/zh/Development/ContributeGuide.md b/docs/zh/Development/ContributeGuide.md
index 1e0ba35..3890d2c 100644
--- a/docs/zh/Development/ContributeGuide.md
+++ b/docs/zh/Development/ContributeGuide.md
@@ -120,7 +120,7 @@ import -> Maven -> Existing Maven Projects
* 服务器的 rpc 实现（主要用来客户端和服务器通信，一般在这里开始打断点）：```server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl```
* jdbc所有语句：executeStatement(TSExecuteStatementReq req)
* jdbc查询语句：executeQueryStatement(TSExecuteStatementReq req)
- * native写入接口：insert(TSInsertReq req)
+ * native写入接口：insertRecord(TSInsertRecordReq req)
* 存储引擎 org.apache.iotdb.db.engine.StorageEngine
* 查询引擎 org.apache.iotdb.db.qp.Planner
diff --git a/docs/zh/SystemDesign/1-TsFile/3-Write.md b/docs/zh/SystemDesign/1-TsFile/3-Write.md
index cb219e9..0cbb1e4 100644
--- a/docs/zh/SystemDesign/1-TsFile/3-Write.md
+++ b/docs/zh/SystemDesign/1-TsFile/3-Write.md
@@ -43,7 +43,7 @@ TsFile 文件层的写入接口有两种
写入一个设备一个时间戳多个测点。
-* TsFileWriter.write(RowBatch rowBatch)
+* TsFileWriter.write(Tablet tablet)
写入一个设备多个时间戳多个测点。
diff --git a/docs/zh/SystemDesign/4-StorageEngine/6-DataManipulation.md b/docs/zh/SystemDesign/4-StorageEngine/6-DataManipulation.md
index 51f024b..23ca674 100644
--- a/docs/zh/SystemDesign/4-StorageEngine/6-DataManipulation.md
+++ b/docs/zh/SystemDesign/4-StorageEngine/6-DataManipulation.md
@@ -29,7 +29,7 @@
* 对应的接口
* JDBC 的 execute 和 executeBatch 接口
- * Session 的 insert 和 insertInBatch
+ * Session 的 insertRecord 和 insertRecords
* 总入口: public void insert(InsertPlan insertPlan) StorageEngine.java
* 找到对应的 StorageGroupProcessor
diff --git a/docs/zh/UserGuide/4-Client/2-Programming - Native API.md b/docs/zh/UserGuide/4-Client/2-Programming - Native API.md
index f52dafb..8db4eeb 100644
--- a/docs/zh/UserGuide/4-Client/2-Programming - Native API.md
+++ b/docs/zh/UserGuide/4-Client/2-Programming - Native API.md
@@ -21,8 +21,6 @@
# 编程 - 原生接口
-## 使用
-
## 依赖
* JDK >= 1.8
@@ -48,73 +46,119 @@
## 原生接口使用示例
下面将给出Session对应的接口的简要介绍和对应参数：
-### 建立连接
-
* 初始化Session
+
+ ```
Session(String host, int port)
Session(String host, String port, String username, String password)
Session(String host, int port, String username, String password)
-
+ ```
+
* 开启Session
- Session.open()
+ ```
+ Session.open()
+ ```
+
* 关闭Session
- Session.close()
-
-### 数据操作接口
-
+
+ ```
+ Session.close()
+ ```
+
* 设置存储组
```
- TSStatus setStorageGroup(String storageGroupId)
+ void setStorageGroup(String storageGroupId)
```
* 删除单个或多个存储组
```
- TSStatus deleteStorageGroup(String storageGroup)
- TSStatus deleteStorageGroups(List<String> storageGroups)
+ void deleteStorageGroup(String storageGroup)
+ void deleteStorageGroups(List<String> storageGroups)
```
-* 创建单个时间序列
+* 创建单个或多个时间序列
```
- TSStatus createTimeseries(String path, TSDataType dataType, TSEncoding encoding, CompressionType compressor)
+ void createTimeseries(String path, TSDataType dataType,
+ TSEncoding encoding, CompressionType compressor, Map<String, String> props,
+ Map<String, String> tags, Map<String, String> attributes, String measurementAlias)
+
+ void createMultiTimeseries(List<String> paths, List<TSDataType> dataTypes,
+ List<TSEncoding> encodings, List<CompressionType> compressors,
+ List<Map<String, String>> propsList, List<Map<String, String>> tagsList,
+ List<Map<String, String>> attributesList, List<String> measurementAliasList)
```
* 删除一个或多个时间序列
```
- TSStatus deleteTimeseries(String path)
- TSStatus deleteTimeseries(List<String> paths)
+ void deleteTimeseries(String path)
+ void deleteTimeseries(List<String> paths)
```
-* 删除某一特定时间前的时间序列
+* 删除一个或多个时间序列在某个时间点前的数据
```
- TSStatus deleteData(String path, long time)
- TSStatus deleteData(List<String> paths, long time)
+ void deleteData(String path, long time)
+ void deleteData(List<String> paths, long time)
```
-* 插入时序数据
+* 插入一个 Record，一个 Record 是一个设备一个时间戳下多个测点的数据
```
- TSStatus insert(String deviceId, long time, List<String> measurements, List<String> values)
+ void insertRecord(String deviceId, long time, List<String> measurements, List<String> values)
```
-* 批量插入时序数据
+* 插入一个 Tablet，Tablet 是一个设备若干行非空数据块，每一行的列都相同
```
- TSExecuteBatchStatementResp insertBatch(RowBatch rowBatch)
+ void insertTablet(Tablet tablet)
```
-### 示例代码
+* 插入多个 Tablet
+
+ ```
+ void insertTablets(Map<String, Tablet> tablet)
+ ```
+
+* 插入多个 Record
+
+ ```
+ void insertRecords(List<String> deviceIds, List<Long> times,
+ List<List<String>> measurementsList, List<List<String>> valuesList)
+ ```
+
+## 测试客户端逻辑+网络传输代价的接口
+
+* 测试 testInsertInBatch，不实际写入数据，只将数据传输到 server 即返回。
+
+ ```
+ TSStatus testInsertInBatch(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList)
+ ```
+
+* 测试 insertRecord，不实际写入数据，只将数据传输到 server 即返回。
+
+ ```
+ void testInsertRecord(String deviceId, long time, List<String> measurements, List<String> values)
+ ```
+
+* 测试 insertTablet，不实际写入数据，只将数据传输到 server 即返回。
+
+ ```
+ void testInsertTablet(Tablet tablet)
+ ```
+
+
+## 示例代码
浏览上述接口的详细信息，请参阅代码 ```session/src/main/java/org/apache/iotdb/session/Session.java```
使用上述接口的示例代码在 ```example/session/src/main/java/org/apache/iotdb/SessionExample.java```
-# 针对原生接口的连接池
+## 针对原生接口的连接池
我们提供了一个针对原生接口的连接池(`SessionPool`)，使用该接口时，你只需要指定连接池的大小，就可以在使用时从池中获取连接。
如果超过60s都没得到一个连接的话，那么会打印一条警告日志，但是程序仍将继续等待。
diff --git a/docs/zh/UserGuide/4-Client/3-Programming - JDBC.md b/docs/zh/UserGuide/4-Client/3-Programming - JDBC.md
index 00d7084..ecf96c0 100644
--- a/docs/zh/UserGuide/4-Client/3-Programming - JDBC.md
+++ b/docs/zh/UserGuide/4-Client/3-Programming - JDBC.md
@@ -61,7 +61,7 @@ mvn clean install -pl jdbc -am -Dmaven.test.skip=true
要求您已经在工程中包含了数据库编程所需引入的包和 JDBC class.
-**注意:为了更快地插入，建议使用 insertBatch()**
+**注意:为了更快地插入，建议使用 executeBatch()**
```Java
import java.sql.*;
diff --git a/docs/zh/UserGuide/4-Client/5-Programming - TsFile API.md b/docs/zh/UserGuide/4-Client/5-Programming - TsFile API.md
index b2fb0c3..f657664 100644
--- a/docs/zh/UserGuide/4-Client/5-Programming - TsFile API.md
+++ b/docs/zh/UserGuide/4-Client/5-Programming - TsFile API.md
@@ -274,146 +274,11 @@ mvn clean install -pl tsfile -am -DskipTests
更详细的例子可以在`/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithTSRecord.java`中查看
-```java
-package org.apache.iotdb.tsfile;
-
-import java.io.File;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.write.TsFileWriter;
-import org.apache.iotdb.tsfile.write.record.TSRecord;
-import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
-import org.apache.iotdb.tsfile.write.record.datapoint.FloatDataPoint;
-import org.apache.iotdb.tsfile.write.record.datapoint.IntDataPoint;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-/**
- * An example of writing data to TsFile
- * It uses the interface:
- * public void addMeasurement(MeasurementSchema MeasurementSchema) throws WriteProcessException
- */
-public class TsFileWriteWithTSRecord {
-
- public static void main(String args[]) {
- try {
- String path = "test.tsfile";
- File f = new File(path);
- if (f.exists()) {
- f.delete();
- }
- TsFileWriter tsFileWriter = new TsFileWriter(f);
-
- // add measurements into file schema
- tsFileWriter
- .addMeasurement(new MeasurementSchema("sensor_1", TSDataType.INT64, TSEncoding.RLE));
- tsFileWriter
- .addMeasurement(new MeasurementSchema("sensor_2", TSDataType.INT64, TSEncoding.RLE));
- tsFileWriter
- .addMeasurement(new MeasurementSchema("sensor_3", TSDataType.INT64, TSEncoding.RLE));
-
- // construct TSRecord
- TSRecord tsRecord = new TSRecord(1, "device_1");
- DataPoint dPoint1 = new LongDataPoint("sensor_1", 1);
- DataPoint dPoint2 = new LongDataPoint("sensor_2", 2);
- DataPoint dPoint3 = new LongDataPoint("sensor_3", 3);
- tsRecord.addTuple(dPoint1);
- tsRecord.addTuple(dPoint2);
- tsRecord.addTuple(dPoint3);
-
- // write TSRecord
- tsFileWriter.write(tsRecord);
-
- // close TsFile
- tsFileWriter.close();
- } catch (Throwable e) {
- e.printStackTrace();
- System.out.println(e.getMessage());
- }
- }
-}
-
-```
-如果所有时序数据都是**对齐**的，您可以通过构造**RowBatch**来写入数据。
+如果所有时序数据都是**对齐**的，您可以通过构造**Tablet**来写入数据。
-更详细的例子可以在`/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithRowBatch.java`中查看
+更详细的例子可以在`/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithTablet.java`中查看
-```java
-package org.apache.iotdb.tsfile;
-
-import java.io.File;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.write.TsFileWriter;
-import org.apache.iotdb.tsfile.write.schema.Schema;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
-/**
- * An example of writing data with RowBatch to TsFile
- */
-public class TsFileWriteWithRowBatch {
-
- public static void main(String[] args) {
- try {
- String path = "test.tsfile";
- File f = new File(path);
- if (f.exists()) {
- f.delete();
- }
-
- Schema schema = new Schema();
-
- // the number of rows to include in the row batch
- int rowNum = 1000000;
- // the number of values to include in the row batch
- int sensorNum = 10;
-
- // add measurements into file schema (all with INT64 data type)
- for (int i = 0; i < sensorNum; i++) {
- schema.registerMeasurement(
- new MeasurementSchema("sensor_" + (i + 1), TSDataType.INT64, TSEncoding.TS_2DIFF));
- }
-
- // add measurements into TSFileWriter
- TsFileWriter tsFileWriter = new TsFileWriter(f, schema);
-
- // construct the row batch
- RowBatch rowBatch = schema.createRowBatch("device_1");
-
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
-
- long timestamp = 1;
- long value = 1000000L;
-
- for (int r = 0; r < rowNum; r++, value++) {
- int row = rowBatch.batchSize++;
- timestamps[row] = timestamp++;
- for (int i = 0; i < sensorNum; i++) {
- long[] sensor = (long[]) values[i];
- sensor[row] = value;
- }
- // write RowBatch to TsFile
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
- }
- }
- // write RowBatch to TsFile
- if (rowBatch.batchSize != 0) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
- }
-
- // close TsFile
- tsFileWriter.close();
- } catch (Throwable e) {
- e.printStackTrace();
- System.out.println(e.getMessage());
- }
- }
-}
-
-```
### 读取 TsFile 接口
diff --git a/example/hadoop/src/main/java/org/apache/iotdb/hadoop/tsfile/TsFileHelper.java b/example/hadoop/src/main/java/org/apache/iotdb/hadoop/tsfile/TsFileHelper.java
index 7a02a76..63e2997 100644
--- a/example/hadoop/src/main/java/org/apache/iotdb/hadoop/tsfile/TsFileHelper.java
+++ b/example/hadoop/src/main/java/org/apache/iotdb/hadoop/tsfile/TsFileHelper.java
@@ -21,12 +21,14 @@ package org.apache.iotdb.hadoop.tsfile;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.write.TsFileWriter;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.Schema;
import org.slf4j.Logger;
@@ -52,39 +54,47 @@ public class TsFileHelper {
Schema schema = new Schema();
- // the number of rows to include in the row batch
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+
+ // the number of rows to include in the tablet
int rowNum = 1000000;
- // the number of values to include in the row batch
+ // the number of values to include in the tablet
int sensorNum = 10;
// add measurements into file schema (all with INT64 data type)
for (int i = 0; i < 2; i++) {
+ MeasurementSchema measurementSchema = new MeasurementSchema(
+ Constant.SENSOR_PREFIX + (i + 1), TSDataType.INT64,
+ TSEncoding.TS_2DIFF);
schema.registerTimeseries(new Path(Constant.DEVICE_1, Constant.SENSOR_PREFIX + (i + 1)),
- new MeasurementSchema(Constant.SENSOR_PREFIX + (i + 1), TSDataType.INT64,
- TSEncoding.TS_2DIFF));
+ measurementSchema);
+ schemaList.add(measurementSchema);
}
for (int i = 2; i < sensorNum; i++) {
+ MeasurementSchema measurementSchema = new MeasurementSchema(
+ Constant.SENSOR_PREFIX + (i + 1), TSDataType.DOUBLE,
+ TSEncoding.TS_2DIFF);
schema.registerTimeseries(new Path(Constant.DEVICE_1, Constant.SENSOR_PREFIX + (i + 1)),
- new MeasurementSchema(Constant.SENSOR_PREFIX + (i + 1), TSDataType.DOUBLE,
- TSEncoding.TS_2DIFF));
+ measurementSchema);
+ schemaList.add(measurementSchema);
}
// add measurements into TSFileWriter
TsFileWriter tsFileWriter = new TsFileWriter(file, schema);
- // construct the row batch
- RowBatch rowBatch = schema.createRowBatch(Constant.DEVICE_1);
+ // construct the tablet
+ Tablet tablet = new Tablet(Constant.DEVICE_1, schemaList);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
long timestamp = 1;
long value = 1000000L;
double doubleValue = 1.1;
for (int r = 0; r < rowNum; r++, value++, doubleValue = doubleValue + 0.1) {
- int row = rowBatch.batchSize++;
+ int row = tablet.rowSize++;
timestamps[row] = timestamp++;
for (int i = 0; i < 2; i++) {
long[] sensor = (long[]) values[i];
@@ -94,16 +104,16 @@ public class TsFileHelper {
double[] sensor = (double[]) values[i];
sensor[row] = doubleValue;
}
- // write RowBatch to TsFile
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
+ // write Tablet to TsFile
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ tsFileWriter.write(tablet);
+ tablet.reset();
}
}
- // write RowBatch to TsFile
- if (rowBatch.batchSize != 0) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
+ // write Tablet to TsFile
+ if (tablet.rowSize != 0) {
+ tsFileWriter.write(tablet);
+ tablet.reset();
}
// close TsFile
diff --git a/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java
index 7fedab8..696b2b9 100644
--- a/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java
+++ b/example/rocketmq/src/main/java/org/apache/iotdb/rocketmq/RocketMQConsumer.java
@@ -90,7 +90,7 @@ public class RocketMQConsumer {
long time = Long.parseLong(dataArray[1]);
List<String> measurements = Arrays.asList(dataArray[2].split(":"));
List<String> values = Arrays.asList(dataArray[3].split(":"));
- session.insert(device, time, measurements, values);
+ session.insertRecord(device, time, measurements, values);
}
public void start() throws MQClientException {
diff --git a/example/session/src/main/java/org/apache/iotdb/SessionExample.java b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
index 020188f..65fb37f 100644
--- a/example/session/src/main/java/org/apache/iotdb/SessionExample.java
+++ b/example/session/src/main/java/org/apache/iotdb/SessionExample.java
@@ -26,8 +26,7 @@ import org.apache.iotdb.session.SessionDataSet;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.read.common.Path;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.Schema;
import java.util.ArrayList;
@@ -55,9 +54,10 @@ public class SessionExample {
createTimeseries();
createMultiTimeseries();
- insert();
- insertInBatch();
- insertRowBatch();
+ insertRecord();
+ insertTablet();
+ insertTablets();
+ insertRecords();
nonQuery();
query();
deleteData();
@@ -81,6 +81,7 @@ public class SessionExample {
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");
@@ -94,41 +95,43 @@ public class SessionExample {
private static void createMultiTimeseries()
throws IoTDBConnectionException, BatchExecutionException {
- 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.DOUBLE);
- tsDataTypes.add(TSDataType.DOUBLE);
- 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);
+ 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 insert() throws IoTDBConnectionException, StatementExecutionException {
+ private static void insertRecord() throws IoTDBConnectionException, StatementExecutionException {
String deviceId = "root.sg1.d1";
List<String> measurements = new ArrayList<>();
measurements.add("s1");
@@ -139,7 +142,7 @@ public class SessionExample {
values.add("1");
values.add("2");
values.add("3");
- session.insert(deviceId, time, measurements, values);
+ session.insertRecord(deviceId, time, measurements, values);
}
}
@@ -151,11 +154,11 @@ public class SessionExample {
measurements.add("s2");
measurements.add("s3");
for (long time = 0; time < 100; time++) {
- session.insert(deviceId, time, measurements, 1L, 1L, 1L);
+ session.insertRecord(deviceId, time, measurements, 1L, 1L, 1L);
}
}
- private static void insertInBatch() throws IoTDBConnectionException, BatchExecutionException {
+ private static void insertRecords() throws IoTDBConnectionException, BatchExecutionException {
String deviceId = "root.sg1.d1";
List<String> measurements = new ArrayList<>();
measurements.add("s1");
@@ -177,7 +180,7 @@ public class SessionExample {
valuesList.add(values);
timestamps.add(time);
if (time != 0 && time % 100 == 0) {
- session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList);
+ session.insertRecords(deviceIds, timestamps, measurementsList, valuesList);
deviceIds.clear();
measurementsList.clear();
valuesList.clear();
@@ -185,14 +188,13 @@ public class SessionExample {
}
}
- session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList);
+ session.insertRecords(deviceIds, timestamps, measurementsList, valuesList);
}
/**
- * insert a batch data of one device, each batch contains multiple timestamps with values of
- * sensors
+ * insert the data of a device. For each timestamp, the number of measurements is the same.
*
- * a RowBatch example:
+ * a Tablet example:
*
* device1
* time s1, s2, s3
@@ -200,85 +202,66 @@ public class SessionExample {
* 2, 2, 2, 2
* 3, 3, 3, 3
*
- * Users need to control the count of RowBatch and write a batch when it reaches the maxBatchSize
+ * Users need to control the count of Tablet and write a batch when it reaches the maxBatchSize
*/
- private static void insertRowBatch() throws IoTDBConnectionException, BatchExecutionException {
+ private static void insertTablet() throws IoTDBConnectionException, BatchExecutionException {
// The schema of sensors of one device
- Schema schema = new Schema();
- schema.registerTimeseries(new Path("root.sg1.d1", "s1"),
- new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path("root.sg1.d1", "s2"),
- new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path("root.sg1.d1", "s3"),
- new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
- RowBatch rowBatch = schema.createRowBatch("root.sg1.d1", 100);
+ Tablet tablet = new Tablet("root.sg1.d1", schemaList, 100);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
for (long time = 0; time < 100; time++) {
- int row = rowBatch.batchSize++;
+ int row = tablet.rowSize++;
timestamps[row] = time;
for (int i = 0; i < 3; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = i;
}
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet, true);
+ tablet.reset();
}
}
- if (rowBatch.batchSize != 0) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize != 0) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
- private static void insertMultipleDeviceRowBatch()
- throws IoTDBConnectionException, BatchExecutionException {
+ private static void insertTablets() throws IoTDBConnectionException, BatchExecutionException {
// The schema of sensors of one device
- Schema schema1 = new Schema();
- schema1.registerTimeseries(new Path("root.sg1.d1.s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema1.registerTimeseries(new Path("root.sg1.d1.s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema1.registerTimeseries(new Path("root.sg1.d1.s3"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
-
- RowBatch rowBatch1 = schema1.createRowBatch("root.sg1.d1", 100);
-
- Schema schema2 = new Schema();
- schema2.registerTimeseries(new Path("root.sg1.d2.s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema2.registerTimeseries(new Path("root.sg1.d2.s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema2.registerTimeseries(new Path("root.sg1.d2.s3"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
-
- RowBatch rowBatch2 = schema2.createRowBatch("root.sg1.d2", 100);
-
- Schema schema3 = new Schema();
- Map<String, MeasurementSchema> template = new HashMap<>();
- template.put("s1", new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- template.put("s2", new MeasurementSchema("s2", TSDataType.INT32, TSEncoding.RLE));
- schema3.registerDeviceTemplate("template3", template);
- schema3.extendTemplate("template3", new MeasurementSchema("s3", TSDataType.FLOAT, TSEncoding.RLE));
- schema3.registerDevice("root.sg1.d3", "template3");
-
- RowBatch rowBatch3 = schema3.createRowBatch("root.sg1.d3", 100);
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
+
+ Tablet tablet1 = new Tablet("root.sg1.d1", schemaList, 100);
+ Tablet tablet2 = new Tablet("root.sg1.d2", schemaList, 100);
+ Tablet tablet3 = new Tablet("root.sg1.d3", schemaList, 100);
- Map<String, RowBatch> rowBatchMap = new HashMap<>();
- rowBatchMap.put("root.sg1.d1", rowBatch1);
- rowBatchMap.put("root.sg1.d2", rowBatch2);
- rowBatchMap.put("root.sg1.d3", rowBatch3);
-
- long[] timestamps1 = rowBatch1.timestamps;
- Object[] values1 = rowBatch1.values;
- long[] timestamps2 = rowBatch2.timestamps;
- Object[] values2 = rowBatch2.values;
- long[] timestamps3 = rowBatch3.timestamps;
- Object[] values3 = rowBatch3.values;
+ Map<String, Tablet> tabletMap = new HashMap<>();
+ tabletMap.put("root.sg1.d1", tablet1);
+ tabletMap.put("root.sg1.d2", tablet2);
+ tabletMap.put("root.sg1.d3", tablet3);
+
+ long[] timestamps1 = tablet1.timestamps;
+ Object[] values1 = tablet1.values;
+ long[] timestamps2 = tablet2.timestamps;
+ Object[] values2 = tablet2.values;
+ long[] timestamps3 = tablet3.timestamps;
+ Object[] values3 = tablet3.values;
for (long time = 0; time < 100; time++) {
- int row1 = rowBatch1.batchSize++;
- int row2 = rowBatch2.batchSize++;
- int row3 = rowBatch3.batchSize++;
+ int row1 = tablet1.rowSize++;
+ int row2 = tablet2.rowSize++;
+ int row3 = tablet3.rowSize++;
timestamps1[row1] = time;
timestamps2[row2] = time;
timestamps3[row3] = time;
@@ -290,20 +273,20 @@ public class SessionExample {
long[] sensor3 = (long[]) values3[i];
sensor3[row3] = i;
}
- if (rowBatch1.batchSize == rowBatch1.getMaxBatchSize()) {
- session.insertMultipleDeviceBatch(rowBatchMap);
+ if (tablet1.rowSize == tablet1.getMaxRowNumber()) {
+ session.insertTablets(tabletMap, true);
- rowBatch1.reset();
- rowBatch2.reset();
- rowBatch3.reset();
+ tablet1.reset();
+ tablet2.reset();
+ tablet3.reset();
}
}
- if (rowBatch1.batchSize != 0) {
- session.insertMultipleDeviceBatch(rowBatchMap);
- rowBatch1.reset();
- rowBatch2.reset();
- rowBatch3.reset();
+ if (tablet1.rowSize != 0) {
+ session.insertTablets(tabletMap, true);
+ tablet1.reset();
+ tablet2.reset();
+ tablet3.reset();
}
}
diff --git a/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileRead.java b/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileRead.java
index 18efc3b..8cea9fc 100644
--- a/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileRead.java
+++ b/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileRead.java
@@ -34,8 +34,8 @@ import java.util.ArrayList;
/**
* The class is to show how to read TsFile file named "test.tsfile".
- * The TsFile file "test.tsfile" is generated from class TsFileWriteWithTSRecord or TsFileWriteWithRowBatch.
- * Run TsFileWriteWithTSRecord or TsFileWriteWithRowBatch to generate the test.tsfile first
+ * The TsFile file "test.tsfile" is generated from class TsFileWriteWithTSRecord or TsFileWriteWithTablet.
+ * Run TsFileWriteWithTSRecord or TsFileWriteWithTablet to generate the test.tsfile first
*/
public class TsFileRead {
private static void queryAndPrint(ArrayList<Path> paths, ReadOnlyTsFile readTsFile, IExpression statement)
diff --git a/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithRowBatch.java b/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithTablet.java
similarity index 52%
rename from example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithRowBatch.java
rename to example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithTablet.java
index bd12a32..7dfe074 100644
--- a/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithRowBatch.java
+++ b/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileWriteWithTablet.java
@@ -19,80 +19,86 @@
package org.apache.iotdb.tsfile;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
-import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.write.TsFileWriter;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.Schema;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import java.io.File;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
- * An example of writing data with RowBatch to TsFile
+ * An example of writing data with Tablet to TsFile
*/
-public class TsFileWriteWithRowBatch {
+public class TsFileWriteWithTablet {
+
+ private static final Logger logger = LoggerFactory.getLogger(TsFileWriteWithTablet.class);
public static void main(String[] args) {
try {
String path = "test.tsfile";
File f = FSFactoryProducer.getFSFactory().getFile(path);
if (f.exists()) {
- f.delete();
+ if (!f.delete()) {
+ throw new RuntimeException("can not delete " + f.getAbsolutePath());
+ }
}
Schema schema = new Schema();
- // the number of rows to include in the row batch
+ // the number of rows to include in the tablet
int rowNum = 1000000;
- // the number of values to include in the row batch
+ // the number of values to include in the tablet
int sensorNum = 10;
+ List<MeasurementSchema> measurementSchemas = new ArrayList<>();
// add measurements into file schema (all with INT64 data type)
for (int i = 0; i < sensorNum; i++) {
- schema.registerTimeseries(new Path("device_1", "sensor_" + (i + 1)),
+ measurementSchemas.add(
new MeasurementSchema("sensor_" + (i + 1), TSDataType.INT64, TSEncoding.TS_2DIFF));
}
// add measurements into TSFileWriter
- TsFileWriter tsFileWriter = new TsFileWriter(f, schema);
+ try (TsFileWriter tsFileWriter = new TsFileWriter(f, schema)) {
- // construct the row batch
- RowBatch rowBatch = schema.createRowBatch("device_1");
+ // construct the tablet
+ Tablet tablet = new Tablet("device_1", measurementSchemas);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
- long timestamp = 1;
- long value = 1000000L;
+ long timestamp = 1;
+ long value = 1000000L;
- for (int r = 0; r < rowNum; r++, value++) {
- int row = rowBatch.batchSize++;
- timestamps[row] = timestamp++;
- for (int i = 0; i < sensorNum; i++) {
- long[] sensor = (long[]) values[i];
- sensor[row] = value;
+ for (int r = 0; r < rowNum; r++, value++) {
+ int row = tablet.rowSize++;
+ timestamps[row] = timestamp++;
+ for (int i = 0; i < sensorNum; i++) {
+ long[] sensor = (long[]) values[i];
+ sensor[row] = value;
+ }
+ // write Tablet to TsFile
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ tsFileWriter.write(tablet);
+ tablet.reset();
+ }
}
- // write RowBatch to TsFile
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
+ // write Tablet to TsFile
+ if (tablet.rowSize != 0) {
+ tsFileWriter.write(tablet);
+ tablet.reset();
}
}
- // write RowBatch to TsFile
- if (rowBatch.batchSize != 0) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
- }
- // close TsFile
- tsFileWriter.close();
- } catch (Throwable e) {
- e.printStackTrace();
- System.out.println(e.getMessage());
+ } catch (Exception e) {
+ logger.error("meet error in TsFileWrite with tablet", e);
}
}
}
diff --git a/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java b/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java
index eebc3ce..bb3e042 100644
--- a/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java
+++ b/flink-iotdb-connector/src/main/java/org/apache/iotdb/flink/IoTDBSink.java
@@ -115,7 +115,7 @@ public class IoTDBSink<IN> extends RichSinkFunction<IN> {
}
convertText(event.getDevice(), event.getMeasurements(), event.getValues());
- pool.insert(event.getDevice(), event.getTimestamp(), event.getMeasurements(),
+ pool.insertRecord(event.getDevice(), event.getTimestamp(), event.getMeasurements(),
event.getValues());
LOG.debug("send event successfully");
}
@@ -182,7 +182,7 @@ public class IoTDBSink<IN> extends RichSinkFunction<IN> {
measurementsList.add(event.getMeasurements());
valuesList.add(event.getValues());
}
- pool.insertInBatch(deviceIds, timestamps, measurementsList, valuesList);
+ pool.insertRecords(deviceIds, timestamps, measurementsList, valuesList);
LOG.debug("send event successfully");
batchList.clear();
}
diff --git a/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchInsertTest.java b/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchInsertTest.java
index 8a83cff..7f6fe5e 100644
--- a/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchInsertTest.java
+++ b/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchInsertTest.java
@@ -73,7 +73,7 @@ public class IoTDBSinkBatchInsertTest {
tuple.put("values", "37.1");
ioTDBSink.invoke(tuple, null);
- verify(pool).insertInBatch(any(List.class), any(List.class), any(List.class), any(List.class));
+ verify(pool).insertRecords(any(List.class), any(List.class), any(List.class), any(List.class));
tuple = new HashMap();
tuple.put("device", "root.sg.D01");
@@ -96,7 +96,7 @@ public class IoTDBSinkBatchInsertTest {
verifyZeroInteractions(pool);
ioTDBSink.close();
- verify(pool).insertInBatch(any(List.class), any(List.class), any(List.class), any(List.class));
+ verify(pool).insertRecords(any(List.class), any(List.class), any(List.class), any(List.class));
verify(pool).close();
}
}
diff --git a/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchTimerTest.java b/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchTimerTest.java
index c4112ae..f8c7f7e 100644
--- a/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchTimerTest.java
+++ b/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkBatchTimerTest.java
@@ -59,7 +59,7 @@ public class IoTDBSinkBatchTimerTest {
Thread.sleep(2500);
- verify(pool).insertInBatch(any(List.class), any(List.class), any(List.class), any(List.class));
+ verify(pool).insertRecords(any(List.class), any(List.class), any(List.class), any(List.class));
Thread.sleep(1000);
diff --git a/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkInsertTest.java b/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkInsertTest.java
index 07a38f7..3bcb367 100644
--- a/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkInsertTest.java
+++ b/flink-iotdb-connector/src/test/java/org/apache/iotdb/flink/IoTDBSinkInsertTest.java
@@ -55,7 +55,7 @@ public class IoTDBSinkInsertTest {
tuple.put("values", "36.5");
ioTDBSink.invoke(tuple, null);
- verify(pool).insert(any(String.class), any(Long.class), any(List.class), any(List.class));
+ verify(pool).insertRecord(any(String.class), any(Long.class), any(List.class), any(List.class));
}
@Test
diff --git a/hadoop/src/test/java/org/apache/iotdb/hadoop/tsfile/TsFileTestHelper.java b/hadoop/src/test/java/org/apache/iotdb/hadoop/tsfile/TsFileTestHelper.java
index 663cfcc..3b349cd 100644
--- a/hadoop/src/test/java/org/apache/iotdb/hadoop/tsfile/TsFileTestHelper.java
+++ b/hadoop/src/test/java/org/apache/iotdb/hadoop/tsfile/TsFileTestHelper.java
@@ -18,12 +18,14 @@
*/
package org.apache.iotdb.hadoop.tsfile;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.write.TsFileWriter;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.Schema;
import org.slf4j.Logger;
@@ -53,47 +55,51 @@ public class TsFileTestHelper {
}
Schema schema = new Schema();
+ List<MeasurementSchema> schemaList = new ArrayList<>();
- // the number of rows to include in the row batch
+ // the number of rows to include in the tablet
int rowNum = 1000000;
- // the number of values to include in the row batch
+ // the number of values to include in the tablet
int sensorNum = 10;
// add timeseries into file schema (all with INT64 data type)
for (int i = 0; i < sensorNum; i++) {
+ MeasurementSchema measurementSchema = new MeasurementSchema("sensor_" + (i + 1),
+ TSDataType.INT64, TSEncoding.TS_2DIFF);
schema.registerTimeseries(new Path("device_1", "sensor_" + (i + 1)),
- new MeasurementSchema("sensor_" + (i + 1), TSDataType.INT64, TSEncoding.TS_2DIFF));
+ measurementSchema);
+ schemaList.add(measurementSchema);
}
// add timeseries into TSFileWriter
TsFileWriter tsFileWriter = new TsFileWriter(file, schema);
- // construct the row batch
- RowBatch rowBatch = schema.createRowBatch("device_1");
+ // construct the tablet
+ Tablet tablet = new Tablet("device_1", schemaList);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
long timestamp = 1;
long value = 1000000L;
for (int r = 0; r < rowNum; r++, value++) {
- int row = rowBatch.batchSize++;
+ int row = tablet.rowSize++;
timestamps[row] = timestamp++;
for (int i = 0; i < sensorNum; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = value;
}
- // write RowBatch to TsFile
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
+ // write Tablet to TsFile
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ tsFileWriter.write(tablet);
+ tablet.reset();
}
}
- // write RowBatch to TsFile
- if (rowBatch.batchSize != 0) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
+ // write Tablet to TsFile
+ if (tablet.rowSize != 0) {
+ tsFileWriter.write(tablet);
+ tablet.reset();
}
// close TsFile
diff --git a/hive-connector/src/test/java/org/apache/iotdb/hive/TsFileTestHelper.java b/hive-connector/src/test/java/org/apache/iotdb/hive/TsFileTestHelper.java
index a3ccca9..9ef22cb 100644
--- a/hive-connector/src/test/java/org/apache/iotdb/hive/TsFileTestHelper.java
+++ b/hive-connector/src/test/java/org/apache/iotdb/hive/TsFileTestHelper.java
@@ -21,13 +21,15 @@ package org.apache.iotdb.hive;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.write.TsFileWriter;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.schema.Schema;
import org.slf4j.Logger;
@@ -53,48 +55,51 @@ public class TsFileTestHelper {
}
Schema schema = new Schema();
+ List<MeasurementSchema> schemaList = new ArrayList<>();
- // the number of rows to include in the row batch
+ // the number of rows to include in the tablet
int rowNum = 1000000;
- // the number of values to include in the row batch
+ // the number of values to include in the tablet
int sensorNum = 10;
// add measurements into file schema (all with INT64 data type)
for (int i = 0; i < sensorNum; i++) {
+ MeasurementSchema measurementSchema = new MeasurementSchema("sensor_" + (i + 1),
+ TSDataType.INT64, TSEncoding.TS_2DIFF, CompressionType.UNCOMPRESSED);
schema.registerTimeseries(new Path("device_1", "sensor_" + (i + 1)),
- new MeasurementSchema("sensor_" + (i + 1), TSDataType.INT64, TSEncoding.TS_2DIFF,
- CompressionType.UNCOMPRESSED));
+ measurementSchema);
+ schemaList.add(measurementSchema);
}
// add measurements into TSFileWriter
TsFileWriter tsFileWriter = new TsFileWriter(file, schema);
- // construct the row batch
- RowBatch rowBatch = schema.createRowBatch("device_1");
+ // construct the tablet
+ Tablet tablet = new Tablet("device_1", schemaList);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
long timestamp = 1;
long value = 1000000L;
for (int r = 0; r < rowNum; r++, value++) {
- int row = rowBatch.batchSize++;
+ int row = tablet.rowSize++;
timestamps[row] = timestamp++;
for (int i = 0; i < sensorNum; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = value;
}
- // write RowBatch to TsFile
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
+ // write Tablet to TsFile
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ tsFileWriter.write(tablet);
+ tablet.reset();
}
}
- // write RowBatch to TsFile
- if (rowBatch.batchSize != 0) {
- tsFileWriter.write(rowBatch);
- rowBatch.reset();
+ // write Tablet to TsFile
+ if (tablet.rowSize != 0) {
+ tsFileWriter.write(tablet);
+ tablet.reset();
}
// close TsFile
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index 695d5cf..8fbf967 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -56,7 +56,7 @@ import org.apache.iotdb.db.exception.StorageGroupProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryFileManager;
@@ -263,20 +263,20 @@ public class StorageEngine implements IService {
*
* @return result of each row
*/
- public TSStatus[] insertBatch(BatchInsertPlan batchInsertPlan) throws StorageEngineException {
+ public TSStatus[] insertTablet(InsertTabletPlan insertTabletPlan) throws StorageEngineException {
StorageGroupProcessor storageGroupProcessor;
try {
- storageGroupProcessor = getProcessor(batchInsertPlan.getDeviceId());
+ storageGroupProcessor = getProcessor(insertTabletPlan.getDeviceId());
} catch (StorageEngineException e) {
logger.warn("get StorageGroupProcessor of device {} failed, because {}",
- batchInsertPlan.getDeviceId(),
+ insertTabletPlan.getDeviceId(),
e.getMessage(), e);
throw new StorageEngineException(e);
}
// TODO monitor: update statistics
try {
- return storageGroupProcessor.insertBatch(batchInsertPlan);
+ return storageGroupProcessor.insertTablet(insertTabletPlan);
} catch (WriteProcessException e) {
throw new StorageEngineException(e);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
index 7297c61..5292938 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/AbstractMemTable.java
@@ -29,7 +29,7 @@ import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.rescon.TVListAllocator;
import org.apache.iotdb.db.utils.CommonUtils;
@@ -104,11 +104,11 @@ public abstract class AbstractMemTable implements IMemTable {
}
@Override
- public void insertBatch(BatchInsertPlan batchInsertPlan, int start, int end)
+ public void insertTablet(InsertTabletPlan insertTabletPlan, int start, int end)
throws WriteProcessException {
try {
- write(batchInsertPlan, start, end);
- long recordSizeInByte = MemUtils.getRecordSize(batchInsertPlan, start, end);
+ write(insertTabletPlan, start, end);
+ long recordSizeInByte = MemUtils.getRecordSize(insertTabletPlan, start, end);
memSize += recordSizeInByte;
} catch (RuntimeException e) {
throw new WriteProcessException(e.getMessage());
@@ -124,12 +124,12 @@ public abstract class AbstractMemTable implements IMemTable {
}
@Override
- public void write(BatchInsertPlan batchInsertPlan, int start, int end) {
- for (int i = 0; i < batchInsertPlan.getMeasurements().length; i++) {
- IWritableMemChunk memSeries = createIfNotExistAndGet(batchInsertPlan.getDeviceId(),
- batchInsertPlan.getMeasurements()[i], batchInsertPlan.getSchemas()[i]);
- memSeries.write(batchInsertPlan.getTimes(), batchInsertPlan.getColumns()[i],
- batchInsertPlan.getDataTypes()[i], start, end);
+ public void write(InsertTabletPlan insertTabletPlan, int start, int end) {
+ for (int i = 0; i < insertTabletPlan.getMeasurements().length; i++) {
+ IWritableMemChunk memSeries = createIfNotExistAndGet(insertTabletPlan.getDeviceId(),
+ insertTabletPlan.getMeasurements()[i], insertTabletPlan.getSchemas()[i]);
+ memSeries.write(insertTabletPlan.getTimes(), insertTabletPlan.getColumns()[i],
+ insertTabletPlan.getDataTypes()[i], start, end);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
index fa67d79..c84d497 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/memtable/IMemTable.java
@@ -24,7 +24,7 @@ import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
@@ -44,7 +44,7 @@ public interface IMemTable {
void write(String deviceId, String measurement, MeasurementSchema schema,
long insertTime, Object objectValue);
- void write(BatchInsertPlan batchInsertPlan, int start, int end);
+ void write(InsertTabletPlan insertTabletPlan, int start, int end);
/**
* @return the number of points
@@ -61,7 +61,7 @@ public interface IMemTable {
/**
* [start, end)
*/
- void insertBatch(BatchInsertPlan batchInsertPlan, int start, int end)
+ void insertTablet(InsertTabletPlan insertTabletPlan, int start, int end)
throws WriteProcessException;
ReadOnlyMemChunk query(String deviceId, String measurement, TSDataType dataType,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 7c979ff..c3a6495 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -73,7 +73,7 @@ import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.metadata.mnode.LeafMNode;
import org.apache.iotdb.db.metadata.mnode.MNode;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
@@ -489,17 +489,17 @@ public class StorageGroupProcessor {
}
}
- public TSStatus[] insertBatch(BatchInsertPlan batchInsertPlan) throws WriteProcessException {
+ public TSStatus[] insertTablet(InsertTabletPlan insertTabletPlan) throws WriteProcessException {
writeLock();
try {
- TSStatus[] results = new TSStatus[batchInsertPlan.getRowCount()];
+ TSStatus[] results = new TSStatus[insertTabletPlan.getRowCount()];
/*
* assume that batch has been sorted by client
*/
int loc = 0;
- while (loc < batchInsertPlan.getRowCount()) {
- long currTime = batchInsertPlan.getTimes()[loc];
+ while (loc < insertTabletPlan.getRowCount()) {
+ long currTime = insertTabletPlan.getTimes()[loc];
// skip points that do not satisfy TTL
if (!checkTTL(currTime)) {
results[loc] = RpcUtils.getStatus(TSStatusCode.OUT_OF_TTL_ERROR,
@@ -510,34 +510,34 @@ public class StorageGroupProcessor {
}
}
// loc pointing at first legal position
- if (loc == batchInsertPlan.getRowCount()) {
+ if (loc == insertTabletPlan.getRowCount()) {
return results;
}
// before is first start point
int before = loc;
// before time partition
- long beforeTimePartition = StorageEngine.getTimePartition(batchInsertPlan.getTimes()[before]);
+ long beforeTimePartition = StorageEngine.getTimePartition(insertTabletPlan.getTimes()[before]);
// init map
long lastFlushTime = partitionLatestFlushedTimeForEachDevice.
computeIfAbsent(beforeTimePartition, id -> new HashMap<>()).
- computeIfAbsent(batchInsertPlan.getDeviceId(), id -> Long.MIN_VALUE);
+ computeIfAbsent(insertTabletPlan.getDeviceId(), id -> Long.MIN_VALUE);
// if is sequence
boolean isSequence = false;
- while (loc < batchInsertPlan.getRowCount()) {
- long time = batchInsertPlan.getTimes()[loc];
+ while (loc < insertTabletPlan.getRowCount()) {
+ long time = insertTabletPlan.getTimes()[loc];
long curTimePartition = StorageEngine.getTimePartition(time);
results[loc] = RpcUtils.SUCCESS_STATUS;
// start next partition
if (curTimePartition != beforeTimePartition) {
// insert last time partition
- insertBatchToTsFileProcessor(batchInsertPlan, before, loc, isSequence, results,
+ insertTabletToTsFileProcessor(insertTabletPlan, before, loc, isSequence, results,
beforeTimePartition);
// re initialize
before = loc;
beforeTimePartition = curTimePartition;
lastFlushTime = partitionLatestFlushedTimeForEachDevice.
computeIfAbsent(beforeTimePartition, id -> new HashMap<>()).
- computeIfAbsent(batchInsertPlan.getDeviceId(), id -> Long.MIN_VALUE);
+ computeIfAbsent(insertTabletPlan.getDeviceId(), id -> Long.MIN_VALUE);
isSequence = false;
}
// still in this partition
@@ -545,7 +545,7 @@ public class StorageGroupProcessor {
// judge if we should insert sequence
if (!isSequence && time > lastFlushTime) {
// insert into unsequence and then start sequence
- insertBatchToTsFileProcessor(batchInsertPlan, before, loc, false, results,
+ insertTabletToTsFileProcessor(insertTabletPlan, before, loc, false, results,
beforeTimePartition);
before = loc;
isSequence = true;
@@ -556,7 +556,7 @@ public class StorageGroupProcessor {
// do not forget last part
if (before < loc) {
- insertBatchToTsFileProcessor(batchInsertPlan, before, loc, isSequence, results,
+ insertTabletToTsFileProcessor(insertTabletPlan, before, loc, isSequence, results,
beforeTimePartition);
}
@@ -576,12 +576,12 @@ public class StorageGroupProcessor {
/**
* insert batch to tsfile processor thread-safety that the caller need to guarantee
*
- * @param batchInsertPlan batch insert plan
+ * @param insertTabletPlan insert a tablet of a device
* @param sequence whether is sequence
* @param results result array
* @param timePartitionId time partition id
*/
- private void insertBatchToTsFileProcessor(BatchInsertPlan batchInsertPlan,
+ private void insertTabletToTsFileProcessor(InsertTabletPlan insertTabletPlan,
int start, int end, boolean sequence, TSStatus[] results, long timePartitionId)
throws WriteProcessException {
// return when start >= end
@@ -599,7 +599,7 @@ public class StorageGroupProcessor {
}
try {
- tsFileProcessor.insertBatch(batchInsertPlan, start, end, results);
+ tsFileProcessor.insertTablet(insertTabletPlan, start, end, results);
} catch (WriteProcessException e) {
logger.error("insert to TsFileProcessor error ", e);
return;
@@ -608,14 +608,14 @@ public class StorageGroupProcessor {
latestTimeForEachDevice.computeIfAbsent(timePartitionId, t -> new HashMap<>());
// try to update the latest time of the device of this tsRecord
if (sequence && latestTimeForEachDevice.get(timePartitionId)
- .getOrDefault(batchInsertPlan.getDeviceId(), Long.MIN_VALUE)
- < batchInsertPlan.getTimes()[end - 1]) {
+ .getOrDefault(insertTabletPlan.getDeviceId(), Long.MIN_VALUE)
+ < insertTabletPlan.getTimes()[end - 1]) {
latestTimeForEachDevice.get(timePartitionId)
- .put(batchInsertPlan.getDeviceId(), batchInsertPlan.getTimes()[end - 1]);
+ .put(insertTabletPlan.getDeviceId(), insertTabletPlan.getTimes()[end - 1]);
}
long globalLatestFlushedTime = globalLatestFlushedTimeForEachDevice.getOrDefault(
- batchInsertPlan.getDeviceId(), Long.MIN_VALUE);
- tryToUpdateBatchInsertLastCache(batchInsertPlan, globalLatestFlushedTime);
+ insertTabletPlan.getDeviceId(), Long.MIN_VALUE);
+ tryToUpdateBatchInsertLastCache(insertTabletPlan, globalLatestFlushedTime);
// check memtable size and may async try to flush the work memtable
if (tsFileProcessor.shouldFlush()) {
@@ -623,7 +623,7 @@ public class StorageGroupProcessor {
}
}
- public void tryToUpdateBatchInsertLastCache(BatchInsertPlan plan, Long latestFlushedTime)
+ public void tryToUpdateBatchInsertLastCache(InsertTabletPlan plan, Long latestFlushedTime)
throws WriteProcessException {
try {
MNode node =
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index ae81a7d..5b2234e 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -47,7 +47,7 @@ import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.UpdateEndTi
import org.apache.iotdb.db.engine.version.VersionController;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.rescon.MemTablePool;
@@ -177,7 +177,7 @@ public class TsFileProcessor {
}
}
- public void insertBatch(BatchInsertPlan batchInsertPlan, int start, int end,
+ public void insertTablet(InsertTabletPlan insertTabletPlan, int start, int end,
TSStatus[] results) throws WriteProcessException {
if (workMemTable == null) {
@@ -186,11 +186,11 @@ public class TsFileProcessor {
// insert insertPlan to the work memtable
try {
- workMemTable.insertBatch(batchInsertPlan, start, end);
+ workMemTable.insertTablet(insertTabletPlan, start, end);
if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
- batchInsertPlan.setStart(start);
- batchInsertPlan.setEnd(end);
- getLogNode().write(batchInsertPlan);
+ insertTabletPlan.setStart(start);
+ insertTabletPlan.setEnd(end);
+ getLogNode().write(insertTabletPlan);
}
} catch (Exception e) {
for (int i = start; i < end; i++) {
@@ -204,13 +204,14 @@ public class TsFileProcessor {
}
tsFileResource
- .updateStartTime(batchInsertPlan.getDeviceId(), batchInsertPlan.getTimes()[start]);
+ .updateStartTime(insertTabletPlan.getDeviceId(), insertTabletPlan.getTimes()[start]);
//for sequence tsfile, we update the endTime only when the file is prepared to be closed.
//for unsequence tsfile, we have to update the endTime for each insertion.
if (!sequence) {
tsFileResource
- .updateEndTime(batchInsertPlan.getDeviceId(), batchInsertPlan.getTimes()[end - 1]);
+ .updateEndTime(
+ insertTabletPlan.getDeviceId(), insertTabletPlan.getTimes()[end - 1]);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
index cf09d6a..addf757 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/IPlanExecutor.java
@@ -24,7 +24,7 @@ import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
@@ -91,5 +91,5 @@ public interface IPlanExecutor {
*
* @return result of each row
*/
- TSStatus[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException;
+ TSStatus[] insertTablet(InsertTabletPlan insertTabletPlan) throws QueryProcessException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
index b275838..908d3e5 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/executor/PlanExecutor.java
@@ -773,12 +773,12 @@ public class PlanExecutor implements IPlanExecutor {
}
@Override
- public TSStatus[] insertBatch(BatchInsertPlan batchInsertPlan) throws QueryProcessException {
+ public TSStatus[] insertTablet(InsertTabletPlan insertTabletPlan) throws QueryProcessException {
try {
- String[] measurementList = batchInsertPlan.getMeasurements();
- String deviceId = batchInsertPlan.getDeviceId();
+ String[] measurementList = insertTabletPlan.getMeasurements();
+ String deviceId = insertTabletPlan.getDeviceId();
MNode node = mManager.getDeviceNodeWithAutoCreateStorageGroup(deviceId);
- TSDataType[] dataTypes = batchInsertPlan.getDataTypes();
+ TSDataType[] dataTypes = insertTabletPlan.getDataTypes();
IoTDBConfig conf = IoTDBDescriptor.getInstance().getConfig();
MeasurementSchema[] schemas = new MeasurementSchema[measurementList.length];
@@ -797,16 +797,16 @@ public class PlanExecutor implements IPlanExecutor {
LeafMNode measurementNode = (LeafMNode) node.getChild(measurementList[i]);
// check data type
- if (measurementNode.getSchema().getType() != batchInsertPlan.getDataTypes()[i]) {
+ if (measurementNode.getSchema().getType() != insertTabletPlan.getDataTypes()[i]) {
throw new QueryProcessException(String
.format("Datatype mismatch, Insert measurement %s type %s, metadata tree type %s",
- measurementList[i], batchInsertPlan.getDataTypes()[i],
+ measurementList[i], insertTabletPlan.getDataTypes()[i],
measurementNode.getSchema().getType()));
}
schemas[i] = measurementNode.getSchema();
}
- batchInsertPlan.setSchemas(schemas);
- return StorageEngine.getInstance().insertBatch(batchInsertPlan);
+ insertTabletPlan.setSchemas(schemas);
+ return StorageEngine.getInstance().insertTablet(insertTabletPlan);
} catch (StorageEngineException | MetadataException e) {
throw new QueryProcessException(e);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index f2d3fc0..0ad6470 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -24,7 +24,7 @@ import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import org.apache.iotdb.db.qp.logical.Operator;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
@@ -149,7 +149,7 @@ public abstract class PhysicalPlan {
plan.deserializeFrom(buffer);
break;
case BATCHINSERT:
- plan = new BatchInsertPlan();
+ plan = new InsertTabletPlan();
plan.deserializeFrom(buffer);
break;
case SET_STORAGE_GROUP:
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
rename to server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
index 6878af6..a26ac3b 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
@@ -43,7 +43,7 @@ import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsInt;
import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsLong;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-public class BatchInsertPlan extends PhysicalPlan {
+public class InsertTabletPlan extends PhysicalPlan {
private static final String DATATYPE_UNSUPPORTED = "Data type %s is not supported.";
@@ -67,22 +67,22 @@ public class BatchInsertPlan extends PhysicalPlan {
private int start;
private int end;
- public BatchInsertPlan() {
+ public InsertTabletPlan() {
super(false, OperatorType.BATCHINSERT);
}
- public BatchInsertPlan(String deviceId, List<String> measurements) {
+ public InsertTabletPlan(String deviceId, List<String> measurements) {
super(false, OperatorType.BATCHINSERT);
this.deviceId = deviceId;
setMeasurements(measurements);
}
- public BatchInsertPlan(String deviceId, String[] measurements) {
+ public InsertTabletPlan(String deviceId, String[] measurements) {
super(false, OperatorType.BATCHINSERT);
this.deviceId = deviceId;
setMeasurements(measurements);
}
- public BatchInsertPlan(String deviceId, String[] measurements, List<Integer> dataTypes) {
+ public InsertTabletPlan(String deviceId, String[] measurements, List<Integer> dataTypes) {
super(false, OperatorType.BATCHINSERT);
this.deviceId = deviceId;
this.measurements = measurements;
diff --git a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 1b98a70..575e10c 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -1039,7 +1039,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
@Override
- public TSExecuteBatchStatementResp insertRowInBatch(TSInsertInBatchReq req) {
+ public TSExecuteBatchStatementResp insertRecords(TSInsertRecordsReq req) {
TSExecuteBatchStatementResp resp = new TSExecuteBatchStatementResp();
if (!checkLogin(req.getSessionId())) {
logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
@@ -1065,25 +1065,25 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
@Override
- public TSExecuteBatchStatementResp testInsertBatch(TSBatchInsertionReq req) {
+ public TSExecuteBatchStatementResp testInsertTablet(TSInsertTabletReq req) {
logger.debug("Test insert batch request receive.");
return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.SUCCESS_STATUS);
}
@Override
- public TSStatus testInsertRow(TSInsertReq req) {
+ public TSStatus testInsertRecord(TSInsertRecordReq req) {
logger.debug("Test insert row request receive.");
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
}
@Override
- public TSExecuteBatchStatementResp testInsertRowInBatch(TSInsertInBatchReq req) {
+ public TSExecuteBatchStatementResp testInsertRecords(TSInsertRecordsReq req) {
logger.debug("Test insert row in batch request receive.");
return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.SUCCESS_STATUS);
}
@Override
- public TSStatus insert(TSInsertReq req) {
+ public TSStatus insertRecord(TSInsertRecordReq req) {
try {
if (!checkLogin(req.getSessionId())) {
logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
@@ -1130,7 +1130,7 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
@Override
- public TSExecuteBatchStatementResp insertBatch(TSBatchInsertionReq req) {
+ public TSExecuteBatchStatementResp insertTablet(TSInsertTabletReq req) {
long t1 = System.currentTimeMillis();
try {
if (!checkLogin(req.getSessionId())) {
@@ -1138,20 +1138,20 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.NOT_LOGIN_ERROR);
}
- BatchInsertPlan batchInsertPlan = new BatchInsertPlan(req.deviceId, req.measurements);
- batchInsertPlan.setTimes(QueryDataSetUtils.readTimesFromBuffer(req.timestamps, req.size));
- batchInsertPlan.setColumns(
+ InsertTabletPlan insertTabletPlan = new InsertTabletPlan(req.deviceId, req.measurements);
+ insertTabletPlan.setTimes(QueryDataSetUtils.readTimesFromBuffer(req.timestamps, req.size));
+ insertTabletPlan.setColumns(
QueryDataSetUtils.readValuesFromBuffer(
req.values, req.types, req.measurements.size(), req.size));
- batchInsertPlan.setRowCount(req.size);
- batchInsertPlan.setDataTypes(req.types);
+ insertTabletPlan.setRowCount(req.size);
+ insertTabletPlan.setDataTypes(req.types);
boolean isAllSuccessful = true;
- TSStatus status = checkAuthority(batchInsertPlan, req.getSessionId());
+ TSStatus status = checkAuthority(insertTabletPlan, req.getSessionId());
if (status != null) {
return RpcUtils.getTSBatchExecuteStatementResp(status);
}
- TSStatus[] tsStatusArray = executor.insertBatch(batchInsertPlan);
+ TSStatus[] tsStatusArray = executor.insertTablet(insertTabletPlan);
for (TSStatus tsStatus : tsStatusArray) {
if (tsStatus.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
@@ -1162,11 +1162,11 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
if (isAllSuccessful) {
if (logger.isDebugEnabled()) {
- logger.debug("Insert one RowBatch successfully");
+ logger.debug("Insert one Tablet successfully");
}
return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.SUCCESS_STATUS);
} else {
- logger.debug("Insert one RowBatch failed!");
+ logger.debug("Insert one Tablet failed!");
return RpcUtils.getTSBatchExecuteStatementResp(Arrays.asList(tsStatusArray));
}
} catch (Exception e) {
@@ -1179,6 +1179,60 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext {
}
@Override
+ public TSExecuteBatchStatementResp insertTablets(TSInsertTabletsReq req) {
+ long t1 = System.currentTimeMillis();
+ try {
+ if (!checkLogin(req.getSessionId())) {
+ logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
+ return RpcUtils.getTSBatchExecuteStatementResp(TSStatusCode.NOT_LOGIN_ERROR);
+ }
+
+ List<TSStatus> statusList = new ArrayList<>();
+ for (int i = 0; i < req.deviceIds.size(); i++) {
+ InsertTabletPlan insertTabletPlan = new InsertTabletPlan(req.deviceIds.get(i),
+ req.measurementsList.get(i));
+ insertTabletPlan.setTimes(
+ QueryDataSetUtils.readTimesFromBuffer(req.timestampsList.get(i), req.sizeList.get(i)));
+ insertTabletPlan.setColumns(
+ QueryDataSetUtils.readValuesFromBuffer(
+ req.valuesList.get(i), req.typesList.get(i), req.measurementsList.get(i).size(), req.sizeList.get(i)));
+ insertTabletPlan.setRowCount(req.sizeList.get(i));
+ insertTabletPlan.setDataTypes(req.typesList.get(i));
+
+ boolean isCurrentTabletSuccessful = true;
+ TSStatus status = checkAuthority(insertTabletPlan, req.getSessionId());
+ if (status != null) {
+ statusList.add(status);
+ continue;
+ }
+ TSStatus[] tsStatusArray = executor.insertTablet(insertTabletPlan);
+ TSStatus failed = RpcUtils.getStatus(TSStatusCode.INTERNAL_SERVER_ERROR);
+
+ for (TSStatus tsStatus : tsStatusArray) {
+ if (tsStatus.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ isCurrentTabletSuccessful = false;
+ failed = tsStatus;
+ break;
+ }
+ }
+
+ if (isCurrentTabletSuccessful) {
+ statusList.add(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));
+ } else {
+ statusList.add(failed);
+ }
+ }
+ return RpcUtils.getTSBatchExecuteStatementResp(statusList);
+ } catch (Exception e) {
+ logger.info("{}: error occurs when insertTablets", IoTDBConstant.GLOBAL_DB_NAME, e);
+ return RpcUtils
+ .getTSBatchExecuteStatementResp(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());
+ } finally {
+ Measurement.INSTANCE.addOperationLatency(Operation.EXECUTE_RPC_BATCH_INSERT, t1);
+ }
+ }
+
+ @Override
public TSStatus setStorageGroup(long sessionId, String storageGroup) {
if (!checkLogin(sessionId)) {
logger.info(INFO_NOT_LOGIN, IoTDBConstant.GLOBAL_DB_NAME);
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
index 7946ba6..6f5b494 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/MemUtils.java
@@ -19,8 +19,8 @@
package org.apache.iotdb.db.utils;
import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.engine.cache.RamUsageEstimator;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.write.record.TSRecord;
@@ -70,13 +70,13 @@ public class MemUtils {
.sizeOf(value.getValues());
}
- public static long getRecordSize(BatchInsertPlan batchInsertPlan, int start, int end) {
+ public static long getRecordSize(InsertTabletPlan insertTabletPlan, int start, int end) {
if (start >= end) {
return 0L;
}
long memSize = 0;
- for (int i = 0; i < batchInsertPlan.getMeasurements().length; i++) {
- switch (batchInsertPlan.getDataTypes()[i]) {
+ for (int i = 0; i < insertTabletPlan.getMeasurements().length; i++) {
+ switch (insertTabletPlan.getDataTypes()[i]) {
case INT32:
memSize += (end - start) * (8L + 4L); break;
case INT64:
@@ -90,7 +90,7 @@ public class MemUtils {
case TEXT:
memSize += (end - start) * 8L;
for (int j = start; j < end; j++) {
- memSize += getBinarySize(((Binary[]) batchInsertPlan.getColumns()[i])[j]);
+ memSize += getBinarySize(((Binary[]) insertTabletPlan.getColumns()[i])[j]);
}
break;
default:
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
index e6c9960..0f32e8e 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
@@ -34,7 +34,7 @@ import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
@@ -96,8 +96,8 @@ public class LogReplayer {
replayDelete((DeletePlan) plan);
} else if (plan instanceof UpdatePlan) {
replayUpdate((UpdatePlan) plan);
- } else if (plan instanceof BatchInsertPlan) {
- replayBatchInsert((BatchInsertPlan) plan);
+ } else if (plan instanceof InsertTabletPlan) {
+ replayBatchInsert((InsertTabletPlan) plan);
}
}
} catch (IOException | WriteProcessException | QueryProcessException e) {
@@ -123,32 +123,33 @@ public class LogReplayer {
}
}
- private void replayBatchInsert(BatchInsertPlan batchInsertPlan)
+ private void replayBatchInsert(InsertTabletPlan insertTabletPlan)
throws WriteProcessException, QueryProcessException {
if (currentTsFileResource != null) {
// the last chunk group may contain the same data with the logs, ignore such logs in seq file
- Long lastEndTime = currentTsFileResource.getEndTimeMap().get(batchInsertPlan.getDeviceId());
- if (lastEndTime != null && lastEndTime >= batchInsertPlan.getMinTime() &&
+ Long lastEndTime = currentTsFileResource.getEndTimeMap().get(insertTabletPlan.getDeviceId());
+ if (lastEndTime != null && lastEndTime >= insertTabletPlan.getMinTime() &&
!acceptDuplication) {
return;
}
- Long startTime = tempStartTimeMap.get(batchInsertPlan.getDeviceId());
- if (startTime == null || startTime > batchInsertPlan.getMinTime()) {
- tempStartTimeMap.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMinTime());
+ Long startTime = tempStartTimeMap.get(insertTabletPlan.getDeviceId());
+ if (startTime == null || startTime > insertTabletPlan.getMinTime()) {
+ tempStartTimeMap.put(insertTabletPlan.getDeviceId(), insertTabletPlan.getMinTime());
}
- Long endTime = tempEndTimeMap.get(batchInsertPlan.getDeviceId());
- if (endTime == null || endTime < batchInsertPlan.getMaxTime()) {
- tempEndTimeMap.put(batchInsertPlan.getDeviceId(), batchInsertPlan.getMaxTime());
+ Long endTime = tempEndTimeMap.get(insertTabletPlan.getDeviceId());
+ if (endTime == null || endTime < insertTabletPlan.getMaxTime()) {
+ tempEndTimeMap.put(insertTabletPlan.getDeviceId(), insertTabletPlan.getMaxTime());
}
}
MeasurementSchema[] schemas;
try {
- schemas = MManager.getInstance().getSchemas(batchInsertPlan.getDeviceId(), batchInsertPlan.getMeasurements());
+ schemas = MManager.getInstance().getSchemas(insertTabletPlan.getDeviceId(), insertTabletPlan
+ .getMeasurements());
} catch (MetadataException e) {
throw new QueryProcessException(e);
}
- batchInsertPlan.setSchemas(schemas);
- recoverMemTable.insertBatch(batchInsertPlan, 0, batchInsertPlan.getRowCount());
+ insertTabletPlan.setSchemas(schemas);
+ recoverMemTable.insertTablet(insertTabletPlan, 0, insertTabletPlan.getRowCount());
}
private void replayInsert(InsertPlan insertPlan) {
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
index 4e2aff7..6e11479 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
@@ -25,11 +25,10 @@ import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
-import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.exception.StorageGroupProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
-import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.utils.EnvironmentUtils;
@@ -152,7 +151,7 @@ public class StorageGroupProcessorTest {
}
@Test
- public void testIoTDBRowBatchWriteAndSyncClose()
+ public void testIoTDBTabletWriteAndSyncClose()
throws WriteProcessException, QueryProcessException {
String[] measurements = new String[2];
@@ -166,9 +165,9 @@ public class StorageGroupProcessorTest {
schemas[0] = new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.PLAIN);
schemas[1] = new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN);
- BatchInsertPlan batchInsertPlan1 = new BatchInsertPlan("root.vehicle.d0", measurements,
+ InsertTabletPlan insertTabletPlan1 = new InsertTabletPlan("root.vehicle.d0", measurements,
dataTypes);
- batchInsertPlan1.setSchemas(schemas);
+ insertTabletPlan1.setSchemas(schemas);
long[] times = new long[100];
Object[] columns = new Object[2];
@@ -180,27 +179,27 @@ public class StorageGroupProcessorTest {
((int[]) columns[0])[r] = 1;
((long[]) columns[1])[r] = 1;
}
- batchInsertPlan1.setTimes(times);
- batchInsertPlan1.setColumns(columns);
- batchInsertPlan1.setRowCount(times.length);
+ insertTabletPlan1.setTimes(times);
+ insertTabletPlan1.setColumns(columns);
+ insertTabletPlan1.setRowCount(times.length);
- processor.insertBatch(batchInsertPlan1);
+ processor.insertTablet(insertTabletPlan1);
processor.asyncCloseAllWorkingTsFileProcessors();
- BatchInsertPlan batchInsertPlan2 = new BatchInsertPlan("root.vehicle.d0", measurements,
+ InsertTabletPlan insertTabletPlan2 = new InsertTabletPlan("root.vehicle.d0", measurements,
dataTypes);
- batchInsertPlan2.setSchemas(schemas);
+ insertTabletPlan2.setSchemas(schemas);
for (int r = 50; r < 149; r++) {
times[r - 50] = r;
((int[]) columns[0])[r - 50] = 1;
((long[]) columns[1])[r - 50] = 1;
}
- batchInsertPlan2.setTimes(times);
- batchInsertPlan2.setColumns(columns);
- batchInsertPlan2.setRowCount(times.length);
+ insertTabletPlan2.setTimes(times);
+ insertTabletPlan2.setColumns(columns);
+ insertTabletPlan2.setRowCount(times.length);
- processor.insertBatch(batchInsertPlan2);
+ processor.insertTablet(insertTabletPlan2);
processor.asyncCloseAllWorkingTsFileProcessors();
processor.syncCloseAllWorkingTsFileProcessors();
diff --git a/service-rpc/rpc-changelist.md b/service-rpc/rpc-changelist.md
index 5a15ea8..b2e528f 100644
--- a/service-rpc/rpc-changelist.md
+++ b/service-rpc/rpc-changelist.md
@@ -19,10 +19,42 @@
-->
-# 0.8.0 (version-0) -> version-1
-Last Updated on October 27th, 2019 by Lei Rui.
+# 0.10.0 (version-1) -> version-2
+
+Last Updated on 2020-4-24 by Jialin Qiao.
+
+
+## 1. Delete Old
+
+| Latest Changes | Related Committers |
+| ---------------------------------- | ------------------ |
+| Remove TS_SessionHandle,TSHandleIdentifier | Tian Jiang |
+| Remove TSStatus,TSExecuteInsertRowInBatchResp | Jialin Qiao|
+
+## 2. Add New
+
+| Latest Changes | Related Committers |
+| ------------------------------------------------------------ | ---------------------------------- |
+| Add parameter sessionId in getTimeZone, getProperties, setStorageGroup, createTimeseries... | Tian Jiang|
+| Add struct TSQueryNonAlignDataSet | Haonan Hou|
+| Add struct TSInsertTabletsReq | Jialin Qiao|
+| Add method insertTablets | Jialin Qiao|
+
+## 3. Update
+
+| Latest Changes | Related Committers |
+| ------------------------------------------------------------ | ---------------------- |
+| Replace TS_SessionHandles with SessionIds, TSOperationHandle with queryIds | Tian Jiang |
+| Add optional TSQueryNonAlignDataSet in TSExecuteStatementResp, TSFetchResultsResp and required bool isAlign in TSFetchResultsReq | Haonan Hou |
+| Rename TSStatusType to TSStatus | Jialin Qiao |
+| Remove sessionId in TSExecuteBatchStatementResp | Jialin Qiao |
+| Rename insertRows to insertReords, insert to insertRecord, insertBatch to insertTablet | Jialin Qiao |
+
+
+# 0.8.0 (version-0) -> version-1
+Last Updated on 2019-10-27 by Lei Rui.
## 1. Delete Old
@@ -40,8 +72,6 @@ Last Updated on October 27th, 2019 by Lei Rui.
| Remove optional i32 nodeLevel in TSFetchMetadataReq | Genius_pig |
-
-
## 2. Add New
| Latest Changes | Related Committers |
@@ -91,33 +121,3 @@ Last Updated on October 27th, 2019 by Lei Rui.
| Add required i64 statementId in TSExecuteStatementReq | Yuan Tian |
| Add required binary time, required list<binary> valueList, required list<binary> bitmapList and remove required binary values, required i32 rowCount in TSQueryDataSet| Yuan Tian |
| Add optional i32 fetchSize in TSExecuteStatementReq,<br />Add optional TSQueryDataSet in TSExecuteStatementResp| liutaohua |
-| Add optional map<string, string> props, optional map<string, string> tags, optional map<string, string> attributes and optional string aliasPath in TSCreateTimeseriesReq | Yuan Tian |
-
-
-# 0.10.0 (version-1) -> version-2
-
-Last Updated on November 12th, 2019 by Tian Jiang.
-
-
-## 1. Delete Old
-
-| Latest Changes | Related Committers |
-| ---------------------------------- | ------------------ |
-| Remove TS_SessionHandle,TSHandleIdentifier | Tian Jiang |
-| Remove TSStatus,TSExecuteInsertRowInBatchResp | Jialin Qiao|
-
-## 2. Add New
-
-| Latest Changes | Related Committers |
-| ------------------------------------------------------------ | ---------------------------------- |
-| Add parameter sessionId in getTimeZone, getProperties, setStorageGroup, createTimeseries... | Tian Jiang|
-| Add struct TSQueryNonAlignDataSet | Haonan Hou|
-
-## 3. Update
-
-| Latest Changes | Related Committers |
-| ------------------------------------------------------------ | ---------------------- |
-| Replace TS_SessionHandles with SessionIds, TSOperationHandle with queryIds | Tian Jiang |
-| Add optional TSQueryNonAlignDataSet in TSExecuteStatementResp, TSFetchResultsResp and required bool isAlign in TSFetchResultsReq | Haonan Hou |
-| Rename TSStatusType to TSStatus | Jialin Qiao |
-| Remove sessionId in TSExecuteBatchStatementResp | Jialin Qiao |
\ No newline at end of file
diff --git a/service-rpc/src/main/thrift/rpc.thrift b/service-rpc/src/main/thrift/rpc.thrift
index a0faa41..7c8e026 100644
--- a/service-rpc/src/main/thrift/rpc.thrift
+++ b/service-rpc/src/main/thrift/rpc.thrift
@@ -165,7 +165,7 @@ struct TSSetTimeZoneReq {
}
// for session
-struct TSInsertReq {
+struct TSInsertRecordReq {
1: required i64 sessionId
2: required string deviceId
3: required list<string> measurements
@@ -173,7 +173,7 @@ struct TSInsertReq {
5: required i64 timestamp
}
-struct TSBatchInsertionReq {
+struct TSInsertTabletReq {
1: required i64 sessionId
2: required string deviceId
3: required list<string> measurements
@@ -183,7 +183,17 @@ struct TSBatchInsertionReq {
7: required i32 size
}
-struct TSInsertInBatchReq {
+struct TSInsertTabletsReq {
+ 1: required i64 sessionId
+ 2: required list<string> deviceIds
+ 3: required list<list<string>> measurementsList
+ 4: required list<binary> valuesList
+ 5: required list<binary> timestampsList
+ 6: required list<list<i32>> typesList
+ 7: required list<i32> sizeList
+}
+
+struct TSInsertRecordsReq {
1: required i64 sessionId
2: required list<string> deviceIds
3: required list<list<string>> measurementsList
@@ -281,17 +291,19 @@ service TSIService {
TSStatus deleteStorageGroups(1:i64 sessionId, 2:list<string> storageGroup);
- TSStatus insert(1:TSInsertReq req);
+ TSStatus insertRecord(1:TSInsertRecordReq req);
+
+ TSExecuteBatchStatementResp insertTablet(1:TSInsertTabletReq req);
- TSExecuteBatchStatementResp insertBatch(1:TSBatchInsertionReq req);
+ TSExecuteBatchStatementResp insertTablets(1:TSInsertTabletsReq req);
- TSExecuteBatchStatementResp insertRowInBatch(1:TSInsertInBatchReq req);
+ TSExecuteBatchStatementResp insertRecords(1:TSInsertRecordsReq req);
- TSExecuteBatchStatementResp testInsertBatch(1:TSBatchInsertionReq req);
+ TSExecuteBatchStatementResp testInsertTablet(1:TSInsertTabletReq req);
- TSStatus testInsertRow(1:TSInsertReq req);
+ TSStatus testInsertRecord(1:TSInsertRecordReq req);
- TSExecuteBatchStatementResp testInsertRowInBatch(1:TSInsertInBatchReq req);
+ TSExecuteBatchStatementResp testInsertRecords(1:TSInsertRecordsReq req);
TSStatus deleteData(1:TSDeleteDataReq req);
diff --git a/session/src/main/java/org/apache/iotdb/session/Session.java b/session/src/main/java/org/apache/iotdb/session/Session.java
index 8ea83f6..0407093 100644
--- a/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -28,7 +28,7 @@ import org.apache.iotdb.rpc.BatchExecutionException;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.service.rpc.thrift.TSBatchInsertionReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertTabletReq;
import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateMultiTimeseriesReq;
import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq;
@@ -37,8 +37,9 @@ import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq;
import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
import org.apache.iotdb.service.rpc.thrift.TSGetTimeZoneResp;
import org.apache.iotdb.service.rpc.thrift.TSIService;
-import org.apache.iotdb.service.rpc.thrift.TSInsertInBatchReq;
-import org.apache.iotdb.service.rpc.thrift.TSInsertReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq;
+import org.apache.iotdb.service.rpc.thrift.TSInsertTabletsReq;
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
@@ -49,7 +50,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.utils.Binary;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
@@ -185,182 +186,166 @@ public class Session {
}
/**
- * check whether the batch has been sorted
+ * insert data in one row, if you want to improve your performance, please use insertRecords method
+ * or insertTablet method
*
- * @return whether the batch has been sorted
+ * @see Session#insertRecords(List, List, List, List)
+ * @see Session#insertTablet(Tablet)
*/
- private boolean checkSorted(RowBatch rowBatch) {
- for (int i = 1; i < rowBatch.batchSize; i++) {
- if (rowBatch.timestamps[i] < rowBatch.timestamps[i - 1]) {
- return false;
- }
+ public void insertRecord(String deviceId, long time, List<String> measurements,
+ Object... values) throws IoTDBConnectionException, StatementExecutionException {
+ List<String> stringValues = new ArrayList<>();
+ for (Object o : values) {
+ stringValues.add(o.toString());
}
- return true;
+ insertRecord(deviceId, time, measurements, stringValues);
}
/**
- * use batch interface to insert sorted data
+ * insert data in one row, if you want to improve your performance, please use insertRecords method
+ * or insertTablet method
*
- * @param rowBatch data batch
+ * @see Session#insertRecords(List, List, List, List)
+ * @see Session#insertTablet(Tablet)
*/
- private void insertSortedBatchIntern(RowBatch rowBatch)
- throws IoTDBConnectionException, BatchExecutionException {
- TSBatchInsertionReq request = new TSBatchInsertionReq();
+ public void insertRecord(String deviceId, long time, List<String> measurements,
+ List<String> values) throws IoTDBConnectionException, StatementExecutionException {
+ TSInsertRecordReq request = new TSInsertRecordReq();
request.setSessionId(sessionId);
- request.deviceId = rowBatch.deviceId;
- for (MeasurementSchema measurementSchema : rowBatch.getSchemas()) {
- request.addToMeasurements(measurementSchema.getMeasurementId());
- request.addToTypes(measurementSchema.getType().ordinal());
- }
- request.setTimestamps(SessionUtils.getTimeBuffer(rowBatch));
- request.setValues(SessionUtils.getValueBuffer(rowBatch));
- request.setSize(rowBatch.batchSize);
+ request.setDeviceId(deviceId);
+ request.setTimestamp(time);
+ request.setMeasurements(measurements);
+ request.setValues(values);
try {
- RpcUtils.verifySuccess(client.insertBatch(request).statusList);
+ RpcUtils.verifySuccess(client.insertRecord(request));
} catch (TException e) {
throw new IoTDBConnectionException(e);
}
}
+
/**
- * use batch interface to insert sorted data times in row batch must be sorted before!
+ * insert the data of a device. For each timestamp, the number of measurements is the same.
+ *
+ * a Tablet example:
+ *
+ * device1
+ * time s1, s2, s3
+ * 1, 1, 1, 1
+ * 2, 2, 2, 2
+ * 3, 3, 3, 3
*
- * @param rowBatch data batch
+ * times in Tablet may be not in ascending order
+ *
+ * @param tablet data batch
*/
- public void insertSortedBatch(RowBatch rowBatch)
+ public void insertTablet(Tablet tablet)
throws BatchExecutionException, IoTDBConnectionException {
- if (!checkSorted(rowBatch)) {
- throw new BatchExecutionException(
- "Row batch has't been sorted when calling insertSortedBatch");
- }
- insertSortedBatchIntern(rowBatch);
+ insertTablet(tablet, false);
}
/**
- * use batch interface to insert data in multiple device
+ * insert a Tablet
*
- * @param rowBatchMap data batch in multiple device
+ * @param tablet data batch
+ * @param sorted whether times in Tablet are in ascending order
*/
- public void insertMultipleDeviceBatch
- (Map<String, RowBatch> rowBatchMap) throws IoTDBConnectionException, BatchExecutionException {
- for (Map.Entry<String, RowBatch> dataInOneDevice : rowBatchMap.entrySet()) {
- sortRowBatch(dataInOneDevice.getValue());
- insertBatch(dataInOneDevice.getValue());
+ public void insertTablet(Tablet tablet, boolean sorted)
+ throws IoTDBConnectionException, BatchExecutionException {
+ if (sorted) {
+ if (!checkSorted(tablet)) {
+ throw new BatchExecutionException("Times in Tablet are not in ascending order");
+ }
+ } else {
+ sortTablet(tablet);
+ }
+
+ TSInsertTabletReq request = new TSInsertTabletReq();
+ request.setSessionId(sessionId);
+ request.deviceId = tablet.deviceId;
+ for (MeasurementSchema measurementSchema : tablet.getSchemas()) {
+ request.addToMeasurements(measurementSchema.getMeasurementId());
+ request.addToTypes(measurementSchema.getType().ordinal());
+ }
+ request.setTimestamps(SessionUtils.getTimeBuffer(tablet));
+ request.setValues(SessionUtils.getValueBuffer(tablet));
+ request.setSize(tablet.rowSize);
+
+ try {
+ RpcUtils.verifySuccess(client.insertTablet(request).statusList);
+ } catch (TException e) {
+ throw new IoTDBConnectionException(e);
}
}
/**
- * use batch interface to insert sorted data in multiple device times in row batch must be sorted
- * before!
+ * insert the data of several deivces.
+ * Given a deivce, for each timestamp, the number of measurements is the same.
+ *
+ * Times in each Tablet may not be in ascending order
*
- * @param rowBatchMap data batch in multiple device
+ * @param tablets data batch in multiple device
*/
- public void insertMultipleDeviceSortedBatch
- (Map<String, RowBatch> rowBatchMap) throws IoTDBConnectionException, BatchExecutionException {
- for (Map.Entry<String, RowBatch> dataInOneDevice : rowBatchMap.entrySet()) {
- checkSorted(dataInOneDevice.getValue());
- insertSortedBatchIntern(dataInOneDevice.getValue());
- }
+ public void insertTablets(Map<String, Tablet> tablets)
+ throws IoTDBConnectionException, BatchExecutionException {
+ insertTablets(tablets, false);
}
/**
- * use batch interface to insert data
+ * insert the data of several devices.
+ * Given a device, for each timestamp, the number of measurements is the same.
*
- * @param rowBatch data batch
+ * @param tablets data batch in multiple device
+ * @param sorted whether times in each Tablet are in ascending order
*/
- public void insertBatch(RowBatch rowBatch)
+ public void insertTablets(Map<String, Tablet> tablets, boolean sorted)
throws IoTDBConnectionException, BatchExecutionException {
- sortRowBatch(rowBatch);
+ TSInsertTabletsReq request = new TSInsertTabletsReq();
+ request.setSessionId(sessionId);
- insertSortedBatchIntern(rowBatch);
- }
+ for (Tablet tablet : tablets.values()) {
+ if (sorted) {
+ if (!checkSorted(tablet)) {
+ throw new BatchExecutionException("Times in Tablet are not in ascending order");
+ }
+ } else {
+ sortTablet(tablet);
+ }
- private void sortRowBatch(RowBatch rowBatch) {
- /*
- * following part of code sort the batch data by time,
- * so we can insert continuous data in value list to get a better performance
- */
- // sort to get index, and use index to sort value list
- Integer[] index = new Integer[rowBatch.batchSize];
- for (int i = 0; i < rowBatch.batchSize; i++) {
- index[i] = i;
- }
- Arrays.sort(index, Comparator.comparingLong(o -> rowBatch.timestamps[o]));
- Arrays.sort(rowBatch.timestamps, 0, rowBatch.batchSize);
- for (int i = 0; i < rowBatch.getSchemas().size(); i++) {
- rowBatch.values[i] =
- sortList(rowBatch.values[i], rowBatch.getSchemas().get(i).getType(), index);
- }
- }
+ request.addToDeviceIds(tablet.deviceId);
+ List<String> measurements = new ArrayList<>();
+ List<Integer> dataTypes = new ArrayList<>();
+ for (MeasurementSchema measurementSchema : tablet.getSchemas()) {
+ measurements.add(measurementSchema.getMeasurementId());
+ dataTypes.add(measurementSchema.getType().ordinal());
+ }
+ request.addToMeasurementsList(measurements);
+ request.addToTypesList(dataTypes);
+ request.addToTimestampsList(SessionUtils.getTimeBuffer(tablet));
+ request.addToValuesList(SessionUtils.getValueBuffer(tablet));
+ request.addToSizeList(tablet.rowSize);
- /**
- * sort value list by index
- *
- * @param valueList value list
- * @param dataType data type
- * @param index index
- * @return sorted list
- */
- private Object sortList(Object valueList, TSDataType dataType, Integer[] index) {
- switch (dataType) {
- case BOOLEAN:
- boolean[] boolValues = (boolean[]) valueList;
- boolean[] sortedValues = new boolean[boolValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedValues[index[i]] = boolValues[i];
- }
- return sortedValues;
- case INT32:
- int[] intValues = (int[]) valueList;
- int[] sortedIntValues = new int[intValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedIntValues[index[i]] = intValues[i];
- }
- return sortedIntValues;
- case INT64:
- long[] longValues = (long[]) valueList;
- long[] sortedLongValues = new long[longValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedLongValues[index[i]] = longValues[i];
- }
- return sortedLongValues;
- case FLOAT:
- float[] floatValues = (float[]) valueList;
- float[] sortedFloatValues = new float[floatValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedFloatValues[index[i]] = floatValues[i];
- }
- return sortedFloatValues;
- case DOUBLE:
- double[] doubleValues = (double[]) valueList;
- double[] sortedDoubleValues = new double[doubleValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedDoubleValues[index[i]] = doubleValues[i];
- }
- return sortedDoubleValues;
- case TEXT:
- Binary[] binaryValues = (Binary[]) valueList;
- Binary[] sortedBinaryValues = new Binary[binaryValues.length];
- for (int i = 0; i < index.length; i++) {
- sortedBinaryValues[index[i]] = binaryValues[i];
- }
- return sortedBinaryValues;
- default:
- throw new UnSupportedDataTypeException("Unsupported data type:" + dataType);
+ try {
+ RpcUtils.verifySuccess(client.insertTablets(request).statusList);
+ } catch (TException e) {
+ throw new IoTDBConnectionException(e);
+ }
}
}
/**
- * Insert data in batch format, which can reduce the overhead of network. This method is just like
- * jdbc batch insert, we pack some insert request in batch and send them to server If you want
- * improve your performance, please see insertBatch method
+ * Insert multiple rows, which can reduce the overhead of network. This method is just like
+ * jdbc executeBatch, we pack some insert request in batch and send them to server.
+ * If you want improve your performance, please see insertTablet method
+ *
+ * Each row is independent, which could have different deviceId, time, number of measurements
*
- * @see Session#insertBatch(RowBatch)
+ * @see Session#insertTablet(Tablet)
*/
- public void insertInBatch(List<String> deviceIds, List<Long> times,
+ public void insertRecords(List<String> deviceIds, List<Long> times,
List<List<String>> measurementsList, List<List<String>> valuesList)
throws IoTDBConnectionException, BatchExecutionException {
// check params size
@@ -370,7 +355,7 @@ public class Session {
"deviceIds, times, measurementsList and valuesList's size should be equal");
}
- TSInsertInBatchReq request = new TSInsertInBatchReq();
+ TSInsertRecordsReq request = new TSInsertRecordsReq();
request.setSessionId(sessionId);
request.setDeviceIds(deviceIds);
request.setTimestamps(times);
@@ -378,75 +363,51 @@ public class Session {
request.setValuesList(valuesList);
try {
- RpcUtils.verifySuccess(client.insertRowInBatch(request).statusList);
+ RpcUtils.verifySuccess(client.insertRecords(request).statusList);
} catch (TException e) {
throw new IoTDBConnectionException(e);
}
}
/**
- * insert data in one row, if you want improve your performance, please use insertInBatch method
- * or insertBatch method
- *
- * @see Session#insertInBatch(List, List, List, List)
- * @see Session#insertBatch(RowBatch)
- */
- public TSStatus insert(String deviceId, long time, List<String> measurements,
- Object... values) throws IoTDBConnectionException, StatementExecutionException {
- List<String> stringValues = new ArrayList<>();
- for (Object o : values) {
- stringValues.add(o.toString());
- }
-
- return insert(deviceId, time, measurements, stringValues);
- }
-
- /**
- * insert data in one row, if you want improve your performance, please use insertInBatch method
- * or insertBatch method
- *
- * @see Session#insertInBatch(List, List, List, List)
- * @see Session#insertBatch(RowBatch)
+ * This method NOT insert data into database and the server just return after accept the request,
+ * this method should be used to test other time cost in client
*/
- public TSStatus insert(String deviceId, long time, List<String> measurements,
+ public void testInsertRecord(String deviceId, long time, List<String> measurements,
List<String> values) throws IoTDBConnectionException, StatementExecutionException {
- TSInsertReq request = new TSInsertReq();
+ TSInsertRecordReq request = new TSInsertRecordReq();
request.setSessionId(sessionId);
request.setDeviceId(deviceId);
request.setTimestamp(time);
request.setMeasurements(measurements);
request.setValues(values);
- TSStatus result;
try {
- result = client.insert(request);
- RpcUtils.verifySuccess(result);
+ RpcUtils.verifySuccess(client.testInsertRecord(request));
} catch (TException e) {
throw new IoTDBConnectionException(e);
}
-
- return result;
}
/**
* This method NOT insert data into database and the server just return after accept the request,
* this method should be used to test other time cost in client
*/
- public void testInsertBatch(RowBatch rowBatch)
+ public void testInsertTablet(Tablet tablet)
throws IoTDBConnectionException, BatchExecutionException {
- TSBatchInsertionReq request = new TSBatchInsertionReq();
+ TSInsertTabletReq request = new TSInsertTabletReq();
request.setSessionId(sessionId);
- request.deviceId = rowBatch.deviceId;
- for (MeasurementSchema measurementSchema : rowBatch.getSchemas()) {
+ request.deviceId = tablet.deviceId;
+ for (MeasurementSchema measurementSchema : tablet.getSchemas()) {
request.addToMeasurements(measurementSchema.getMeasurementId());
request.addToTypes(measurementSchema.getType().ordinal());
}
- request.setTimestamps(SessionUtils.getTimeBuffer(rowBatch));
- request.setValues(SessionUtils.getValueBuffer(rowBatch));
- request.setSize(rowBatch.batchSize);
+ request.setTimestamps(SessionUtils.getTimeBuffer(tablet));
+ request.setValues(SessionUtils.getValueBuffer(tablet));
+ request.setSize(tablet.rowSize);
try {
- RpcUtils.verifySuccess(client.testInsertBatch(request).statusList);
+ RpcUtils.verifySuccess(client.testInsertTablet(request).statusList);
} catch (TException e) {
throw new IoTDBConnectionException(e);
}
@@ -456,7 +417,7 @@ public class Session {
* This method NOT insert data into database and the server just return after accept the request,
* this method should be used to test other time cost in client
*/
- public void testInsertInBatch(List<String> deviceIds, List<Long> times,
+ public void testInsertRecords(List<String> deviceIds, List<Long> times,
List<List<String>> measurementsList, List<List<String>> valuesList)
throws IoTDBConnectionException, BatchExecutionException {
// check params size
@@ -466,7 +427,7 @@ public class Session {
"deviceIds, times, measurementsList and valuesList's size should be equal");
}
- TSInsertInBatchReq request = new TSInsertInBatchReq();
+ TSInsertRecordsReq request = new TSInsertRecordsReq();
request.setSessionId(sessionId);
request.setDeviceIds(deviceIds);
request.setTimestamps(times);
@@ -474,27 +435,7 @@ public class Session {
request.setValuesList(valuesList);
try {
- RpcUtils.verifySuccess(client.testInsertRowInBatch(request).statusList);
- } catch (TException e) {
- throw new IoTDBConnectionException(e);
- }
- }
-
- /**
- * This method NOT insert data into database and the server just return after accept the request,
- * this method should be used to test other time cost in client
- */
- public void testInsert(String deviceId, long time, List<String> measurements,
- List<String> values) throws IoTDBConnectionException, StatementExecutionException {
- TSInsertReq request = new TSInsertReq();
- request.setSessionId(sessionId);
- request.setDeviceId(deviceId);
- request.setTimestamp(time);
- request.setMeasurements(measurements);
- request.setValues(values);
-
- try {
- RpcUtils.verifySuccess(client.testInsertRow(request));
+ RpcUtils.verifySuccess(client.testInsertRecords(request).statusList);
} catch (TException e) {
throw new IoTDBConnectionException(e);
}
@@ -730,4 +671,94 @@ public class Session {
}
}
+ /**
+ * check whether the batch has been sorted
+ *
+ * @return whether the batch has been sorted
+ */
+ private boolean checkSorted(Tablet tablet) {
+ for (int i = 1; i < tablet.rowSize; i++) {
+ if (tablet.timestamps[i] < tablet.timestamps[i - 1]) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private void sortTablet(Tablet tablet) {
+ /*
+ * following part of code sort the batch data by time,
+ * so we can insert continuous data in value list to get a better performance
+ */
+ // sort to get index, and use index to sort value list
+ Integer[] index = new Integer[tablet.rowSize];
+ for (int i = 0; i < tablet.rowSize; i++) {
+ index[i] = i;
+ }
+ Arrays.sort(index, Comparator.comparingLong(o -> tablet.timestamps[o]));
+ Arrays.sort(tablet.timestamps, 0, tablet.rowSize);
+ for (int i = 0; i < tablet.getSchemas().size(); i++) {
+ tablet.values[i] =
+ sortList(tablet.values[i], tablet.getSchemas().get(i).getType(), index);
+ }
+ }
+
+ /**
+ * sort value list by index
+ *
+ * @param valueList value list
+ * @param dataType data type
+ * @param index index
+ * @return sorted list
+ */
+ private Object sortList(Object valueList, TSDataType dataType, Integer[] index) {
+ switch (dataType) {
+ case BOOLEAN:
+ boolean[] boolValues = (boolean[]) valueList;
+ boolean[] sortedValues = new boolean[boolValues.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedValues[index[i]] = boolValues[i];
+ }
+ return sortedValues;
+ case INT32:
+ int[] intValues = (int[]) valueList;
+ int[] sortedIntValues = new int[intValues.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedIntValues[index[i]] = intValues[i];
+ }
+ return sortedIntValues;
+ case INT64:
+ long[] longValues = (long[]) valueList;
+ long[] sortedLongValues = new long[longValues.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedLongValues[index[i]] = longValues[i];
+ }
+ return sortedLongValues;
+ case FLOAT:
+ float[] floatValues = (float[]) valueList;
+ float[] sortedFloatValues = new float[floatValues.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedFloatValues[index[i]] = floatValues[i];
+ }
+ return sortedFloatValues;
+ case DOUBLE:
+ double[] doubleValues = (double[]) valueList;
+ double[] sortedDoubleValues = new double[doubleValues.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedDoubleValues[index[i]] = doubleValues[i];
+ }
+ return sortedDoubleValues;
+ case TEXT:
+ Binary[] binaryValues = (Binary[]) valueList;
+ Binary[] sortedBinaryValues = new Binary[binaryValues.length];
+ for (int i = 0; i < index.length; i++) {
+ sortedBinaryValues[index[i]] = binaryValues[i];
+ }
+ return sortedBinaryValues;
+ default:
+ throw new UnSupportedDataTypeException("Unsupported data type:" + dataType);
+ }
+ }
+
}
diff --git a/session/src/main/java/org/apache/iotdb/session/SessionUtils.java b/session/src/main/java/org/apache/iotdb/session/SessionUtils.java
index 093d712..7fda683 100644
--- a/session/src/main/java/org/apache/iotdb/session/SessionUtils.java
+++ b/session/src/main/java/org/apache/iotdb/session/SessionUtils.java
@@ -22,59 +22,59 @@ import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.BytesUtils;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import java.nio.ByteBuffer;
public class SessionUtils {
- public static ByteBuffer getTimeBuffer(RowBatch rowBatch) {
- ByteBuffer timeBuffer = ByteBuffer.allocate(rowBatch.getTimeBytesSize());
- for (int i = 0; i < rowBatch.batchSize; i++) {
- timeBuffer.putLong(rowBatch.timestamps[i]);
+ public static ByteBuffer getTimeBuffer(Tablet tablet) {
+ ByteBuffer timeBuffer = ByteBuffer.allocate(tablet.getTimeBytesSize());
+ for (int i = 0; i < tablet.rowSize; i++) {
+ timeBuffer.putLong(tablet.timestamps[i]);
}
timeBuffer.flip();
return timeBuffer;
}
- public static ByteBuffer getValueBuffer(RowBatch rowBatch) {
- ByteBuffer valueBuffer = ByteBuffer.allocate(rowBatch.getValueBytesSize());
- for (int i = 0; i < rowBatch.getSchemas().size(); i++) {
- TSDataType dataType = rowBatch.getSchemas().get(i).getType();
+ public static ByteBuffer getValueBuffer(Tablet tablet) {
+ ByteBuffer valueBuffer = ByteBuffer.allocate(tablet.getValueBytesSize());
+ for (int i = 0; i < tablet.getSchemas().size(); i++) {
+ TSDataType dataType = tablet.getSchemas().get(i).getType();
switch (dataType) {
case INT32:
- int[] intValues = (int[]) rowBatch.values[i];
- for (int index = 0; index < rowBatch.batchSize; index++) {
+ int[] intValues = (int[]) tablet.values[i];
+ for (int index = 0; index < tablet.rowSize; index++) {
valueBuffer.putInt(intValues[index]);
}
break;
case INT64:
- long[] longValues = (long[]) rowBatch.values[i];
- for (int index = 0; index < rowBatch.batchSize; index++) {
+ long[] longValues = (long[]) tablet.values[i];
+ for (int index = 0; index < tablet.rowSize; index++) {
valueBuffer.putLong(longValues[index]);
}
break;
case FLOAT:
- float[] floatValues = (float[]) rowBatch.values[i];
- for (int index = 0; index < rowBatch.batchSize; index++) {
+ float[] floatValues = (float[]) tablet.values[i];
+ for (int index = 0; index < tablet.rowSize; index++) {
valueBuffer.putFloat(floatValues[index]);
}
break;
case DOUBLE:
- double[] doubleValues = (double[]) rowBatch.values[i];
- for (int index = 0; index < rowBatch.batchSize; index++) {
+ double[] doubleValues = (double[]) tablet.values[i];
+ for (int index = 0; index < tablet.rowSize; index++) {
valueBuffer.putDouble(doubleValues[index]);
}
break;
case BOOLEAN:
- boolean[] boolValues = (boolean[]) rowBatch.values[i];
- for (int index = 0; index < rowBatch.batchSize; index++) {
+ boolean[] boolValues = (boolean[]) tablet.values[i];
+ for (int index = 0; index < tablet.rowSize; index++) {
valueBuffer.put(BytesUtils.boolToByte(boolValues[index]));
}
break;
case TEXT:
- Binary[] binaryValues = (Binary[]) rowBatch.values[i];
- for (int index = 0; index < rowBatch.batchSize; index++) {
+ Binary[] binaryValues = (Binary[]) tablet.values[i];
+ for (int index = 0; index < tablet.rowSize; index++) {
valueBuffer.putInt(binaryValues[index].getLength());
valueBuffer.put(binaryValues[index].getValues());
}
diff --git a/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java b/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
index e87cb03..1a61d3c 100644
--- a/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
+++ b/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java
@@ -26,14 +26,13 @@ import java.util.concurrent.ConcurrentMap;
import org.apache.iotdb.rpc.BatchExecutionException;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.iotdb.service.rpc.thrift.TSStatus;
import org.apache.iotdb.session.Config;
import org.apache.iotdb.session.Session;
import org.apache.iotdb.session.SessionDataSet;
import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -223,42 +222,47 @@ public class SessionPool {
}
/**
- * use batch interface to insert sorted data times in row batch must be sorted before!
+ * insert the data of a device. For each timestamp, the number of measurements is the same.
*
- * @param rowBatch data batch
+ * a Tablet example:
+ *
+ * device1
+ * time s1, s2, s3
+ * 1, 1, 1, 1
+ * 2, 2, 2, 2
+ * 3, 3, 3, 3
+ *
+ * times in Tablet may be not in ascending order
+ *
+ * @param tablet data batch
*/
- public void insertSortedBatch(RowBatch rowBatch)
+ public void insertTablet(Tablet tablet)
throws IoTDBConnectionException, BatchExecutionException {
- for (int i = 0; i < RETRY; i++) {
- Session session = getSession();
- try {
- session.insertSortedBatch(rowBatch);
- putBack(session);
- } catch (IoTDBConnectionException e) {
- // TException means the connection is broken, remove it and get a new one.
- closeSession(session);
- removeSession();
- } catch (BatchExecutionException e) {
- putBack(session);
- throw e;
- }
- }
- throw new IoTDBConnectionException(
- String.format("retry to execute statement on %s:%s failed %d times", ip, port, RETRY));
+ insertTablet(tablet, false);
}
-
/**
- * use batch interface to insert data
+ * insert the data of a device. For each timestamp, the number of measurements is the same.
+ *
+ * a Tablet example:
*
- * @param rowBatch data batch
+ * device1
+ * time s1, s2, s3
+ * 1, 1, 1, 1
+ * 2, 2, 2, 2
+ * 3, 3, 3, 3
+ *
+ * Users need to control the count of Tablet and write a batch when it reaches the maxBatchSize
+ *
+ * @param tablet a tablet data of one device
+ * @param sorted whether times in Tablet are in ascending order
*/
- public void insertBatch(RowBatch rowBatch)
+ public void insertTablet(Tablet tablet, boolean sorted)
throws IoTDBConnectionException, BatchExecutionException {
for (int i = 0; i < RETRY; i++) {
Session session = getSession();
try {
- session.insertBatch(rowBatch);
+ session.insertTablet(tablet, sorted);
putBack(session);
return;
} catch (IoTDBConnectionException e) {
@@ -278,40 +282,24 @@ public class SessionPool {
/**
* use batch interface to insert data
*
- * @param rowBatchMap data batch
+ * @param tablets multiple batch
*/
- public void insertMultipleDeviceSortedBatch(Map<String, RowBatch> rowBatchMap)
+ public void insertTablets(Map<String, Tablet> tablets)
throws IoTDBConnectionException, BatchExecutionException {
- for (int i = 0; i < RETRY; i++) {
- Session session = getSession();
- try {
- session.insertMultipleDeviceSortedBatch(rowBatchMap);
- putBack(session);
- return;
- } catch (IoTDBConnectionException e) {
- // TException means the connection is broken, remove it and get a new one.
- closeSession(session);
- removeSession();
- } catch (BatchExecutionException e) {
- putBack(session);
- throw e;
- }
- }
- throw new IoTDBConnectionException(
- String.format("retry to execute statement on %s:%s failed %d times", ip, port, RETRY));
+ insertTablets(tablets, false);
}
/**
* use batch interface to insert data
*
- * @param rowBatchMap data batch
+ * @param tablets multiple batch
*/
- public void insertMultipleDeviceBatch(Map<String, RowBatch> rowBatchMap)
+ public void insertTablets(Map<String, Tablet> tablets, boolean sorted)
throws IoTDBConnectionException, BatchExecutionException {
for (int i = 0; i < RETRY; i++) {
Session session = getSession();
try {
- session.insertMultipleDeviceBatch(rowBatchMap);
+ session.insertTablets(tablets, sorted);
putBack(session);
return;
} catch (IoTDBConnectionException e) {
@@ -330,17 +318,17 @@ public class SessionPool {
/**
* Insert data in batch format, which can reduce the overhead of network. This method is just like
* jdbc batch insert, we pack some insert request in batch and send them to server If you want
- * improve your performance, please see insertBatch method
+ * improve your performance, please see insertTablet method
*
- * @see Session#insertBatch(RowBatch)
+ * @see Session#insertTablet(Tablet)
*/
- public void insertInBatch(List<String> deviceIds, List<Long> times,
+ public void insertRecords(List<String> deviceIds, List<Long> times,
List<List<String>> measurementsList, List<List<String>> valuesList)
throws IoTDBConnectionException, BatchExecutionException {
for (int i = 0; i < RETRY; i++) {
Session session = getSession();
try {
- session.insertInBatch(deviceIds, times, measurementsList, valuesList);
+ session.insertRecords(deviceIds, times, measurementsList, valuesList);
putBack(session);
return;
} catch (IoTDBConnectionException e) {
@@ -357,20 +345,20 @@ public class SessionPool {
}
/**
- * insert data in one row, if you want improve your performance, please use insertInBatch method
- * or insertBatch method
+ * insert data in one row, if you want improve your performance, please use insertRecords method
+ * or insertTablet method
*
- * @see Session#insertInBatch(List, List, List, List)
- * @see Session#insertBatch(RowBatch)
+ * @see Session#insertRecords(List, List, List, List)
+ * @see Session#insertTablet(Tablet)
*/
- public TSStatus insert(String deviceId, long time, List<String> measurements, List<String> values)
+ public void insertRecord(String deviceId, long time, List<String> measurements, List<String> values)
throws IoTDBConnectionException, StatementExecutionException {
for (int i = 0; i < RETRY; i++) {
Session session = getSession();
try {
- TSStatus resp = session.insert(deviceId, time, measurements, values);
+ session.insertRecord(deviceId, time, measurements, values);
putBack(session);
- return resp;
+ return;
} catch (IoTDBConnectionException e) {
// TException means the connection is broken, remove it and get a new one.
closeSession(session);
@@ -388,12 +376,12 @@ public class SessionPool {
* This method NOT insert data into database and the server just return after accept the request,
* this method should be used to test other time cost in client
*/
- public void testInsertBatch(RowBatch rowBatch)
+ public void testInsertTablet(Tablet tablet)
throws IoTDBConnectionException, BatchExecutionException {
for (int i = 0; i < RETRY; i++) {
Session session = getSession();
try {
- session.testInsertBatch(rowBatch);
+ session.testInsertTablet(tablet);
putBack(session);
return;
} catch (IoTDBConnectionException e) {
@@ -413,14 +401,13 @@ public class SessionPool {
* This method NOT insert data into database and the server just return after accept the request,
* this method should be used to test other time cost in client
*/
- public void testInsertInBatch(List<String> deviceIds, List<Long> times,
+ public void testInsertRecords(List<String> deviceIds, List<Long> times,
List<List<String>> measurementsList, List<List<String>> valuesList)
throws IoTDBConnectionException, BatchExecutionException {
for (int i = 0; i < RETRY; i++) {
Session session = getSession();
try {
- session
- .testInsertInBatch(deviceIds, times, measurementsList, valuesList);
+ session.testInsertRecords(deviceIds, times, measurementsList, valuesList);
putBack(session);
return;
} catch (IoTDBConnectionException e) {
@@ -440,12 +427,12 @@ public class SessionPool {
* This method NOT insert data into database and the server just return after accept the request,
* this method should be used to test other time cost in client
*/
- public void testInsert(String deviceId, long time, List<String> measurements,
+ public void testInsertRecord(String deviceId, long time, List<String> measurements,
List<String> values) throws IoTDBConnectionException, StatementExecutionException {
for (int i = 0; i < RETRY; i++) {
Session session = getSession();
try {
- session.testInsert(deviceId, time, measurements, values);
+ session.testInsertRecord(deviceId, time, measurements, values);
putBack(session);
return;
} catch (IoTDBConnectionException e) {
diff --git a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionIT.java b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionIT.java
index fa0d79c..fcc03b9 100644
--- a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionIT.java
+++ b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionIT.java
@@ -33,9 +33,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBConstant;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.jdbc.Config;
import org.apache.iotdb.rpc.BatchExecutionException;
@@ -45,10 +43,8 @@ import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.common.Field;
-import org.apache.iotdb.tsfile.read.common.Path;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-import org.apache.iotdb.tsfile.write.schema.Schema;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -102,7 +98,7 @@ public class IoTDBSessionIT {
createTimeseries();
- insertRowBatchTest2("root.sg1.d1");
+ insertTabletTest2("root.sg1.d1");
queryForAlignByDevice();
queryForAlignByDevice2();
@@ -118,7 +114,7 @@ public class IoTDBSessionIT {
createTimeseriesForTime();
- insertRowBatchTestForTime("root.sg1.d1");
+ insertTabletTestForTime("root.sg1.d1");
}
@Test
@@ -131,7 +127,7 @@ public class IoTDBSessionIT {
createTimeseries();
- insertRowBatchTest2("root.sg1.d1");
+ insertTabletTest2("root.sg1.d1");
// flush
Class.forName(Config.JDBC_DRIVER_NAME);
try (Connection connection = DriverManager
@@ -140,7 +136,7 @@ public class IoTDBSessionIT {
statement.execute("FLUSH");
}
//
- insertRowBatchTest3("root.sg1.d1");
+ insertTabletTest3("root.sg1.d1");
queryForBatchSeqAndUnseq();
}
@@ -155,7 +151,7 @@ public class IoTDBSessionIT {
createTimeseries();
- insertRowBatchTest2("root.sg1.d1");
+ insertTabletTest2("root.sg1.d1");
queryForBatch();
}
@@ -205,17 +201,17 @@ public class IoTDBSessionIT {
createTimeseries();
- // test insert batch
- Schema schema = new Schema();
- schema.registerTimeseries(new Path(deviceId, "s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId, "s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId, "s3"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
+ // test insert tablet
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
- RowBatch rowBatch = schema.createRowBatch("root.sg1.d1", 100);
+ Tablet tablet = new Tablet("root.sg1.d1", schemaList, 100);
- session.testInsertBatch(rowBatch);
+ session.testInsertTablet(tablet);
- // test insert row
+ // test insert record
List<String> measurements = new ArrayList<>();
measurements.add("s1");
measurements.add("s2");
@@ -225,10 +221,10 @@ public class IoTDBSessionIT {
values.add("1");
values.add("2");
values.add("3");
- session.testInsert(deviceId, time, measurements, values);
+ session.testInsertRecord(deviceId, time, measurements, values);
}
- // test insert row in batch
+ // test insert records
measurements = new ArrayList<>();
measurements.add("s1");
measurements.add("s2");
@@ -249,7 +245,7 @@ public class IoTDBSessionIT {
valuesList.add(values);
timestamps.add(time);
if (time != 0 && time % 100 == 0) {
- session.testInsertInBatch(deviceIds, timestamps, measurementsList, valuesList);
+ session.testInsertRecords(deviceIds, timestamps, measurementsList, valuesList);
deviceIds.clear();
measurementsList.clear();
valuesList.clear();
@@ -257,7 +253,7 @@ public class IoTDBSessionIT {
}
}
- session.testInsertInBatch(deviceIds, timestamps, measurementsList, valuesList);
+ session.testInsertRecords(deviceIds, timestamps, measurementsList, valuesList);
}
@Test
@@ -304,7 +300,7 @@ public class IoTDBSessionIT {
query3();
-// insertRowBatchTest1();
+// insertTabletTest1();
deleteData();
query();
@@ -313,7 +309,7 @@ public class IoTDBSessionIT {
query2();
- insertInBatch();
+ insertRecords();
query4();
@@ -338,7 +334,7 @@ public class IoTDBSessionIT {
// set storage group but do not create timeseries
session.setStorageGroup("root.sg3");
- insertRowBatchTest1("root.sg3.d1");
+ insertTabletTest1("root.sg3.d1");
// create timeseries but do not set storage group
session.createTimeseries("root.sg4.d1.s1", TSDataType.INT64, TSEncoding.RLE,
@@ -347,10 +343,10 @@ public class IoTDBSessionIT {
CompressionType.SNAPPY);
session.createTimeseries("root.sg4.d1.s3", TSDataType.INT64, TSEncoding.RLE,
CompressionType.SNAPPY);
- insertRowBatchTest1("root.sg4.d1");
+ insertTabletTest1("root.sg4.d1");
// do not set storage group and create timeseries
- insertRowBatchTest1("root.sg5.d1");
+ insertTabletTest1("root.sg5.d1");
session.close();
}
@@ -415,12 +411,12 @@ public class IoTDBSessionIT {
List<String> values = new ArrayList<>();
measurements.add(sensorId);
values.add("100");
- session.insert(deviceId, i, measurements, values);
+ session.insertRecord(deviceId, i, measurements, values);
}
}
}
- private void insertInBatch() throws IoTDBConnectionException, BatchExecutionException {
+ private void insertRecords() throws IoTDBConnectionException, BatchExecutionException {
String deviceId = "root.sg1.d2";
List<String> measurements = new ArrayList<>();
measurements.add("s1");
@@ -442,7 +438,7 @@ public class IoTDBSessionIT {
valuesList.add(values);
timestamps.add(time);
if (time != 0 && time % 100 == 0) {
- session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList);
+ session.insertRecords(deviceIds, timestamps, measurementsList, valuesList);
deviceIds.clear();
measurementsList.clear();
valuesList.clear();
@@ -450,7 +446,7 @@ public class IoTDBSessionIT {
}
}
- session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList);
+ session.insertRecords(deviceIds, timestamps, measurementsList, valuesList);
}
private void insertInObject() throws IoTDBConnectionException, StatementExecutionException {
@@ -460,7 +456,7 @@ public class IoTDBSessionIT {
measurements.add("s2");
measurements.add("s3");
for (long time = 0; time < 100; time++) {
- session.insert(deviceId, time, measurements, 1L, 2L, 3L);
+ session.insertRecord(deviceId, time, measurements, 1L, 2L, 3L);
}
}
@@ -475,38 +471,38 @@ public class IoTDBSessionIT {
values.add("1");
values.add("2");
values.add("3");
- session.insert(deviceId, time, measurements, values);
+ session.insertRecord(deviceId, time, measurements, values);
}
}
- private void insertRowBatchTest1(String deviceId)
+ private void insertTabletTest1(String deviceId)
throws IoTDBConnectionException, BatchExecutionException {
- Schema schema = new Schema();
- schema.registerTimeseries(new Path(deviceId, "s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId, "s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId, "s3"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
- RowBatch rowBatch = schema.createRowBatch(deviceId, 100);
+ Tablet tablet = new Tablet(deviceId, schemaList, 100);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
for (long time = 0; time < 100; time++) {
- int row = rowBatch.batchSize++;
+ int row = tablet.rowSize++;
timestamps[row] = time;
for (int i = 0; i < 3; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = i;
}
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
- if (rowBatch.batchSize != 0) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize != 0) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
@@ -751,114 +747,101 @@ public class IoTDBSessionIT {
assertEquals(correctStatus, status);
}
- private void insertRowBatchTest2(String deviceId)
+ private void insertTabletTest2(String deviceId)
throws IoTDBConnectionException, BatchExecutionException {
- Schema schema = new Schema();
- schema.registerTimeseries(new Path(deviceId,"s1"),
- new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s2"),
- new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s3"),
- new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
- RowBatch rowBatch = schema.createRowBatch(deviceId, 256);
+ Tablet tablet = new Tablet(deviceId, schemaList, 256);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
for (long time = 0; time < 1000; time++) {
- int row = rowBatch.batchSize++;
+ int row = tablet.rowSize++;
timestamps[row] = time;
for (int i = 0; i < 3; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = i;
}
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
- if (rowBatch.batchSize != 0) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize != 0) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
- private void insertRowBatchTest3(String deviceId)
+ private void insertTabletTest3(String deviceId)
throws IoTDBConnectionException, BatchExecutionException {
- Schema schema = new Schema();
- schema.registerTimeseries(new Path(deviceId,"s1"),
- new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s2"),
- new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s3"),
- new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
- RowBatch rowBatch = schema.createRowBatch(deviceId, 200);
+ Tablet tablet = new Tablet(deviceId, schemaList, 200);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
for (long time = 500; time < 1500; time++) {
- int row = rowBatch.batchSize++;
+ int row = tablet.rowSize++;
timestamps[row] = time;
for (int i = 0; i < 3; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = i;
}
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
- if (rowBatch.batchSize != 0) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize != 0) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
- private void insertRowBatchTestForTime(String deviceId)
+ private void insertTabletTestForTime(String deviceId)
throws IoTDBConnectionException, BatchExecutionException {
- Schema schema = new Schema();
- schema.registerTimeseries(new Path(deviceId,"s1"),
- new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s2"),
- new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s3"),
- new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s4"),
- new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s5"),
- new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
- schema.registerTimeseries(new Path(deviceId,"s6"),
- new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
+ List<MeasurementSchema> schemaList = new ArrayList<>();
+ schemaList.add(new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s4", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s5", TSDataType.INT64, TSEncoding.RLE));
+ schemaList.add(new MeasurementSchema("s6", TSDataType.INT64, TSEncoding.RLE));
long count = 10000000;
long begin = 0;
- //long begin = 1579414903000L;
- RowBatch rowBatch = schema.createRowBatch(deviceId, 1000);
+ Tablet tablet = new Tablet(deviceId, schemaList, 1000);
- long[] timestamps = rowBatch.timestamps;
- Object[] values = rowBatch.values;
+ long[] timestamps = tablet.timestamps;
+ Object[] values = tablet.values;
for (long time = begin; time < count + begin; time++) {
- int row = rowBatch.batchSize++;
+ int row = tablet.rowSize++;
timestamps[row] = time;
for (int i = 0; i < 6; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = i;
}
- if (rowBatch.batchSize == rowBatch.getMaxBatchSize()) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize == tablet.getMaxRowNumber()) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
- if (rowBatch.batchSize != 0) {
- session.insertBatch(rowBatch);
- rowBatch.reset();
+ if (tablet.rowSize != 0) {
+ session.insertTablet(tablet);
+ tablet.reset();
}
}
diff --git a/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java b/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
index 9b649aa..00d03ff 100644
--- a/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
+++ b/session/src/test/java/org/apache/iotdb/session/pool/SessionPoolTest.java
@@ -19,19 +19,14 @@
package org.apache.iotdb.session.pool;
import static org.junit.Assert.*;
-
-import java.sql.SQLException;
-import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
-import org.apache.thrift.TException;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -60,7 +55,7 @@ public class SessionPoolTest {
final int no = i;
service.submit(() -> {
try {
- pool.insert("root.sg1.d1", 1, Collections.singletonList("s" + no), Collections.singletonList("3"));
+ pool.insertRecord("root.sg1.d1", 1, Collections.singletonList("s" + no), Collections.singletonList("3"));
} catch (IoTDBConnectionException | StatementExecutionException e) {
fail();
}
@@ -83,7 +78,7 @@ public class SessionPoolTest {
SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3);
assertEquals(0, pool.currentAvailableSize());
try {
- pool.insert(".root.sg1.d1", 1, Collections.singletonList("s" ), Collections.singletonList("3"));
+ pool.insertRecord(".root.sg1.d1", 1, Collections.singletonList("s" ), Collections.singletonList("3"));
} catch (IoTDBConnectionException | StatementExecutionException e) {
//do nothing
}
@@ -98,7 +93,7 @@ public class SessionPoolTest {
ExecutorService service = Executors.newFixedThreadPool(10);
for (int i = 0; i < 10; i++) {
try {
- pool.insert("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
+ pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
} catch (IoTDBConnectionException | StatementExecutionException e) {
fail();
}
@@ -139,7 +134,7 @@ public class SessionPoolTest {
ExecutorService service = Executors.newFixedThreadPool(10);
for (int i = 0; i < 10; i++) {
try {
- pool.insert("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
+ pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
} catch (IoTDBConnectionException | StatementExecutionException e) {
fail();
}
@@ -174,7 +169,7 @@ public class SessionPoolTest {
SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3, 1, 6000, false);
for (int i = 0; i < 10; i++) {
try {
- pool.insert("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
+ pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
} catch (IoTDBConnectionException | StatementExecutionException e) {
fail();
}
@@ -209,7 +204,7 @@ public class SessionPoolTest {
SessionPool pool = new SessionPool("127.0.0.1", 6667, "root", "root", 3, 1, 60000, false);
for (int i = 0; i < 10; i++) {
try {
- pool.insert("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
+ pool.insertRecord("root.sg1.d1", i, Collections.singletonList("s" + i), Collections.singletonList("" + i));
} catch (IoTDBConnectionException | StatementExecutionException e) {
fail();
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/TsFileWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/TsFileWriter.java
index 2ded552..5346847 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/TsFileWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/TsFileWriter.java
@@ -25,7 +25,7 @@ import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.write.chunk.ChunkGroupWriterImpl;
import org.apache.iotdb.tsfile.write.chunk.IChunkGroupWriter;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
import org.apache.iotdb.tsfile.write.schema.Schema;
@@ -206,24 +206,24 @@ public class TsFileWriter implements AutoCloseable {
}
/**
- * Confirm whether the row batch is legal.
+ * Confirm whether the tablet is legal.
*
- * @param rowBatch - a row batch responding multiple columns
- * @return - whether the row batch has been added into RecordWriter legally
+ * @param tablet - a tablet data responding multiple columns
+ * @return - whether the tablet's measurements have been added into RecordWriter legally
* @throws WriteProcessException exception
*/
- private void checkIsTimeSeriesExist(RowBatch rowBatch) throws WriteProcessException {
+ private void checkIsTimeSeriesExist(Tablet tablet) throws WriteProcessException {
IChunkGroupWriter groupWriter;
- if (!groupWriters.containsKey(rowBatch.deviceId)) {
- groupWriter = new ChunkGroupWriterImpl(rowBatch.deviceId);
- groupWriters.put(rowBatch.deviceId, groupWriter);
+ if (!groupWriters.containsKey(tablet.deviceId)) {
+ groupWriter = new ChunkGroupWriterImpl(tablet.deviceId);
+ groupWriters.put(tablet.deviceId, groupWriter);
} else {
- groupWriter = groupWriters.get(rowBatch.deviceId);
+ groupWriter = groupWriters.get(tablet.deviceId);
}
- String deviceId = rowBatch.deviceId;
+ String deviceId = tablet.deviceId;
- // add all SeriesWriter of measurements in this RowBatch to this ChunkGroupWriter
- for (MeasurementSchema timeseries : rowBatch.getSchemas()) {
+ // add all SeriesWriter of measurements in this Tablet to this ChunkGroupWriter
+ for (MeasurementSchema timeseries : tablet.getSchemas()) {
String measurementId = timeseries.getMeasurementId();
if (schema.containsTimeseries(new Path(deviceId, measurementId))) {
groupWriter.tryToAddSeriesWriter(schema.getSeriesSchema(new Path(deviceId, measurementId)),
@@ -252,18 +252,18 @@ public class TsFileWriter implements AutoCloseable {
}
/**
- * write a row batch
+ * write a tablet
*
- * @param rowBatch - multiple time series of one device that share a time column
+ * @param tablet - multiple time series of one device that share a time column
* @throws IOException exception in IO
* @throws WriteProcessException exception in write process
*/
- public boolean write(RowBatch rowBatch) throws IOException, WriteProcessException {
- // make sure the ChunkGroupWriter for this RowBatch exist
- checkIsTimeSeriesExist(rowBatch);
- // get corresponding ChunkGroupWriter and write this RowBatch
- groupWriters.get(rowBatch.deviceId).write(rowBatch);
- recordCount += rowBatch.batchSize;
+ public boolean write(Tablet tablet) throws IOException, WriteProcessException {
+ // make sure the ChunkGroupWriter for this Tablet exist
+ checkIsTimeSeriesExist(tablet);
+ // get corresponding ChunkGroupWriter and write this Tablet
+ groupWriters.get(tablet.deviceId).write(tablet);
+ recordCount += tablet.rowSize;
return checkMemorySizeAndMayFlushChunks();
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkGroupWriterImpl.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkGroupWriterImpl.java
index d658c8e..e92a7dc 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkGroupWriterImpl.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkGroupWriterImpl.java
@@ -28,7 +28,7 @@ import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
@@ -75,39 +75,39 @@ public class ChunkGroupWriterImpl implements IChunkGroupWriter {
}
@Override
- public void write(RowBatch rowBatch) throws WriteProcessException, IOException {
- List<MeasurementSchema> timeseries = rowBatch.getSchemas();
+ public void write(Tablet tablet) throws WriteProcessException, IOException {
+ List<MeasurementSchema> timeseries = tablet.getSchemas();
for (int i = 0; i < timeseries.size(); i++) {
String measurementId = timeseries.get(i).getMeasurementId();
TSDataType dataType = timeseries.get(i).getType();
if (!chunkWriters.containsKey(measurementId)) {
throw new NoMeasurementException("measurement id" + measurementId + " not found!");
}
- writeByDataType(rowBatch, measurementId, dataType, i);
+ writeByDataType(tablet, measurementId, dataType, i);
}
}
private void writeByDataType(
- RowBatch rowBatch, String measurementId, TSDataType dataType, int index) throws IOException {
- int batchSize = rowBatch.batchSize;
+ Tablet tablet, String measurementId, TSDataType dataType, int index) throws IOException {
+ int batchSize = tablet.rowSize;
switch (dataType) {
case INT32:
- chunkWriters.get(measurementId).write(rowBatch.timestamps, (int[]) rowBatch.values[index], batchSize);
+ chunkWriters.get(measurementId).write(tablet.timestamps, (int[]) tablet.values[index], batchSize);
break;
case INT64:
- chunkWriters.get(measurementId).write(rowBatch.timestamps, (long[]) rowBatch.values[index], batchSize);
+ chunkWriters.get(measurementId).write(tablet.timestamps, (long[]) tablet.values[index], batchSize);
break;
case FLOAT:
- chunkWriters.get(measurementId).write(rowBatch.timestamps, (float[]) rowBatch.values[index], batchSize);
+ chunkWriters.get(measurementId).write(tablet.timestamps, (float[]) tablet.values[index], batchSize);
break;
case DOUBLE:
- chunkWriters.get(measurementId).write(rowBatch.timestamps, (double[]) rowBatch.values[index], batchSize);
+ chunkWriters.get(measurementId).write(tablet.timestamps, (double[]) tablet.values[index], batchSize);
break;
case BOOLEAN:
- chunkWriters.get(measurementId).write(rowBatch.timestamps, (boolean[]) rowBatch.values[index], batchSize);
+ chunkWriters.get(measurementId).write(tablet.timestamps, (boolean[]) tablet.values[index], batchSize);
break;
case TEXT:
- chunkWriters.get(measurementId).write(rowBatch.timestamps, (Binary[]) rowBatch.values[index], batchSize);
+ chunkWriters.get(measurementId).write(tablet.timestamps, (Binary[]) tablet.values[index], batchSize);
break;
default:
throw new UnSupportedDataTypeException(
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/IChunkGroupWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/IChunkGroupWriter.java
index ebe6a41..f6e3fe7 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/IChunkGroupWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/IChunkGroupWriter.java
@@ -22,7 +22,7 @@ import java.io.IOException;
import java.util.List;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
@@ -49,16 +49,12 @@ public interface IChunkGroupWriter {
void write(long time, List<DataPoint> data) throws WriteProcessException, IOException;
/**
- * receive a row batch, write it to timeseries writers
+ * receive a tablet, write it to chunk writers
*
- * @param rowBatch
- * - row batch to input
- * @throws WriteProcessException
- * exception in write process
- * @throws IOException
- * exception in IO
+ * @throws WriteProcessException exception in write process
+ * @throws IOException exception in IO
*/
- void write(RowBatch rowBatch) throws WriteProcessException, IOException;
+ void write(Tablet tablet) throws WriteProcessException, IOException;
/**
* flushing method for serializing to local file system or HDFS.
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/RowBatch.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/Tablet.java
similarity index 62%
rename from tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/RowBatch.java
rename to tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/Tablet.java
index 16b12c6..1ead641 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/RowBatch.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/Tablet.java
@@ -19,76 +19,81 @@
package org.apache.iotdb.tsfile.write.record;
import java.util.List;
-
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
/**
- * Multiple time series of one device that share a time column
+ * A tablet data of one device, the tablet contains multiple measurements of this device that share
+ * the same time column.
+ *
+ * for example: device root.sg1.d1
+ *
+ * time, m1, m2, m3
+ * 1, 1, 2, 3
+ * 2, 1, 2, 3
+ * 3, 1, 2, 3
+ *
+ * Notice: The tablet should not have empty cell
+ *
*/
-public class RowBatch {
+public class Tablet {
private static final int DEFAULT_SIZE = 1024;
/**
- * deviceId of this row batch
+ * deviceId of this tablet
*/
public String deviceId;
+
/**
- * the list of measurement schemas for creating the row batch
+ * the list of measurement schemas for creating the tablet
*/
private List<MeasurementSchema> schemas;
/**
- * timestamps in this row batch
+ * timestamps in this tablet
*/
public long[] timestamps;
/**
- * each object is a primitive type array, which represents values of one
- * measurement
+ * each object is a primitive type array, which represents values of one measurement
*/
public Object[] values;
/**
- * the number of rows to include in this row batch
- */
- public int batchSize;
- /**
- * the maximum number of rows for this row batch
+ * the number of rows to include in this tablet
*/
- private int maxBatchSize;
-
+ public int rowSize;
/**
- * total byte size that values occupies
+ * the maximum number of rows for this tablet
*/
- private int valueOccupation = -1;
+ private int maxRowNumber;
/**
- * Return a row batch with default specified row number. This is the standard
- * constructor (all RowBatch should be the same size).
+ * Return a tablet with default specified row number. This is the standard
+ * constructor (all Tablet should be the same size).
*
* @param deviceId the name of the device specified to be written in
- * @param timeseries the list of measurement schemas for creating the row batch
+ * @param timeseries the list of measurement schemas for creating the tablet
*/
- public RowBatch(String deviceId, List<MeasurementSchema> timeseries) {
+ public Tablet(String deviceId, List<MeasurementSchema> timeseries) {
this(deviceId, timeseries, DEFAULT_SIZE);
}
/**
- * Return a row batch with the specified number of rows (maxBatchSize). Only
- * call this constructor directly for testing purposes. RowBatch should normally
+ * Return a tablet with the specified number of rows (maxBatchSize). Only
+ * call this constructor directly for testing purposes. Tablet should normally
* always be default size.
*
* @param deviceId the name of the device specified to be written in
* @param schemas the list of measurement schemas for creating the row
* batch
- * @param maxBatchSize the maximum number of rows for this row batch
+ * @param maxRowNumber the maximum number of rows for this tablet
*/
- public RowBatch(String deviceId, List<MeasurementSchema> schemas, int maxBatchSize) {
+ public Tablet(String deviceId, List<MeasurementSchema> schemas, int maxRowNumber) {
this.deviceId = deviceId;
this.schemas = schemas;
- this.maxBatchSize = maxBatchSize;
+ this.maxRowNumber = maxRowNumber;
createColumns();
@@ -100,44 +105,44 @@ public class RowBatch {
}
/**
- * Return the maximum number of rows for this row batch
+ * Return the maximum number of rows for this tablet
*/
- public int getMaxBatchSize() {
- return maxBatchSize;
+ public int getMaxRowNumber() {
+ return maxRowNumber;
}
/**
- * Reset RowBatch to the default state - set the batchSize to 0
+ * Reset Tablet to the default state - set the rowSize to 0
*/
public void reset() {
- batchSize = 0;
+ rowSize = 0;
}
private void createColumns() {
// create timestamp column
- timestamps = new long[maxBatchSize];
+ timestamps = new long[maxRowNumber];
values = new Object[schemas.size()];
// create value columns
for (int i = 0; i < schemas.size(); i++) {
TSDataType dataType = schemas.get(i).getType();
switch (dataType) {
case INT32:
- values[i] = new int[maxBatchSize];
+ values[i] = new int[maxRowNumber];
break;
case INT64:
- values[i] = new long[maxBatchSize];
+ values[i] = new long[maxRowNumber];
break;
case FLOAT:
- values[i] = new float[maxBatchSize];
+ values[i] = new float[maxRowNumber];
break;
case DOUBLE:
- values[i] = new double[maxBatchSize];
+ values[i] = new double[maxRowNumber];
break;
case BOOLEAN:
- values[i] = new boolean[maxBatchSize];
+ values[i] = new boolean[maxRowNumber];
break;
case TEXT:
- values[i] = new Binary[maxBatchSize];
+ values[i] = new Binary[maxRowNumber];
break;
default:
throw new UnSupportedDataTypeException(String.format("Data type %s is not supported.", dataType));
@@ -146,33 +151,36 @@ public class RowBatch {
}
public int getTimeBytesSize() {
- return batchSize * 8;
+ return rowSize * 8;
}
/**
* @return total bytes of values
*/
public int getValueBytesSize() {
- valueOccupation = 0;
+ /**
+ * total byte size that values occupies
+ */
+ int valueOccupation = 0;
for (int i = 0; i < schemas.size(); i++) {
switch (schemas.get(i).getType()) {
case BOOLEAN:
- valueOccupation += batchSize;
+ valueOccupation += rowSize;
break;
case INT32:
- valueOccupation += batchSize * 4;
+ valueOccupation += rowSize * 4;
break;
case INT64:
- valueOccupation += batchSize * 8;
+ valueOccupation += rowSize * 8;
break;
case FLOAT:
- valueOccupation += batchSize * 4;
+ valueOccupation += rowSize * 4;
break;
case DOUBLE:
- valueOccupation += batchSize * 8;
+ valueOccupation += rowSize * 8;
break;
case TEXT:
- valueOccupation += batchSize * 4;
+ valueOccupation += rowSize * 4;
for (Binary value : (Binary[]) values[i]) {
valueOccupation += value.getLength();
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/Schema.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/Schema.java
index 171fb2a..a1cb556 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/Schema.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schema/Schema.java
@@ -20,9 +20,6 @@ package org.apache.iotdb.tsfile.write.schema;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Path;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
-
-import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -52,25 +49,6 @@ public class Schema {
this.registeredTimeseries = knownSchema;
}
- /**
- * Create a row batch to write aligned data
- *
- * @param deviceId the name of the device specified to be written in
- */
- public RowBatch createRowBatch(String deviceId) {
- return new RowBatch(deviceId, new ArrayList<>(registeredTimeseries.values()));
- }
-
- /**
- * Create a row batch to write aligned data
- *
- * @param deviceId the name of the device specified to be written in
- * @param maxBatchSize max size of rows in batch
- */
- public RowBatch createRowBatch(String deviceId, int maxBatchSize) {
- return new RowBatch(deviceId, new ArrayList<>(registeredTimeseries.values()), maxBatchSize);
- }
-
public void registerTimeseries(Path path, MeasurementSchema descriptor) {
this.registeredTimeseries.put(path, descriptor);
}
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/TsFileWriterTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/TsFileWriterTest.java
index a9c7580..ddcec06 100644
--- a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/TsFileWriterTest.java
+++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/TsFileWriterTest.java
@@ -31,8 +31,7 @@ import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.read.expression.QueryExpression;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
-import org.apache.iotdb.tsfile.utils.Pair;
-import org.apache.iotdb.tsfile.write.record.RowBatch;
+import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.FloatDataPoint;
import org.apache.iotdb.tsfile.write.record.datapoint.IntDataPoint;
@@ -192,30 +191,28 @@ public class TsFileWriterTest {
}
@Test
- public void writeRowBatch() throws IOException, WriteProcessException {
- RowBatch rowBatch = new RowBatch("d1", Arrays.asList(new MeasurementSchema[]{
+ public void writeTablet() throws IOException, WriteProcessException {
+ Tablet tablet = new Tablet("d1", Arrays.asList(
new MeasurementSchema("s1", TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY),
- new MeasurementSchema("s2", TSDataType.INT32, TSEncoding.RLE, CompressionType.SNAPPY)
- }));
- rowBatch.timestamps[0] = 10000;
- ((float[])rowBatch.values[0])[0] = 5.0f;
- ((int[])rowBatch.values[1])[0] = 5;
- rowBatch.batchSize = 1;
- writer.write(rowBatch);
+ new MeasurementSchema("s2", TSDataType.INT32, TSEncoding.RLE, CompressionType.SNAPPY)));
+ tablet.timestamps[0] = 10000;
+ ((float[]) tablet.values[0])[0] = 5.0f;
+ ((int[]) tablet.values[1])[0] = 5;
+ tablet.rowSize = 1;
+ writer.write(tablet);
closeFile();
readOneRow();
}
@Test
- public void writeRowBatch2() throws IOException, WriteProcessException {
- RowBatch rowBatch = new RowBatch("d1", Arrays.asList(new MeasurementSchema[]{
+ public void writeTablet2() throws IOException, WriteProcessException {
+ Tablet tablet = new Tablet("d1", Arrays.asList(
new MeasurementSchema("s1", TSDataType.FLOAT, TSEncoding.RLE, CompressionType.SNAPPY),
- new MeasurementSchema("s2", TSDataType.INT32, TSEncoding.RLE, CompressionType.SNAPPY)
- }));
- rowBatch.timestamps[0] = 10000;
- ((float[])rowBatch.values[0])[0] = 5.0f;
- rowBatch.batchSize = 1;
- writer.write(rowBatch);
+ new MeasurementSchema("s2", TSDataType.INT32, TSEncoding.RLE, CompressionType.SNAPPY)));
+ tablet.timestamps[0] = 10000;
+ ((float[]) tablet.values[0])[0] = 5.0f;
+ tablet.rowSize = 1;
+ writer.write(tablet);
closeFile();
//in this case, the value of s2 = 0 at time 10000.
readOneRow(0);