This is an automated email from the ASF dual-hosted git repository. sunzesong pushed a commit to branch jira_520 in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit c4fc9c3ee437712e9b5ca4116dc7290d43de44e1 Author: samperson1997 <[email protected]> AuthorDate: Wed Feb 26 18:55:04 2020 +0800 [IOTDB-520] Result of IBatchReader should not cross partition --- .../SystemDesign/5-DataQuery/2-SeriesReader.md | 2 +- .../engine/storagegroup/StorageGroupProcessor.java | 4 ++-- .../reader/universal/PriorityMergeReader.java | 26 ++++++++++++++++++---- .../iotdb/db/integration/IoTDBSeriesReaderIT.java | 4 ++++ 4 files changed, 29 insertions(+), 7 deletions(-) diff --git a/docs/Documentation-CHN/SystemDesign/5-DataQuery/2-SeriesReader.md b/docs/Documentation-CHN/SystemDesign/5-DataQuery/2-SeriesReader.md index 32a9964..54b0f38 100644 --- a/docs/Documentation-CHN/SystemDesign/5-DataQuery/2-SeriesReader.md +++ b/docs/Documentation-CHN/SystemDesign/5-DataQuery/2-SeriesReader.md @@ -78,7 +78,7 @@ BatchData nextBatch() throws IOException; #### 一般使用流程 ``` -while (batchReader. hasNextBatch()) { +while (batchReader.hasNextBatch()) { BatchData batchData = batchReader.nextBatch(); // use batchData to do some work diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java index 547b9c8..c1f83c3 100755 --- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java @@ -143,8 +143,8 @@ public class StorageGroupProcessor { // includes sealed and unsealed sequence TsFiles private TreeSet<TsFileResource> sequenceFileTreeSet = new TreeSet<>( (o1, o2) -> { - int rangeCompare = o1.getFile().getParentFile().getName() - .compareTo(o2.getFile().getParentFile().getName()); + int rangeCompare = Long.compare(Long.parseLong(o1.getFile().getParentFile().getName()), + Long.parseLong(o2.getFile().getParentFile().getName())); return rangeCompare == 0 ? compareFileName(o1.getFile(), o2.getFile()) : rangeCompare; }); diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java index 6901250..33b9fc4 100644 --- a/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java +++ b/server/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReader.java @@ -18,12 +18,13 @@ */ package org.apache.iotdb.db.query.reader.universal; -import org.apache.iotdb.tsfile.read.reader.IPointReader; -import org.apache.iotdb.tsfile.read.TimeValuePair; - import java.io.IOException; import java.util.List; import java.util.PriorityQueue; +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.tsfile.read.TimeValuePair; +import org.apache.iotdb.tsfile.read.reader.IPointReader; /** * This class implements {@link IPointReader} for data sources with different priorities. @@ -33,6 +34,8 @@ public class PriorityMergeReader implements IPointReader { // largest end time of all added readers private long currentLargestEndTime; + private final static IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + PriorityQueue<Element> heap = new PriorityQueue<>((o1, o2) -> { int timeCompare = Long.compare(o1.timeValuePair.getTimestamp(), o2.timeValuePair.getTimestamp()); @@ -58,9 +61,24 @@ public class PriorityMergeReader implements IPointReader { } public void addReader(IPointReader reader, long priority, long endTime) throws IOException { + long partitionInterval = config.getPartitionInterval(); + switch (config.getTimestampPrecision()) { + case "ns": + partitionInterval *= 1000_000_000L; + break; + case "us": + partitionInterval *= 1000_000L; + break; + default: + partitionInterval *= 1000; + break; + } if (reader.hasNextTimeValuePair()) { heap.add(new Element(reader, reader.nextTimeValuePair(), priority)); - currentLargestEndTime = Math.max(currentLargestEndTime, endTime); + int partition = Math.round(reader.currentTimeValuePair().getTimestamp() / partitionInterval); + // set end time before current partition ends + currentLargestEndTime = Math.min((partition + 1) * partitionInterval - 1, + Math.max(currentLargestEndTime, endTime)); } else { reader.close(); } diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java index f722651..b63feb7 100644 --- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java +++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBSeriesReaderIT.java @@ -80,6 +80,9 @@ public class IoTDBSeriesReaderIT { tsFileConfig.setGroupSizeInByte(1024 * 1024 * 150); IoTDBDescriptor.getInstance().getConfig().setMemtableSizeThreshold(1024 * 16); + // test result of IBatchReader should not cross partition + IoTDBDescriptor.getInstance().getConfig().setPartitionInterval(6); + EnvironmentUtils.envSetUp(); insertData(); @@ -96,6 +99,7 @@ public class IoTDBSeriesReaderIT { tsFileConfig.setPageSizeInByte(pageSizeInByte); tsFileConfig.setGroupSizeInByte(groupSizeInByte); IoTDBDescriptor.getInstance().getConfig().setMemtableSizeThreshold(groupSizeInByte); + IoTDBDescriptor.getInstance().getConfig().setPartitionInterval(604800); EnvironmentUtils.cleanEnv(); }
