You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2019/11/29 06:18:30 UTC
[incubator-iotdb] branch refactor_tsfile updated: add v2
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch refactor_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/refactor_tsfile by this push:
new c0c82b6 add v2
c0c82b6 is described below
commit c0c82b61be7ca72e7b66f79171f5a5d8f3fd99d2
Author: qiaojialin <64...@qq.com>
AuthorDate: Fri Nov 29 14:18:18 2019 +0800
add v2
---
.../tsfile/file/metadataV2/TsFileMetadataV2.java | 154 ++++++++++++++
.../iotdb/tsfile/write/record/RowBatchV2.java | 5 +
.../iotdb/tsfile/write/schemaV2/SchemaV2.java | 67 ++++++
.../tsfile/write/schemaV2/TimeseriesSchema.java | 233 +++++++++++++++++++++
4 files changed, 459 insertions(+)
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadataV2/TsFileMetadataV2.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadataV2/TsFileMetadataV2.java
new file mode 100644
index 0000000..13f0f44
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadataV2/TsFileMetadataV2.java
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.tsfile.file.metadataV2;
+
+
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import java.util.TreeMap;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.utils.BloomFilter;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.schemaV2.SchemaV2;
+
+/**
+ * TSFileMetaData collects all metadata info and saves in its data structure.
+ */
+public class TsFileMetadataV2 {
+
+ private SchemaV2 schemaV2;
+
+ // fields below are IoTDB extensions and they does not affect TsFile's stand-alone functionality
+ private int totalChunkNum;
+ // invalid means a chunk has been rewritten by merge and the chunk's data is in
+ // another new chunk
+ private int invalidChunkNum;
+
+ // bloom filter
+ private BloomFilter bloomFilter;
+
+ private List<ChunkMetaData> chunkMetaDataList;
+
+ private TreeMap<Path, List<ChunkMetaData>> timeseriesMetadata;
+
+ private List<Pair<Path, Long>> seriesMetadataIndex;
+
+ /**
+ * deserialize data from the buffer.
+ *
+ * @param buffer -buffer use to deserialize
+ * @return -a instance of TsFileMetaData
+ */
+ public static TsFileMetadataV2 deserializeFrom(ByteBuffer buffer, boolean isOldVersion)
+ throws IOException {
+ TsFileMetadataV2 fileMetaData = new TsFileMetadataV2();
+
+
+
+ return fileMetaData;
+ }
+
+ public BloomFilter getBloomFilter() {
+ return bloomFilter;
+ }
+
+ /**
+ * use the given outputStream to serialize.
+ *
+ * @param outputStream -output stream to determine byte length
+ * @return -byte length
+ */
+ public int serializeTo(OutputStream outputStream) throws IOException {
+ int byteLen = 0;
+
+ byteLen += ReadWriteIOUtils.write(totalChunkNum, outputStream);
+ byteLen += ReadWriteIOUtils.write(invalidChunkNum, outputStream);
+
+ return byteLen;
+ }
+
+ /**
+ * use the given outputStream to serialize bloom filter.
+ *
+ * @param outputStream -output stream to determine byte length
+ * @return -byte length
+ */
+ public int serializeBloomFilter(OutputStream outputStream,
+ List<ChunkMetaData> chunkGroupMetaDataList)
+ throws IOException {
+ int byteLen = 0;
+ BloomFilter filter = buildBloomFilter(chunkGroupMetaDataList);
+
+ byte[] bytes = filter.serialize();
+ byteLen += ReadWriteIOUtils.write(bytes.length, outputStream);
+ outputStream.write(bytes);
+ byteLen += bytes.length;
+ byteLen += ReadWriteIOUtils.write(filter.getSize(), outputStream);
+ byteLen += ReadWriteIOUtils.write(filter.getHashFunctionSize(), outputStream);
+ return byteLen;
+ }
+
+
+ /**
+ * build bloom filter
+ *
+ * @return bloom filter
+ */
+ private BloomFilter buildBloomFilter(List<ChunkMetaData> chunkGroupMetaDataList) {
+ List<String> paths = getAllPath(chunkGroupMetaDataList);
+ BloomFilter bloomFilter = BloomFilter
+ .getEmptyBloomFilter(TSFileDescriptor.getInstance().getConfig().getBloomFilterErrorRate(),
+ paths.size());
+ for (String path : paths) {
+ bloomFilter.add(path);
+ }
+ return bloomFilter;
+ }
+
+ public int getTotalChunkNum() {
+ return totalChunkNum;
+ }
+
+ public void setTotalChunkNum(int totalChunkNum) {
+ this.totalChunkNum = totalChunkNum;
+ }
+
+ public int getInvalidChunkNum() {
+ return invalidChunkNum;
+ }
+
+ public void setInvalidChunkNum(int invalidChunkNum) {
+ this.invalidChunkNum = invalidChunkNum;
+ }
+
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/RowBatchV2.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/RowBatchV2.java
new file mode 100644
index 0000000..8daeb8f
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/record/RowBatchV2.java
@@ -0,0 +1,5 @@
+package org.apache.iotdb.tsfile.write.record;
+
+public class RowBatchV2 {
+
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schemaV2/SchemaV2.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schemaV2/SchemaV2.java
new file mode 100644
index 0000000..60c3d0b
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schemaV2/SchemaV2.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.tsfile.write.schemaV2;
+
+import org.apache.iotdb.tsfile.read.common.Path;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Schema stores the schema of the measurements and devices that exist in this file. All
+ * devices written to the same TsFile shall have the same schema. Schema takes the JSON schema
+ * file as a parameter and registers measurements in such JSON. Schema also records all existing
+ * device IDs in this file.
+ */
+public class SchemaV2 {
+
+ /**
+ * Path (device + measurement) -> TimeseriesSchema
+ * By default, use the LinkedHashMap to store the order of insertion
+ */
+ private Map<Path, TimeseriesSchema> timeseriesSchemaMap;
+
+ /**
+ * register a measurement schema map.
+ */
+ public void registerTimeseries(Path path, TimeseriesSchema descriptor) {
+
+ }
+
+
+ public void regieterDeviceTemplate () {
+
+ }
+
+ public void regiesterDevice () {
+
+ }
+
+
+ public TimeseriesSchema getSeriesSchema(Path path) {
+ return timeseriesSchemaMap.get(path);
+ }
+
+ /**
+ * check if this schema contains a measurement named measurementId.
+ */
+ public boolean hasTimeseries(Path path) {
+ return timeseriesSchemaMap.containsKey(path);
+ }
+
+}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schemaV2/TimeseriesSchema.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schemaV2/TimeseriesSchema.java
new file mode 100644
index 0000000..67b3559
--- /dev/null
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/schemaV2/TimeseriesSchema.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.tsfile.write.schemaV2;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.encoding.encoder.Encoder;
+import org.apache.iotdb.tsfile.encoding.encoder.TSEncodingBuilder;
+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.ReadWriteIOUtils;
+import org.apache.iotdb.tsfile.utils.StringContainer;
+
+/**
+ * This class describes a measurement's information registered in {@linkplain SchemaV2 FilSchema},
+ * including measurement id, data type, encoding and compressor type. For each TSEncoding,
+ * MeasurementSchema maintains respective TSEncodingBuilder; For TSDataType, only ENUM has
+ * TSDataTypeConverter up to now.
+ */
+public class TimeseriesSchema implements Comparable<TimeseriesSchema>, Serializable {
+
+ private TSDataType type;
+
+ private TSEncoding encoding;
+
+ private CompressionType compressionType;
+
+ private TSEncodingBuilder encodingConverter;
+
+ private Map<String, String> props = new HashMap<>();
+
+ public TimeseriesSchema() {
+ }
+
+ /**
+ * Constructor of MeasurementSchema.
+ *
+ * <p>props - information in encoding method. For RLE, Encoder.MAX_POINT_NUMBER For PLAIN,
+ * Encoder.maxStringLength
+ */
+ public TimeseriesSchema(TSDataType type, TSEncoding encoding,
+ CompressionType compressionType, Map<String, String> props) {
+ this.type = type;
+ this.encoding = encoding;
+ this.compressionType = compressionType;
+ this.props = props == null ? Collections.emptyMap() : props;
+ }
+
+
+ /**
+ * function for deserializing data from byte buffer.
+ */
+ public static TimeseriesSchema deserializeFrom(ByteBuffer buffer) {
+ TimeseriesSchema measurementSchema = new TimeseriesSchema();
+
+ measurementSchema.measurementId = ReadWriteIOUtils.readString(buffer);
+
+ measurementSchema.type = ReadWriteIOUtils.readDataType(buffer);
+
+ measurementSchema.encoding = ReadWriteIOUtils.readEncoding(buffer);
+
+ measurementSchema.compressionType = ReadWriteIOUtils.readCompressionType(buffer);
+
+ int size = ReadWriteIOUtils.readInt(buffer);
+ if (size > 0) {
+ measurementSchema.props = new HashMap<>();
+ String key;
+ String value;
+ for (int i = 0; i < size; i++) {
+ key = ReadWriteIOUtils.readString(buffer);
+ value = ReadWriteIOUtils.readString(buffer);
+ measurementSchema.props.put(key, value);
+ }
+ }
+
+ return measurementSchema;
+ }
+
+ public String getDeviceId() {
+ return deviceId;
+ }
+
+ public void setDeviceId(String deviceId) {
+ this.deviceId = deviceId;
+ }
+
+ public String getMeasurementId() {
+ return measurementId;
+ }
+
+ public void setMeasurementId(String measurementId) {
+ this.measurementId = measurementId;
+ }
+
+ public Map<String, String> getProps() {
+ return props;
+ }
+
+ public TSEncoding getEncodingType() {
+ return encoding;
+ }
+
+ public TSDataType getType() {
+ return type;
+ }
+
+ public void setProps(Map<String, String> props) {
+ this.props = props;
+ }
+
+ /**
+
+ /**
+ * function for getting time encoder.
+ * TODO can I be optimized?
+ */
+ public Encoder getTimeEncoder() {
+ TSEncoding timeSeriesEncoder = TSEncoding.valueOf(TSFileDescriptor.getInstance().getConfig().getTimeEncoder());
+ TSDataType timeType = TSDataType.valueOf(TSFileDescriptor.getInstance().getConfig().getTimeSeriesDataType());
+ return TSEncodingBuilder.getConverter(timeSeriesEncoder).getEncoder(timeType);
+ }
+
+ /**
+ * get Encoder of value from encodingConverter by measurementID and data type.
+ * TODO can I be optimized?
+ * @return Encoder for value
+ */
+ public Encoder getValueEncoder() {
+ //it is ok even if encodingConverter is constructed two instances for concurrent scenario..
+ if (encodingConverter == null) {
+ // initialize TSEncoding. e.g. set max error for PLA and SDT
+ encodingConverter = TSEncodingBuilder.getConverter(encoding);
+ encodingConverter.initFromProps(props);
+ }
+ return encodingConverter.getEncoder(type);
+ }
+
+ public CompressionType getCompressionType() {
+ return compressionType;
+ }
+
+ /**
+ * function for serializing data to output stream.
+ */
+ public int serializeTo(OutputStream outputStream) throws IOException {
+ int byteLen = 0;
+
+ byteLen += ReadWriteIOUtils.write(measurementId, outputStream);
+
+ byteLen += ReadWriteIOUtils.write(type, outputStream);
+
+ byteLen += ReadWriteIOUtils.write(encoding, outputStream);
+
+ byteLen += ReadWriteIOUtils.write(compressionType, outputStream);
+
+ if (props == null) {
+ byteLen += ReadWriteIOUtils.write(0, outputStream);
+ } else {
+ byteLen += ReadWriteIOUtils.write(props.size(), outputStream);
+ for (Map.Entry<String, String> entry : props.entrySet()) {
+ byteLen += ReadWriteIOUtils.write(entry.getKey(), outputStream);
+ byteLen += ReadWriteIOUtils.write(entry.getValue(), outputStream);
+ }
+ }
+
+ return byteLen;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TimeseriesSchema that = (TimeseriesSchema) o;
+ return type == that.type && encoding == that.encoding && Objects
+ .equals(measurementId, that.measurementId)
+ && Objects.equals(compressionType, that.compressionType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(type, encoding, measurementId, compressionType);
+ }
+
+ /**
+ * compare by measurementID.
+ */
+ @Override
+ public int compareTo(TimeseriesSchema o) {
+ if (equals(o)) {
+ return 0;
+ } else {
+ return this.measurementId.compareTo(o.measurementId);
+ }
+ }
+
+ @Override
+ public String toString() {
+ StringContainer sc = new StringContainer("");
+ sc.addTail("[", measurementId, ",", type.toString(), ",", encoding.toString(), ",",
+ props.toString(), ",",
+ compressionType.toString());
+ sc.addTail("]");
+ return sc.toString();
+ }
+}