This is an automated email from the ASF dual-hosted git repository. xuekaifeng pushed a commit to branch IOTDB-615-Use-binary-rather-than-string-in-insert-plan in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit d4b3ed687b3c2c9b0a995ec442cd0371708992d9 Author: 151250176 <[email protected]> AuthorDate: Thu Apr 23 13:44:46 2020 +0800 init --- .../main/java/org/apache/iotdb/SessionExample.java | 82 +++++---- .../iotdb/db/engine/memtable/AbstractMemTable.java | 32 ++-- .../db/engine/storagegroup/TsFileProcessor.java | 3 +- .../apache/iotdb/db/qp/executor/PlanExecutor.java | 68 +++++-- .../iotdb/db/qp/physical/crud/InsertPlan.java | 200 ++++++++++++++++++--- .../iotdb/db/qp/strategy/PhysicalGenerator.java | 75 ++++++-- .../org/apache/iotdb/db/service/TSServiceImpl.java | 102 ++++++++--- .../org/apache/iotdb/db/utils/CommonUtils.java | 44 ++++- .../iotdb/db/engine/storagegroup/TTLTest.java | 23 ++- .../org/apache/iotdb/db/tools/WalCheckerTest.java | 11 +- .../apache/iotdb/db/writelog/PerformanceTest.java | 5 +- .../iotdb/db/writelog/WriteLogNodeManagerTest.java | 2 + .../apache/iotdb/db/writelog/WriteLogNodeTest.java | 6 + .../iotdb/db/writelog/io/LogWriterReaderTest.java | 5 +- .../iotdb/db/writelog/recover/LogReplayerTest.java | 19 +- .../db/writelog/recover/SeqTsFileRecoverTest.java | 7 +- .../writelog/recover/UnseqTsFileRecoverTest.java | 26 +-- service-rpc/rpc-changelist.md | 3 +- service-rpc/src/main/thrift/rpc.thrift | 4 +- .../java/org/apache/iotdb/session/Session.java | 107 +++++++++-- .../org/apache/iotdb/session/pool/SessionPool.java | 47 +++-- .../org/apache/iotdb/session/IoTDBSessionIT.java | 96 ++++++---- .../apache/iotdb/session/pool/SessionPoolTest.java | 45 +++-- .../java/org/apache/iotdb/tsfile/utils/Binary.java | 9 +- .../iotdb/tsfile/utils/ReadWriteIOUtils.java | 54 +++++- 25 files changed, 806 insertions(+), 269 deletions(-) diff --git a/example/session/src/main/java/org/apache/iotdb/SessionExample.java b/example/session/src/main/java/org/apache/iotdb/SessionExample.java index 020188f..f3d5b55 100644 --- a/example/session/src/main/java/org/apache/iotdb/SessionExample.java +++ b/example/session/src/main/java/org/apache/iotdb/SessionExample.java @@ -131,15 +131,20 @@ public class SessionExample { private static void insert() throws IoTDBConnectionException, StatementExecutionException { String deviceId = "root.sg1.d1"; List<String> measurements = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); measurements.add("s1"); measurements.add("s2"); measurements.add("s3"); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + for (long time = 0; time < 100; time++) { - List<String> values = new ArrayList<>(); - values.add("1"); - values.add("2"); - values.add("3"); - session.insert(deviceId, time, measurements, values); + List<Object> values = new ArrayList<>(); + values.add(1L); + values.add(2L); + values.add(3L); + session.insert(deviceId, time, measurements, types, values); } } @@ -147,11 +152,16 @@ public class SessionExample { throws IoTDBConnectionException, StatementExecutionException { String deviceId = "root.sg1.d1"; List<String> measurements = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); measurements.add("s1"); measurements.add("s2"); measurements.add("s3"); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + for (long time = 0; time < 100; time++) { - session.insert(deviceId, time, measurements, 1L, 1L, 1L); + session.insert(deviceId, time, measurements, types, 1L, 1L, 1L); } } @@ -163,21 +173,27 @@ public class SessionExample { measurements.add("s3"); List<String> deviceIds = new ArrayList<>(); List<List<String>> measurementsList = new ArrayList<>(); - List<List<String>> valuesList = new ArrayList<>(); + List<List<Object>> valuesList = new ArrayList<>(); List<Long> timestamps = new ArrayList<>(); + List<List<TSDataType>> typesList = new ArrayList<>(); for (long time = 0; time < 500; time++) { - List<String> values = new ArrayList<>(); - values.add("1"); - values.add("2"); - values.add("3"); + List<Object> values = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); + values.add(1L); + values.add(2L); + values.add(3L); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); deviceIds.add(deviceId); measurementsList.add(measurements); valuesList.add(values); + typesList.add(types); timestamps.add(time); if (time != 0 && time % 100 == 0) { - session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList); + session.insertInBatch(deviceIds, timestamps, measurementsList, typesList, valuesList); deviceIds.clear(); measurementsList.clear(); valuesList.clear(); @@ -185,21 +201,16 @@ public class SessionExample { } } - session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList); + session.insertInBatch(deviceIds, timestamps, measurementsList, typesList, valuesList); } - /** * insert a batch data of one device, each batch contains multiple timestamps with values of * sensors - * + * <p> * a RowBatch example: - * - * device1 - * time s1, s2, s3 - * 1, 1, 1, 1 - * 2, 2, 2, 2 - * 3, 3, 3, 3 - * + * <p> + * device1 time s1, s2, s3 1, 1, 1, 1 2, 2, 2, 2 3, 3, 3, 3 + * <p> * Users need to control the count of RowBatch and write a batch when it reaches the maxBatchSize */ private static void insertRowBatch() throws IoTDBConnectionException, BatchExecutionException { @@ -240,29 +251,36 @@ public class SessionExample { 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)); + 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)); + 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 + .extendTemplate("template3", new MeasurementSchema("s3", TSDataType.FLOAT, TSEncoding.RLE)); schema3.registerDevice("root.sg1.d3", "template3"); - + RowBatch rowBatch3 = schema3.createRowBatch("root.sg1.d3", 100); - + Map<String, RowBatch> rowBatchMap = new HashMap<>(); rowBatchMap.put("root.sg1.d1", rowBatch1); rowBatchMap.put("root.sg1.d2", rowBatch2); 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..7a3b2f5 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 @@ -32,7 +32,6 @@ 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.InsertPlan; import org.apache.iotdb.db.rescon.TVListAllocator; -import org.apache.iotdb.db.utils.CommonUtils; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.db.utils.datastructure.TVList; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; @@ -41,12 +40,9 @@ import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; public abstract class AbstractMemTable implements IMemTable { + private final Map<String, Map<String, IWritableMemChunk>> memTableMap; private long version = Long.MAX_VALUE; - private List<Modification> modifications = new ArrayList<>(); - - private final Map<String, Map<String, IWritableMemChunk>> memTableMap; - private long memSize = 0; public AbstractMemTable() { @@ -86,20 +82,14 @@ public abstract class AbstractMemTable implements IMemTable { protected abstract IWritableMemChunk genMemSeries(MeasurementSchema schema); @Override - public void insert(InsertPlan insertPlan) throws WriteProcessException { - try { - for (int i = 0; i < insertPlan.getValues().length; i++) { - - Object value = CommonUtils.parseValue(insertPlan.getSchemas()[i].getType(), - insertPlan.getValues()[i]); + public void insert(InsertPlan insertPlan) { + for (int i = 0; i < insertPlan.getValues().length; i++) { - memSize += MemUtils.getRecordSize(insertPlan.getSchemas()[i].getType(), value); + Object value = insertPlan.getValues()[i]; + memSize += MemUtils.getRecordSize(insertPlan.getSchemas()[i].getType(), value); - write(insertPlan.getDeviceId(), insertPlan.getMeasurements()[i], - insertPlan.getSchemas()[i], insertPlan.getTime(), value); - } - } catch (QueryProcessException e) { - throw new WriteProcessException(e.getMessage()); + write(insertPlan.getDeviceId(), insertPlan.getMeasurements()[i], + insertPlan.getSchemas()[i], insertPlan.getTime(), value); } } @@ -209,14 +199,14 @@ public abstract class AbstractMemTable implements IMemTable { this.modifications.add(deletion); } - public void setVersion(long version) { - this.version = version; - } - public long getVersion() { return version; } + public void setVersion(long version) { + this.version = version; + } + @Override public void release() { for (Entry<String, Map<String, IWritableMemChunk>> entry : memTableMap.entrySet()) { 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..020de24 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 @@ -33,7 +33,6 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.conf.adapter.ActiveTimeSeriesCounter; import org.apache.iotdb.db.conf.adapter.CompressionRatio; import org.apache.iotdb.db.conf.adapter.IoTDBConfigDynamicAdapter; -import org.apache.iotdb.db.engine.cache.RamUsageEstimator; import org.apache.iotdb.db.engine.flush.FlushManager; import org.apache.iotdb.db.engine.flush.MemTableFlushTask; import org.apache.iotdb.db.engine.flush.NotifyFlushMemTable; @@ -56,8 +55,8 @@ import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager; import org.apache.iotdb.db.writelog.node.WriteLogNode; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; -import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata; import org.apache.iotdb.service.rpc.thrift.TSStatus; +import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; import org.apache.iotdb.tsfile.utils.Pair; 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..838a6e7 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 @@ -18,6 +18,37 @@ */ package org.apache.iotdb.db.qp.executor; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_CHILD_PATHS; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_COLUMN; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_COUNT; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_DEVICES; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_ITEM; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PARAMETER; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_PRIVILEGE; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_ROLE; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_STORAGE_GROUP; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES_ALIAS; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES_COMPRESSION; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES_DATATYPE; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TIMESERIES_ENCODING; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_TTL; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_USER; +import static org.apache.iotdb.db.conf.IoTDBConstant.COLUMN_VALUE; +import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR; +import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; import org.apache.iotdb.db.auth.AuthException; import org.apache.iotdb.db.auth.authorizer.IAuthorizer; import org.apache.iotdb.db.auth.authorizer.LocalFileAuthorizer; @@ -45,8 +76,32 @@ import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode; import org.apache.iotdb.db.qp.logical.sys.AuthorOperator; import org.apache.iotdb.db.qp.logical.sys.AuthorOperator.AuthorType; import org.apache.iotdb.db.qp.physical.PhysicalPlan; -import org.apache.iotdb.db.qp.physical.crud.*; -import org.apache.iotdb.db.qp.physical.sys.*; +import org.apache.iotdb.db.qp.physical.crud.AggregationPlan; +import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan; +import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan; +import org.apache.iotdb.db.qp.physical.crud.DeletePlan; +import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.GroupByFillPlan; +import org.apache.iotdb.db.qp.physical.crud.GroupByPlan; +import org.apache.iotdb.db.qp.physical.crud.InsertPlan; +import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.QueryPlan; +import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.UpdatePlan; +import org.apache.iotdb.db.qp.physical.sys.AuthorPlan; +import org.apache.iotdb.db.qp.physical.sys.CountPlan; +import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan; +import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan; +import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan; +import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan; +import org.apache.iotdb.db.qp.physical.sys.OperateFilePlan; +import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan; +import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowChildPathsPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowTTLPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan; import org.apache.iotdb.db.query.context.QueryContext; import org.apache.iotdb.db.query.dataset.AlignByDeviceDataSet; import org.apache.iotdb.db.query.dataset.ListDataSet; @@ -78,12 +133,6 @@ import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.File; -import java.io.IOException; -import java.util.*; -import static org.apache.iotdb.db.conf.IoTDBConstant.*; -import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR; -import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX; public class PlanExecutor implements IPlanExecutor { @@ -709,7 +758,6 @@ public class PlanExecutor implements IPlanExecutor { String[] measurementList = insertPlan.getMeasurements(); String deviceId = insertPlan.getDeviceId(); MNode node = mManager.getDeviceNodeWithAutoCreateStorageGroup(deviceId); - String[] strValues = insertPlan.getValues(); MeasurementSchema[] schemas = new MeasurementSchema[measurementList.length]; for (int i = 0; i < measurementList.length; i++) { @@ -718,7 +766,7 @@ public class PlanExecutor implements IPlanExecutor { if (!IoTDBDescriptor.getInstance().getConfig().isAutoCreateSchemaEnabled()) { throw new PathNotExistException(deviceId + PATH_SEPARATOR + measurement); } - TSDataType dataType = TypeInferenceUtils.getPredictedDataType(strValues[i]); + TSDataType dataType = TypeInferenceUtils.getPredictedDataType(insertPlan.getValues()[i]); Path path = new Path(deviceId, measurement); internalCreateTimeseries(path.toString(), dataType); } diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java index 5f334af..5924a0e 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertPlan.java @@ -25,16 +25,18 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Objects; - import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.qp.logical.Operator; import org.apache.iotdb.db.qp.logical.Operator.OperatorType; import org.apache.iotdb.db.qp.physical.PhysicalPlan; import org.apache.iotdb.db.utils.CommonUtils; import org.apache.iotdb.db.utils.TestOnly; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; import org.apache.iotdb.tsfile.read.TimeValuePair; import org.apache.iotdb.tsfile.read.common.Path; +import org.apache.iotdb.tsfile.utils.Binary; +import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils; import org.apache.iotdb.tsfile.utils.TsPrimitiveType; import org.apache.iotdb.tsfile.write.record.TSRecord; import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; @@ -44,8 +46,11 @@ public class InsertPlan extends PhysicalPlan { private long time; private String deviceId; private String[] measurements; - private String[] values; + private Object[] values; + private TSDataType[] types; private MeasurementSchema[] schemas; + // only for sql + private String[] strValueList; public InsertPlan() { super(false, OperatorType.INSERT); @@ -53,12 +58,38 @@ public class InsertPlan extends PhysicalPlan { } @TestOnly - public InsertPlan(String deviceId, long insertTime, String measurement, String insertValue) { + public InsertPlan(String deviceId, long insertTime, String[] measurements, TSDataType[] types, + String[] insertValues) { + super(false, OperatorType.INSERT); + this.time = insertTime; + this.deviceId = deviceId; + this.measurements = measurements; + + this.types = types; + this.values = new Object[measurements.length]; + for (int i = 0; i < measurements.length; i++) { + try { + values[i] = CommonUtils.parseValueForTest(types[i], insertValues[i]); + } catch (QueryProcessException e) { + e.printStackTrace(); + } + } + canbeSplit = false; + } + + @TestOnly + public InsertPlan(String deviceId, long insertTime, String measurement, TSDataType type, String insertValue) { super(false, OperatorType.INSERT); this.time = insertTime; this.deviceId = deviceId; - this.measurements = new String[] {measurement}; - this.values = new String[] {insertValue}; + this.measurements = new String[]{measurement}; + this.types = new TSDataType[]{type}; + this.values = new Object[1]; + try { + values[0] = CommonUtils.parseValueForTest(types[0], insertValue); + } catch (QueryProcessException e) { + e.printStackTrace(); + } canbeSplit = false; } @@ -68,25 +99,43 @@ public class InsertPlan extends PhysicalPlan { this.time = tsRecord.time; this.measurements = new String[tsRecord.dataPointList.size()]; this.schemas = new MeasurementSchema[tsRecord.dataPointList.size()]; - this.values = new String[tsRecord.dataPointList.size()]; + this.types = new TSDataType[tsRecord.dataPointList.size()]; + this.values = new Object[tsRecord.dataPointList.size()]; for (int i = 0; i < tsRecord.dataPointList.size(); i++) { measurements[i] = tsRecord.dataPointList.get(i).getMeasurementId(); - schemas[i] = new MeasurementSchema(measurements[i], tsRecord.dataPointList.get(i).getType(), TSEncoding.PLAIN); - values[i] = tsRecord.dataPointList.get(i).getValue().toString(); + schemas[i] = new MeasurementSchema(measurements[i], tsRecord.dataPointList.get(i).getType(), + TSEncoding.PLAIN); + types[i] = tsRecord.dataPointList.get(i).getType(); + values[i] = tsRecord.dataPointList.get(i).getValue(); } canbeSplit = false; } + public InsertPlan(String deviceId, long insertTime, String[] measurementList, TSDataType[] types, + Object[] insertValues) { + super(false, Operator.OperatorType.INSERT); + this.time = insertTime; + this.deviceId = deviceId; + this.measurements = measurementList; + this.types = types; + this.values = insertValues; + canbeSplit = false; + } + public InsertPlan(String deviceId, long insertTime, String[] measurementList, String[] insertValues) { super(false, Operator.OperatorType.INSERT); this.time = insertTime; this.deviceId = deviceId; this.measurements = measurementList; - this.values = insertValues; + // build types and values + this.types = new TSDataType[measurements.length]; + this.values = new Object[measurements.length]; + this.strValueList = insertValues; canbeSplit = false; } + public long getTime() { return time; } @@ -101,6 +150,17 @@ public class InsertPlan extends PhysicalPlan { public void setSchemas(MeasurementSchema[] schemas) { this.schemas = schemas; + if (strValueList != null) { + for (int i = 0; i < schemas.length; i++) { + types[i] = schemas[i].getType(); + try { + values[i] = CommonUtils.parseValue(types[i], strValueList[i]); + } catch (QueryProcessException e) { + e.printStackTrace(); + } + } + strValueList = null; + } } @Override @@ -129,11 +189,11 @@ public class InsertPlan extends PhysicalPlan { this.measurements = measurements; } - public String[] getValues() { + public Object[] getValues() { return this.values; } - public void setValues(String[] values) { + public void setValues(Object[] values) { this.values = values; } @@ -174,8 +234,102 @@ public class InsertPlan extends PhysicalPlan { schema.serializeTo(stream); } - for (String m : values) { - putString(stream, m); + try { + putValues(stream); + } catch (QueryProcessException e) { + throw new IOException(e); + } + } + + private void putValues(DataOutputStream outputStream) throws QueryProcessException, IOException { + for (int i = 0; i < values.length; i++) { + ReadWriteIOUtils.write(types[i], outputStream); + switch (types[i]) { + case BOOLEAN: + ReadWriteIOUtils.write((Boolean) values[i], outputStream); + break; + case INT32: + ReadWriteIOUtils.write((Integer) values[i], outputStream); + break; + case INT64: + ReadWriteIOUtils.write((Long) values[i], outputStream); + break; + case FLOAT: + ReadWriteIOUtils.write((Float) values[i], outputStream); + break; + case DOUBLE: + ReadWriteIOUtils.write((Double) values[i], outputStream); + break; + case TEXT: + ReadWriteIOUtils.write((Binary) values[i], outputStream); + break; + default: + throw new QueryProcessException("Unsupported data type:" + types[i]); + } + } + } + + private void putValues(ByteBuffer buffer) throws QueryProcessException { + for (int i = 0; i < values.length; i++) { + ReadWriteIOUtils.write(types[i], buffer); + switch (types[i]) { + case BOOLEAN: + ReadWriteIOUtils.write((Boolean) values[i], buffer); + break; + case INT32: + ReadWriteIOUtils.write((Integer) values[i], buffer); + break; + case INT64: + ReadWriteIOUtils.write((Long) values[i], buffer); + break; + case FLOAT: + ReadWriteIOUtils.write((Float) values[i], buffer); + break; + case DOUBLE: + ReadWriteIOUtils.write((Double) values[i], buffer); + break; + case TEXT: + ReadWriteIOUtils.write((Binary) values[i], buffer); + break; + default: + throw new QueryProcessException("Unsupported data type:" + types[i]); + } + } + } + + public TSDataType[] getTypes() { + return types; + } + + public void setTypes(TSDataType[] types) { + this.types = types; + } + + public void getValues(ByteBuffer buffer) throws QueryProcessException { + for (int i = 0; i < measurements.length; i++) { + types[i] = ReadWriteIOUtils.readDataType(buffer); + switch (types[i]) { + case BOOLEAN: + values[i] = ReadWriteIOUtils.readBool(buffer); + break; + case INT32: + values[i] = ReadWriteIOUtils.readInt(buffer); + break; + case INT64: + values[i] = ReadWriteIOUtils.readLong(buffer); + break; + case FLOAT: + values[i] = ReadWriteIOUtils.readFloat(buffer); + break; + case DOUBLE: + values[i] = ReadWriteIOUtils.readDouble(buffer); + break; + case TEXT: + values[i] = ReadWriteIOUtils.readBinary(buffer); + break; + default: + throw new QueryProcessException("Unsupported data type:" + types[i]); + } } } @@ -193,8 +347,10 @@ public class InsertPlan extends PhysicalPlan { putString(buffer, m); } - for (String m : values) { - putString(buffer, m); + try { + putValues(buffer); + } catch (QueryProcessException e) { + e.printStackTrace(); } } @@ -210,9 +366,12 @@ public class InsertPlan extends PhysicalPlan { measurements[i] = readString(buffer); } - this.values = new String[measurementSize]; - for (int i = 0; i < measurementSize; i++) { - values[i] = readString(buffer); + this.types = new TSDataType[measurementSize]; + this.values = new Object[measurementSize]; + try { + getValues(buffer); + } catch (QueryProcessException e) { + e.printStackTrace(); } } @@ -225,7 +384,8 @@ public class InsertPlan extends PhysicalPlan { if (measurementIndex >= values.length) { return null; } - Object value = CommonUtils.parseValue(schemas[measurementIndex].getType(), values[measurementIndex]); - return new TimeValuePair(time, TsPrimitiveType.getByType(schemas[measurementIndex].getType(), value)); + Object value = values[measurementIndex]; + return new TimeValuePair(time, + TsPrimitiveType.getByType(schemas[measurementIndex].getType(), value)); } } diff --git a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java index 5cfc547..f192091 100644 --- a/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java +++ b/server/src/main/java/org/apache/iotdb/db/qp/strategy/PhysicalGenerator.java @@ -18,6 +18,14 @@ */ package org.apache.iotdb.db.qp.strategy; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; import org.apache.iotdb.db.auth.AuthException; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.query.LogicalOperatorException; @@ -27,21 +35,62 @@ import org.apache.iotdb.db.metadata.MManager; import org.apache.iotdb.db.qp.constant.SQLConstant; import org.apache.iotdb.db.qp.logical.Operator; import org.apache.iotdb.db.qp.logical.Operator.OperatorType; -import org.apache.iotdb.db.qp.logical.crud.*; -import org.apache.iotdb.db.qp.logical.sys.*; +import org.apache.iotdb.db.qp.logical.crud.BasicFunctionOperator; +import org.apache.iotdb.db.qp.logical.crud.DeleteDataOperator; +import org.apache.iotdb.db.qp.logical.crud.FilterOperator; +import org.apache.iotdb.db.qp.logical.crud.InsertOperator; +import org.apache.iotdb.db.qp.logical.crud.QueryOperator; +import org.apache.iotdb.db.qp.logical.sys.AuthorOperator; +import org.apache.iotdb.db.qp.logical.sys.CountOperator; +import org.apache.iotdb.db.qp.logical.sys.CreateTimeSeriesOperator; +import org.apache.iotdb.db.qp.logical.sys.DataAuthOperator; +import org.apache.iotdb.db.qp.logical.sys.DeleteStorageGroupOperator; +import org.apache.iotdb.db.qp.logical.sys.DeleteTimeSeriesOperator; +import org.apache.iotdb.db.qp.logical.sys.LoadDataOperator; +import org.apache.iotdb.db.qp.logical.sys.LoadFilesOperator; +import org.apache.iotdb.db.qp.logical.sys.MoveFileOperator; +import org.apache.iotdb.db.qp.logical.sys.RemoveFileOperator; +import org.apache.iotdb.db.qp.logical.sys.SetStorageGroupOperator; +import org.apache.iotdb.db.qp.logical.sys.SetTTLOperator; +import org.apache.iotdb.db.qp.logical.sys.ShowChildPathsOperator; +import org.apache.iotdb.db.qp.logical.sys.ShowDevicesOperator; +import org.apache.iotdb.db.qp.logical.sys.ShowTTLOperator; +import org.apache.iotdb.db.qp.logical.sys.ShowTimeSeriesOperator; import org.apache.iotdb.db.qp.physical.PhysicalPlan; -import org.apache.iotdb.db.qp.physical.crud.*; +import org.apache.iotdb.db.qp.physical.crud.AggregationPlan; +import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan; import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType; -import org.apache.iotdb.db.qp.physical.sys.*; +import org.apache.iotdb.db.qp.physical.crud.DeletePlan; +import org.apache.iotdb.db.qp.physical.crud.FillQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.GroupByFillPlan; +import org.apache.iotdb.db.qp.physical.crud.GroupByPlan; +import org.apache.iotdb.db.qp.physical.crud.InsertPlan; +import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.QueryPlan; +import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan; +import org.apache.iotdb.db.qp.physical.sys.AuthorPlan; +import org.apache.iotdb.db.qp.physical.sys.CountPlan; +import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan; +import org.apache.iotdb.db.qp.physical.sys.DataAuthPlan; +import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan; +import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan; +import org.apache.iotdb.db.qp.physical.sys.LoadConfigurationPlan; +import org.apache.iotdb.db.qp.physical.sys.LoadDataPlan; +import org.apache.iotdb.db.qp.physical.sys.OperateFilePlan; +import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan; +import org.apache.iotdb.db.qp.physical.sys.SetTTLPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowChildPathsPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowDevicesPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowPlan; import org.apache.iotdb.db.qp.physical.sys.ShowPlan.ShowContentType; +import org.apache.iotdb.db.qp.physical.sys.ShowTTLPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowTimeSeriesPlan; import org.apache.iotdb.db.utils.SchemaUtils; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.common.Path; import org.apache.iotdb.tsfile.read.expression.IExpression; import org.apache.iotdb.tsfile.utils.Pair; -import java.util.*; - /** * Used to convert logical operator to physical plan */ @@ -93,6 +142,7 @@ public class PhysicalGenerator { throw new LogicalOperatorException( "For Insert command, cannot specified more than one seriesPath: " + paths); } + return new InsertPlan(paths.get(0).getFullPath(), insert.getTime(), insert.getMeasurementList(), insert.getValueList()); @@ -188,14 +238,14 @@ public class PhysicalGenerator { ((GroupByFillPlan) queryPlan).setSlidingStep(queryOperator.getSlidingStep()); ((GroupByFillPlan) queryPlan).setLeftCRightO(queryOperator.isLeftCRightO()); if (!queryOperator.isLeftCRightO()) { - ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime()+1); - ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime()+1); + ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime() + 1); + ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime() + 1); } else { ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime()); ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime()); } ((GroupByFillPlan) queryPlan) - .setAggregations(queryOperator.getSelectOperator().getAggregations()); + .setAggregations(queryOperator.getSelectOperator().getAggregations()); for (String aggregation : queryPlan.getAggregations()) { if (!SQLConstant.LAST_VALUE.equals(aggregation)) { throw new QueryProcessException("Group By Fill only support last_value function"); @@ -208,8 +258,8 @@ public class PhysicalGenerator { ((GroupByPlan) queryPlan).setSlidingStep(queryOperator.getSlidingStep()); ((GroupByPlan) queryPlan).setLeftCRightO(queryOperator.isLeftCRightO()); if (!queryOperator.isLeftCRightO()) { - ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime()+1); - ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime()+1); + ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime() + 1); + ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime() + 1); } else { ((GroupByPlan) queryPlan).setStartTime(queryOperator.getStartTime()); ((GroupByPlan) queryPlan).setEndTime(queryOperator.getEndTime()); @@ -490,7 +540,8 @@ public class PhysicalGenerator { Set<String> columnSet = new HashSet<>(); int index = 0; for (Pair<Path, Integer> indexedPath : indexedPaths) { - String column = aggregations.get(indexedPath.right) + "(" + indexedPath.left.toString() + ")"; + String column = + aggregations.get(indexedPath.right) + "(" + indexedPath.left.toString() + ")"; if (!columnSet.contains(column)) { aggregationPlan.addDeduplicatedPaths(indexedPath.left); TSDataType seriesType = dataTypes.get(indexedPath.right); 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..ac65ac0 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java @@ -18,6 +18,24 @@ */ package org.apache.iotdb.db.service; +import static org.apache.iotdb.db.conf.IoTDBConfig.PATH_PATTERN; +import static org.apache.iotdb.db.qp.physical.sys.ShowPlan.ShowContentType.TIMESERIES; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.sql.SQLException; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; import org.antlr.v4.runtime.misc.ParseCancellationException; import org.apache.iotdb.db.auth.AuthException; import org.apache.iotdb.db.auth.AuthorityChecker; @@ -43,9 +61,19 @@ import org.apache.iotdb.db.qp.executor.IPlanExecutor; import org.apache.iotdb.db.qp.executor.PlanExecutor; import org.apache.iotdb.db.qp.logical.Operator.OperatorType; import org.apache.iotdb.db.qp.physical.PhysicalPlan; -import org.apache.iotdb.db.qp.physical.crud.*; +import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan; import org.apache.iotdb.db.qp.physical.crud.AlignByDevicePlan.MeasurementType; -import org.apache.iotdb.db.qp.physical.sys.*; +import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan; +import org.apache.iotdb.db.qp.physical.crud.DeletePlan; +import org.apache.iotdb.db.qp.physical.crud.InsertPlan; +import org.apache.iotdb.db.qp.physical.crud.LastQueryPlan; +import org.apache.iotdb.db.qp.physical.crud.QueryPlan; +import org.apache.iotdb.db.qp.physical.sys.AuthorPlan; +import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan; +import org.apache.iotdb.db.qp.physical.sys.DeleteStorageGroupPlan; +import org.apache.iotdb.db.qp.physical.sys.DeleteTimeSeriesPlan; +import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan; +import org.apache.iotdb.db.qp.physical.sys.ShowPlan; import org.apache.iotdb.db.query.context.QueryContext; import org.apache.iotdb.db.query.control.QueryResourceManager; import org.apache.iotdb.db.query.dataset.NonAlignEngineDataSet; @@ -55,9 +83,34 @@ import org.apache.iotdb.db.tools.watermark.WatermarkEncoder; import org.apache.iotdb.db.utils.QueryDataSetUtils; import org.apache.iotdb.db.utils.SchemaUtils; import org.apache.iotdb.rpc.RpcUtils; -import org.apache.iotdb.rpc.StatementExecutionException; import org.apache.iotdb.rpc.TSStatusCode; -import org.apache.iotdb.service.rpc.thrift.*; +import org.apache.iotdb.service.rpc.thrift.ServerProperties; +import org.apache.iotdb.service.rpc.thrift.TSBatchInsertionReq; +import org.apache.iotdb.service.rpc.thrift.TSCancelOperationReq; +import org.apache.iotdb.service.rpc.thrift.TSCloseOperationReq; +import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq; +import org.apache.iotdb.service.rpc.thrift.TSCreateMultiTimeseriesReq; +import org.apache.iotdb.service.rpc.thrift.TSCreateTimeseriesReq; +import org.apache.iotdb.service.rpc.thrift.TSDeleteDataReq; +import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementReq; +import org.apache.iotdb.service.rpc.thrift.TSExecuteBatchStatementResp; +import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementReq; +import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp; +import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataReq; +import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataResp; +import org.apache.iotdb.service.rpc.thrift.TSFetchResultsReq; +import org.apache.iotdb.service.rpc.thrift.TSFetchResultsResp; +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.TSOpenSessionReq; +import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp; +import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion; +import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet; +import org.apache.iotdb.service.rpc.thrift.TSQueryNonAlignDataSet; +import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq; +import org.apache.iotdb.service.rpc.thrift.TSStatus; import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException; import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; @@ -70,18 +123,6 @@ import org.apache.thrift.server.ServerContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.nio.ByteBuffer; -import java.sql.SQLException; -import java.time.ZoneId; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicLong; -import java.util.stream.Collectors; - -import static org.apache.iotdb.db.conf.IoTDBConfig.PATH_PATTERN; -import static org.apache.iotdb.db.qp.physical.sys.ShowPlan.ShowContentType.TIMESERIES; - /** * Thrift RPC implementation at server side. @@ -1049,15 +1090,22 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { InsertPlan plan = new InsertPlan(); for (int i = 0; i < req.deviceIds.size(); i++) { - plan.setDeviceId(req.getDeviceIds().get(i)); - plan.setTime(req.getTimestamps().get(i)); - plan.setMeasurements(req.getMeasurementsList().get(i).toArray(new String[0])); - plan.setValues(req.getValuesList().get(i).toArray(new String[0])); - TSStatus status = checkAuthority(plan, req.getSessionId()); - if (status != null) { - resp.addToStatusList(status); - } else { - resp.addToStatusList(executePlan(plan)); + try { + plan.setDeviceId(req.getDeviceIds().get(i)); + plan.setTime(req.getTimestamps().get(i)); + plan.setMeasurements(req.getMeasurementsList().get(i).toArray(new String[0])); + plan.setTypes(new TSDataType[plan.getMeasurements().length]); + plan.setValues(new Object[plan.getMeasurements().length]); + plan.getValues(req.valuesList.get(i)); + TSStatus status = checkAuthority(plan, req.getSessionId()); + if (status != null) { + resp.addToStatusList(status); + } else { + resp.addToStatusList(executePlan(plan)); + } + } + catch (Exception e) { + logger.error("meet error when insert in batch", e); } } @@ -1094,7 +1142,9 @@ public class TSServiceImpl implements TSIService.Iface, ServerContext { plan.setDeviceId(req.getDeviceId()); plan.setTime(req.getTimestamp()); plan.setMeasurements(req.getMeasurements().toArray(new String[0])); - plan.setValues(req.getValues().toArray(new String[0])); + plan.setTypes(new TSDataType[plan.getMeasurements().length]); + plan.setValues(new Object[plan.getMeasurements().length]); + plan.getValues(req.values); TSStatus status = checkAuthority(plan, req.getSessionId()); if (status != null) { diff --git a/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java index 3f9e43a..c6575de 100644 --- a/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java +++ b/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java @@ -31,7 +31,8 @@ import org.apache.iotdb.tsfile.utils.Binary; public class CommonUtils { - private CommonUtils(){} + private CommonUtils() { + } /** * get JDK version. @@ -72,7 +73,8 @@ public class CommonUtils { switch (dataType) { case BOOLEAN: value = value.toLowerCase(); - if (SQLConstant.BOOLEAN_FALSE_NUM.equals(value) || SQLConstant.BOOLEN_FALSE.equals(value)) { + if (SQLConstant.BOOLEAN_FALSE_NUM.equals(value) || SQLConstant.BOOLEN_FALSE + .equals(value)) { return false; } if (SQLConstant.BOOLEAN_TRUE_NUM.equals(value) || SQLConstant.BOOLEN_TRUE.equals(value)) { @@ -89,14 +91,48 @@ public class CommonUtils { return Double.parseDouble(value); case TEXT: if ((value.startsWith(SQLConstant.QUOTE) && value.endsWith(SQLConstant.QUOTE)) - || (value.startsWith(SQLConstant.DQUOTE) && value.endsWith(SQLConstant.DQUOTE))) { + || (value.startsWith(SQLConstant.DQUOTE) && value.endsWith(SQLConstant.DQUOTE))) { if (value.length() == 1) { return new Binary(value); } else { return new Binary(value.substring(1, value.length() - 1)); } } - throw new QueryProcessException("The TEXT data type should be covered by \" or '"); + + return new Binary(value); + default: + throw new QueryProcessException("Unsupported data type:" + dataType); + } + } catch (NumberFormatException e) { + throw new QueryProcessException(e.getMessage()); + } + } + + @TestOnly + public static Object parseValueForTest(TSDataType dataType, String value) + throws QueryProcessException { + try { + switch (dataType) { + case BOOLEAN: + value = value.toLowerCase(); + if (SQLConstant.BOOLEAN_FALSE_NUM.equals(value) || SQLConstant.BOOLEN_FALSE + .equals(value)) { + return false; + } + if (SQLConstant.BOOLEAN_TRUE_NUM.equals(value) || SQLConstant.BOOLEN_TRUE.equals(value)) { + return true; + } + throw new QueryProcessException("The BOOLEAN should be true/TRUE, false/FALSE or 0/1"); + case INT32: + return Integer.parseInt(value); + case INT64: + return Long.parseLong(value); + case FLOAT: + return Float.parseFloat(value); + case DOUBLE: + return Double.parseDouble(value); + case TEXT: + return new Binary(value); default: throw new QueryProcessException("Unsupported data type:" + dataType); } diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java index f627484..18598a4 100644 --- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java +++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java @@ -20,6 +20,17 @@ package org.apache.iotdb.db.engine.storagegroup; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; import org.apache.iotdb.db.conf.IoTDBConstant; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.conf.directories.DirectoryManager; @@ -57,12 +68,6 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; -import java.io.File; -import java.io.IOException; -import java.util.*; - -import static org.junit.Assert.*; - public class TTLTest { private String sg1 = "root.TTL_SG1"; @@ -126,7 +131,8 @@ public class TTLTest { insertPlan.setDeviceId(sg1); insertPlan.setTime(System.currentTimeMillis()); insertPlan.setMeasurements(new String[]{"s1"}); - insertPlan.setValues(new String[]{"1"}); + insertPlan.setTypes(new TSDataType[]{TSDataType.INT64}); + insertPlan.setValues(new Object[]{1L}); insertPlan.setSchemas( new MeasurementSchema[]{new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN)}); @@ -152,7 +158,8 @@ public class TTLTest { insertPlan.setDeviceId(sg1); insertPlan.setTime(System.currentTimeMillis()); insertPlan.setMeasurements(new String[]{"s1"}); - insertPlan.setValues(new String[]{"1"}); + insertPlan.setTypes(new TSDataType[]{TSDataType.INT64}); + insertPlan.setValues(new Object[]{1L}); insertPlan.setSchemas( new MeasurementSchema[]{new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN)}); diff --git a/server/src/test/java/org/apache/iotdb/db/tools/WalCheckerTest.java b/server/src/test/java/org/apache/iotdb/db/tools/WalCheckerTest.java index 1e7401a..2dc30bc 100644 --- a/server/src/test/java/org/apache/iotdb/db/tools/WalCheckerTest.java +++ b/server/src/test/java/org/apache/iotdb/db/tools/WalCheckerTest.java @@ -32,6 +32,7 @@ import org.apache.iotdb.db.constant.TestConstant; import org.apache.iotdb.db.exception.SystemCheckException; import org.apache.iotdb.db.qp.physical.crud.InsertPlan; import org.apache.iotdb.db.writelog.io.LogWriter; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.junit.Test; public class WalCheckerTest { @@ -73,12 +74,13 @@ public class WalCheckerTest { LogWriter logWriter = new LogWriter(subDir.getPath() + File.separator + WAL_FILE_NAME); - ByteBuffer binaryPlans = ByteBuffer.allocate(64*1024); + ByteBuffer binaryPlans = ByteBuffer.allocate(64 * 1024); String deviceId = "device1"; String[] measurements = new String[]{"s1", "s2", "s3"}; + TSDataType[] types = new TSDataType[]{TSDataType.INT64, TSDataType.INT64, TSDataType.INT64}; String[] values = new String[]{"5", "6", "7"}; for (int j = 0; j < 10; j++) { - new InsertPlan(deviceId, j, measurements, values).serializeTo(binaryPlans); + new InsertPlan(deviceId, j, measurements, types, values).serializeTo(binaryPlans); } binaryPlans.flip(); logWriter.write(binaryPlans); @@ -106,12 +108,13 @@ public class WalCheckerTest { LogWriter logWriter = new LogWriter(subDir.getPath() + File.separator + WAL_FILE_NAME); - ByteBuffer binaryPlans = ByteBuffer.allocate(64*1024); + ByteBuffer binaryPlans = ByteBuffer.allocate(64 * 1024); String deviceId = "device1"; String[] measurements = new String[]{"s1", "s2", "s3"}; + TSDataType[] types = new TSDataType[]{TSDataType.INT64, TSDataType.INT64, TSDataType.INT64}; String[] values = new String[]{"5", "6", "7"}; for (int j = 0; j < 10; j++) { - new InsertPlan(deviceId, j, measurements, values).serializeTo(binaryPlans); + new InsertPlan(deviceId, j, measurements, types, values).serializeTo(binaryPlans); } if (i > 2) { binaryPlans.put("not a wal".getBytes()); diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/PerformanceTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/PerformanceTest.java index b118f05..269f93c 100644 --- a/server/src/test/java/org/apache/iotdb/db/writelog/PerformanceTest.java +++ b/server/src/test/java/org/apache/iotdb/db/writelog/PerformanceTest.java @@ -88,6 +88,7 @@ public class PerformanceTest { for (int i = 0; i < 1000000; i++) { InsertPlan bwInsertPlan = new InsertPlan("logTestDevice", 100, new String[]{"s1", "s2", "s3", "s4"}, + new TSDataType[]{TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN}, new String[]{"1.0", "15", "str", "false"}); UpdatePlan updatePlan = new UpdatePlan(0, 100, "2.0", new Path("root.logTestDevice.s1")); @@ -147,7 +148,9 @@ public class PerformanceTest { for (int i = 0; i < 1000000; i++) { InsertPlan bwInsertPlan = new InsertPlan("root.logTestDevice", 100, - new String[]{"s1", "s2", "s3", "s4"}, new String[]{"1.0", "15", "str", "false"}); + new String[]{"s1", "s2", "s3", "s4"}, + new TSDataType[]{TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN}, + new String[]{"1.0", "15", "str", "false"}); UpdatePlan updatePlan = new UpdatePlan(0, 100, "2.0", new Path("root.logTestDevice.s1")); DeletePlan deletePlan = new DeletePlan(50, new Path("root.logTestDevice.s1")); diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeManagerTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeManagerTest.java index 4aa8297..7125d88 100644 --- a/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeManagerTest.java +++ b/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeManagerTest.java @@ -33,6 +33,7 @@ import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager; import org.apache.iotdb.db.writelog.manager.WriteLogNodeManager; import org.apache.iotdb.db.writelog.node.WriteLogNode; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.common.Path; import org.junit.After; import org.junit.Before; @@ -86,6 +87,7 @@ public class WriteLogNodeManagerTest { InsertPlan bwInsertPlan = new InsertPlan("logTestDevice", 100, new String[]{"s1", "s2", "s3", "s4"}, + new TSDataType[]{TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN}, new String[]{"1.0", "15", "str", "false"}); DeletePlan deletePlan = new DeletePlan(50, new Path("root.logTestDevice.s1")); diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java index 63a5931..953a76e 100644 --- a/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java +++ b/server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java @@ -32,6 +32,7 @@ import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.writelog.io.ILogReader; import org.apache.iotdb.db.writelog.node.ExclusiveWriteLogNode; import org.apache.iotdb.db.writelog.node.WriteLogNode; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.common.Path; import org.junit.After; import org.junit.Before; @@ -66,6 +67,7 @@ public class WriteLogNodeTest { InsertPlan bwInsertPlan = new InsertPlan(identifier, 100, new String[]{"s1", "s2", "s3", "s4"}, + new TSDataType[]{TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN}, new String[]{"1.0", "15", "str", "false"}); DeletePlan deletePlan = new DeletePlan(50, new Path(identifier + ".s1")); @@ -96,6 +98,7 @@ public class WriteLogNodeTest { InsertPlan bwInsertPlan = new InsertPlan(identifier, 100, new String[]{"s1", "s2", "s3", "s4"}, + new TSDataType[]{TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN}, new String[]{"1.0", "15", "str", "false"}); DeletePlan deletePlan = new DeletePlan(50, new Path(identifier + ".s1")); @@ -132,6 +135,7 @@ public class WriteLogNodeTest { InsertPlan bwInsertPlan = new InsertPlan("root.logTestDevice", 100, new String[]{"s1", "s2", "s3", "s4"}, + new TSDataType[]{TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN}, new String[]{"1.0", "15", "str", "false"}); DeletePlan deletePlan = new DeletePlan(50, new Path("root.logTestDevice.s1")); @@ -157,6 +161,7 @@ public class WriteLogNodeTest { InsertPlan bwInsertPlan = new InsertPlan("logTestDevice", 100, new String[]{"s1", "s2", "s3", "s4"}, + new TSDataType[]{TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN}, new String[]{"1.0", "15", "str", "false"}); DeletePlan deletePlan = new DeletePlan(50, new Path("root.logTestDevice.s1")); @@ -181,6 +186,7 @@ public class WriteLogNodeTest { InsertPlan bwInsertPlan = new InsertPlan("root.logTestDevice.oversize", 100, new String[]{"s1", "s2", "s3", "s4"}, + new TSDataType[]{TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN}, new String[]{"1.0", "15", new String(new char[65 * 1024 * 1024]), "false"}); boolean caught = false; diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/io/LogWriterReaderTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/io/LogWriterReaderTest.java index 6dea9ac..b2e087e 100644 --- a/server/src/test/java/org/apache/iotdb/db/writelog/io/LogWriterReaderTest.java +++ b/server/src/test/java/org/apache/iotdb/db/writelog/io/LogWriterReaderTest.java @@ -29,6 +29,7 @@ import java.util.List; import org.apache.iotdb.db.qp.physical.PhysicalPlan; import org.apache.iotdb.db.qp.physical.crud.DeletePlan; import org.apache.iotdb.db.qp.physical.crud.InsertPlan; +import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.common.Path; import org.junit.Before; import org.junit.Test; @@ -36,7 +37,7 @@ import org.junit.Test; public class LogWriterReaderTest { private static String filePath = "logtest.test"; - ByteBuffer logsBuffer = ByteBuffer.allocate(64*1024); + ByteBuffer logsBuffer = ByteBuffer.allocate(64 * 1024); List<PhysicalPlan> plans = new ArrayList<>(); @Before @@ -45,8 +46,10 @@ public class LogWriterReaderTest { new File(filePath).delete(); } InsertPlan insertPlan1 = new InsertPlan("d1", 10L, new String[]{"s1", "s2"}, + new TSDataType[]{TSDataType.INT64, TSDataType.INT64}, new String[]{"1", "2"}); InsertPlan insertPlan2 = new InsertPlan("d1", 10L, new String[]{"s1", "s2"}, + new TSDataType[]{TSDataType.INT64, TSDataType.INT64}, new String[]{"1", "2"}); DeletePlan deletePlan = new DeletePlan(10L, new Path("root.d1.s1")); plans.add(insertPlan1); diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/recover/LogReplayerTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/recover/LogReplayerTest.java index fc1005c..37e263b 100644 --- a/server/src/test/java/org/apache/iotdb/db/writelog/recover/LogReplayerTest.java +++ b/server/src/test/java/org/apache/iotdb/db/writelog/recover/LogReplayerTest.java @@ -36,10 +36,10 @@ import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk; import org.apache.iotdb.db.engine.storagegroup.TsFileResource; import org.apache.iotdb.db.engine.version.VersionController; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.StorageGroupProcessorException; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.metadata.MManager; -import org.apache.iotdb.db.exception.StorageGroupProcessorException; import org.apache.iotdb.db.qp.physical.crud.DeletePlan; import org.apache.iotdb.db.qp.physical.crud.InsertPlan; import org.apache.iotdb.db.utils.EnvironmentUtils; @@ -48,10 +48,9 @@ import org.apache.iotdb.db.writelog.node.WriteLogNode; import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; -import org.apache.iotdb.tsfile.read.reader.IPointReader; import org.apache.iotdb.tsfile.read.TimeValuePair; import org.apache.iotdb.tsfile.read.common.Path; -import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.apache.iotdb.tsfile.read.reader.IPointReader; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -105,10 +104,13 @@ public class LogReplayerTest { WriteLogNode node = MultiFileLogNodeManager.getInstance().getNode(logNodePrefix + tsFile.getName()); - node.write(new InsertPlan("root.sg.device0", 100, "sensor0", String.valueOf(0))); - node.write(new InsertPlan("root.sg.device0", 2, "sensor1", String.valueOf(0))); + node.write( + new InsertPlan("root.sg.device0", 100, "sensor0", TSDataType.INT64, String.valueOf(0))); + node.write( + new InsertPlan("root.sg.device0", 2, "sensor1", TSDataType.INT64, String.valueOf(0))); for (int i = 1; i < 5; i++) { - node.write(new InsertPlan("root.sg.device" + i, i, "sensor" + i, String.valueOf(i))); + node.write(new InsertPlan("root.sg.device" + i, i, "sensor" + i, TSDataType.INT64, + String.valueOf(i))); } DeletePlan deletePlan = new DeletePlan(200, new Path("root.sg.device0", "sensor0")); node.write(deletePlan); @@ -117,8 +119,9 @@ public class LogReplayerTest { replayer.replayLogs(); for (int i = 0; i < 5; i++) { - ReadOnlyMemChunk memChunk = memTable.query("root.sg.device" + i, "sensor" + i, TSDataType.INT64, - TSEncoding.RLE, Collections.emptyMap(), Long.MIN_VALUE); + ReadOnlyMemChunk memChunk = memTable + .query("root.sg.device" + i, "sensor" + i, TSDataType.INT64, + TSEncoding.RLE, Collections.emptyMap(), Long.MIN_VALUE); IPointReader iterator = memChunk.getPointReader(); if (i == 0) { assertFalse(iterator.hasNextTimeValuePair()); diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/recover/SeqTsFileRecoverTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/recover/SeqTsFileRecoverTest.java index 5c51d9a..ac5ae05 100644 --- a/server/src/test/java/org/apache/iotdb/db/writelog/recover/SeqTsFileRecoverTest.java +++ b/server/src/test/java/org/apache/iotdb/db/writelog/recover/SeqTsFileRecoverTest.java @@ -142,17 +142,20 @@ public class SeqTsFileRecoverTest { for (int i = 10; i < 20; i++) { for (int j = 0; j < 10; j++) { String[] measurements = new String[10]; + TSDataType[] types = new TSDataType[10]; String[] values = new String[10]; for (int k = 0; k < 10; k++) { measurements[k] = "sensor" + k; + types[k] = TSDataType.INT64; values[k] = String.valueOf(k); } - InsertPlan insertPlan = new InsertPlan("root.sg.device" + j, i, measurements, values); + InsertPlan insertPlan = new InsertPlan("root.sg.device" + j, i, measurements, types, + values); node.write(insertPlan); } node.notifyStartFlush(); } - + resource = new TsFileResource(tsF); } diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/recover/UnseqTsFileRecoverTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/recover/UnseqTsFileRecoverTest.java index b285dd0..016fb60 100644 --- a/server/src/test/java/org/apache/iotdb/db/writelog/recover/UnseqTsFileRecoverTest.java +++ b/server/src/test/java/org/apache/iotdb/db/writelog/recover/UnseqTsFileRecoverTest.java @@ -19,6 +19,10 @@ package org.apache.iotdb.db.writelog.recover; +import static org.junit.Assert.assertEquals; + +import java.io.File; +import java.io.IOException; import java.util.Collections; import org.apache.commons.io.FileUtils; import org.apache.iotdb.db.conf.adapter.ActiveTimeSeriesCounter; @@ -53,17 +57,12 @@ import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader; 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.schema.Schema; import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; +import org.apache.iotdb.tsfile.write.schema.Schema; import org.junit.After; import org.junit.Before; import org.junit.Test; -import java.io.File; -import java.io.IOException; - -import static org.junit.Assert.assertEquals; - public class UnseqTsFileRecoverTest { private File tsF; @@ -96,7 +95,8 @@ public class UnseqTsFileRecoverTest { for (int i = 0; i < 10; i++) { for (int j = 0; j < 10; j++) { Path path = new Path(("root.sg.device" + i), ("sensor" + j)); - MeasurementSchema measurementSchema = new MeasurementSchema("sensor" + j, TSDataType.INT64, TSEncoding.PLAIN); + MeasurementSchema measurementSchema = new MeasurementSchema("sensor" + j, TSDataType.INT64, + TSEncoding.PLAIN); schema.registerTimeseries(path, measurementSchema); MManager.getInstance().createTimeseries(path.getFullPath(), measurementSchema.getType(), measurementSchema.getEncodingType(), measurementSchema.getCompressor(), @@ -145,19 +145,22 @@ public class UnseqTsFileRecoverTest { for (int i = 0; i < 10; i++) { for (int j = 0; j < 10; j++) { String[] measurements = new String[10]; + TSDataType[] types = new TSDataType[10]; String[] values = new String[10]; for (int k = 0; k < 10; k++) { measurements[k] = "sensor" + k; + types[k] = TSDataType.INT64; values[k] = String.valueOf(k + 10); } - InsertPlan insertPlan = new InsertPlan("root.sg.device" + j, i, measurements, values); + InsertPlan insertPlan = new InsertPlan("root.sg.device" + j, i, measurements, types, + values); node.write(insertPlan); } node.notifyStartFlush(); } - InsertPlan insertPlan = new InsertPlan("root.sg.device99", 1, "sensor4", "4"); + InsertPlan insertPlan = new InsertPlan("root.sg.device99", 1, "sensor4", TSDataType.INT64, "4"); node.write(insertPlan); - insertPlan = new InsertPlan("root.sg.device99", 300, "sensor2", "2"); + insertPlan = new InsertPlan("root.sg.device99", 300, "sensor2", TSDataType.INT64, "2"); node.write(insertPlan); node.close(); @@ -176,7 +179,8 @@ public class UnseqTsFileRecoverTest { public void test() throws StorageGroupProcessorException, IOException { TsFileRecoverPerformer performer = new TsFileRecoverPerformer(logNodePrefix, versionController, resource, true, false); - ActiveTimeSeriesCounter.getInstance().init(resource.getFile().getParentFile().getParentFile().getName()); + ActiveTimeSeriesCounter.getInstance() + .init(resource.getFile().getParentFile().getParentFile().getName()); performer.recover(); assertEquals(1, (long) resource.getStartTimeMap().get("root.sg.device99")); diff --git a/service-rpc/rpc-changelist.md b/service-rpc/rpc-changelist.md index 5a15ea8..43f5fcc 100644 --- a/service-rpc/rpc-changelist.md +++ b/service-rpc/rpc-changelist.md @@ -120,4 +120,5 @@ Last Updated on November 12th, 2019 by Tian Jiang. | 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 +| Remove sessionId in TSExecuteBatchStatementResp | Jialin Qiao | +| Use TsDataType and binary rather than string in TSInsertInBatchReq and TSInsertReq | Kaifeng Xue | \ 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..47bb22c 100644 --- a/service-rpc/src/main/thrift/rpc.thrift +++ b/service-rpc/src/main/thrift/rpc.thrift @@ -169,7 +169,7 @@ struct TSInsertReq { 1: required i64 sessionId 2: required string deviceId 3: required list<string> measurements - 4: required list<string> values + 4: required binary values 5: required i64 timestamp } @@ -187,7 +187,7 @@ struct TSInsertInBatchReq { 1: required i64 sessionId 2: required list<string> deviceIds 3: required list<list<string>> measurementsList - 4: required list<list<string>> valuesList + 4: required list<binary> valuesList 5: required list<i64> timestamps } 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..0210d10 100644 --- a/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/session/src/main/java/org/apache/iotdb/session/Session.java @@ -18,6 +18,7 @@ */ package org.apache.iotdb.session; +import java.nio.ByteBuffer; import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; @@ -44,11 +45,13 @@ import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp; import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion; import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq; import org.apache.iotdb.service.rpc.thrift.TSStatus; +import org.apache.iotdb.tsfile.common.conf.TSFileConfig; import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException; 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.utils.ReadWriteIOUtils; import org.apache.iotdb.tsfile.write.record.RowBatch; import org.apache.iotdb.tsfile.write.schema.MeasurementSchema; import org.apache.thrift.TException; @@ -361,7 +364,8 @@ public class Session { * @see Session#insertBatch(RowBatch) */ public void insertInBatch(List<String> deviceIds, List<Long> times, - List<List<String>> measurementsList, List<List<String>> valuesList) + List<List<String>> measurementsList, List<List<TSDataType>> typesList, + List<List<Object>> valuesList) throws IoTDBConnectionException, BatchExecutionException { // check params size int len = deviceIds.size(); @@ -375,7 +379,14 @@ public class Session { request.setDeviceIds(deviceIds); request.setTimestamps(times); request.setMeasurementsList(measurementsList); - request.setValuesList(valuesList); + List<ByteBuffer> buffersList = new ArrayList<>(); + for (int i = 0; i < measurementsList.size(); i++) { + ByteBuffer buffer = ByteBuffer.allocate(calculateLength(typesList.get(i), valuesList.get(i))); + putValues(typesList.get(i), valuesList.get(i), buffer); + buffer.flip(); + buffersList.add(buffer); + } + request.setValuesList(buffersList); try { RpcUtils.verifySuccess(client.insertRowInBatch(request).statusList); @@ -388,34 +399,36 @@ public class Session { * 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#insertInBatch(List, List, List, List, List) * @see Session#insertBatch(RowBatch) */ public TSStatus insert(String deviceId, long time, List<String> measurements, + List<TSDataType> types, Object... values) throws IoTDBConnectionException, StatementExecutionException { - List<String> stringValues = new ArrayList<>(); - for (Object o : values) { - stringValues.add(o.toString()); - } + List<Object> valuesList = new ArrayList<>(Arrays.asList(values)); - return insert(deviceId, time, measurements, stringValues); + return insert(deviceId, time, measurements, types, valuesList); } /** * 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#insertInBatch(List, List, List, List, List) * @see Session#insertBatch(RowBatch) */ public TSStatus insert(String deviceId, long time, List<String> measurements, - List<String> values) throws IoTDBConnectionException, StatementExecutionException { + List<TSDataType> types, + List<Object> values) throws IoTDBConnectionException, StatementExecutionException { TSInsertReq request = new TSInsertReq(); request.setSessionId(sessionId); request.setDeviceId(deviceId); request.setTimestamp(time); request.setMeasurements(measurements); - request.setValues(values); + ByteBuffer buffer = ByteBuffer.allocate(calculateLength(types, values)); + putValues(types, values, buffer); + buffer.flip(); + request.setValues(buffer); TSStatus result; try { @@ -428,6 +441,68 @@ public class Session { return result; } + private void putValues(List<TSDataType> types, List<Object> values, ByteBuffer buffer) + throws IoTDBConnectionException { + for (int i = 0; i < values.size(); i++) { + ReadWriteIOUtils.write(types.get(i), buffer); + switch (types.get(i)) { + case BOOLEAN: + ReadWriteIOUtils.write((Boolean) values.get(i), buffer); + break; + case INT32: + ReadWriteIOUtils.write((Integer) values.get(i), buffer); + break; + case INT64: + ReadWriteIOUtils.write((Long) values.get(i), buffer); + break; + case FLOAT: + ReadWriteIOUtils.write((Float) values.get(i), buffer); + break; + case DOUBLE: + ReadWriteIOUtils.write((Double) values.get(i), buffer); + break; + case TEXT: + ReadWriteIOUtils.write(new Binary((String) values.get(i)), buffer); + break; + default: + throw new IoTDBConnectionException("Unsupported data type:" + types.get(i)); + } + } + } + + private int calculateLength(List<TSDataType> types, List<Object> values) + throws IoTDBConnectionException { + int res = 0; + for (int i = 0; i < types.size(); i++) { + // types + res += Short.BYTES; + switch (types.get(i)) { + case BOOLEAN: + res += 1; + break; + case INT32: + res += Integer.BYTES; + break; + case INT64: + res += Long.BYTES; + break; + case FLOAT: + res += Float.BYTES; + break; + case DOUBLE: + res += Double.BYTES; + break; + case TEXT: + res += ((String) values.get(i)).getBytes(TSFileConfig.STRING_CHARSET).length; + break; + default: + throw new IoTDBConnectionException("Unsupported data type:" + types.get(i)); + } + } + + return res; + } + /** * 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 @@ -471,7 +546,7 @@ public class Session { request.setDeviceIds(deviceIds); request.setTimestamps(times); request.setMeasurementsList(measurementsList); - request.setValuesList(valuesList); + request.setValuesList(new ArrayList<>()); try { RpcUtils.verifySuccess(client.testInsertRowInBatch(request).statusList); @@ -491,7 +566,7 @@ public class Session { request.setDeviceId(deviceId); request.setTimestamp(time); request.setMeasurements(measurements); - request.setValues(values); + request.setValues(ByteBuffer.allocate(1)); try { RpcUtils.verifySuccess(client.testInsertRow(request)); @@ -624,19 +699,19 @@ public class Session { request.setPaths(paths); List<Integer> dataTypeOrdinals = new ArrayList<>(paths.size()); - for (TSDataType dataType: dataTypes) { + for (TSDataType dataType : dataTypes) { dataTypeOrdinals.add(dataType.ordinal()); } request.setDataTypes(dataTypeOrdinals); List<Integer> encodingOrdinals = new ArrayList<>(paths.size()); - for (TSEncoding encoding: encodings) { + for (TSEncoding encoding : encodings) { encodingOrdinals.add(encoding.ordinal()); } request.setEncodings(encodingOrdinals); List<Integer> compressionOrdinals = new ArrayList<>(paths.size()); - for (CompressionType compression: compressors) { + for (CompressionType compression : compressors) { compressionOrdinals.add(compression.ordinal()); } request.setCompressors(compressionOrdinals); 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..8fef612 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 @@ -38,45 +38,40 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * SessionPool is a wrapper of a Session Set. - * Using SessionPool, the user do not need to consider how to reuse a session connection. - * Even if the session is disconnected, the session pool can recognize it and remove the broken - * session connection and create a new one. - * + * SessionPool is a wrapper of a Session Set. Using SessionPool, the user do not need to consider + * how to reuse a session connection. Even if the session is disconnected, the session pool can + * recognize it and remove the broken session connection and create a new one. + * <p> * If there is no available connections and the pool reaches its max size, the all methods will hang * until there is a available connection. - * + * <p> * If a user has waited for a session for more than 60 seconds, a warn log will be printed. - * + * <p> * The only thing you have to remember is that: - * + * <p> * For a query, if you have get all data, i.e., SessionDataSetWrapper.hasNext() == false, it is ok. - * Otherwise, i.e., you want to stop the query before you get all data (SessionDataSetWrapper.hasNext() == true), - * then you have to call closeResultSet(SessionDataSetWrapper wrapper) manually. - * Otherwise the connection is occupied by the query. - * - * Another case that you have to manually call closeResultSet() is that when there is exception - * when you call SessionDataSetWrapper.hasNext() or next() - * + * Otherwise, i.e., you want to stop the query before you get all data + * (SessionDataSetWrapper.hasNext() == true), then you have to call closeResultSet(SessionDataSetWrapper + * wrapper) manually. Otherwise the connection is occupied by the query. + * <p> + * Another case that you have to manually call closeResultSet() is that when there is exception when + * you call SessionDataSetWrapper.hasNext() or next() */ public class SessionPool { private static final Logger logger = LoggerFactory.getLogger(SessionPool.class); + private static int RETRY = 3; private ConcurrentLinkedDeque<Session> queue = new ConcurrentLinkedDeque<>(); //for session whose resultSet is not released. private ConcurrentMap<Session, Session> occupied = new ConcurrentHashMap<>(); - private int size = 0; private int maxSize = 0; private String ip; private int port; private String user; private String password; - private int fetchSize; - private long timeout; //ms - private static int RETRY = 3; private boolean enableCompression = false; public SessionPool(String ip, int port, String user, String password, int maxSize) { @@ -335,12 +330,13 @@ public class SessionPool { * @see Session#insertBatch(RowBatch) */ public void insertInBatch(List<String> deviceIds, List<Long> times, - List<List<String>> measurementsList, List<List<String>> valuesList) + List<List<String>> measurementsList, List<List<TSDataType>> typesList, + List<List<Object>> valuesList) throws IoTDBConnectionException, BatchExecutionException { for (int i = 0; i < RETRY; i++) { Session session = getSession(); try { - session.insertInBatch(deviceIds, times, measurementsList, valuesList); + session.insertInBatch(deviceIds, times, measurementsList, typesList, valuesList); putBack(session); return; } catch (IoTDBConnectionException e) { @@ -360,15 +356,16 @@ public class SessionPool { * 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#insertInBatch(List, List, List, List, List) * @see Session#insertBatch(RowBatch) */ - public TSStatus insert(String deviceId, long time, List<String> measurements, List<String> values) + public TSStatus insert(String deviceId, long time, List<String> measurements, + List<TSDataType> types, List<Object> values) throws IoTDBConnectionException, StatementExecutionException { for (int i = 0; i < RETRY; i++) { Session session = getSession(); try { - TSStatus resp = session.insert(deviceId, time, measurements, values); + TSStatus resp = session.insert(deviceId, time, measurements, types, values); putBack(session); return resp; } catch (IoTDBConnectionException e) { @@ -544,7 +541,7 @@ public class SessionPool { * delete data <= time in multiple timeseries * * @param paths data in which time series to delete - * @param time data with time stamp less than or equal to time will be deleted + * @param time data with time stamp less than or equal to time will be deleted */ public void deleteData(List<String> paths, long time) throws IoTDBConnectionException, StatementExecutionException { 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..b24b82a 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; @@ -57,6 +55,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class IoTDBSessionIT { + private static Logger logger = LoggerFactory.getLogger(IoTDBSessionIT.class); private Session session; @@ -207,9 +206,12 @@ public class IoTDBSessionIT { // 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)); + 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)); RowBatch rowBatch = schema.createRowBatch("root.sg1.d1", 100); @@ -412,10 +414,13 @@ public class IoTDBSessionIT { } String sensorId = ss[ss.length - 1]; List<String> measurements = new ArrayList<>(); - List<String> values = new ArrayList<>(); + List<Object> values = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); + measurements.add(sensorId); - values.add("100"); - session.insert(deviceId, i, measurements, values); + types.add(TSDataType.INT64); + values.add(100L); + session.insert(deviceId, i, measurements, types, values); } } } @@ -428,21 +433,27 @@ public class IoTDBSessionIT { measurements.add("s3"); List<String> deviceIds = new ArrayList<>(); List<List<String>> measurementsList = new ArrayList<>(); - List<List<String>> valuesList = new ArrayList<>(); + List<List<Object>> valuesList = new ArrayList<>(); List<Long> timestamps = new ArrayList<>(); + List<List<TSDataType>> typesList = new ArrayList<>(); for (long time = 0; time < 500; time++) { - List<String> values = new ArrayList<>(); - values.add("1"); - values.add("2"); - values.add("3"); + List<Object> values = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); + values.add(1L); + values.add(2L); + values.add(3L); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); deviceIds.add(deviceId); measurementsList.add(measurements); valuesList.add(values); + typesList.add(types); timestamps.add(time); if (time != 0 && time % 100 == 0) { - session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList); + session.insertInBatch(deviceIds, timestamps, measurementsList, typesList, valuesList); deviceIds.clear(); measurementsList.clear(); valuesList.clear(); @@ -450,41 +461,54 @@ public class IoTDBSessionIT { } } - session.insertInBatch(deviceIds, timestamps, measurementsList, valuesList); + session.insertInBatch(deviceIds, timestamps, measurementsList, typesList, valuesList); } private void insertInObject() throws IoTDBConnectionException, StatementExecutionException { String deviceId = "root.sg1.d1"; List<String> measurements = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); measurements.add("s1"); measurements.add("s2"); measurements.add("s3"); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + for (long time = 0; time < 100; time++) { - session.insert(deviceId, time, measurements, 1L, 2L, 3L); + session.insert(deviceId, time, measurements, types,1L, 2L, 3L); } } private void insert() throws IoTDBConnectionException, StatementExecutionException { String deviceId = "root.sg1.d1"; List<String> measurements = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); measurements.add("s1"); measurements.add("s2"); measurements.add("s3"); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + types.add(TSDataType.INT64); + for (long time = 0; time < 100; time++) { - List<String> values = new ArrayList<>(); - values.add("1"); - values.add("2"); - values.add("3"); - session.insert(deviceId, time, measurements, values); + List<Object> values = new ArrayList<>(); + values.add(1L); + values.add(2L); + values.add(3L); + session.insert(deviceId, time, measurements, types, values); } } private void insertRowBatchTest1(String deviceId) throws IoTDBConnectionException, BatchExecutionException { Schema schema = new Schema(); - schema.registerTimeseries(new Path(deviceId, "s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId, "s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId, "s3"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE)); + 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)); RowBatch rowBatch = schema.createRowBatch(deviceId, 100); @@ -754,11 +778,11 @@ public class IoTDBSessionIT { private void insertRowBatchTest2(String deviceId) throws IoTDBConnectionException, BatchExecutionException { Schema schema = new Schema(); - schema.registerTimeseries(new Path(deviceId,"s1"), + schema.registerTimeseries(new Path(deviceId, "s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s2"), + schema.registerTimeseries(new Path(deviceId, "s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s3"), + schema.registerTimeseries(new Path(deviceId, "s3"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE)); RowBatch rowBatch = schema.createRowBatch(deviceId, 256); @@ -788,11 +812,11 @@ public class IoTDBSessionIT { private void insertRowBatchTest3(String deviceId) throws IoTDBConnectionException, BatchExecutionException { Schema schema = new Schema(); - schema.registerTimeseries(new Path(deviceId,"s1"), + schema.registerTimeseries(new Path(deviceId, "s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s2"), + schema.registerTimeseries(new Path(deviceId, "s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s3"), + schema.registerTimeseries(new Path(deviceId, "s3"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE)); RowBatch rowBatch = schema.createRowBatch(deviceId, 200); @@ -822,17 +846,17 @@ public class IoTDBSessionIT { private void insertRowBatchTestForTime(String deviceId) throws IoTDBConnectionException, BatchExecutionException { Schema schema = new Schema(); - schema.registerTimeseries(new Path(deviceId,"s1"), + schema.registerTimeseries(new Path(deviceId, "s1"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s2"), + schema.registerTimeseries(new Path(deviceId, "s2"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s3"), + schema.registerTimeseries(new Path(deviceId, "s3"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s4"), + schema.registerTimeseries(new Path(deviceId, "s4"), new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s5"), + schema.registerTimeseries(new Path(deviceId, "s5"), new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); - schema.registerTimeseries(new Path(deviceId,"s6"), + schema.registerTimeseries(new Path(deviceId, "s6"), new MeasurementSchema("s3", TSDataType.INT64, TSEncoding.RLE)); long count = 10000000; long begin = 0; 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..5923577 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 @@ -18,20 +18,20 @@ */ package org.apache.iotdb.session.pool; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; -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.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -60,7 +60,8 @@ public class SessionPoolTest { final int no = i; service.submit(() -> { try { - pool.insert("root.sg1.d1", 1, Collections.singletonList("s" + no), Collections.singletonList("3")); + pool.insert("root.sg1.d1", 1, Collections.singletonList("s" + no), + Collections.singletonList(TSDataType.INT64), Collections.singletonList(3L)); } catch (IoTDBConnectionException | StatementExecutionException e) { fail(); } @@ -83,7 +84,9 @@ 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.insert(".root.sg1.d1", 1, Collections.singletonList("s"), + Collections.singletonList(TSDataType.INT64), + Collections.singletonList(3L)); } catch (IoTDBConnectionException | StatementExecutionException e) { //do nothing } @@ -98,7 +101,9 @@ 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.insert("root.sg1.d1", i, Collections.singletonList("s" + i), + Collections.singletonList(TSDataType.INT64), + Collections.singletonList((long) i)); } catch (IoTDBConnectionException | StatementExecutionException e) { fail(); } @@ -108,7 +113,8 @@ public class SessionPoolTest { final int no = i; service.submit(() -> { try { - SessionDataSetWrapper wrapper = pool.executeQueryStatement("select * from root.sg1.d1 where time = " + no); + SessionDataSetWrapper wrapper = pool + .executeQueryStatement("select * from root.sg1.d1 where time = " + no); //this is incorrect becasue wrapper is not closed. //so all other 7 queries will be blocked } catch (IoTDBConnectionException | StatementExecutionException e) { @@ -139,7 +145,9 @@ 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.insert("root.sg1.d1", i, Collections.singletonList("s" + i), + Collections.singletonList(TSDataType.INT64), + Collections.singletonList((long) i)); } catch (IoTDBConnectionException | StatementExecutionException e) { fail(); } @@ -149,7 +157,8 @@ public class SessionPoolTest { final int no = i; service.submit(() -> { try { - SessionDataSetWrapper wrapper = pool.executeQueryStatement("select * from root.sg1.d1 where time = " + no); + SessionDataSetWrapper wrapper = pool + .executeQueryStatement("select * from root.sg1.d1 where time = " + no); pool.closeResultSet(wrapper); pool.closeResultSet(wrapper); } catch (Exception e) { @@ -174,7 +183,9 @@ 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.insert("root.sg1.d1", i, Collections.singletonList("s" + i), + Collections.singletonList(TSDataType.INT64), + Collections.singletonList((long) i)); } catch (IoTDBConnectionException | StatementExecutionException e) { fail(); } @@ -184,10 +195,10 @@ public class SessionPoolTest { wrapper = pool.executeQueryStatement("select * from root.sg1.d1 where time > 1"); EnvironmentUtils.stopDaemon(); //user does not know what happens. - while(wrapper.hasNext()) { + while (wrapper.hasNext()) { wrapper.next(); } - } catch (IoTDBConnectionException e) { + } catch (IoTDBConnectionException e) { try { pool.closeResultSet(wrapper); } catch (StatementExecutionException ex) { @@ -209,7 +220,9 @@ 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.insert("root.sg1.d1", i, Collections.singletonList("s" + i), + Collections.singletonList(TSDataType.INT64), + Collections.singletonList((long) i)); } catch (IoTDBConnectionException | StatementExecutionException e) { fail(); } @@ -221,7 +234,7 @@ public class SessionPoolTest { //user does not know what happens. assertEquals(0, pool.currentAvailableSize()); assertEquals(1, pool.currentOccupiedSize()); - while(wrapper.hasNext()) { + while (wrapper.hasNext()) { wrapper.next(); } assertEquals(1, pool.currentAvailableSize()); diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Binary.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Binary.java index 58e1260..1903a09 100644 --- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Binary.java +++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/Binary.java @@ -24,13 +24,12 @@ import java.util.Arrays; import org.apache.iotdb.tsfile.common.conf.TSFileConfig; /** - * Override compareTo() and equals() function to Binary class. This class is - * used to accept Java String type + * Override compareTo() and equals() function to Binary class. This class is used to accept Java + * String type */ public class Binary implements Comparable<Binary>, Serializable { private static final long serialVersionUID = 6394197743397020735L; - private byte[] values; /** @@ -117,4 +116,8 @@ public class Binary implements Comparable<Binary>, Serializable { public byte[] getValues() { return values; } + + public void setValues(byte[] values) { + this.values = values; + } } diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java index 5003803..eef02eb 100644 --- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java +++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteIOUtils.java @@ -19,11 +19,14 @@ package org.apache.iotdb.tsfile.utils; -import org.apache.iotdb.tsfile.common.conf.TSFileConfig; -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.reader.TsFileInput; +import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.BINARY; +import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.BOOLEAN; +import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.DOUBLE; +import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.FLOAT; +import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.INTEGER; +import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.LONG; +import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.NULL; +import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.STRING; import java.io.DataOutputStream; import java.io.IOException; @@ -31,10 +34,17 @@ import java.io.InputStream; import java.io.OutputStream; import java.nio.ByteBuffer; import java.nio.charset.CharacterCodingException; -import java.util.*; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.Map.Entry; - -import static org.apache.iotdb.tsfile.utils.ReadWriteIOUtils.ClassSerializeId.*; +import org.apache.iotdb.tsfile.common.conf.TSFileConfig; +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.reader.TsFileInput; /** * ConverterUtils is a utility class. It provide conversion between normal datatype and byte array. @@ -206,6 +216,17 @@ public class ReadWriteIOUtils { } /** + * write a short n to byteBuffer. + * + * @return The number of bytes used to represent n. + */ + public static int write(Binary n, ByteBuffer buffer) { + buffer.putInt(n.getLength()); + buffer.put(n.getValues()); + return INT_LEN + n.getLength(); + } + + /** * write a int n to outputStream. * * @return The number of bytes used to represent n. @@ -280,6 +301,23 @@ public class ReadWriteIOUtils { } /** + * write a float n to byteBuffer. + */ + public static int write(float n, ByteBuffer buffer) { + buffer.putFloat(n); + return FLOAT_LEN; + } + + /** + * write a double n to byteBuffer. + */ + public static int write(double n, ByteBuffer buffer) { + buffer.putDouble(n); + return DOUBLE_LEN; + } + + + /** * write string to outputStream. * * @return the length of string represented by byte[].
