You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2019/11/26 05:07:33 UTC
[incubator-iotdb] branch improve_chunkWriter created (now ad72d5f)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a change to branch improve_chunkWriter
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at ad72d5f refine chunk writer
This branch includes the following new commits:
new ad72d5f refine chunk writer
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 01/01: refine chunk writer
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch improve_chunkWriter
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit ad72d5fe02413604512ae32ced69623aa2090e2f
Author: qiaojialin <64...@qq.com>
AuthorDate: Tue Nov 26 13:07:11 2019 +0800
refine chunk writer
---
.../iotdb/tsfile/write/chunk/ChunkWriterImpl.java | 45 +++++++---------------
.../apache/iotdb/tsfile/write/page/PageWriter.java | 9 ++---
2 files changed, 18 insertions(+), 36 deletions(-)
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 6a7e8e2..2beff8d 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
@@ -38,19 +38,12 @@ import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/**
- * @see IChunkWriter IChunkWriter
- */
public class ChunkWriterImpl implements IChunkWriter {
private static final Logger logger = LoggerFactory.getLogger(ChunkWriterImpl.class);
private MeasurementSchema measurementSchema;
- /**
- * help to encode data of this series.
- */
- //private final ChunkBuffer chunkBuffer;
private ICompressor compressor;
/**
@@ -63,8 +56,9 @@ public class ChunkWriterImpl implements IChunkWriter {
private long chunkMinTime = Long.MIN_VALUE;
private int numOfPages;
+
/**
- * value writer to encode data.
+ * write data into current page
*/
private PageWriter pageWriter;
@@ -75,23 +69,18 @@ public class ChunkWriterImpl implements IChunkWriter {
private final int maxNumberOfPointsInPage;
- // initial value for this.valueCountInOnePageForNextCheck
- private static final int MINIMUM_RECORD_COUNT_FOR_CHECK = 1500;
-
/**
- * value count in a page. It will be reset after calling {@code writePageHeaderAndDataIntoBuff()}
+ * value count in current page.
*/
private int valueCountInOnePageForNextCheck;
- /**
- * statistic on a stage. It will be reset after calling {@code writeAllPagesOfSeriesToTsFile()}
- */
- private Statistics<?> chunkStatistics;
+ // initial value for valueCountInOnePageForNextCheck
+ private static final int MINIMUM_RECORD_COUNT_FOR_CHECK = 1500;
/**
- * statistic on a page. It will be reset after calling {@code writePageHeaderAndDataIntoBuff()}
+ * statistic of this chunk.
*/
- //private Statistics<?> pageStatistics;
+ private Statistics<?> chunkStatistics;
/**
* @param schema schema of this measurement
@@ -107,7 +96,7 @@ public class ChunkWriterImpl implements IChunkWriter {
// initial check of memory usage. So that we have enough data to make an initial prediction
this.valueCountInOnePageForNextCheck = MINIMUM_RECORD_COUNT_FOR_CHECK;
- // init statistics for this series and page
+ // init statistics for this chunk and page
this.chunkStatistics = Statistics.getStatsByType(measurementSchema.getType());
this.pageWriter = new PageWriter(measurementSchema);
@@ -251,8 +240,11 @@ public class ChunkWriterImpl implements IChunkWriter {
public void writeToFileWriter(TsFileIOWriter tsfileWriter) throws IOException {
sealCurrentPage();
writeAllPagesOfChunkToTsFile(tsfileWriter, chunkStatistics);
- this.reset();
- // reset series_statistics
+
+ // reinit this chunk writer
+ pageBuffer.reset();
+ chunkPointCount = 0;
+ chunkMinTime = Long.MIN_VALUE;
this.chunkStatistics = Statistics.getStatsByType(measurementSchema.getType());
}
@@ -333,7 +325,7 @@ public class ChunkWriterImpl implements IChunkWriter {
* write the page to specified IOWriter.
*
* @param writer the specified IOWriter
- * @param statistics the statistic information provided by series writer
+ * @param statistics the chunk statistics
* @return the data size of this chunk
* @throws IOException exception in IO
*/
@@ -367,15 +359,6 @@ public class ChunkWriterImpl implements IChunkWriter {
}
/**
- * reset exist data in page for next stage.
- */
- public void reset() {
- chunkMinTime = Long.MIN_VALUE;
- pageBuffer.reset();
- chunkPointCount = 0;
- }
-
- /**
* estimate max page memory size.
*
* @return the max possible allocated size currently
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 da6a0dc..048baaf 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
@@ -25,7 +25,6 @@ import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import org.apache.iotdb.tsfile.compress.ICompressor;
import org.apache.iotdb.tsfile.encoding.encoder.Encoder;
-import org.apache.iotdb.tsfile.exception.write.PageException;
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;
@@ -38,8 +37,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This function is used to write time-value into a page. It consists of a time encoder, a value encoder and respective
- * OutputStream.
+ * This writer is used to write time-value into a page. It consists of a time encoder,
+ * a value encoder and respective OutputStream.
*/
public class PageWriter {
@@ -49,7 +48,7 @@ public class PageWriter {
private long pageMaxTime;
private long pageMinTime = Long.MIN_VALUE;
- ICompressor compressor;
+ private ICompressor compressor;
// time
private Encoder timeEncoder;
@@ -74,7 +73,7 @@ public class PageWriter {
this.compressor = ICompressor.getCompressor(measurementSchema.getCompressor());
}
- public PageWriter(Encoder timeEncoder, Encoder valueEncoder) {
+ private PageWriter(Encoder timeEncoder, Encoder valueEncoder) {
this.timeOut = new PublicBAOS();
this.valueOut = new PublicBAOS();
this.timeEncoder = timeEncoder;