You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2020/12/02 08:04:25 UTC
[iotdb] 01/11: init
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch NewTsFile
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3d01eec2597b8e313ff418746d787002c194a8ed
Author: JackieTien97 <Ja...@foxmail.com>
AuthorDate: Wed Nov 25 13:48:39 2020 +0800
init
---
.../apache/iotdb/tsfile/TsFileSequenceRead.java | 13 +-
.../apache/iotdb/hadoop/fileSystem/HDFSInput.java | 11 +
.../apache/iotdb/hadoop/fileSystem/HDFSOutput.java | 5 +
.../iotdb/db/engine/flush/MemTableFlushTask.java | 2 +-
.../iotdb/db/query/control/FileReaderManager.java | 2 +-
.../apache/iotdb/db/tools/TsFileSketchTool.java | 16 +-
.../db/tools/upgrade/TsFileOnlineUpgradeTool.java | 14 +-
.../iotdb/tsfile/common/conf/TSFileConfig.java | 6 +-
.../org/apache/iotdb/tsfile/file/MetaMarker.java | 9 +-
...ChunkGroupFooter.java => ChunkGroupHeader.java} | 69 +---
.../iotdb/tsfile/file/header/ChunkHeader.java | 114 +++---
.../iotdb/tsfile/file/header/PageHeader.java | 22 +-
.../file/metadata/statistics/BinaryStatistics.java | 9 +-
.../metadata/statistics/BooleanStatistics.java | 36 +-
.../metadata/statistics/IntegerStatistics.java | 44 ++-
.../file/metadata/statistics/Statistics.java | 18 +-
.../iotdb/tsfile/read/TsFileSequenceReader.java | 42 +--
.../iotdb/tsfile/read/reader/LocalTsFileInput.java | 10 +
.../iotdb/tsfile/read/reader/TsFileInput.java | 5 +
.../tsfile/utils/ReadWriteForEncodingUtils.java | 90 ++++-
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 94 ++++-
.../v1/file/metadata/ChunkGroupMetaDataV1.java | 118 ------
.../tsfile/v1/file/metadata/ChunkMetadataV1.java | 131 -------
.../v1/file/metadata/TimeseriesMetadataForV1.java | 42 ---
.../v1/file/metadata/TsDeviceMetadataIndexV1.java | 78 ----
.../v1/file/metadata/TsDeviceMetadataV1.java | 87 -----
.../iotdb/tsfile/v1/file/metadata/TsDigestV1.java | 75 ----
.../tsfile/v1/file/metadata/TsFileMetadataV1.java | 106 ------
.../metadata/statistics/BinaryStatisticsV1.java | 84 -----
.../metadata/statistics/BooleanStatisticsV1.java | 80 ----
.../metadata/statistics/DoubleStatisticsV1.java | 79 ----
.../metadata/statistics/FloatStatisticsV1.java | 79 ----
.../metadata/statistics/IntegerStatisticsV1.java | 79 ----
.../file/metadata/statistics/LongStatisticsV1.java | 80 ----
.../v1/file/metadata/statistics/StatisticsV1.java | 225 ------------
.../iotdb/tsfile/v1/file/utils/HeaderUtils.java | 141 -------
.../tsfile/v1/read/TsFileSequenceReaderForV1.java | 409 ---------------------
.../iotdb/tsfile/write/chunk/ChunkWriterImpl.java | 53 +--
.../apache/iotdb/tsfile/write/page/PageWriter.java | 31 +-
.../tsfile/write/writer/LocalTsFileOutput.java | 5 +
.../iotdb/tsfile/write/writer/TsFileIOWriter.java | 19 +-
.../iotdb/tsfile/write/writer/TsFileOutput.java | 8 +
.../tsfile/read/TsFileSequenceReaderTest.java | 11 +-
.../iotdb/tsfile/write/TsFileIOWriterTest.java | 19 +-
.../write/writer/RestorableTsFileIOWriterTest.java | 8 +-
45 files changed, 508 insertions(+), 2170 deletions(-)
diff --git a/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileSequenceRead.java b/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileSequenceRead.java
index da5bae6..818ac3b 100644
--- a/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileSequenceRead.java
+++ b/example/tsfile/src/main/java/org/apache/iotdb/tsfile/TsFileSequenceRead.java
@@ -26,7 +26,7 @@ import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.encoding.decoder.Decoder;
import org.apache.iotdb.tsfile.file.MetaMarker;
-import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
+import org.apache.iotdb.tsfile.file.footer.ChunkGroupHeader;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
import org.apache.iotdb.tsfile.file.header.PageHeader;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
@@ -55,7 +55,7 @@ public class TsFileSequenceRead {
// first SeriesChunks (headers and data) in one ChunkGroup, then the CHUNK_GROUP_FOOTER
// Because we do not know how many chunks a ChunkGroup may have, we should read one byte (the marker) ahead and
// judge accordingly.
- reader.position((long) TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER
+ reader.position((long) TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER_V2
.getBytes().length);
System.out.println("[Chunk Group]");
System.out.println("position: " + reader.position());
@@ -63,9 +63,10 @@ public class TsFileSequenceRead {
while ((marker = reader.readMarker()) != MetaMarker.SEPARATOR) {
switch (marker) {
case MetaMarker.CHUNK_HEADER:
+ case MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER:
System.out.println("\t[Chunk]");
System.out.println("\tposition: " + reader.position());
- ChunkHeader header = reader.readChunkHeader();
+ ChunkHeader header = reader.readChunkHeader(marker);
System.out.println("\tMeasurement: " + header.getMeasurementID());
Decoder defaultTimeDecoder = Decoder.getDecoderByType(
TSEncoding.valueOf(TSFileDescriptor.getInstance().getConfig().getTimeEncoder()),
@@ -92,10 +93,10 @@ public class TsFileSequenceRead {
}
}
break;
- case MetaMarker.CHUNK_GROUP_FOOTER:
+ case MetaMarker.CHUNK_GROUP_HEADER:
System.out.println("Chunk Group Footer position: " + reader.position());
- ChunkGroupFooter chunkGroupFooter = reader.readChunkGroupFooter();
- System.out.println("device: " + chunkGroupFooter.getDeviceID());
+ ChunkGroupHeader chunkGroupHeader = reader.readChunkGroupFooter();
+ System.out.println("device: " + chunkGroupHeader.getDeviceID());
break;
case MetaMarker.VERSION:
long version = reader.readVersion();
diff --git a/hadoop/src/main/java/org/apache/iotdb/hadoop/fileSystem/HDFSInput.java b/hadoop/src/main/java/org/apache/iotdb/hadoop/fileSystem/HDFSInput.java
index 0b7da82..f453d3a 100644
--- a/hadoop/src/main/java/org/apache/iotdb/hadoop/fileSystem/HDFSInput.java
+++ b/hadoop/src/main/java/org/apache/iotdb/hadoop/fileSystem/HDFSInput.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.iotdb.tsfile.read.reader.TsFileInput;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
public class HDFSInput implements TsFileInput {
@@ -126,4 +127,14 @@ public class HDFSInput implements TsFileInput {
public int readInt() throws IOException {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public String readVarIntString(long position) throws IOException {
+ long srcPosition = fsDataInputStream.getPos();
+
+ fsDataInputStream.seek(position);
+ String res = ReadWriteIOUtils.readVarIntString(fsDataInputStream);
+ fsDataInputStream.seek(srcPosition);
+ return res;
+ }
}
diff --git a/hadoop/src/main/java/org/apache/iotdb/hadoop/fileSystem/HDFSOutput.java b/hadoop/src/main/java/org/apache/iotdb/hadoop/fileSystem/HDFSOutput.java
index aec1a58..2920311 100644
--- a/hadoop/src/main/java/org/apache/iotdb/hadoop/fileSystem/HDFSOutput.java
+++ b/hadoop/src/main/java/org/apache/iotdb/hadoop/fileSystem/HDFSOutput.java
@@ -61,6 +61,11 @@ public class HDFSOutput implements TsFileOutput {
fsDataOutputStream.write(b);
}
+ @Override
+ public void write(byte b) throws IOException {
+ fsDataOutputStream.write(b);
+ }
+
public void write(ByteBuffer b) throws IOException {
throw new UnsupportedOperationException("Unsupported operation.");
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
index e9f4c92..fbc9877 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java
@@ -75,7 +75,7 @@ public class MemTableFlushTask {
* the function for flushing memtable.
*/
public void syncFlushMemTable()
- throws ExecutionException, InterruptedException, IOException {
+ throws ExecutionException, InterruptedException {
logger.info("The memTable size of SG {} is {}, the avg series points num in chunk is {} ",
storageGroup,
memTable.memSize(),
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
index 2b60229..776fcdb 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/FileReaderManager.java
@@ -172,7 +172,7 @@ public class FileReaderManager implements IService {
tsFileReader.close();
tsFileReader = new TsFileSequenceReaderForV1(filePath);
break;
- case TSFileConfig.VERSION_NUMBER:
+ case TSFileConfig.VERSION_NUMBER_V2:
break;
default:
throw new IOException("The version of this TsFile is not corrent. ");
diff --git a/server/src/main/java/org/apache/iotdb/db/tools/TsFileSketchTool.java b/server/src/main/java/org/apache/iotdb/db/tools/TsFileSketchTool.java
index a558015..ca1b26c 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/TsFileSketchTool.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/TsFileSketchTool.java
@@ -28,7 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
-import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
+import org.apache.iotdb.tsfile.file.footer.ChunkGroupHeader;
import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetadata;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.MetadataIndexEntry;
@@ -105,25 +105,25 @@ public class TsFileSketchTool {
}
// chunkGroupFooter begins
printlnBoth(pw, String.format("%20s", chunkEndPos) + "|\t[Chunk Group Footer]");
- ChunkGroupFooter chunkGroupFooter = reader.readChunkGroupFooter(chunkEndPos, false);
+ ChunkGroupHeader chunkGroupHeader = reader.readChunkGroupFooter(chunkEndPos, false);
printlnBoth(pw, String.format("%20s", "") + "|\t\t[marker] 0");
printlnBoth(pw,
- String.format("%20s", "") + "|\t\t[deviceID] " + chunkGroupFooter.getDeviceID());
+ String.format("%20s", "") + "|\t\t[deviceID] " + chunkGroupHeader.getDeviceID());
printlnBoth(pw,
- String.format("%20s", "") + "|\t\t[dataSize] " + chunkGroupFooter.getDataSize());
- printlnBoth(pw, String.format("%20s", "") + "|\t\t[num of chunks] " + chunkGroupFooter
+ String.format("%20s", "") + "|\t\t[dataSize] " + chunkGroupHeader.getDataSize());
+ printlnBoth(pw, String.format("%20s", "") + "|\t\t[num of chunks] " + chunkGroupHeader
.getNumberOfChunks());
printlnBoth(pw, str1.toString() + "\t[Chunk Group] of "
+ chunkGroupMetadata.getDevice() + " ends");
// versionInfo begins if there is a versionInfo
- if (versionMap.containsKey(chunkEndPos + chunkGroupFooter.getSerializedSize())) {
+ if (versionMap.containsKey(chunkEndPos + chunkGroupHeader.getSerializedSize())) {
printlnBoth(pw,
- String.format("%20s", chunkEndPos + chunkGroupFooter.getSerializedSize())
+ String.format("%20s", chunkEndPos + chunkGroupHeader.getSerializedSize())
+ "|\t[Version Info]");
printlnBoth(pw, String.format("%20s", "") + "|\t\t[marker] 3");
printlnBoth(pw,
String.format("%20s", "") + "|\t\t[version] "
- + versionMap.get(chunkEndPos + chunkGroupFooter.getSerializedSize()));
+ + versionMap.get(chunkEndPos + chunkGroupHeader.getSerializedSize()));
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/tools/upgrade/TsFileOnlineUpgradeTool.java b/server/src/main/java/org/apache/iotdb/db/tools/upgrade/TsFileOnlineUpgradeTool.java
index 405b4c2..52b7eb3 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/upgrade/TsFileOnlineUpgradeTool.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/upgrade/TsFileOnlineUpgradeTool.java
@@ -36,7 +36,7 @@ import org.apache.iotdb.tsfile.exception.write.PageException;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.file.MetaMarker;
-import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
+import org.apache.iotdb.tsfile.file.footer.ChunkGroupHeader;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
import org.apache.iotdb.tsfile.file.header.PageHeader;
import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
@@ -185,7 +185,7 @@ public class TsFileOnlineUpgradeTool implements AutoCloseable {
*/
public String readVersionNumber() throws IOException {
ByteBuffer versionNumberBytes = ByteBuffer
- .allocate(TSFileConfig.VERSION_NUMBER.getBytes().length);
+ .allocate(TSFileConfig.VERSION_NUMBER_V2.getBytes().length);
tsFileInput.position(TSFileConfig.MAGIC_STRING.getBytes().length);
tsFileInput.read(versionNumberBytes);
versionNumberBytes.flip();
@@ -213,8 +213,8 @@ public class TsFileOnlineUpgradeTool implements AutoCloseable {
* @return a CHUNK_GROUP_FOOTER
* @throws IOException io error
*/
- public ChunkGroupFooter readChunkGroupFooter() throws IOException {
- return ChunkGroupFooter.deserializeFrom(tsFileInput.wrapAsInputStream(), true);
+ public ChunkGroupHeader readChunkGroupFooter() throws IOException {
+ return ChunkGroupHeader.deserializeFrom(tsFileInput.wrapAsInputStream(), true);
}
/**
@@ -374,10 +374,10 @@ public class TsFileOnlineUpgradeTool implements AutoCloseable {
pageDataInChunkGroup.add(dataInChunk);
pagePartitionInfoInChunkGroup.add(pagePartitionInfo);
break;
- case MetaMarker.CHUNK_GROUP_FOOTER:
+ case MetaMarker.CHUNK_GROUP_HEADER:
// this is the footer of a ChunkGroup.
- ChunkGroupFooter chunkGroupFooter = this.readChunkGroupFooter();
- String deviceID = chunkGroupFooter.getDeviceID();
+ ChunkGroupHeader chunkGroupHeader = this.readChunkGroupFooter();
+ String deviceID = chunkGroupHeader.getDeviceID();
rewrite(oldTsFile, deviceID, measurementSchemaList, pageHeadersInChunkGroup,
pageDataInChunkGroup, versionOfChunkGroup, pagePartitionInfoInChunkGroup);
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
index 24e4dde..2193786 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/conf/TSFileConfig.java
@@ -61,8 +61,12 @@ public class TSFileConfig implements Serializable {
public static final Charset STRING_CHARSET = Charset.forName(STRING_ENCODING);
public static final String CONFIG_FILE_NAME = "iotdb-engine.properties";
public static final String MAGIC_STRING = "TsFile";
- public static final String VERSION_NUMBER = "000002";
+ public static final String VERSION_NUMBER_V2 = "000002";
public static final String VERSION_NUMBER_V1 = "000001";
+ /**
+ * version number is changed to use 1 byte to represent since version 3
+ */
+ public static final byte VERSION_NUMBER = 0x03;
/**
* Bloom filter constrain
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/MetaMarker.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/MetaMarker.java
index 758f0d5..c9086b3 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/MetaMarker.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/MetaMarker.java
@@ -26,10 +26,17 @@ import java.io.IOException;
*/
public class MetaMarker {
- public static final byte CHUNK_GROUP_FOOTER = 0;
+ public static final byte CHUNK_GROUP_HEADER = 0;
+ /**
+ * means this chunk has more than one page
+ */
public static final byte CHUNK_HEADER = 1;
public static final byte SEPARATOR = 2;
public static final byte VERSION = 3;
+ /**
+ * means this chunk has only one page
+ */
+ public static final byte ONLY_ONE_PAGE_CHUNK_HEADER = 4;
private MetaMarker() {
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/footer/ChunkGroupFooter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/footer/ChunkGroupHeader.java
similarity index 56%
rename from tsfile/src/main/java/org/apache/iotdb/tsfile/file/footer/ChunkGroupFooter.java
rename to tsfile/src/main/java/org/apache/iotdb/tsfile/file/footer/ChunkGroupHeader.java
index 3bc85d0..66afbd8 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/footer/ChunkGroupFooter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/footer/ChunkGroupHeader.java
@@ -22,40 +22,28 @@ package org.apache.iotdb.tsfile.file.footer;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.io.UnsupportedEncodingException;
-import java.nio.ByteBuffer;
-
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.file.MetaMarker;
import org.apache.iotdb.tsfile.read.reader.TsFileInput;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-public class ChunkGroupFooter {
-
- private static final byte MARKER = MetaMarker.CHUNK_GROUP_FOOTER;
+public class ChunkGroupHeader {
- private String deviceID;
+ private static final byte MARKER = MetaMarker.CHUNK_GROUP_HEADER;
- private long dataSize;
-
- private int numberOfChunks;
+ private final String deviceID;
// this field does not need to be serialized.
private int serializedSize;
/**
* constructor of CHUNK_GROUP_FOOTER.
+ * @param deviceID device ID
*
- * @param deviceID device ID
- * @param dataSize data size
- * @param numberOfChunks number of chunks
*/
- public ChunkGroupFooter(String deviceID, long dataSize, int numberOfChunks) throws UnsupportedEncodingException {
+ public ChunkGroupHeader(String deviceID) {
this.deviceID = deviceID;
- this.dataSize = dataSize;
- this.numberOfChunks = numberOfChunks;
- this.serializedSize = Byte.BYTES + Integer.BYTES + deviceID.getBytes(TSFileConfig.STRING_CHARSET).length
- + Long.BYTES + Integer.BYTES;
+ this.serializedSize = Byte.BYTES + Integer.BYTES + deviceID.getBytes(TSFileConfig.STRING_CHARSET).length;
}
public static int getSerializedSize(String deviceID) {
@@ -71,7 +59,7 @@ public class ChunkGroupFooter {
*
* @param markerRead Whether the marker of the CHUNK_GROUP_FOOTER is read ahead.
*/
- public static ChunkGroupFooter deserializeFrom(InputStream inputStream, boolean markerRead) throws IOException {
+ public static ChunkGroupHeader deserializeFrom(InputStream inputStream, boolean markerRead) throws IOException {
if (!markerRead) {
byte marker = (byte) inputStream.read();
if (marker != MARKER) {
@@ -79,10 +67,8 @@ public class ChunkGroupFooter {
}
}
- String deviceID = ReadWriteIOUtils.readString(inputStream);
- long dataSize = ReadWriteIOUtils.readLong(inputStream);
- int numOfChunks = ReadWriteIOUtils.readInt(inputStream);
- return new ChunkGroupFooter(deviceID, dataSize, numOfChunks);
+ String deviceID = ReadWriteIOUtils.readVarIntString(inputStream);
+ return new ChunkGroupHeader(deviceID);
}
/**
@@ -90,24 +76,14 @@ public class ChunkGroupFooter {
*
* @param markerRead Whether the marker of the CHUNK_GROUP_FOOTER is read ahead.
*/
- public static ChunkGroupFooter deserializeFrom(TsFileInput input, long offset, boolean markerRead)
+ public static ChunkGroupHeader deserializeFrom(TsFileInput input, long offset, boolean markerRead)
throws IOException {
long offsetVar = offset;
if (!markerRead) {
offsetVar++;
}
- ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES);
- input.read(buffer, offsetVar);
- buffer.flip();
- int size = buffer.getInt();
- offsetVar += Integer.BYTES;
- buffer = ByteBuffer.allocate(getSerializedSize(size));
- ReadWriteIOUtils.readAsPossible(input, offsetVar, buffer);
- buffer.flip();
- String deviceID = ReadWriteIOUtils.readStringWithLength(buffer, size);
- long dataSize = ReadWriteIOUtils.readLong(buffer);
- int numOfChunks = ReadWriteIOUtils.readInt(buffer);
- return new ChunkGroupFooter(deviceID, dataSize, numOfChunks);
+ String deviceID = input.readVarIntString(offsetVar);
+ return new ChunkGroupHeader(deviceID);
}
public int getSerializedSize() {
@@ -118,17 +94,6 @@ public class ChunkGroupFooter {
return deviceID;
}
- public long getDataSize() {
- return dataSize;
- }
-
- public void setDataSize(long dataSize) {
- this.dataSize = dataSize;
- }
-
- public int getNumberOfChunks() {
- return numberOfChunks;
- }
/**
* serialize to outputStream.
@@ -140,15 +105,15 @@ public class ChunkGroupFooter {
public int serializeTo(OutputStream outputStream) throws IOException {
int length = 0;
length += ReadWriteIOUtils.write(MARKER, outputStream);
- length += ReadWriteIOUtils.write(deviceID, outputStream);
- length += ReadWriteIOUtils.write(dataSize, outputStream);
- length += ReadWriteIOUtils.write(numberOfChunks, outputStream);
+ length += ReadWriteIOUtils.writeVar(deviceID, outputStream);
return length;
}
@Override
public String toString() {
- return "CHUNK_GROUP_FOOTER{" + "deviceID='" + deviceID + '\'' + ", dataSize=" + dataSize + ", numberOfChunks="
- + numberOfChunks + ", serializedSize=" + serializedSize + '}';
+ return "ChunkGroupHeader{" +
+ "deviceID='" + deviceID + '\'' +
+ ", serializedSize=" + serializedSize +
+ '}';
}
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/ChunkHeader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/ChunkHeader.java
index 96556b8..c318d44 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/ChunkHeader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/ChunkHeader.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.reader.TsFileInput;
+import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import java.io.IOException;
import java.io.InputStream;
@@ -33,101 +34,118 @@ import java.nio.ByteBuffer;
public class ChunkHeader {
+
private String measurementID;
private int dataSize;
private TSDataType dataType;
private CompressionType compressionType;
private TSEncoding encodingType;
- private int numOfPages;
- // this field does not need to be serialized.
+ // the following fields do not need to be serialized.
+ /**
+ * 1 means this chunk has more than one page, so each page has its own page statistic 4 means this
+ * chunk has only one page, and this page has no page statistic
+ */
+ private byte chunkType;
+ private int numOfPages;
private int serializedSize;
public ChunkHeader(String measurementID, int dataSize, TSDataType dataType,
- CompressionType compressionType,
- TSEncoding encoding, int numOfPages) {
- this(measurementID, dataSize, getSerializedSize(measurementID), dataType, compressionType,
- encoding, numOfPages);
+ CompressionType compressionType, TSEncoding encoding, int numOfPages) {
+ this(numOfPages <= 1 ? MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER : MetaMarker.CHUNK_HEADER,
+ measurementID, dataSize, getSerializedSize(measurementID, dataSize), dataType,
+ compressionType,
+ encoding);
+ this.numOfPages = numOfPages;
}
- private ChunkHeader(String measurementID, int dataSize, int headerSize, TSDataType dataType,
- CompressionType compressionType, TSEncoding encoding, int numOfPages) {
+ public ChunkHeader(byte chunkType, String measurementID, int dataSize, TSDataType dataType,
+ CompressionType compressionType, TSEncoding encoding) {
+ this(chunkType, measurementID, dataSize, getSerializedSize(measurementID, dataSize), dataType,
+ compressionType, encoding);
+ }
+
+ private ChunkHeader(byte chunkType, String measurementID, int dataSize, int headerSize,
+ TSDataType dataType, CompressionType compressionType, TSEncoding encoding) {
+ this.chunkType = chunkType;
this.measurementID = measurementID;
this.dataSize = dataSize;
this.dataType = dataType;
this.compressionType = compressionType;
- this.numOfPages = numOfPages;
this.encodingType = encoding;
this.serializedSize = headerSize;
}
+ /**
+ * the exact serialized size of chunk header
+ */
+ public static int getSerializedSize(String measurementID, int dataSize) {
+ int measurementIdLength = measurementID.getBytes(TSFileConfig.STRING_CHARSET).length;
+ return ReadWriteForEncodingUtils.varIntSize(measurementIdLength) // measurementID length
+ + measurementIdLength // measurementID
+ + ReadWriteForEncodingUtils.varIntSize(dataSize) // dataSize
+ + TSDataType.getSerializedSize() // dataType
+ + CompressionType.getSerializedSize() // compressionType
+ + TSEncoding.getSerializedSize(); // encodingType
+ }
+
+ /**
+ * The estimated serialized size of chunk header. Only used when we don't know the actual dataSize
+ * attribute
+ */
public static int getSerializedSize(String measurementID) {
- return Byte.BYTES // marker
- + Integer.BYTES // measurementID length
- + measurementID.getBytes(TSFileConfig.STRING_CHARSET).length // measurementID
- + Integer.BYTES // dataSize
+
+ int measurementIdLength = measurementID.getBytes(TSFileConfig.STRING_CHARSET).length;
+ return ReadWriteForEncodingUtils.varIntSize(measurementIdLength) // measurementID length
+ + measurementIdLength // measurementID
+ + Integer.BYTES + 1 // varInr dataSize
+ TSDataType.getSerializedSize() // dataType
+ CompressionType.getSerializedSize() // compressionType
- + TSEncoding.getSerializedSize() // encodingType
- + Integer.BYTES; // numOfPages
+ + TSEncoding.getSerializedSize(); // encodingType
}
/**
* deserialize from inputStream.
- *
- * @param markerRead Whether the marker of the CHUNK_HEADER has been read
*/
- public static ChunkHeader deserializeFrom(InputStream inputStream, boolean markerRead)
+ public static ChunkHeader deserializeFrom(InputStream inputStream, byte chunkType)
throws IOException {
- if (!markerRead) {
- byte marker = (byte) inputStream.read();
- if (marker != MetaMarker.CHUNK_HEADER) {
- MetaMarker.handleUnexpectedMarker(marker);
- }
- }
-
- String measurementID = ReadWriteIOUtils.readString(inputStream);
- int dataSize = ReadWriteIOUtils.readInt(inputStream);
- TSDataType dataType = TSDataType.deserialize(ReadWriteIOUtils.readShort(inputStream));
- int numOfPages = ReadWriteIOUtils.readInt(inputStream);
+ // read measurementID
+ String measurementID = ReadWriteIOUtils.readVarIntString(inputStream);
+ int dataSize = ReadWriteForEncodingUtils.readVarInt(inputStream);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(inputStream);
CompressionType type = ReadWriteIOUtils.readCompressionType(inputStream);
TSEncoding encoding = ReadWriteIOUtils.readEncoding(inputStream);
- return new ChunkHeader(measurementID, dataSize, dataType, type, encoding, numOfPages);
+ return new ChunkHeader(chunkType, measurementID, dataSize, dataType, type, encoding);
}
/**
* deserialize from TsFileInput.
*
- * @param input TsFileInput
- * @param offset offset
- * @param chunkHeaderSize the size of chunk's header
- * @param markerRead read marker (boolean type)
+ * @param input TsFileInput
+ * @param offset offset
+ * @param chunkHeaderSize the estimated size of chunk's header
* @return CHUNK_HEADER object
* @throws IOException IOException
*/
- public static ChunkHeader deserializeFrom(TsFileInput input, long offset, int chunkHeaderSize,
- boolean markerRead)
- throws IOException {
- long offsetVar = offset;
- if (!markerRead) {
- offsetVar++;
- }
+ public static ChunkHeader deserializeFrom(TsFileInput input, long offset, int chunkHeaderSize) throws IOException {
// read chunk header from input to buffer
ByteBuffer buffer = ByteBuffer.allocate(chunkHeaderSize);
- input.read(buffer, offsetVar);
+ input.read(buffer, offset);
buffer.flip();
+ byte chunkType = buffer.get();
// read measurementID
- int size = buffer.getInt();
+ int size = ReadWriteForEncodingUtils.readVarInt(buffer);
String measurementID = ReadWriteIOUtils.readStringWithLength(buffer, size);
- int dataSize = ReadWriteIOUtils.readInt(buffer);
- TSDataType dataType = TSDataType.deserialize(ReadWriteIOUtils.readShort(buffer));
- int numOfPages = ReadWriteIOUtils.readInt(buffer);
+ int dataSize = ReadWriteForEncodingUtils.readVarInt(buffer);
+ TSDataType dataType = ReadWriteIOUtils.readDataType(buffer);
CompressionType type = ReadWriteIOUtils.readCompressionType(buffer);
TSEncoding encoding = ReadWriteIOUtils.readEncoding(buffer);
- return new ChunkHeader(measurementID, dataSize, chunkHeaderSize, dataType, type, encoding,
- numOfPages);
+ chunkHeaderSize =
+ chunkHeaderSize - Integer.BYTES + ReadWriteForEncodingUtils.varIntSize(dataSize);
+ return new ChunkHeader(chunkType, measurementID, dataSize, chunkHeaderSize, dataType, type,
+ encoding);
}
public int getSerializedSize() {
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/PageHeader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/PageHeader.java
index fa24f93..0b3e4cd 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/PageHeader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/PageHeader.java
@@ -23,10 +23,9 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
-
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
public class PageHeader {
@@ -41,21 +40,24 @@ public class PageHeader {
this.statistics = statistics;
}
- public static int calculatePageHeaderSizeWithoutStatistics() {
- return 2 * Integer.BYTES; // uncompressedSize, compressedSize
+ /**
+ * max page header size without statistics
+ */
+ public static int estimateMaxPageHeaderSizeWithoutStatistics() {
+ return 2 * (Integer.BYTES + 1); // uncompressedSize, compressedSize
}
public static PageHeader deserializeFrom(InputStream inputStream, TSDataType dataType)
throws IOException {
- int uncompressedSize = ReadWriteIOUtils.readInt(inputStream);
- int compressedSize = ReadWriteIOUtils.readInt(inputStream);
+ int uncompressedSize = ReadWriteForEncodingUtils.readUnsignedVarInt(inputStream);
+ int compressedSize = ReadWriteForEncodingUtils.readUnsignedVarInt(inputStream);
Statistics statistics = Statistics.deserialize(inputStream, dataType);
return new PageHeader(uncompressedSize, compressedSize, statistics);
}
public static PageHeader deserializeFrom(ByteBuffer buffer, TSDataType dataType) {
- int uncompressedSize = ReadWriteIOUtils.readInt(buffer);
- int compressedSize = ReadWriteIOUtils.readInt(buffer);
+ int uncompressedSize = ReadWriteForEncodingUtils.readUnsignedVarInt(buffer);
+ int compressedSize = ReadWriteForEncodingUtils.readUnsignedVarInt(buffer);
Statistics statistics = Statistics.deserialize(buffer, dataType);
return new PageHeader(uncompressedSize, compressedSize, statistics);
}
@@ -93,8 +95,8 @@ public class PageHeader {
}
public void serializeTo(OutputStream outputStream) throws IOException {
- ReadWriteIOUtils.write(uncompressedSize, outputStream);
- ReadWriteIOUtils.write(compressedSize, outputStream);
+ ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, outputStream);
+ ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, outputStream);
statistics.serialize(outputStream);
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/BinaryStatistics.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/BinaryStatistics.java
index 607589b..6934d17 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/BinaryStatistics.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/BinaryStatistics.java
@@ -101,8 +101,13 @@ public class BinaryStatistics extends Statistics<Binary> {
}
@Override
- public double getSumValue() {
- throw new StatisticsClassException(String.format(BINARY_STATS_UNSUPPORTED_MSG, "sum"));
+ public double getSumDoubleValue() {
+ throw new StatisticsClassException(String.format(BINARY_STATS_UNSUPPORTED_MSG, "double sum"));
+ }
+
+ @Override
+ public long getSumLongValue() {
+ throw new StatisticsClassException(String.format(BINARY_STATS_UNSUPPORTED_MSG, "long sum"));
}
@Override
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/BooleanStatistics.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/BooleanStatistics.java
index f7c2d1c..e24ef46 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/BooleanStatistics.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/BooleanStatistics.java
@@ -32,8 +32,9 @@ public class BooleanStatistics extends Statistics<Boolean> {
private boolean firstValue;
private boolean lastValue;
+ private long sumValue;
- static final int BOOLEAN_STATISTICS_FIXED_RAM_SIZE = 48;
+ static final int BOOLEAN_STATISTICS_FIXED_RAM_SIZE = 56;
@Override
@@ -43,7 +44,7 @@ public class BooleanStatistics extends Statistics<Boolean> {
@Override
public int getStatsSize() {
- return 2;
+ return 10;
}
/**
@@ -52,16 +53,18 @@ public class BooleanStatistics extends Statistics<Boolean> {
* @param firstValue first boolean value
* @param lastValue last boolean value
*/
- public void initializeStats(boolean firstValue, boolean lastValue) {
+ public void initializeStats(boolean firstValue, boolean lastValue, long sum) {
this.firstValue = firstValue;
this.lastValue = lastValue;
+ this.sumValue = sum;
}
- private void updateStats(boolean firstValue, boolean lastValue) {
+ private void updateStats(boolean firstValue, boolean lastValue, long sum) {
this.lastValue = lastValue;
+ this.sumValue += sum;
}
- private void updateStats(boolean firstValue, boolean lastValue, long startTime, long endTime) {
+ private void updateStats(boolean firstValue, boolean lastValue, long startTime, long endTime, long sum) {
// only if endTime greater or equals to the current endTime need we update the last value
// only if startTime less or equals to the current startTime need we update the first value
// otherwise, just ignore
@@ -71,15 +74,16 @@ public class BooleanStatistics extends Statistics<Boolean> {
if (endTime >= this.getEndTime()) {
this.lastValue = lastValue;
}
+ this.sumValue += sum;
}
@Override
void updateStats(boolean value) {
if (isEmpty) {
- initializeStats(value, value);
+ initializeStats(value, value, value ? 1 : 0);
isEmpty = false;
} else {
- updateStats(value, value);
+ updateStats(value, value, value ? 1 : 0);
}
}
@@ -120,8 +124,13 @@ public class BooleanStatistics extends Statistics<Boolean> {
}
@Override
- public double getSumValue() {
- throw new StatisticsClassException("Boolean statistics does not support: sum");
+ public double getSumDoubleValue() {
+ throw new StatisticsClassException("Boolean statistics does not support: double sum");
+ }
+
+ @Override
+ public long getSumLongValue() {
+ return sumValue;
}
@Override
@@ -146,17 +155,18 @@ public class BooleanStatistics extends Statistics<Boolean> {
@Override
public ByteBuffer getSumValueBuffer() {
- throw new StatisticsClassException("Boolean statistics do not support: sum");
+ return ReadWriteIOUtils.getByteBuffer(sumValue);
}
@Override
protected void mergeStatisticsValue(Statistics stats) {
BooleanStatistics boolStats = (BooleanStatistics) stats;
if (isEmpty) {
- initializeStats(boolStats.getFirstValue(), boolStats.getLastValue());
+ initializeStats(boolStats.getFirstValue(), boolStats.getLastValue(), boolStats.sumValue);
isEmpty = false;
} else {
- updateStats(boolStats.getFirstValue(), boolStats.getLastValue(), stats.getStartTime(), stats.getEndTime());
+ updateStats(boolStats.getFirstValue(), boolStats.getLastValue(), stats.getStartTime(),
+ stats.getEndTime(), boolStats.sumValue);
}
}
@@ -182,7 +192,7 @@ public class BooleanStatistics extends Statistics<Boolean> {
@Override
public byte[] getSumValueBytes() {
- throw new StatisticsClassException("Boolean statistics does not support: sum");
+ return BytesUtils.longToBytes(sumValue);
}
@Override
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/IntegerStatistics.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/IntegerStatistics.java
index 4ca9e81..dec73a1 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/IntegerStatistics.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/IntegerStatistics.java
@@ -18,14 +18,14 @@
*/
package org.apache.iotdb.tsfile.file.metadata.statistics;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.utils.BytesUtils;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
+import org.apache.iotdb.tsfile.exception.filter.StatisticsClassException;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.utils.BytesUtils;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
/**
* Statistics for int type.
@@ -36,7 +36,7 @@ public class IntegerStatistics extends Statistics<Integer> {
private int maxValue;
private int firstValue;
private int lastValue;
- private double sumValue;
+ private long sumValue;
static final int INTEGER_STATISTICS_FIXED_RAM_SIZE = 64;
@@ -51,7 +51,7 @@ public class IntegerStatistics extends Statistics<Integer> {
return 24;
}
- public void initializeStats(int min, int max, int first, int last, double sum) {
+ public void initializeStats(int min, int max, int first, int last, long sum) {
this.minValue = min;
this.maxValue = max;
this.firstValue = first;
@@ -59,7 +59,7 @@ public class IntegerStatistics extends Statistics<Integer> {
this.sumValue = sum;
}
- private void updateStats(int minValue, int maxValue, int lastValue, double sumValue) {
+ private void updateStats(int minValue, int maxValue, int lastValue, long sumValue) {
if (minValue < this.minValue) {
this.minValue = minValue;
}
@@ -70,7 +70,8 @@ public class IntegerStatistics extends Statistics<Integer> {
this.lastValue = lastValue;
}
- private void updateStats(int minValue, int maxValue, int firstValue, int lastValue, double sumValue, long startTime, long endTime) {
+ private void updateStats(int minValue, int maxValue, int firstValue, int lastValue,
+ long sumValue, long startTime, long endTime) {
if (minValue < this.minValue) {
this.minValue = minValue;
}
@@ -102,7 +103,6 @@ public class IntegerStatistics extends Statistics<Integer> {
isEmpty = false;
} else {
updateStats(value, value, value, value);
- isEmpty = false;
}
}
@@ -139,7 +139,12 @@ public class IntegerStatistics extends Statistics<Integer> {
}
@Override
- public double getSumValue() {
+ public double getSumDoubleValue() {
+ throw new StatisticsClassException("Integer statistics does not support: double sum");
+ }
+
+ @Override
+ public long getSumLongValue() {
return sumValue;
}
@@ -147,14 +152,13 @@ public class IntegerStatistics extends Statistics<Integer> {
protected void mergeStatisticsValue(Statistics stats) {
IntegerStatistics intStats = (IntegerStatistics) stats;
if (isEmpty) {
- initializeStats(intStats.getMinValue(), intStats.getMaxValue(), intStats.getFirstValue(), intStats.getLastValue(),
- intStats.getSumValue());
+ initializeStats(intStats.getMinValue(), intStats.getMaxValue(), intStats.getFirstValue(),
+ intStats.getLastValue(), intStats.sumValue);
isEmpty = false;
} else {
- updateStats(intStats.getMinValue(), intStats.getMaxValue(), intStats.getFirstValue(), intStats.getLastValue(),
- intStats.getSumValue(), stats.getStartTime(), stats.getEndTime());
+ updateStats(intStats.getMinValue(), intStats.getMaxValue(), intStats.getFirstValue(),
+ intStats.getLastValue(), intStats.sumValue, stats.getStartTime(), stats.getEndTime());
}
-
}
@Override
@@ -204,7 +208,7 @@ public class IntegerStatistics extends Statistics<Integer> {
@Override
public byte[] getSumValueBytes() {
- return BytesUtils.doubleToBytes(sumValue);
+ return BytesUtils.longToBytes(sumValue);
}
@Override
@@ -224,7 +228,7 @@ public class IntegerStatistics extends Statistics<Integer> {
this.maxValue = ReadWriteIOUtils.readInt(inputStream);
this.firstValue = ReadWriteIOUtils.readInt(inputStream);
this.lastValue = ReadWriteIOUtils.readInt(inputStream);
- this.sumValue = ReadWriteIOUtils.readDouble(inputStream);
+ this.sumValue = ReadWriteIOUtils.readLong(inputStream);
}
@Override
@@ -233,12 +237,12 @@ public class IntegerStatistics extends Statistics<Integer> {
this.maxValue = ReadWriteIOUtils.readInt(byteBuffer);
this.firstValue = ReadWriteIOUtils.readInt(byteBuffer);
this.lastValue = ReadWriteIOUtils.readInt(byteBuffer);
- this.sumValue = ReadWriteIOUtils.readDouble(byteBuffer);
+ this.sumValue = ReadWriteIOUtils.readLong(byteBuffer);
}
@Override
public String toString() {
- return super.toString() + " [minValue:" + minValue + ",maxValue:" + maxValue + ",firstValue:" + firstValue +
- ",lastValue:" + lastValue + ",sumValue:" + sumValue + "]";
+ return super.toString() + " [minValue:" + minValue + ",maxValue:" + maxValue + ",firstValue:"
+ + firstValue + ",lastValue:" + lastValue + ",sumValue:" + sumValue + "]";
}
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/Statistics.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/Statistics.java
index 7a31494..acbc339 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/Statistics.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/metadata/statistics/Statistics.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.tsfile.exception.filter.StatisticsClassException;
import org.apache.iotdb.tsfile.exception.write.UnknownColumnTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,7 +51,7 @@ public abstract class Statistics<T> {
/**
* number of time-value points
*/
- private long count = 0;
+ private int count = 0;
private long startTime = Long.MAX_VALUE;
private long endTime = Long.MIN_VALUE;
@@ -102,7 +103,8 @@ public abstract class Statistics<T> {
public abstract TSDataType getType();
public int getSerializedSize() {
- return 24 // count, startTime, endTime
+ return ReadWriteForEncodingUtils.uVarIntSize(count) // count
+ + 16 // startTime, endTime
+ getStatsSize();
}
@@ -110,7 +112,7 @@ public abstract class Statistics<T> {
public int serialize(OutputStream outputStream) throws IOException {
int byteLen = 0;
- byteLen += ReadWriteIOUtils.write(count, outputStream);
+ byteLen += ReadWriteForEncodingUtils.writeUnsignedVarInt(count, outputStream);
byteLen += ReadWriteIOUtils.write(startTime, outputStream);
byteLen += ReadWriteIOUtils.write(endTime, outputStream);
// value statistics of different data type
@@ -137,7 +139,9 @@ public abstract class Statistics<T> {
public abstract T getLastValue();
- public abstract double getSumValue();
+ public abstract double getSumDoubleValue();
+
+ public abstract long getSumLongValue();
public abstract byte[] getMinValueBytes();
@@ -389,7 +393,7 @@ public abstract class Statistics<T> {
public static Statistics deserialize(InputStream inputStream, TSDataType dataType)
throws IOException {
Statistics statistics = getStatsByType(dataType);
- statistics.setCount(ReadWriteIOUtils.readLong(inputStream));
+ statistics.setCount(ReadWriteForEncodingUtils.readUnsignedVarInt(inputStream));
statistics.setStartTime(ReadWriteIOUtils.readLong(inputStream));
statistics.setEndTime(ReadWriteIOUtils.readLong(inputStream));
statistics.deserialize(inputStream);
@@ -399,7 +403,7 @@ public abstract class Statistics<T> {
public static Statistics deserialize(ByteBuffer buffer, TSDataType dataType) {
Statistics statistics = getStatsByType(dataType);
- statistics.setCount(ReadWriteIOUtils.readLong(buffer));
+ statistics.setCount(ReadWriteForEncodingUtils.readUnsignedVarInt(buffer));
statistics.setStartTime(ReadWriteIOUtils.readLong(buffer));
statistics.setEndTime(ReadWriteIOUtils.readLong(buffer));
statistics.deserialize(buffer);
@@ -427,7 +431,7 @@ public abstract class Statistics<T> {
this.endTime = endTime;
}
- public void setCount(long count) {
+ public void setCount(int count) {
this.count = count;
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
index 16f932e..9b9b3f4 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
@@ -39,7 +39,7 @@ import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.compress.IUnCompressor;
import org.apache.iotdb.tsfile.file.MetaMarker;
-import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
+import org.apache.iotdb.tsfile.file.footer.ChunkGroupHeader;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
import org.apache.iotdb.tsfile.file.header.PageHeader;
import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetadata;
@@ -207,7 +207,7 @@ public class TsFileSequenceReader implements AutoCloseable {
*/
public boolean isComplete() throws IOException {
return tsFileInput.size() >= TSFileConfig.MAGIC_STRING.getBytes().length * 2
- + TSFileConfig.VERSION_NUMBER.getBytes().length
+ + TSFileConfig.VERSION_NUMBER_V2.getBytes().length
&& (readTailMagic().equals(readHeadMagic()) || readTailMagic()
.equals(TSFileConfig.VERSION_NUMBER_V1));
}
@@ -228,7 +228,7 @@ public class TsFileSequenceReader implements AutoCloseable {
*/
public String readVersionNumber() throws IOException {
ByteBuffer versionNumberBytes = ByteBuffer
- .allocate(TSFileConfig.VERSION_NUMBER.getBytes().length);
+ .allocate(TSFileConfig.VERSION_NUMBER_V2.getBytes().length);
tsFileInput.read(versionNumberBytes, TSFileConfig.MAGIC_STRING.getBytes().length);
versionNumberBytes.flip();
return new String(versionNumberBytes.array());
@@ -654,8 +654,8 @@ public class TsFileSequenceReader implements AutoCloseable {
* @return a CHUNK_GROUP_FOOTER
* @throws IOException io error
*/
- public ChunkGroupFooter readChunkGroupFooter() throws IOException {
- return ChunkGroupFooter.deserializeFrom(tsFileInput.wrapAsInputStream(), true);
+ public ChunkGroupHeader readChunkGroupFooter() throws IOException {
+ return ChunkGroupHeader.deserializeFrom(tsFileInput.wrapAsInputStream(), true);
}
/**
@@ -666,9 +666,9 @@ public class TsFileSequenceReader implements AutoCloseable {
* @return a CHUNK_GROUP_FOOTER
* @throws IOException io error
*/
- public ChunkGroupFooter readChunkGroupFooter(long position, boolean markerRead)
+ public ChunkGroupHeader readChunkGroupFooter(long position, boolean markerRead)
throws IOException {
- return ChunkGroupFooter.deserializeFrom(tsFileInput, position, markerRead);
+ return ChunkGroupHeader.deserializeFrom(tsFileInput, position, markerRead);
}
public long readVersion() throws IOException {
@@ -687,20 +687,17 @@ public class TsFileSequenceReader implements AutoCloseable {
* @return a CHUNK_HEADER
* @throws IOException io error
*/
- public ChunkHeader readChunkHeader() throws IOException {
- return ChunkHeader.deserializeFrom(tsFileInput.wrapAsInputStream(), true);
+ public ChunkHeader readChunkHeader(byte chunkType) throws IOException {
+ return ChunkHeader.deserializeFrom(tsFileInput.wrapAsInputStream(), chunkType);
}
/**
* read the chunk's header.
- *
- * @param position the file offset of this chunk's header
+ * @param position the file offset of this chunk's header
* @param chunkHeaderSize the size of chunk's header
- * @param markerRead true if the offset does not contains the marker , otherwise false
*/
- private ChunkHeader readChunkHeader(long position, int chunkHeaderSize, boolean markerRead)
- throws IOException {
- return ChunkHeader.deserializeFrom(tsFileInput, position, chunkHeaderSize, markerRead);
+ private ChunkHeader readChunkHeader(long position, int chunkHeaderSize) throws IOException {
+ return ChunkHeader.deserializeFrom(tsFileInput, position, chunkHeaderSize);
}
/**
@@ -722,7 +719,7 @@ public class TsFileSequenceReader implements AutoCloseable {
*/
public Chunk readMemChunk(ChunkMetadata metaData) throws IOException {
int chunkHeadSize = ChunkHeader.getSerializedSize(metaData.getMeasurementUid());
- ChunkHeader header = readChunkHeader(metaData.getOffsetOfChunkHeader(), chunkHeadSize, false);
+ ChunkHeader header = readChunkHeader(metaData.getOffsetOfChunkHeader(), chunkHeadSize);
ByteBuffer buffer = readChunk(metaData.getOffsetOfChunkHeader() + header.getSerializedSize(),
header.getDataSize());
return new Chunk(header, buffer, metaData.getDeleteIntervalList());
@@ -903,12 +900,12 @@ public class TsFileSequenceReader implements AutoCloseable {
List<ChunkMetadata> chunkMetadataList = null;
String deviceID;
- int headerLength = TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER
+ int headerLength = TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER_V2
.getBytes().length;
if (fileSize < headerLength) {
return TsFileCheckStatus.INCOMPATIBLE_FILE;
}
- if (!TSFileConfig.MAGIC_STRING.equals(readHeadMagic()) || !TSFileConfig.VERSION_NUMBER
+ if (!TSFileConfig.MAGIC_STRING.equals(readHeadMagic()) || !TSFileConfig.VERSION_NUMBER_V2
.equals(readVersionNumber())) {
return TsFileCheckStatus.INCOMPATIBLE_FILE;
}
@@ -932,6 +929,7 @@ public class TsFileSequenceReader implements AutoCloseable {
while ((marker = this.readMarker()) != MetaMarker.SEPARATOR) {
switch (marker) {
case MetaMarker.CHUNK_HEADER:
+ case MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER:
// this is the first chunk of a new ChunkGroup.
if (newChunkGroup) {
newChunkGroup = false;
@@ -941,7 +939,7 @@ public class TsFileSequenceReader implements AutoCloseable {
// if there is something wrong with a chunk, we will drop the whole ChunkGroup
// as different chunks may be created by the same insertions(sqls), and partial
// insertion is not tolerable
- ChunkHeader chunkHeader = this.readChunkHeader();
+ ChunkHeader chunkHeader = this.readChunkHeader(marker);
measurementID = chunkHeader.getMeasurementID();
MeasurementSchema measurementSchema = new MeasurementSchema(measurementID,
chunkHeader.getDataType(),
@@ -960,12 +958,12 @@ public class TsFileSequenceReader implements AutoCloseable {
chunkMetadataList.add(currentChunk);
chunkCnt++;
break;
- case MetaMarker.CHUNK_GROUP_FOOTER:
+ case MetaMarker.CHUNK_GROUP_HEADER:
// this is a chunk group
// if there is something wrong with the ChunkGroup Footer, we will drop this ChunkGroup
// because we can not guarantee the correctness of the deviceId.
- ChunkGroupFooter chunkGroupFooter = this.readChunkGroupFooter();
- deviceID = chunkGroupFooter.getDeviceID();
+ ChunkGroupHeader chunkGroupHeader = this.readChunkGroupFooter();
+ deviceID = chunkGroupHeader.getDeviceID();
if (newSchema != null) {
for (MeasurementSchema tsSchema : measurementSchemaList) {
newSchema.putIfAbsent(new Path(deviceID, tsSchema.getMeasurementId()), tsSchema);
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/LocalTsFileInput.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/LocalTsFileInput.java
index de314b1..ae05d7d 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/LocalTsFileInput.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/LocalTsFileInput.java
@@ -25,6 +25,7 @@ import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -125,4 +126,13 @@ public class LocalTsFileInput implements TsFileInput {
public int readInt() {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public String readVarIntString(long offset) throws IOException {
+ long position = channel.position();
+ channel.position(offset);
+ String res = ReadWriteIOUtils.readVarIntString(wrapAsInputStream());
+ channel.position(position);
+ return res;
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/TsFileInput.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/TsFileInput.java
index b948e0c..e215d3c 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/TsFileInput.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/reader/TsFileInput.java
@@ -143,4 +143,9 @@ public interface TsFileInput {
* read 4 bytes from the Input and convert it to a integer.
*/
int readInt() throws IOException;
+
+ /**
+ * read a string from the Input at the given position
+ */
+ String readVarIntString(long offset) throws IOException;
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteForEncodingUtils.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteForEncodingUtils.java
index 0b1f8b0..578ab4c 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteForEncodingUtils.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/utils/ReadWriteForEncodingUtils.java
@@ -106,6 +106,15 @@ public class ReadWriteForEncodingUtils {
return value | (b << i);
}
+ public static int readVarInt(InputStream in) throws IOException {
+ int value = readUnsignedVarInt(in);
+ int x = value >>> 1;
+ if ((value & 1) != 0) {
+ x = ~x;
+ }
+ return x;
+ }
+
/**
* read an unsigned var int in stream and transform it to int format.
*
@@ -123,6 +132,15 @@ public class ReadWriteForEncodingUtils {
return value | (b << i);
}
+ public static int readVarInt(ByteBuffer buffer) {
+ int value = readUnsignedVarInt(buffer);
+ int x = value >>> 1;
+ if ((value & 1) != 0) {
+ x = ~x;
+ }
+ return x;
+ }
+
/**
* write a value to stream using unsigned var int format. for example, int
* 123456789 has its binary format 00000111-01011011-11001101-00010101 (if we
@@ -134,12 +152,42 @@ public class ReadWriteForEncodingUtils {
* @param value value to write into stream
* @param out output stream
*/
- public static void writeUnsignedVarInt(int value, ByteArrayOutputStream out) {
+ public static int writeUnsignedVarInt(int value, ByteArrayOutputStream out) {
+ int position = 1;
+ while ((value & 0xFFFFFF80) != 0L) {
+ out.write((value & 0x7F) | 0x80);
+ value >>>= 7;
+ position++;
+ }
+ out.write(value & 0x7F);
+ return position;
+ }
+
+ public static int writeVarInt(int value, ByteArrayOutputStream out) {
+ int uValue = value << 1;
+ if (value < 0) {
+ uValue = ~uValue;
+ }
+ return writeUnsignedVarInt(uValue, out);
+ }
+
+ public static int writeUnsignedVarInt(int value, OutputStream out) throws IOException {
+ int position = 1;
while ((value & 0xFFFFFF80) != 0L) {
out.write((value & 0x7F) | 0x80);
value >>>= 7;
+ position++;
}
out.write(value & 0x7F);
+ return position;
+ }
+
+ public static int writeVarInt(int value, OutputStream out) throws IOException {
+ int uValue = value << 1;
+ if (value < 0) {
+ uValue = ~uValue;
+ }
+ return writeUnsignedVarInt(uValue, out);
}
/**
@@ -167,6 +215,46 @@ public class ReadWriteForEncodingUtils {
return position;
}
+ public static int writeVarInt(int value, ByteBuffer buffer) {
+ int uValue = value << 1;
+ if (value < 0) {
+ uValue = ~uValue;
+ }
+ return writeUnsignedVarInt(uValue, buffer);
+ }
+
+ /**
+ * Returns the encoding size in bytes of its input value.
+ * @param value the integer to be measured
+ * @return the encoding size in bytes of its input value
+ */
+ public static int varIntSize(int value) {
+ int uValue = value << 1;
+ if (value < 0) {
+ uValue = ~uValue;
+ }
+ int position = 1;
+ while ((uValue & 0xFFFFFF80) != 0L) {
+ uValue >>>= 7;
+ position++;
+ }
+ return position;
+ }
+
+ /**
+ * Returns the encoding size in bytes of its input value.
+ * @param value the unsigned integer to be measured
+ * @return the encoding size in bytes of its input value
+ */
+ public static int uVarIntSize(int value) {
+ int position = 1;
+ while ((value & 0xFFFFFF80) != 0L) {
+ value >>>= 7;
+ position++;
+ }
+ return position;
+ }
+
/**
* write integer value using special bit to output stream.
*
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 04ab1ba..45f00ae 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
@@ -353,6 +353,25 @@ public class ReadWriteIOUtils {
}
/**
+ * write string to outputStream.
+ *
+ * @return the length of string represented by byte[].
+ */
+ public static int writeVar(String s, OutputStream outputStream) throws IOException {
+ int len = 0;
+ if (s == null) {
+ len += ReadWriteForEncodingUtils.writeVarInt(-1, outputStream);
+ return len;
+ }
+
+ byte[] bytes = s.getBytes();
+ len += write(bytes.length, outputStream);
+ outputStream.write(bytes);
+ len += bytes.length;
+ return len;
+ }
+
+ /**
* write string to byteBuffer.
*
* @return the length of string represented by byte[].
@@ -369,6 +388,18 @@ public class ReadWriteIOUtils {
return len;
}
+ public static int writeVar(String s, ByteBuffer buffer) {
+ if (s == null) {
+ return write(-1, buffer);
+ }
+ int len = 0;
+ byte[] bytes = s.getBytes();
+ len += write(bytes.length, buffer);
+ buffer.put(bytes);
+ len += bytes.length;
+ return len;
+ }
+
/**
* write byteBuffer.capacity and byteBuffer.array to outputStream.
*/
@@ -454,6 +485,13 @@ public class ReadWriteIOUtils {
}
/**
+ * read a byte var from inputStream.
+ */
+ public static byte readByte(InputStream inputStream) throws IOException {
+ return (byte) inputStream.read();
+ }
+
+ /**
* read a short var from inputStream.
*/
public static short readShort(InputStream inputStream) throws IOException {
@@ -587,6 +625,23 @@ public class ReadWriteIOUtils {
}
/**
+ * string length's type is varInt
+ */
+ public static String readVarIntString(InputStream inputStream) throws IOException {
+ int strLength = ReadWriteForEncodingUtils.readVarInt(inputStream);
+ if (strLength <= 0) {
+ return null;
+ }
+ byte[] bytes = new byte[strLength];
+ int readLen = inputStream.read(bytes, 0, strLength);
+ if (readLen != strLength) {
+ throw new IOException(String.format(RETURN_ERROR,
+ strLength, readLen));
+ }
+ return new String(bytes, 0, strLength);
+ }
+
+ /**
* read string from byteBuffer.
*/
public static String readString(ByteBuffer buffer) {
@@ -602,6 +657,21 @@ public class ReadWriteIOUtils {
}
/**
+ * string length's type is varInt
+ */
+ public static String readVarIntString(ByteBuffer buffer) {
+ int strLength = readInt(buffer);
+ if (strLength < 0) {
+ return null;
+ } else if (strLength == 0) {
+ return "";
+ }
+ byte[] bytes = new byte[strLength];
+ buffer.get(bytes, 0, strLength);
+ return new String(bytes, 0, strLength);
+ }
+
+ /**
* read string from byteBuffer with user define length.
*/
public static String readStringWithLength(ByteBuffer buffer, int length) {
@@ -842,33 +912,33 @@ public class ReadWriteIOUtils {
}
public static CompressionType readCompressionType(InputStream inputStream) throws IOException {
- short n = readShort(inputStream);
- return CompressionType.deserialize(n);
+ byte n = readByte(inputStream);
+ return CompressionType.byteToEnum(n);
}
public static CompressionType readCompressionType(ByteBuffer buffer) {
- short n = readShort(buffer);
- return CompressionType.deserialize(n);
+ byte n = buffer.get();
+ return CompressionType.byteToEnum(n);
}
public static TSDataType readDataType(InputStream inputStream) throws IOException {
- short n = readShort(inputStream);
- return TSDataType.deserialize(n);
+ byte n = readByte(inputStream);
+ return TSDataType.byteToEnum(n);
}
public static TSDataType readDataType(ByteBuffer buffer) {
- short n = readShort(buffer);
- return TSDataType.deserialize(n);
+ byte n = buffer.get();
+ return TSDataType.byteToEnum(n);
}
public static TSEncoding readEncoding(InputStream inputStream) throws IOException {
- short n = readShort(inputStream);
- return TSEncoding.deserialize(n);
+ byte n = readByte(inputStream);
+ return TSEncoding.byteToEnum(n);
}
public static TSEncoding readEncoding(ByteBuffer buffer) {
- short n = readShort(buffer);
- return TSEncoding.deserialize(n);
+ byte n = buffer.get();
+ return TSEncoding.byteToEnum(n);
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/ChunkGroupMetaDataV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/ChunkGroupMetaDataV1.java
deleted file mode 100644
index 64ce030..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/ChunkGroupMetaDataV1.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.tsfile.v1.file.metadata;
-
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Metadata of ChunkGroup.
- */
-public class ChunkGroupMetaDataV1 {
-
- /**
- * Name of device, this field is not serialized.
- */
- private String deviceID;
-
- /**
- * Byte offset of the corresponding data in the file Notice: include the chunk group marker.
- * For Hadoop and Spark.
- */
- private long startOffsetOfChunkGroup;
-
- /**
- * End Byte position of the whole chunk group in the file Notice: position after the chunk group footer.
- * For Hadoop and Spark.
- */
- private long endOffsetOfChunkGroup;
-
- /**
- * All time series chunks in this chunk group.
- */
- private List<ChunkMetadataV1> chunkMetaDataList;
-
- private long version;
-
- private ChunkGroupMetaDataV1() {
- chunkMetaDataList = new ArrayList<>();
- }
-
- /**
- * deserialize from ByteBuffer.
- *
- * @param buffer ByteBuffer
- * @return ChunkGroupMetaData object
- */
- public static ChunkGroupMetaDataV1 deserializeFrom(ByteBuffer buffer) {
- ChunkGroupMetaDataV1 chunkGroupMetaData = new ChunkGroupMetaDataV1();
-
- chunkGroupMetaData.deviceID = ReadWriteIOUtils.readString(buffer);
- chunkGroupMetaData.startOffsetOfChunkGroup = ReadWriteIOUtils.readLong(buffer);
- chunkGroupMetaData.endOffsetOfChunkGroup = ReadWriteIOUtils.readLong(buffer);
- chunkGroupMetaData.version = ReadWriteIOUtils.readLong(buffer);
-
- int size = ReadWriteIOUtils.readInt(buffer);
-
- List<ChunkMetadataV1> chunkMetaDataList = new ArrayList<>();
- for (int i = 0; i < size; i++) {
- ChunkMetadataV1 metaData = ChunkMetadataV1.deserializeFrom(buffer);
- chunkMetaDataList.add(metaData);
- }
- chunkGroupMetaData.chunkMetaDataList = chunkMetaDataList;
-
- return chunkGroupMetaData;
- }
-
- /**
- * add time series chunk metadata to list. THREAD NOT SAFE
- *
- * @param metadata time series metadata to add
- */
- public void addTimeSeriesChunkMetaData(ChunkMetadataV1 metadata) {
- if (chunkMetaDataList == null) {
- chunkMetaDataList = new ArrayList<>();
- }
- chunkMetaDataList.add(metadata);
- }
-
- public List<ChunkMetadataV1> getChunkMetaDataList() {
- return chunkMetaDataList;
- }
-
- public String getDeviceID() {
- return deviceID;
- }
-
- public long getStartOffsetOfChunkGroup() {
- return startOffsetOfChunkGroup;
- }
-
- public long getEndOffsetOfChunkGroup() {
- return endOffsetOfChunkGroup;
- }
-
- public long getVersion() {
- return version;
- }
-
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/ChunkMetadataV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/ChunkMetadataV1.java
deleted file mode 100644
index fee07ab..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/ChunkMetadataV1.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.tsfile.v1.file.metadata;
-
-import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import org.apache.iotdb.tsfile.v1.file.metadata.statistics.StatisticsV1;
-
-import java.nio.ByteBuffer;
-/**
- * MetaData of one chunk.
- */
-public class ChunkMetadataV1 {
-
-
- private String measurementUid;
-
- /**
- * Byte offset of the corresponding data in the file Notice: include the chunk header and marker.
- */
- private long offsetOfChunkHeader;
-
- private long numOfPoints;
-
- private long startTime;
-
- private long endTime;
-
- private TSDataType tsDataType;
-
- /**
- * version is used to define the order of operations(insertion, deletion, update). version is set
- * according to its belonging ChunkGroup only when being queried, so it is not persisted.
- */
- private long version;
-
- private TsDigestV1 valuesStatistics;
-
- private ChunkMetadataV1() {
- }
-
- /**
- * deserialize from ByteBuffer.
- *
- * @param buffer ByteBuffer
- * @return ChunkMetaData object
- */
- public static ChunkMetadataV1 deserializeFrom(ByteBuffer buffer) {
- ChunkMetadataV1 chunkMetaData = new ChunkMetadataV1();
-
- chunkMetaData.measurementUid = ReadWriteIOUtils.readString(buffer);
- chunkMetaData.offsetOfChunkHeader = ReadWriteIOUtils.readLong(buffer);
- chunkMetaData.numOfPoints = ReadWriteIOUtils.readLong(buffer);
- chunkMetaData.startTime = ReadWriteIOUtils.readLong(buffer);
- chunkMetaData.endTime = ReadWriteIOUtils.readLong(buffer);
- chunkMetaData.tsDataType = ReadWriteIOUtils.readDataType(buffer);
-
- chunkMetaData.valuesStatistics = TsDigestV1.deserializeFrom(buffer);
-
- return chunkMetaData;
- }
-
- public long getNumOfPoints() {
- return numOfPoints;
- }
-
- public ChunkMetadata upgradeToChunkMetadata() {
- Statistics<?> statistics = StatisticsV1
- .constructStatisticsFromOldChunkMetadata(this);
- ChunkMetadata chunkMetadata = new ChunkMetadata(this.measurementUid, this.tsDataType,
- this.offsetOfChunkHeader, statistics);
- chunkMetadata.setFromOldTsFile(true);
- return chunkMetadata;
- }
-
- /**
- * get offset of chunk header.
- *
- * @return Byte offset of header of this chunk (includes the marker)
- */
- public long getOffsetOfChunkHeader() {
- return offsetOfChunkHeader;
- }
-
- public String getMeasurementUid() {
- return measurementUid;
- }
-
- public TsDigestV1 getDigest() {
- return valuesStatistics;
- }
-
- public long getStartTime() {
- return startTime;
- }
-
- public long getEndTime() {
- return endTime;
- }
-
- public TSDataType getTsDataType() {
- return tsDataType;
- }
-
- public long getVersion() {
- return version;
- }
-
- public void setVersion(long version) {
- this.version = version;
- }
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TimeseriesMetadataForV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TimeseriesMetadataForV1.java
deleted file mode 100644
index 04fd8e5..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TimeseriesMetadataForV1.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.tsfile.v1.file.metadata;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
-import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
-
-public class TimeseriesMetadataForV1 extends TimeseriesMetadata {
-
- private List<ChunkMetadata> chunkMetadataList;
-
- public void setChunkMetadataList(List<ChunkMetadata> chunkMetadataList) {
- this.chunkMetadataList = chunkMetadataList;
- }
-
- @Override
- public List<ChunkMetadata> loadChunkMetadataList() throws IOException {
- chunkMetadataLoader.setDiskChunkLoader(chunkMetadataList);
- return chunkMetadataList;
- }
-
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDeviceMetadataIndexV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDeviceMetadataIndexV1.java
deleted file mode 100644
index 2258b61..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDeviceMetadataIndexV1.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.tsfile.v1.file.metadata;
-
-import java.nio.ByteBuffer;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-public class TsDeviceMetadataIndexV1 {
-
- /**
- * The offset of the TsDeviceMetadata.
- */
- private long offset;
- /**
- * The size of the TsDeviceMetadata in the disk.
- */
- private int len;
- /**
- * The start time of the device.
- */
- private long startTime;
- /**
- * The end time of the device.
- */
- private long endTime;
-
- public TsDeviceMetadataIndexV1() {
- //do nothing
- }
-
- /**
- * use buffer to get a TsDeviceMetadataIndex.
- *
- * @param buffer -determine the index's source
- * @return -a TsDeviceMetadataIndex
- */
- public static TsDeviceMetadataIndexV1 deserializeFrom(ByteBuffer buffer) {
- TsDeviceMetadataIndexV1 index = new TsDeviceMetadataIndexV1();
- index.offset = ReadWriteIOUtils.readLong(buffer);
- index.len = ReadWriteIOUtils.readInt(buffer);
- index.startTime = ReadWriteIOUtils.readLong(buffer);
- index.endTime = ReadWriteIOUtils.readLong(buffer);
- return index;
- }
-
- public long getOffset() {
- return offset;
- }
-
- public int getLen() {
- return len;
- }
-
- public long getStartTime() {
- return startTime;
- }
-
- public long getEndTime() {
- return endTime;
- }
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDeviceMetadataV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDeviceMetadataV1.java
deleted file mode 100644
index 4441158..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDeviceMetadataV1.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.tsfile.v1.file.metadata;
-
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-public class TsDeviceMetadataV1 {
-
- /**
- * start time for a device.
- **/
- private long startTime = Long.MAX_VALUE;
-
- /**
- * end time for a device.
- **/
- private long endTime = Long.MIN_VALUE;
-
- /**
- * Row groups in this file.
- */
- private List<ChunkGroupMetaDataV1> chunkGroupMetadataList = new ArrayList<>();
-
- public TsDeviceMetadataV1() {
- // allowed to clair an empty TsDeviceMetadata whose fields will be assigned later.
- }
-
-
- /**
- * deserialize from the given buffer.
- *
- * @param buffer -buffer to deserialize
- * @return -device meta data
- */
- public static TsDeviceMetadataV1 deserializeFrom(ByteBuffer buffer) {
- TsDeviceMetadataV1 deviceMetadata = new TsDeviceMetadataV1();
-
- deviceMetadata.startTime = ReadWriteIOUtils.readLong(buffer);
- deviceMetadata.endTime = ReadWriteIOUtils.readLong(buffer);
-
- int size = ReadWriteIOUtils.readInt(buffer);
- if (size > 0) {
- List<ChunkGroupMetaDataV1> chunkGroupMetaDataList = new ArrayList<>();
- for (int i = 0; i < size; i++) {
- chunkGroupMetaDataList.add(ChunkGroupMetaDataV1.deserializeFrom(buffer));
- }
- deviceMetadata.chunkGroupMetadataList = chunkGroupMetaDataList;
- }
-
- return deviceMetadata;
- }
-
- public List<ChunkGroupMetaDataV1> getChunkGroupMetaDataList() {
- return Collections.unmodifiableList(chunkGroupMetadataList);
- }
-
- public long getStartTime() {
- return startTime;
- }
-
- public long getEndTime() {
- return endTime;
- }
-
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDigestV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDigestV1.java
deleted file mode 100644
index a40d3be..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsDigestV1.java
+++ /dev/null
@@ -1,75 +0,0 @@
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.tsfile.v1.file.metadata;
-
-import java.nio.ByteBuffer;
-
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-/**
- * Digest/statistics per chunk group and per page.
- */
-public class TsDigestV1 {
-
- private ByteBuffer[] statistics;
-
- public TsDigestV1() {
- // allowed to declare an empty TsDigest whose fields will be assigned later.
- }
-
- /**
- * use given buffer to deserialize.
- *
- * @param buffer -given buffer
- * @return -an instance of TsDigest
- */
- public static TsDigestV1 deserializeFrom(ByteBuffer buffer) {
- TsDigestV1 digest = new TsDigestV1();
- int size = ReadWriteIOUtils.readInt(buffer);
- if (size > 0) {
- digest.statistics = new ByteBuffer[StatisticType.getTotalTypeNum()];
- ByteBuffer value;
- for (int i = 0; i < size; i++) {
- short n = ReadWriteIOUtils.readShort(buffer);
- value = ReadWriteIOUtils.readByteBufferWithSelfDescriptionLength(buffer);
- digest.statistics[n] = value;
- }
- } // else left digest.statistics as null
-
- return digest;
- }
-
- /**
- * get statistics of the current object.
- */
- public ByteBuffer[] getStatistics() {
- return statistics;
- }
-
- public enum StatisticType {
- MIN_VALUE, MAX_VALUE, FIRST_VALUE, LAST_VALUE, SUM_VALUE;
-
- public static int getTotalTypeNum() {
- return StatisticType.values().length;
- }
-
- }
-}
\ No newline at end of file
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsFileMetadataV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsFileMetadataV1.java
deleted file mode 100644
index 9184154..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/TsFileMetadataV1.java
+++ /dev/null
@@ -1,106 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <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.v1.file.metadata;
-
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.iotdb.tsfile.utils.BloomFilter;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-
-/**
- * TSFileMetaData collects all metadata info and saves in its data structure.
- */
-public class TsFileMetadataV1 {
-
- private Map<String, TsDeviceMetadataIndexV1> deviceIndexMap = new HashMap<>();
-
- // bloom filter
- private BloomFilter bloomFilter;
-
- public TsFileMetadataV1() {
- //do nothing
- }
-
- /**
- * deserialize data from the buffer.
- *
- * @param buffer -buffer use to deserialize
- * @return -a instance of TsFileMetaData
- */
- public static TsFileMetadataV1 deserializeFrom(ByteBuffer buffer) {
- TsFileMetadataV1 fileMetaData = new TsFileMetadataV1();
-
- int size = ReadWriteIOUtils.readInt(buffer);
- if (size > 0) {
- Map<String, TsDeviceMetadataIndexV1> deviceMap = new HashMap<>();
- String key;
- TsDeviceMetadataIndexV1 value;
- for (int i = 0; i < size; i++) {
- key = ReadWriteIOUtils.readString(buffer);
- value = TsDeviceMetadataIndexV1.deserializeFrom(buffer);
- deviceMap.put(key, value);
- }
- fileMetaData.deviceIndexMap = deviceMap;
- }
-
- size = ReadWriteIOUtils.readInt(buffer);
- if (size > 0) {
- for (int i = 0; i < size; i++) {
- ReadWriteIOUtils.readString(buffer);
- MeasurementSchema.deserializeFrom(buffer);
- }
- }
-
- if (ReadWriteIOUtils.readIsNull(buffer)) {
- ReadWriteIOUtils.readString(buffer); // createdBy String
- }
- ReadWriteIOUtils.readInt(buffer); // totalChunkNum
- ReadWriteIOUtils.readInt(buffer); // invalidChunkNum
- // read bloom filter
- if (buffer.hasRemaining()) {
- byte[] bytes = ReadWriteIOUtils.readByteBufferWithSelfDescriptionLength(buffer).array();
- int filterSize = ReadWriteIOUtils.readInt(buffer);
- int hashFunctionSize = ReadWriteIOUtils.readInt(buffer);
- fileMetaData.bloomFilter = BloomFilter.buildBloomFilter(bytes, filterSize, hashFunctionSize);
- }
-
- return fileMetaData;
- }
-
- public BloomFilter getBloomFilter() {
- return bloomFilter;
- }
-
- public Map<String, TsDeviceMetadataIndexV1> getDeviceMap() {
- return deviceIndexMap;
- }
-
- public boolean containsDevice(String deltaObjUid) {
- return this.deviceIndexMap.containsKey(deltaObjUid);
- }
-
- public TsDeviceMetadataIndexV1 getDeviceMetadataIndex(String deviceUid) {
- return this.deviceIndexMap.get(deviceUid);
- }
-
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/BinaryStatisticsV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/BinaryStatisticsV1.java
deleted file mode 100644
index 3c68be1..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/BinaryStatisticsV1.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.tsfile.v1.file.metadata.statistics;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import org.apache.iotdb.tsfile.utils.Binary;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-/**
- * Statistics for string type.
- */
-public class BinaryStatisticsV1 extends StatisticsV1<Binary> {
-
- private Binary min = new Binary("");
- private Binary max = new Binary("");
- private Binary first = new Binary("");
- private Binary last = new Binary("");
- private double sum;
-
- @Override
- public Binary getMin() {
- return min;
- }
-
- @Override
- public Binary getMax() {
- return max;
- }
-
- @Override
- public Binary getFirst() {
- return first;
- }
-
- @Override
- public Binary getLast() {
- return last;
- }
-
- @Override
- public double getSum() {
- return sum;
- }
-
- @Override
- void deserialize(ByteBuffer byteBuffer) throws IOException {
- this.min = new Binary(
- ReadWriteIOUtils.readByteBufferWithSelfDescriptionLength(byteBuffer).array());
- this.max = new Binary(
- ReadWriteIOUtils.readByteBufferWithSelfDescriptionLength(byteBuffer).array());
- this.first = new Binary(
- ReadWriteIOUtils.readByteBufferWithSelfDescriptionLength(byteBuffer).array());
- this.last = new Binary(
- ReadWriteIOUtils.readByteBufferWithSelfDescriptionLength(byteBuffer).array());
- this.sum = ReadWriteIOUtils.readDouble(byteBuffer);
- }
-
- @Override
- void deserialize(InputStream inputStream) throws IOException {
- this.min = new Binary(ReadWriteIOUtils.readBytesWithSelfDescriptionLength(inputStream));
- this.max = new Binary(ReadWriteIOUtils.readBytesWithSelfDescriptionLength(inputStream));
- this.first = new Binary(ReadWriteIOUtils.readBytesWithSelfDescriptionLength(inputStream));
- this.last = new Binary(ReadWriteIOUtils.readBytesWithSelfDescriptionLength(inputStream));
- this.sum = ReadWriteIOUtils.readDouble(inputStream);
- }
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/BooleanStatisticsV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/BooleanStatisticsV1.java
deleted file mode 100644
index 24e2152..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/BooleanStatisticsV1.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.tsfile.v1.file.metadata.statistics;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-/**
- * Boolean Statistics.
- */
-public class BooleanStatisticsV1 extends StatisticsV1<Boolean> {
-
- private boolean min;
- private boolean max;
- private boolean first;
- private boolean last;
- private double sum;
-
- @Override
- public Boolean getMin() {
- return min;
- }
-
- @Override
- public Boolean getMax() {
- return max;
- }
-
- @Override
- public Boolean getFirst() {
- return first;
- }
-
- @Override
- public Boolean getLast() {
- return last;
- }
-
- @Override
- public double getSum() {
- return sum;
- }
-
- @Override
- void deserialize(ByteBuffer byteBuffer) throws IOException {
- this.min = ReadWriteIOUtils.readBool(byteBuffer);
- this.max = ReadWriteIOUtils.readBool(byteBuffer);
- this.first = ReadWriteIOUtils.readBool(byteBuffer);
- this.last = ReadWriteIOUtils.readBool(byteBuffer);
- this.sum = ReadWriteIOUtils.readDouble(byteBuffer);
- }
-
- @Override
- void deserialize(InputStream inputStream) throws IOException {
- this.min = ReadWriteIOUtils.readBool(inputStream);
- this.max = ReadWriteIOUtils.readBool(inputStream);
- this.first = ReadWriteIOUtils.readBool(inputStream);
- this.last = ReadWriteIOUtils.readBool(inputStream);
- this.sum = ReadWriteIOUtils.readDouble(inputStream);
- }
-
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/DoubleStatisticsV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/DoubleStatisticsV1.java
deleted file mode 100644
index e784b91..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/DoubleStatisticsV1.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.tsfile.v1.file.metadata.statistics;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-/**
- * Statistics for double type.
- */
-public class DoubleStatisticsV1 extends StatisticsV1<Double> {
-
- private double min;
- private double max;
- private double first;
- private double last;
- private double sum;
-
- @Override
- public Double getMin() {
- return min;
- }
-
- @Override
- public Double getMax() {
- return max;
- }
-
- @Override
- public Double getFirst() {
- return first;
- }
-
- @Override
- public Double getLast() {
- return last;
- }
-
- @Override
- public double getSum() {
- return sum;
- }
-
- @Override
- void deserialize(ByteBuffer byteBuffer) throws IOException {
- this.min = ReadWriteIOUtils.readDouble(byteBuffer);
- this.max = ReadWriteIOUtils.readDouble(byteBuffer);
- this.first = ReadWriteIOUtils.readDouble(byteBuffer);
- this.last = ReadWriteIOUtils.readDouble(byteBuffer);
- this.sum = ReadWriteIOUtils.readDouble(byteBuffer);
- }
-
- @Override
- void deserialize(InputStream inputStream) throws IOException {
- this.min = ReadWriteIOUtils.readDouble(inputStream);
- this.max = ReadWriteIOUtils.readDouble(inputStream);
- this.first = ReadWriteIOUtils.readDouble(inputStream);
- this.last = ReadWriteIOUtils.readDouble(inputStream);
- this.sum = ReadWriteIOUtils.readDouble(inputStream);
- }
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/FloatStatisticsV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/FloatStatisticsV1.java
deleted file mode 100644
index 6a8d8e8..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/FloatStatisticsV1.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.tsfile.v1.file.metadata.statistics;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-/**
- * Statistics for float type.
- */
-public class FloatStatisticsV1 extends StatisticsV1<Float> {
-
- private float min;
- private float max;
- private float first;
- private double sum;
- private float last;
-
- @Override
- public Float getMin() {
- return min;
- }
-
- @Override
- public Float getMax() {
- return max;
- }
-
- @Override
- public Float getFirst() {
- return first;
- }
-
- @Override
- public Float getLast() {
- return last;
- }
-
- @Override
- public double getSum() {
- return sum;
- }
-
- @Override
- void deserialize(ByteBuffer byteBuffer) throws IOException {
- this.min = ReadWriteIOUtils.readFloat(byteBuffer);
- this.max = ReadWriteIOUtils.readFloat(byteBuffer);
- this.first = ReadWriteIOUtils.readFloat(byteBuffer);
- this.last = ReadWriteIOUtils.readFloat(byteBuffer);
- this.sum = ReadWriteIOUtils.readDouble(byteBuffer);
- }
-
- @Override
- void deserialize(InputStream inputStream) throws IOException {
- this.min = ReadWriteIOUtils.readFloat(inputStream);
- this.max = ReadWriteIOUtils.readFloat(inputStream);
- this.first = ReadWriteIOUtils.readFloat(inputStream);
- this.last = ReadWriteIOUtils.readFloat(inputStream);
- this.sum = ReadWriteIOUtils.readDouble(inputStream);
- }
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/IntegerStatisticsV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/IntegerStatisticsV1.java
deleted file mode 100644
index 2511abd..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/IntegerStatisticsV1.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.tsfile.v1.file.metadata.statistics;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-/**
- * Statistics for int type.
- */
-public class IntegerStatisticsV1 extends StatisticsV1<Integer> {
-
- private int min;
- private int max;
- private int first;
- private int last;
- private double sum;
-
- @Override
- public Integer getMin() {
- return min;
- }
-
- @Override
- public Integer getMax() {
- return max;
- }
-
- @Override
- public Integer getFirst() {
- return first;
- }
-
- @Override
- public Integer getLast() {
- return last;
- }
-
- @Override
- public double getSum() {
- return sum;
- }
-
- @Override
- void deserialize(ByteBuffer byteBuffer) throws IOException {
- this.min = ReadWriteIOUtils.readInt(byteBuffer);
- this.max = ReadWriteIOUtils.readInt(byteBuffer);
- this.first = ReadWriteIOUtils.readInt(byteBuffer);
- this.last = ReadWriteIOUtils.readInt(byteBuffer);
- this.sum = ReadWriteIOUtils.readDouble(byteBuffer);
- }
-
- @Override
- void deserialize(InputStream inputStream) throws IOException {
- this.min = ReadWriteIOUtils.readInt(inputStream);
- this.max = ReadWriteIOUtils.readInt(inputStream);
- this.first = ReadWriteIOUtils.readInt(inputStream);
- this.last = ReadWriteIOUtils.readInt(inputStream);
- this.sum = ReadWriteIOUtils.readDouble(inputStream);
- }
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/LongStatisticsV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/LongStatisticsV1.java
deleted file mode 100644
index 29beeeb..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/LongStatisticsV1.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.tsfile.v1.file.metadata.statistics;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-
-/**
- * Statistics for long type.
- */
-public class LongStatisticsV1 extends StatisticsV1<Long> {
-
- private long min;
- private long max;
- private long first;
- private long last;
- private double sum;
-
- @Override
- public Long getMin() {
- return min;
- }
-
- @Override
- public Long getMax() {
- return max;
- }
-
- @Override
- public Long getFirst() {
- return first;
- }
-
- @Override
- public Long getLast() {
- return last;
- }
-
- @Override
- public double getSum() {
- return sum;
- }
-
- @Override
- void deserialize(ByteBuffer byteBuffer) throws IOException {
- this.min = ReadWriteIOUtils.readLong(byteBuffer);
- this.max = ReadWriteIOUtils.readLong(byteBuffer);
- this.first = ReadWriteIOUtils.readLong(byteBuffer);
- this.last = ReadWriteIOUtils.readLong(byteBuffer);
- this.sum = ReadWriteIOUtils.readDouble(byteBuffer);
- }
-
- @Override
- void deserialize(InputStream inputStream) throws IOException {
- this.min = ReadWriteIOUtils.readLong(inputStream);
- this.max = ReadWriteIOUtils.readLong(inputStream);
- this.first = ReadWriteIOUtils.readLong(inputStream);
- this.last = ReadWriteIOUtils.readLong(inputStream);
- this.sum = ReadWriteIOUtils.readDouble(inputStream);
- }
-
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/StatisticsV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/StatisticsV1.java
deleted file mode 100644
index ea91dd3..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/metadata/statistics/StatisticsV1.java
+++ /dev/null
@@ -1,225 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.tsfile.v1.file.metadata.statistics;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-
-import org.apache.iotdb.tsfile.exception.write.UnknownColumnTypeException;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.statistics.BinaryStatistics;
-import org.apache.iotdb.tsfile.file.metadata.statistics.BooleanStatistics;
-import org.apache.iotdb.tsfile.file.metadata.statistics.DoubleStatistics;
-import org.apache.iotdb.tsfile.file.metadata.statistics.FloatStatistics;
-import org.apache.iotdb.tsfile.file.metadata.statistics.IntegerStatistics;
-import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics;
-import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
-import org.apache.iotdb.tsfile.utils.Binary;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import org.apache.iotdb.tsfile.v1.file.metadata.ChunkMetadataV1;
-import org.apache.iotdb.tsfile.v1.file.metadata.TsDigestV1;
-
-/**
- * This class is used for recording statistic information of each measurement in a delta file. While
- * writing processing, the processor records the digest information. Statistics includes maximum,
- * minimum and null value count up to version 0.0.1.<br> Each data type extends this Statistic as
- * super class.<br>
- *
- * @param <T> data type for Statistics
- */
-public abstract class StatisticsV1<T> {
-
- /**
- * static method providing statistic instance for respective data type.
- *
- * @param type - data type
- * @return Statistics
- */
- public static StatisticsV1 getStatsByType(TSDataType type) {
- switch (type) {
- case INT32:
- return new IntegerStatisticsV1();
- case INT64:
- return new LongStatisticsV1();
- case TEXT:
- return new BinaryStatisticsV1();
- case BOOLEAN:
- return new BooleanStatisticsV1();
- case DOUBLE:
- return new DoubleStatisticsV1();
- case FLOAT:
- return new FloatStatisticsV1();
- default:
- throw new UnknownColumnTypeException(type.toString());
- }
- }
-
- public static StatisticsV1 deserialize(InputStream inputStream, TSDataType dataType)
- throws IOException {
- StatisticsV1<?> statistics = getStatsByType(dataType);
- statistics.deserialize(inputStream);
- return statistics;
- }
-
- public static StatisticsV1 deserialize(ByteBuffer buffer, TSDataType dataType) throws IOException {
- StatisticsV1<?> statistics = getStatsByType(dataType);
- statistics.deserialize(buffer);
- return statistics;
- }
-
- /**
- * For upgrading 0.9.x/v1 -> 0.10/v2
- */
- public static Statistics upgradeOldStatistics(StatisticsV1<?> oldstatistics,
- TSDataType dataType, int numOfValues, long maxTimestamp, long minTimestamp) {
- Statistics<?> statistics = Statistics.getStatsByType(dataType);
- statistics.setStartTime(minTimestamp);
- statistics.setEndTime(maxTimestamp);
- statistics.setCount(numOfValues);
- statistics.setEmpty(false);
- switch (dataType) {
- case INT32:
- ((IntegerStatistics) statistics)
- .initializeStats(((IntegerStatisticsV1) oldstatistics).getMin(),
- ((IntegerStatisticsV1) oldstatistics).getMax(),
- ((IntegerStatisticsV1) oldstatistics).getFirst(),
- ((IntegerStatisticsV1) oldstatistics).getLast(),
- ((IntegerStatisticsV1) oldstatistics).getSum());
- break;
- case INT64:
- ((LongStatistics) statistics)
- .initializeStats(((LongStatisticsV1) oldstatistics).getMin(),
- ((LongStatisticsV1) oldstatistics).getMax(),
- ((LongStatisticsV1) oldstatistics).getFirst(),
- ((LongStatisticsV1) oldstatistics).getLast(),
- ((LongStatisticsV1) oldstatistics).getSum());
- break;
- case TEXT:
- ((BinaryStatistics) statistics)
- .initializeStats(((BinaryStatisticsV1) oldstatistics).getFirst(),
- ((BinaryStatisticsV1) oldstatistics).getLast());
- break;
- case BOOLEAN:
- ((BooleanStatistics) statistics)
- .initializeStats(((BooleanStatisticsV1) oldstatistics).getFirst(),
- ((BooleanStatisticsV1) oldstatistics).getLast());
- break;
- case DOUBLE:
- ((DoubleStatistics) statistics)
- .initializeStats(((DoubleStatisticsV1) oldstatistics).getMin(),
- ((DoubleStatisticsV1) oldstatistics).getMax(),
- ((DoubleStatisticsV1) oldstatistics).getFirst(),
- ((DoubleStatisticsV1) oldstatistics).getLast(),
- ((DoubleStatisticsV1) oldstatistics).getSum());
- break;
- case FLOAT:
- ((FloatStatistics) statistics)
- .initializeStats(((FloatStatisticsV1) oldstatistics).getMin(),
- ((FloatStatisticsV1) oldstatistics).getMax(),
- ((FloatStatisticsV1) oldstatistics).getFirst(),
- ((FloatStatisticsV1) oldstatistics).getLast(),
- ((FloatStatisticsV1) oldstatistics).getSum());
- break;
- default:
- throw new UnknownColumnTypeException(statistics.getType()
- .toString());
- }
- return statistics;
- }
-
- /**
- * For upgrading 0.9.x/v1 -> 0.10.x/v2
- */
- public static Statistics constructStatisticsFromOldChunkMetadata(ChunkMetadataV1 oldChunkMetadata) {
- Statistics<?> statistics = Statistics.getStatsByType(oldChunkMetadata.getTsDataType());
- statistics.setStartTime(oldChunkMetadata.getStartTime());
- statistics.setEndTime(oldChunkMetadata.getEndTime());
- statistics.setCount(oldChunkMetadata.getNumOfPoints());
- statistics.setEmpty(false);
- TsDigestV1 tsDigest = oldChunkMetadata.getDigest();
- ByteBuffer[] buffers = tsDigest.getStatistics();
- switch (statistics.getType()) {
- case INT32:
- ((IntegerStatistics) statistics)
- .initializeStats(ReadWriteIOUtils.readInt(buffers[0]),
- ReadWriteIOUtils.readInt(buffers[1]),
- ReadWriteIOUtils.readInt(buffers[2]),
- ReadWriteIOUtils.readInt(buffers[3]),
- ReadWriteIOUtils.readDouble(buffers[4]));
- break;
- case INT64:
- ((LongStatistics) statistics)
- .initializeStats(ReadWriteIOUtils.readLong(buffers[0]),
- ReadWriteIOUtils.readLong(buffers[1]),
- ReadWriteIOUtils.readLong(buffers[2]),
- ReadWriteIOUtils.readLong(buffers[3]),
- ReadWriteIOUtils.readDouble(buffers[4]));
- break;
- case TEXT:
- ((BinaryStatistics) statistics)
- .initializeStats(new Binary(buffers[2].array()),
- new Binary(buffers[3].array()));
- break;
- case BOOLEAN:
- ((BooleanStatistics) statistics)
- .initializeStats(ReadWriteIOUtils.readBool(buffers[2]),
- ReadWriteIOUtils.readBool(buffers[3]));
- break;
- case DOUBLE:
- ((DoubleStatistics) statistics)
- .initializeStats(ReadWriteIOUtils.readDouble(buffers[0]),
- ReadWriteIOUtils.readDouble(buffers[1]),
- ReadWriteIOUtils.readDouble(buffers[2]),
- ReadWriteIOUtils.readDouble(buffers[3]),
- ReadWriteIOUtils.readDouble(buffers[4]));
- break;
- case FLOAT:
- ((FloatStatistics) statistics)
- .initializeStats(ReadWriteIOUtils.readFloat(buffers[0]),
- ReadWriteIOUtils.readFloat(buffers[1]),
- ReadWriteIOUtils.readFloat(buffers[2]),
- ReadWriteIOUtils.readFloat(buffers[3]),
- ReadWriteIOUtils.readDouble(buffers[4]));
- break;
- default:
- throw new UnknownColumnTypeException(statistics.getType()
- .toString());
- }
- return statistics;
- }
-
- public abstract T getMin();
-
- public abstract T getMax();
-
- public abstract T getFirst();
-
- public abstract T getLast();
-
- public abstract double getSum();
-
- /**
- * read data from the inputStream.
- */
- abstract void deserialize(InputStream inputStream) throws IOException;
-
- abstract void deserialize(ByteBuffer byteBuffer) throws IOException;
-
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/utils/HeaderUtils.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/utils/HeaderUtils.java
deleted file mode 100644
index 403db9a..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/file/utils/HeaderUtils.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.tsfile.v1.file.utils;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.ByteBuffer;
-
-import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
-import org.apache.iotdb.tsfile.file.MetaMarker;
-import org.apache.iotdb.tsfile.file.header.ChunkHeader;
-import org.apache.iotdb.tsfile.file.header.PageHeader;
-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.file.metadata.statistics.Statistics;
-import org.apache.iotdb.tsfile.read.reader.TsFileInput;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import org.apache.iotdb.tsfile.v1.file.metadata.statistics.StatisticsV1;
-
-public class HeaderUtils {
-
- private HeaderUtils() {
- }
-
- public static PageHeader deserializePageHeaderV1(InputStream inputStream, TSDataType dataType)
- throws IOException {
- int uncompressedSize = ReadWriteIOUtils.readInt(inputStream);
- int compressedSize = ReadWriteIOUtils.readInt(inputStream);
- int numOfValues = ReadWriteIOUtils.readInt(inputStream);
- long maxTimestamp = ReadWriteIOUtils.readLong(inputStream);
- long minTimestamp = ReadWriteIOUtils.readLong(inputStream);
- StatisticsV1<?> oldstatistics = StatisticsV1.deserialize(inputStream, dataType);
- Statistics<?> statistics = StatisticsV1.upgradeOldStatistics(oldstatistics, dataType,
- numOfValues, maxTimestamp, minTimestamp);
- return new PageHeader(uncompressedSize, compressedSize, statistics);
- }
-
- public static PageHeader deserializePageHeaderV1(ByteBuffer buffer, TSDataType dataType)
- throws IOException {
- int uncompressedSize = ReadWriteIOUtils.readInt(buffer);
- int compressedSize = ReadWriteIOUtils.readInt(buffer);
- int numOfValues = ReadWriteIOUtils.readInt(buffer);
- long maxTimestamp = ReadWriteIOUtils.readLong(buffer);
- long minTimestamp = ReadWriteIOUtils.readLong(buffer);
- StatisticsV1<?> oldstatistics = StatisticsV1.deserialize(buffer, dataType);
- Statistics<?> statistics = StatisticsV1.upgradeOldStatistics(oldstatistics, dataType,
- numOfValues, maxTimestamp, minTimestamp);
- return new PageHeader(uncompressedSize, compressedSize, statistics);
- }
-
- /**
- * deserialize from inputStream.
- *
- * @param markerRead Whether the marker of the CHUNK_HEADER has been read
- */
- public static ChunkHeader deserializeChunkHeaderV1(InputStream inputStream, boolean markerRead)
- throws IOException {
- if (!markerRead) {
- byte marker = (byte) inputStream.read();
- if (marker != MetaMarker.CHUNK_HEADER) {
- MetaMarker.handleUnexpectedMarker(marker);
- }
- }
-
- String measurementID = ReadWriteIOUtils.readString(inputStream);
- int dataSize = ReadWriteIOUtils.readInt(inputStream);
- TSDataType dataType = TSDataType.deserialize(ReadWriteIOUtils.readShort(inputStream));
- int numOfPages = ReadWriteIOUtils.readInt(inputStream);
- CompressionType type = ReadWriteIOUtils.readCompressionType(inputStream);
- TSEncoding encoding = ReadWriteIOUtils.readEncoding(inputStream);
- // read maxTombstoneTime from old TsFile, has been removed in newer versions of TsFile
- ReadWriteIOUtils.readLong(inputStream);
- return new ChunkHeader(measurementID, dataSize, dataType, type, encoding,
- numOfPages);
- }
-
- /**
- * deserialize from TsFileInput.
- *
- * @param input TsFileInput
- * @param offset offset
- * @param chunkHeaderSize the size of chunk's header
- * @param markerRead read marker (boolean type)
- * @return CHUNK_HEADER object
- * @throws IOException IOException
- */
- public static ChunkHeader deserializeChunkHeaderV1(TsFileInput input, long offset,
- int chunkHeaderSize, boolean markerRead) throws IOException {
- long offsetVar = offset;
- if (!markerRead) {
- offsetVar++;
- }
-
- // read chunk header from input to buffer
- ByteBuffer buffer = ByteBuffer.allocate(chunkHeaderSize);
- input.read(buffer, offsetVar);
- buffer.flip();
-
- // read measurementID
- int size = buffer.getInt();
- String measurementID = ReadWriteIOUtils.readStringWithLength(buffer, size);
- int dataSize = ReadWriteIOUtils.readInt(buffer);
- TSDataType dataType = TSDataType.deserialize(ReadWriteIOUtils.readShort(buffer));
- int numOfPages = ReadWriteIOUtils.readInt(buffer);
- CompressionType type = ReadWriteIOUtils.readCompressionType(buffer);
- TSEncoding encoding = ReadWriteIOUtils.readEncoding(buffer);
- // read maxTombstoneTime from old TsFile, has been removed in newer versions of TsFile
- ReadWriteIOUtils.readLong(buffer);
- return new ChunkHeader(measurementID, dataSize, dataType, type, encoding, numOfPages);
- }
-
- public static int getSerializedSizeV1(String measurementID) {
- return Byte.BYTES // marker
- + Integer.BYTES // measurementID length
- + measurementID.getBytes(TSFileConfig.STRING_CHARSET).length // measurementID
- + Integer.BYTES // dataSize
- + TSDataType.getSerializedSize() // dataType
- + CompressionType.getSerializedSize() // compressionType
- + TSEncoding.getSerializedSize() // encodingType
- + Integer.BYTES // numOfPages
- + Long.BYTES; // maxTombstoneTime
- }
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/read/TsFileSequenceReaderForV1.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/read/TsFileSequenceReaderForV1.java
deleted file mode 100644
index 91990be..0000000
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/v1/read/TsFileSequenceReaderForV1.java
+++ /dev/null
@@ -1,409 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <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.v1.read;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
-import org.apache.iotdb.tsfile.file.header.ChunkHeader;
-import org.apache.iotdb.tsfile.file.header.PageHeader;
-import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
-import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
-import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
-import org.apache.iotdb.tsfile.read.common.Chunk;
-import org.apache.iotdb.tsfile.read.common.Path;
-import org.apache.iotdb.tsfile.read.reader.TsFileInput;
-import org.apache.iotdb.tsfile.utils.BloomFilter;
-import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
-import org.apache.iotdb.tsfile.v1.file.metadata.ChunkGroupMetaDataV1;
-import org.apache.iotdb.tsfile.v1.file.metadata.ChunkMetadataV1;
-import org.apache.iotdb.tsfile.v1.file.metadata.TsDeviceMetadataV1;
-import org.apache.iotdb.tsfile.v1.file.metadata.TsDeviceMetadataIndexV1;
-import org.apache.iotdb.tsfile.v1.file.metadata.TsFileMetadataV1;
-import org.apache.iotdb.tsfile.v1.file.metadata.TimeseriesMetadataForV1;
-import org.apache.iotdb.tsfile.v1.file.utils.HeaderUtils;
-
-public class TsFileSequenceReaderForV1 extends TsFileSequenceReader {
-
- private long fileMetadataPos;
- private int fileMetadataSize;
- private TsFileMetadataV1 oldTsFileMetaData;
- // device -> measurement -> TimeseriesMetadata
- private Map<String, Map<String, TimeseriesMetadata>> cachedDeviceMetadataFromOldFile = new ConcurrentHashMap<>();
- private static final ReadWriteLock cacheLock = new ReentrantReadWriteLock();
- private boolean cacheDeviceMetadata;
-
- /**
- * Create a file reader of the given file. The reader will read the tail of the file to get the
- * file metadata size.Then the reader will skip the first TSFileConfig.MAGIC_STRING.getBytes().length
- * + TSFileConfig.NUMBER_VERSION.getBytes().length bytes of the file for preparing reading real
- * data.
- *
- * @param file the data file
- * @throws IOException If some I/O error occurs
- */
- public TsFileSequenceReaderForV1(String file) throws IOException {
- super(file, true);
- }
-
- /**
- * construct function for TsFileSequenceReader.
- *
- * @param file -given file name
- * @param loadMetadataSize -whether load meta data size
- */
- public TsFileSequenceReaderForV1(String file, boolean loadMetadataSize) throws IOException {
- super(file, loadMetadataSize);
- }
-
- /**
- * Create a file reader of the given file. The reader will read the tail of the file to get the
- * file metadata size.Then the reader will skip the first TSFileConfig.MAGIC_STRING.getBytes().length
- * + TSFileConfig.NUMBER_VERSION.getBytes().length bytes of the file for preparing reading real
- * data.
- *
- * @param input given input
- */
- public TsFileSequenceReaderForV1(TsFileInput input) throws IOException {
- this(input, true);
- }
-
- /**
- * construct function for TsFileSequenceReader.
- *
- * @param input -given input
- * @param loadMetadataSize -load meta data size
- */
- public TsFileSequenceReaderForV1(TsFileInput input, boolean loadMetadataSize) throws IOException {
- super(input, loadMetadataSize);
- }
-
- /**
- * construct function for TsFileSequenceReader.
- *
- * @param input the input of a tsfile. The current position should be a markder and then a chunk
- * Header, rather than the magic number
- * @param fileMetadataPos the position of the file metadata in the TsFileInput from the beginning
- * of the input to the current position
- * @param fileMetadataSize the byte size of the file metadata in the input
- */
- public TsFileSequenceReaderForV1(TsFileInput input, long fileMetadataPos, int fileMetadataSize) {
- super(input, fileMetadataPos, fileMetadataSize);
- this.fileMetadataPos = fileMetadataPos;
- this.fileMetadataSize = fileMetadataSize;
- }
-
- @Override
- public void loadMetadataSize() throws IOException {
- ByteBuffer metadataSize = ByteBuffer.allocate(Integer.BYTES);
- if (readTailMagic().equals(TSFileConfig.MAGIC_STRING)) {
- tsFileInput.read(metadataSize,
- tsFileInput.size() - TSFileConfig.MAGIC_STRING.getBytes().length - Integer.BYTES);
- metadataSize.flip();
- // read file metadata size and position
- fileMetadataSize = ReadWriteIOUtils.readInt(metadataSize);
- fileMetadataPos = tsFileInput.size() - TSFileConfig.MAGIC_STRING.getBytes().length
- - Integer.BYTES - fileMetadataSize;
- }
- }
-
- public TsFileMetadataV1 readOldFileMetadata() throws IOException {
- if (oldTsFileMetaData == null) {
- oldTsFileMetaData = TsFileMetadataV1
- .deserializeFrom(readDataFromOldFile(fileMetadataPos, fileMetadataSize));
- }
- return oldTsFileMetaData;
- }
-
- /**
- * this function does not modify the position of the file reader.
- *
- * @throws IOException io error
- */
- @Override
- public BloomFilter readBloomFilter() throws IOException {
- readOldFileMetadata();
- return oldTsFileMetaData.getBloomFilter();
- }
-
- /**
- * this function reads measurements and TimeseriesMetaDatas in given device Thread Safe
- *
- * @param device name
- * @return the map measurementId -> TimeseriesMetaData in one device
- * @throws IOException io error
- */
- @Override
- public Map<String, TimeseriesMetadata> readDeviceMetadata(String device) throws IOException {
- if (!cacheDeviceMetadata) {
- return constructDeviceMetadataFromOldFile(device);
- }
-
- cacheLock.readLock().lock();
- try {
- if (cachedDeviceMetadataFromOldFile.containsKey(device)) {
- return cachedDeviceMetadataFromOldFile.get(device);
- }
- } finally {
- cacheLock.readLock().unlock();
- }
-
- cacheLock.writeLock().lock();
- try {
- if (cachedDeviceMetadataFromOldFile.containsKey(device)) {
- return cachedDeviceMetadataFromOldFile.get(device);
- }
- readOldFileMetadata();
- if (!oldTsFileMetaData.containsDevice(device)) {
- return new HashMap<>();
- }
- Map<String, TimeseriesMetadata> deviceMetadata = constructDeviceMetadataFromOldFile(device);
- cachedDeviceMetadataFromOldFile.put(device, deviceMetadata);
- return deviceMetadata;
- } finally {
- cacheLock.writeLock().unlock();
- }
- }
-
- /**
- * for old TsFile
- * this function does not modify the position of the file reader.
- */
- private Map<String, TimeseriesMetadata> constructDeviceMetadataFromOldFile(String device)
- throws IOException {
- Map<String, TimeseriesMetadata> newDeviceMetadata = new HashMap<>();
- readOldFileMetadata();
- TsDeviceMetadataIndexV1 index = oldTsFileMetaData.getDeviceMetadataIndex(device);
- // read TsDeviceMetadata from file
- TsDeviceMetadataV1 tsDeviceMetadata = readOldTsDeviceMetaData(index);
- if (tsDeviceMetadata == null) {
- return newDeviceMetadata;
- }
-
- Map<String, List<ChunkMetadata>> measurementChunkMetaMap = new HashMap<>();
- // get all ChunkMetaData of this path included in all ChunkGroups of this device
- for (ChunkGroupMetaDataV1 chunkGroupMetaData : tsDeviceMetadata.getChunkGroupMetaDataList()) {
- List<ChunkMetadataV1> chunkMetaDataListInOneChunkGroup = chunkGroupMetaData.getChunkMetaDataList();
- for (ChunkMetadataV1 oldChunkMetadata : chunkMetaDataListInOneChunkGroup) {
- oldChunkMetadata.setVersion(chunkGroupMetaData.getVersion());
- measurementChunkMetaMap.computeIfAbsent(oldChunkMetadata.getMeasurementUid(), key -> new ArrayList<>())
- .add(oldChunkMetadata.upgradeToChunkMetadata());
- }
- }
- measurementChunkMetaMap.forEach((measurementId, chunkMetadataList) -> {
- if (!chunkMetadataList.isEmpty()) {
- TimeseriesMetadataForV1 timeseiresMetadata = new TimeseriesMetadataForV1();
- timeseiresMetadata.setMeasurementId(measurementId);
- timeseiresMetadata.setTSDataType(chunkMetadataList.get(0).getDataType());
- Statistics<?> statistics = Statistics.getStatsByType(chunkMetadataList.get(0).getDataType());
- for (ChunkMetadata chunkMetadata : chunkMetadataList) {
- statistics.mergeStatistics(chunkMetadata.getStatistics());
- }
- timeseiresMetadata.setStatistics(statistics);
- timeseiresMetadata.setChunkMetadataList(chunkMetadataList);
- newDeviceMetadata.put(measurementId, timeseiresMetadata);
- }
- });
- return newDeviceMetadata;
- }
-
- /**
- * for old TsFile
- * this function does not modify the position of the file reader.
- */
- private TsDeviceMetadataV1 readOldTsDeviceMetaData(TsDeviceMetadataIndexV1 index)
- throws IOException {
- if (index == null) {
- return null;
- }
- return TsDeviceMetadataV1.deserializeFrom(readDataFromOldFile(index.getOffset(), index.getLen()));
- }
-
- @Override
- public TimeseriesMetadata readTimeseriesMetadata(Path path) throws IOException {
- return getTimeseriesMetadataFromOldFile(path);
- }
-
- @Override
- public List<TimeseriesMetadata> readTimeseriesMetadata(String device, Set<String> measurements)
- throws IOException {
- return getTimeseriesMetadataFromOldFile(device, measurements);
- }
-
- /**
- * for 0.9.x/v1 TsFile
- */
- private TimeseriesMetadata getTimeseriesMetadataFromOldFile(Path path) throws IOException {
- Map<String, TimeseriesMetadata> deviceMetadata =
- constructDeviceMetadataFromOldFile(path.getDevice());
- return deviceMetadata.get(path.getMeasurement());
- }
-
- /**
- * for 0.9.x/v1 TsFile
- */
- private List<TimeseriesMetadata> getTimeseriesMetadataFromOldFile(String device, Set<String> measurements)
- throws IOException {
- Map<String, TimeseriesMetadata> deviceMetadata =
- constructDeviceMetadataFromOldFile(device);
- List<TimeseriesMetadata> resultTimeseriesMetadataList = new ArrayList<>();
- for (Entry<String, TimeseriesMetadata> entry : deviceMetadata.entrySet()) {
- if (measurements.contains(entry.getKey())) {
- resultTimeseriesMetadataList.add(entry.getValue());
- }
- }
- return resultTimeseriesMetadataList;
- }
-
- /**
- * read the chunk's header.
- *
- * @param position the file offset of this chunk's header
- * @param chunkHeaderSize the size of chunk's header
- * @param markerRead true if the offset does not contains the marker , otherwise false
- */
- private ChunkHeader readChunkHeaderFromOldFile(long position, int chunkHeaderSize, boolean markerRead)
- throws IOException {
- return HeaderUtils.deserializeChunkHeaderV1(tsFileInput, position, chunkHeaderSize, markerRead);
- }
-
- /**
- * notice, this function will modify channel's position.
- *
- * @param dataSize the size of chunkdata
- * @param position the offset of the chunk data
- * @return the pages of this chunk
- */
- private ByteBuffer readChunkFromOldFile(long position, int dataSize) throws IOException {
- return readDataFromOldFile(position, dataSize);
- }
-
- /**
- * read memory chunk.
- *
- * @param metaData -given chunk meta data
- * @return -chunk
- */
- @Override
- public Chunk readMemChunk(ChunkMetadata metaData) throws IOException {
- int chunkHeadSize = HeaderUtils.getSerializedSizeV1(metaData.getMeasurementUid());
- ChunkHeader header = readChunkHeaderFromOldFile(metaData.getOffsetOfChunkHeader(), chunkHeadSize, false);
- ByteBuffer buffer = readChunkFromOldFile(metaData.getOffsetOfChunkHeader() + chunkHeadSize,
- header.getDataSize());
- return new Chunk(header, buffer, metaData.getDeleteIntervalList());
- }
-
- /**
- * not thread safe.
- *
- * @param type given tsfile data type
- */
- @Override
- public PageHeader readPageHeader(TSDataType type) throws IOException {
- return HeaderUtils.deserializePageHeaderV1(tsFileInput.wrapAsInputStream(), type);
- }
-
- /**
- * read data from tsFileInput, from the current position (if position = -1), or the given
- * position. <br> if position = -1, the tsFileInput's position will be changed to the current
- * position + real data size that been read. Other wise, the tsFileInput's position is not
- * changed.
- *
- * @param position the start position of data in the tsFileInput, or the current position if
- * position = -1
- * @param size the size of data that want to read
- * @return data that been read.
- */
- private ByteBuffer readDataFromOldFile(long position, int size) throws IOException {
- ByteBuffer buffer = ByteBuffer.allocate(size);
- if (position < 0) {
- if (ReadWriteIOUtils.readAsPossible(tsFileInput, buffer) != size) {
- throw new IOException("reach the end of the data");
- }
- } else {
- if (ReadWriteIOUtils.readAsPossible(tsFileInput, buffer, position, size) != size) {
- throw new IOException("reach the end of the data");
- }
- }
- buffer.flip();
- return buffer;
- }
-
- /**
- * get ChunkMetaDatas of given path
- *
- * @param path timeseries path
- * @return List of ChunkMetaData
- */
- @Override
- public List<ChunkMetadata> getChunkMetadataList(Path path) throws IOException {
- return getChunkMetadataListFromOldFile(path);
- }
-
- /**
- * For old TsFile
- */
- private List<ChunkMetadata> getChunkMetadataListFromOldFile(Path path) throws IOException {
- readOldFileMetadata();
- if (!oldTsFileMetaData.containsDevice(path.getDevice())) {
- return new ArrayList<>();
- }
-
- // get the index information of TsDeviceMetadata
- TsDeviceMetadataIndexV1 index = oldTsFileMetaData.getDeviceMetadataIndex(path.getDevice());
-
- // read TsDeviceMetadata from file
- TsDeviceMetadataV1 tsDeviceMetadata = readOldTsDeviceMetaData(index);
- if (tsDeviceMetadata == null) {
- return new ArrayList<>();
- }
-
- // get all ChunkMetaData of this path included in all ChunkGroups of this device
- List<ChunkMetadataV1> oldChunkMetaDataList = new ArrayList<>();
- for (ChunkGroupMetaDataV1 chunkGroupMetaData : tsDeviceMetadata.getChunkGroupMetaDataList()) {
- List<ChunkMetadataV1> chunkMetaDataListInOneChunkGroup = chunkGroupMetaData
- .getChunkMetaDataList();
- for (ChunkMetadataV1 chunkMetaData : chunkMetaDataListInOneChunkGroup) {
- if (path.getMeasurement().equals(chunkMetaData.getMeasurementUid())) {
- chunkMetaData.setVersion(chunkGroupMetaData.getVersion());
- oldChunkMetaDataList.add(chunkMetaData);
- }
- }
- }
- oldChunkMetaDataList.sort(Comparator.comparingLong(ChunkMetadataV1::getStartTime));
- List<ChunkMetadata> chunkMetadataList = new ArrayList<>();
- for (ChunkMetadataV1 oldChunkMetaData : oldChunkMetaDataList) {
- chunkMetadataList.add(oldChunkMetaData.upgradeToChunkMetadata());
- }
- return chunkMetadataList;
- }
-}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkWriterImpl.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkWriterImpl.java
index 01a5f80..107f026 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkWriterImpl.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkWriterImpl.java
@@ -78,6 +78,12 @@ public class ChunkWriterImpl implements IChunkWriter {
private Statistics<?> statistics;
/**
+ * first page info
+ */
+ private int sizeWithoutStatistic;
+ private Statistics<?> firstPageStatistics;
+
+ /**
* @param schema schema of this measurement
*/
public ChunkWriterImpl(MeasurementSchema schema) {
@@ -95,8 +101,6 @@ public class ChunkWriterImpl implements IChunkWriter {
this.statistics = Statistics.getStatsByType(measurementSchema.getType());
this.pageWriter = new PageWriter(measurementSchema);
- this.pageWriter.setTimeEncoder(measurementSchema.getTimeEncoder());
- this.pageWriter.setValueEncoder(measurementSchema.getValueEncoder());
}
@Override
@@ -172,8 +176,8 @@ public class ChunkWriterImpl implements IChunkWriter {
}
/**
- * check occupied memory size, if it exceeds the PageSize threshold, construct a page and
- * put it to pageBuffer
+ * check occupied memory size, if it exceeds the PageSize threshold, construct a page and put it
+ * to pageBuffer
*/
private void checkPageSizeAndMayOpenANewPage() {
if (pageWriter.getPointNumber() == maxNumberOfPointsInPage) {
@@ -201,7 +205,19 @@ public class ChunkWriterImpl implements IChunkWriter {
private void writePageToPageBuffer() {
try {
- pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer);
+ int sizeWithoutStatistic = pageWriter
+ .writePageHeaderAndDataIntoBuff(pageBuffer, numOfPages == 0);
+ if (numOfPages == 0) { // record the firstPageStatistics
+ this.firstPageStatistics = pageWriter.getStatistics();
+ this.sizeWithoutStatistic = sizeWithoutStatistic;
+ } else if (numOfPages == 1) { // put the firstPageStatistics into pageBuffer
+ byte[] b = pageBuffer.toByteArray();
+ pageBuffer.reset();
+ pageBuffer.write(b, 0, sizeWithoutStatistic);
+ firstPageStatistics.serialize(pageBuffer);
+ pageBuffer.write(b, sizeWithoutStatistic, b.length - sizeWithoutStatistic);
+ firstPageStatistics = null;
+ }
// update statistics of this chunk
numOfPages++;
@@ -226,7 +242,9 @@ public class ChunkWriterImpl implements IChunkWriter {
@Override
public long estimateMaxSeriesMemSize() {
- return pageWriter.estimateMaxMemSize() + this.estimateMaxPageMemSize();
+ return pageBuffer.size() + pageWriter.estimateMaxMemSize() + PageHeader
+ .estimateMaxPageHeaderSizeWithoutStatistics() + pageWriter.getStatistics()
+ .getSerializedSize();
}
@Override
@@ -235,7 +253,8 @@ public class ChunkWriterImpl implements IChunkWriter {
return 0;
}
// return the serialized size of the chunk header + all pages
- return ChunkHeader.getSerializedSize(measurementSchema.getMeasurementId()) + (long) pageBuffer.size();
+ return ChunkHeader.getSerializedSize(measurementSchema.getMeasurementId(), pageBuffer.size())
+ + (long) pageBuffer.size();
}
@Override
@@ -267,10 +286,12 @@ public class ChunkWriterImpl implements IChunkWriter {
// write the page header to pageBuffer
try {
- logger.debug("start to flush a page header into buffer, buffer position {} ", pageBuffer.size());
+ logger.debug("start to flush a page header into buffer, buffer position {} ",
+ pageBuffer.size());
header.serializeTo(pageBuffer);
- logger.debug("finish to flush a page header {} of {} into buffer, buffer position {} ", header,
- measurementSchema.getMeasurementId(), pageBuffer.size());
+ logger
+ .debug("finish to flush a page header {} of {} into buffer, buffer position {} ", header,
+ measurementSchema.getMeasurementId(), pageBuffer.size());
statistics.mergeStatistics(header.getStatistics());
@@ -319,17 +340,5 @@ public class ChunkWriterImpl implements IChunkWriter {
writer.endCurrentChunk();
}
- /**
- * estimate max page memory size.
- *
- * @return the max possible allocated size currently
- */
- private long estimateMaxPageMemSize() {
- // return the sum of size of buffer and page max size
- return (long) (pageBuffer.size() +
- PageHeader.calculatePageHeaderSizeWithoutStatistics() +
- pageWriter.getStatistics().getSerializedSize());
- }
-
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/page/PageWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/page/PageWriter.java
index ba8001a..63bd3a2 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/page/PageWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/page/PageWriter.java
@@ -22,13 +22,8 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import org.apache.iotdb.tsfile.compress.ICompressor;
import org.apache.iotdb.tsfile.encoding.encoder.Encoder;
-import org.apache.iotdb.tsfile.file.header.PageHeader;
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.statistics.Statistics;
@@ -36,6 +31,8 @@ import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.PublicBAOS;
import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* This writer is used to write time-value into a page. It consists of a time
@@ -233,29 +230,36 @@ public class PageWriter {
/**
* write the page header and data into the PageWriter's output stream.
*/
- public void writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer) throws IOException {
+ public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first) throws IOException {
if (statistics.getCount() == 0) {
- return;
+ return 0;
}
ByteBuffer pageData = getUncompressedBytes();
int uncompressedSize = pageData.remaining();
int compressedSize;
- int compressedPosition = 0;
byte[] compressedBytes = null;
if (compressor.getType().equals(CompressionType.UNCOMPRESSED)) {
- compressedSize = pageData.remaining();
+ compressedSize = uncompressedSize;
} else {
+ // TODO maybe we can control this memory to avoid allocate new heap memory each time
compressedBytes = new byte[compressor.getMaxBytesForCompression(uncompressedSize)];
- compressedPosition = 0;
// data is never a directByteBuffer now, so we can use data.array()
compressedSize = compressor.compress(pageData.array(), pageData.position(), uncompressedSize, compressedBytes);
}
+
// write the page header to IOWriter
- PageHeader header = new PageHeader(uncompressedSize, compressedSize, statistics);
- header.serializeTo(pageBuffer);
+ int sizeWithoutStatistic = 0;
+ if (first) {
+ sizeWithoutStatistic += ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, pageBuffer);
+ sizeWithoutStatistic += ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, pageBuffer);
+ } else {
+ ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, pageBuffer);
+ ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, pageBuffer);
+ statistics.serialize(pageBuffer);
+ }
// write page content to temp PBAOS
logger.trace("start to flush a page data into buffer, buffer position {} ", pageBuffer.size());
@@ -264,9 +268,10 @@ public class PageWriter {
channel.write(pageData);
}
} else {
- pageBuffer.write(compressedBytes, compressedPosition, compressedSize);
+ pageBuffer.write(compressedBytes, 0, compressedSize);
}
logger.trace("start to flush a page data into buffer, buffer position {} ", pageBuffer.size());
+ return sizeWithoutStatistic;
}
/**
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/LocalTsFileOutput.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/LocalTsFileOutput.java
index 8423dfb..43ad692 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/LocalTsFileOutput.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/LocalTsFileOutput.java
@@ -52,6 +52,11 @@ public class LocalTsFileOutput implements TsFileOutput {
}
@Override
+ public void write(byte b) throws IOException {
+ bufferedStream.write(b);
+ }
+
+ @Override
public void write(ByteBuffer b) throws IOException {
bufferedStream.write(b.array());
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
index 7ad89b2..40a9161 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
@@ -30,7 +30,7 @@ import java.util.TreeMap;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.file.MetaMarker;
-import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
+import org.apache.iotdb.tsfile.file.footer.ChunkGroupHeader;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetadata;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
@@ -60,14 +60,14 @@ import org.slf4j.LoggerFactory;
public class TsFileIOWriter {
public static final byte[] magicStringBytes;
- public static final byte[] versionNumberBytes;
+ public static final byte versionNumberByte;
protected static final TSFileConfig config = TSFileDescriptor.getInstance().getConfig();
private static final Logger logger = LoggerFactory.getLogger(TsFileIOWriter.class);
private static final Logger resourceLogger = LoggerFactory.getLogger("FileMonitor");
static {
magicStringBytes = BytesUtils.stringToBytes(TSFileConfig.MAGIC_STRING);
- versionNumberBytes = TSFileConfig.VERSION_NUMBER.getBytes();
+ versionNumberByte = TSFileConfig.VERSION_NUMBER;
}
protected TsFileOutput out;
@@ -85,7 +85,6 @@ public class TsFileIOWriter {
private long markedPosition;
private String currentChunkGroupDeviceId;
- private long currentChunkGroupStartOffset;
protected List<Pair<Long, Long>> versionInfo = new ArrayList<>();
// for upgrade tool
@@ -136,16 +135,17 @@ public class TsFileIOWriter {
protected void startFile() throws IOException {
out.write(magicStringBytes);
- out.write(versionNumberBytes);
+ out.write(versionNumberByte);
}
public void startChunkGroup(String deviceId) throws IOException {
this.currentChunkGroupDeviceId = deviceId;
- currentChunkGroupStartOffset = out.getPosition();
if (logger.isDebugEnabled()) {
logger.debug("start chunk group:{}, file position {}", deviceId, out.getPosition());
}
chunkMetadataList = new ArrayList<>();
+ ChunkGroupHeader chunkGroupHeader = new ChunkGroupHeader(currentChunkGroupDeviceId);
+ chunkGroupHeader.serializeTo(out.wrapAsStream());
}
/**
@@ -155,12 +155,7 @@ public class TsFileIOWriter {
if (currentChunkGroupDeviceId == null || chunkMetadataList.isEmpty()) {
return;
}
- long dataSize = out.getPosition() - currentChunkGroupStartOffset;
- ChunkGroupFooter chunkGroupFooter = new ChunkGroupFooter(currentChunkGroupDeviceId, dataSize,
- chunkMetadataList.size());
- chunkGroupFooter.serializeTo(out.wrapAsStream());
- chunkGroupMetadataList
- .add(new ChunkGroupMetadata(currentChunkGroupDeviceId, chunkMetadataList));
+ chunkGroupMetadataList.add(new ChunkGroupMetadata(currentChunkGroupDeviceId, chunkMetadataList));
currentChunkGroupDeviceId = null;
chunkMetadataList = null;
out.flush();
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileOutput.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileOutput.java
index 77ce3e2..e8de8ca 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileOutput.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileOutput.java
@@ -34,6 +34,14 @@ public interface TsFileOutput {
void write(byte[] b) throws IOException;
/**
+ * Writes 1 byte to this output at the current position.
+ *
+ * @param b the data.
+ * @throws IOException if an I/O error occurs.
+ */
+ void write(byte b) throws IOException;
+
+ /**
* Writes <code>b.remaining()</code> bytes from the specified byte array to this
* output at the current position.
*
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java
index f8c80ec..97a7a33 100644
--- a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java
+++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/TsFileSequenceReaderTest.java
@@ -30,7 +30,7 @@ import java.util.Set;
import java.util.stream.Collectors;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.file.MetaMarker;
-import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
+import org.apache.iotdb.tsfile.file.footer.ChunkGroupHeader;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
import org.apache.iotdb.tsfile.file.header.PageHeader;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
@@ -64,7 +64,7 @@ public class TsFileSequenceReaderTest {
@Test
public void testReadTsFileSequently() throws IOException {
TsFileSequenceReader reader = new TsFileSequenceReader(FILE_PATH);
- reader.position(TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER
+ reader.position(TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER_V2
.getBytes().length);
Map<String, List<Pair<Long, Long>>> deviceChunkGroupMetadataOffsets = new HashMap<>();
@@ -73,14 +73,15 @@ public class TsFileSequenceReaderTest {
while ((marker = reader.readMarker()) != MetaMarker.SEPARATOR) {
switch (marker) {
case MetaMarker.CHUNK_HEADER:
- ChunkHeader header = reader.readChunkHeader();
+ case MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER:
+ ChunkHeader header = reader.readChunkHeader(marker);
for (int j = 0; j < header.getNumOfPages(); j++) {
PageHeader pageHeader = reader.readPageHeader(header.getDataType());
reader.readPage(pageHeader, header.getCompressionType());
}
break;
- case MetaMarker.CHUNK_GROUP_FOOTER:
- ChunkGroupFooter footer = reader.readChunkGroupFooter();
+ case MetaMarker.CHUNK_GROUP_HEADER:
+ ChunkGroupHeader footer = reader.readChunkGroupFooter();
long endOffset = reader.position();
Pair<Long, Long> pair = new Pair<>(startOffset, endOffset);
deviceChunkGroupMetadataOffsets.putIfAbsent(footer.getDeviceID(), new ArrayList<>());
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/TsFileIOWriterTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/TsFileIOWriterTest.java
index 61e4786..a5bb8e6 100644
--- a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/TsFileIOWriterTest.java
+++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/TsFileIOWriterTest.java
@@ -23,7 +23,7 @@ import java.io.IOException;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.constant.TestConstant;
import org.apache.iotdb.tsfile.file.MetaMarker;
-import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
+import org.apache.iotdb.tsfile.file.footer.ChunkGroupHeader;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
import org.apache.iotdb.tsfile.file.metadata.TimeSeriesMetadataTest;
import org.apache.iotdb.tsfile.file.metadata.TsFileMetadata;
@@ -84,20 +84,21 @@ public class TsFileIOWriterTest {
// magic_string
Assert.assertEquals(TSFileConfig.MAGIC_STRING, reader.readHeadMagic());
- Assert.assertEquals(TSFileConfig.VERSION_NUMBER, reader.readVersionNumber());
+ Assert.assertEquals(TSFileConfig.VERSION_NUMBER_V2, reader.readVersionNumber());
Assert.assertEquals(TSFileConfig.MAGIC_STRING, reader.readTailMagic());
+ // chunk group header
+ Assert.assertEquals(MetaMarker.CHUNK_GROUP_HEADER, reader.readMarker());
+ ChunkGroupHeader chunkGroupHeader = reader.readChunkGroupFooter();
+ Assert.assertEquals(deviceId, chunkGroupHeader.getDeviceID());
+
// chunk header
- reader.position(TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER
+ reader.position(TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER_V2
.getBytes().length);
- Assert.assertEquals(MetaMarker.CHUNK_HEADER, reader.readMarker());
- ChunkHeader header = reader.readChunkHeader();
+ Assert.assertEquals(MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER, reader.readMarker());
+ ChunkHeader header = reader.readChunkHeader(MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER);
Assert.assertEquals(TimeSeriesMetadataTest.measurementUID, header.getMeasurementID());
- // chunk group footer
- Assert.assertEquals(MetaMarker.CHUNK_GROUP_FOOTER, reader.readMarker());
- ChunkGroupFooter footer = reader.readChunkGroupFooter();
- Assert.assertEquals(deviceId, footer.getDeviceID());
// separator
Assert.assertEquals(MetaMarker.VERSION, reader.readMarker());
diff --git a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriterTest.java b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriterTest.java
index 87b88fc..68c46dc 100644
--- a/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriterTest.java
+++ b/tsfile/src/test/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriterTest.java
@@ -85,7 +85,7 @@ public class RestorableTsFileIOWriterTest {
RestorableTsFileIOWriter rWriter = new RestorableTsFileIOWriter(file);
writer = new TsFileWriter(rWriter);
writer.close();
- assertEquals(TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER
+ assertEquals(TSFileConfig.MAGIC_STRING.getBytes().length + TSFileConfig.VERSION_NUMBER_V2
.getBytes().length, rWriter.getTruncatedSize());
rWriter = new RestorableTsFileIOWriter(file);
@@ -105,7 +105,7 @@ public class RestorableTsFileIOWriterTest {
RestorableTsFileIOWriter rWriter = new RestorableTsFileIOWriter(file);
writer = new TsFileWriter(rWriter);
writer.close();
- assertEquals(TsFileIOWriter.magicStringBytes.length + TsFileIOWriter.versionNumberBytes.length,
+ assertEquals(TsFileIOWriter.magicStringBytes.length + 1,
rWriter.getTruncatedSize());
assertTrue(file.delete());
}
@@ -119,7 +119,7 @@ public class RestorableTsFileIOWriterTest {
RestorableTsFileIOWriter rWriter = new RestorableTsFileIOWriter(file);
TsFileWriter writer = new TsFileWriter(rWriter);
writer.close();
- assertEquals(TsFileIOWriter.magicStringBytes.length + TsFileIOWriter.versionNumberBytes.length,
+ assertEquals(TsFileIOWriter.magicStringBytes.length + 1,
rWriter.getTruncatedSize());
assertTrue(file.delete());
}
@@ -137,7 +137,7 @@ public class RestorableTsFileIOWriterTest {
RestorableTsFileIOWriter rWriter = new RestorableTsFileIOWriter(file);
writer = new TsFileWriter(rWriter);
writer.close();
- assertEquals(TsFileIOWriter.magicStringBytes.length + TsFileIOWriter.versionNumberBytes.length,
+ assertEquals(TsFileIOWriter.magicStringBytes.length + 1,
rWriter.getTruncatedSize());
assertTrue(file.delete());
}