You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2019/07/17 11:41:15 UTC
[incubator-iotdb] branch dev_merge updated (a301d7b -> 03483d7)
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a change to branch dev_merge
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
from a301d7b add documentation and code refinements
add 647c11a perfect documents * Fix words that are inconsistent with the document * Update some screenshots * Modify the layout of some content
add 9ed439b Synchronized Modification of Chinese Documents
add f16ea72 perfect the README.md document of JDBC project
add 67714b8 Update docs/Documentation-CHN/UserGuideV0.7.0/4-Deployment and Management/2-Configuration.md
add 33af5f1 Merge pull request #249 from lvpengze/perfect_documents
add f967491 Tried to solve issue 140
add 176288e Update 0-Content.md
add c9e2bfa Merge pull request #251 from samperson1997/update_content
add d728162 Fix errors in 1-Deployment.md
add 4e0f99a Merge pull request #252 from y-wan/patch-2
add d977e45 Fixed minor formatting issues in pom files.
add 923d8d9 Fixed a bug where CompressionRatioTest fail on a System with Locale different than English.
add 3c09832 Fixed a bug where all tests in IoTDBAggregationIT fail on a System with Locale different than English.
add d47c41b Fixed a bug where tests in IoTDBDeletionIT fail on a System with Locale different than English.
add a4593e3 Refactor the reader package for readability (#241)
add cd83557 Merge pull request #255 from LeiRui/cherrypick07171438
new 03483d7 Merge branch 'master' into dev_merge
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.
Summary of changes:
.../2-Data Type.md | 12 +-
.../3-Operation Manual/3-Data Import.md | 2 +-
.../3-Operation Manual/4-Data Query.md | 6 +-
.../4-Deployment and Management/1-Deployment.md | 4 +-
.../4-Deployment and Management/2-Configuration.md | 20 +-
.../3-System Monitor.md | 4 +-
docs/Documentation/UserGuideV0.7.0/0-Content.md | 7 +-
.../2-Data Type.md | 10 +-
.../3-Operation Manual/3-Data Import.md | 4 +-
.../3-Operation Manual/4-Data Query.md | 6 +-
.../4-Deployment and Management/2-Configuration.md | 18 +-
.../3-System Monitor.md | 6 +-
grafana/pom.xml | 2 +-
.../iotdb/db/conf/adapter/CompressionRatio.java | 6 +-
.../db/engine/merge/manage/MergeResource.java | 5 +-
.../db/query/aggregation/AggregateFunction.java | 4 +-
.../db/query/aggregation/impl/CountAggrFunc.java | 4 +-
.../db/query/aggregation/impl/FirstAggrFunc.java | 4 +-
.../db/query/aggregation/impl/LastAggrFunc.java | 4 +-
.../db/query/aggregation/impl/MaxTimeAggrFunc.java | 4 +-
.../query/aggregation/impl/MaxValueAggrFunc.java | 4 +-
.../db/query/aggregation/impl/MeanAggrFunc.java | 4 +-
.../db/query/aggregation/impl/MinTimeAggrFunc.java | 4 +-
.../query/aggregation/impl/MinValueAggrFunc.java | 4 +-
.../dataset/EngineDataSetWithValueFilter.java | 10 +-
.../groupby/GroupByWithValueFilterDataSet.java | 15 +-
.../groupby/GroupByWithoutValueFilterDataSet.java | 25 +-
.../db/query/executor/AggregateEngineExecutor.java | 40 +--
.../iotdb/db/query/executor/EngineExecutor.java | 27 +-
.../db/query/factory/ISeriesReaderFactory.java | 90 -------
.../db/query/factory/SeriesReaderFactoryImpl.java | 286 ---------------------
.../java/org/apache/iotdb/db/query/fill/IFill.java | 5 +-
.../iotdb/db/query/reader/IAggregateReader.java | 7 +-
.../apache/iotdb/db/query/reader/IBatchReader.java | 1 -
.../apache/iotdb/db/query/reader/IPointReader.java | 3 +-
...derByTimeStamp.java => IReaderByTimestamp.java} | 17 +-
.../reader/SeriesReaderWithoutValueFilter.java | 134 ----------
.../DiskChunkReader.java | 16 +-
.../DiskChunkReaderByTimestamp.java | 36 ++-
.../{mem => chunkRelated}/MemChunkReader.java | 19 +-
.../MemChunkReaderByTimestamp.java | 19 +-
.../FileSeriesReaderAdapter.java | 11 +-
.../FileSeriesReaderByTimestampAdapter.java | 15 +-
.../fileRelated/UnSealedTsFileIterateReader.java | 130 ++++++++++
.../UnSealedTsFileReaderByTimestamp.java | 96 +++++++
.../reader/merge/SeriesReaderByTimestamp.java | 72 ------
.../resourceRelated/SeqResourceIterateReader.java | 186 ++++++++++++++
.../SeqResourceReaderByTimestamp.java | 189 ++++++++++++++
.../resourceRelated/UnseqResourceMergeReader.java | 141 ++++++++++
.../UnseqResourceReaderByTimestamp.java | 97 +++++++
.../reader/sequence/SequenceSeriesReader.java | 155 -----------
.../sequence/SequenceSeriesReaderByTimestamp.java | 156 -----------
.../reader/sequence/UnSealedTsFileReader.java | 96 -------
.../sequence/UnSealedTsFileReaderByTimestamp.java | 93 -------
.../seriesRelated/SeriesReaderByTimestamp.java | 57 ++++
.../SeriesReaderWithValueFilter.java | 42 ++-
.../SeriesReaderWithoutValueFilter.java | 171 ++++++++++++
.../{sequence => universal}/IterateReader.java | 20 +-
.../PriorityMergeReader.java} | 26 +-
.../universal/PriorityMergeReaderByTimestamp.java | 75 ++++++
.../query/timegenerator/EngineNodeConstructor.java | 6 +-
.../db/conf/adapter/CompressionRatioTest.java | 14 +-
.../iotdb/db/engine/merge/MergeTaskTest.java | 14 +-
.../iotdb/db/integration/IoTDBAggregationIT.java | 14 +-
.../iotdb/db/integration/IoTDBDeletionIT.java | 10 +-
.../iotdb/db/query/reader/ReaderTestHelper.java | 1 +
.../UnSealedTsFileReaderTest.java} | 23 +-
.../SeqResourceReaderTest.java} | 46 ++--
.../resourceRelated/UnseqResourceReaderTest.java | 141 ++++++++++
.../{ => seriesRelated}/FakedIBatchPoint.java | 26 +-
.../{ => seriesRelated}/FakedIPointReader.java | 9 +-
.../SeriesReaderWithValueFilterTest.java | 61 +++++
.../SeriesReaderWithoutValueFilterTest.java | 14 +-
.../FakedSeriesReaderByTimestamp.java | 9 +-
.../PriorityMergeReaderByTimestampTest.java} | 30 +--
.../PriorityMergeReaderTest.java} | 15 +-
.../PriorityMergeReaderTest2.java} | 21 +-
.../UnseqSeriesReaderByTimestampTest.java | 97 -------
.../unsequence/UnsequenceSeriesReaderTest.java | 95 -------
.../writelog/recover/UnseqTsFileRecoverTest.java | 12 +-
jdbc/README.md | 157 +++++++++--
jdbc/pom.xml | 3 +-
pom.xml | 6 +-
service-rpc/pom.xml | 2 +-
.../iotdb/tsfile/file/header/ChunkHeader.java | 2 +-
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 4 +-
86 files changed, 1911 insertions(+), 1647 deletions(-)
delete mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/factory/ISeriesReaderFactory.java
delete mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/factory/SeriesReaderFactoryImpl.java
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{IReaderByTimeStamp.java => IReaderByTimestamp.java} (60%)
delete mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/SeriesReaderWithoutValueFilter.java
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{unsequence => chunkRelated}/DiskChunkReader.java (78%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{unsequence => chunkRelated}/DiskChunkReaderByTimestamp.java (67%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{mem => chunkRelated}/MemChunkReader.java (88%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{mem => chunkRelated}/MemChunkReaderByTimestamp.java (73%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{sequence/adapter => fileRelated}/FileSeriesReaderAdapter.java (76%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{sequence/adapter => fileRelated}/FileSeriesReaderByTimestampAdapter.java (72%)
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/fileRelated/UnSealedTsFileIterateReader.java
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/fileRelated/UnSealedTsFileReaderByTimestamp.java
delete mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/merge/SeriesReaderByTimestamp.java
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/SeqResourceIterateReader.java
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/SeqResourceReaderByTimestamp.java
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceMergeReader.java
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceReaderByTimestamp.java
delete mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/SequenceSeriesReader.java
delete mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/SequenceSeriesReaderByTimestamp.java
delete mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/UnSealedTsFileReader.java
delete mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/UnSealedTsFileReaderByTimestamp.java
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/SeriesReaderByTimestamp.java
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{ => seriesRelated}/SeriesReaderWithValueFilter.java (54%)
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/seriesRelated/SeriesReaderWithoutValueFilter.java
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{sequence => universal}/IterateReader.java (74%)
rename iotdb/src/main/java/org/apache/iotdb/db/query/reader/{unsequence/UnsequenceSeriesReader.java => universal/PriorityMergeReader.java} (79%)
create mode 100644 iotdb/src/main/java/org/apache/iotdb/db/query/reader/universal/PriorityMergeReaderByTimestamp.java
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{sequence/UnsealedSeqReaderTest.java => fileRelated/UnSealedTsFileReaderTest.java} (81%)
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{sequence/SeqDataReaderTest.java => resourceRelated/SeqResourceReaderTest.java} (68%)
create mode 100644 iotdb/src/test/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceReaderTest.java
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{ => seriesRelated}/FakedIBatchPoint.java (92%)
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{ => seriesRelated}/FakedIPointReader.java (90%)
create mode 100644 iotdb/src/test/java/org/apache/iotdb/db/query/reader/seriesRelated/SeriesReaderWithValueFilterTest.java
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{ => seriesRelated}/SeriesReaderWithoutValueFilterTest.java (90%)
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{ => universal}/FakedSeriesReaderByTimestamp.java (93%)
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{merge/SeriesReaderByTimestampTest.java => universal/PriorityMergeReaderByTimestampTest.java} (85%)
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{merge/SeriesMergeSortReaderTest.java => universal/PriorityMergeReaderTest.java} (86%)
rename iotdb/src/test/java/org/apache/iotdb/db/query/reader/{merge/UnsequenceSeriesReaderTest.java => universal/PriorityMergeReaderTest2.java} (83%)
delete mode 100644 iotdb/src/test/java/org/apache/iotdb/db/query/reader/unsequence/UnseqSeriesReaderByTimestampTest.java
delete mode 100644 iotdb/src/test/java/org/apache/iotdb/db/query/reader/unsequence/UnsequenceSeriesReaderTest.java
[incubator-iotdb] 01/01: Merge branch 'master' into dev_merge
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch dev_merge
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 03483d77ca10b9427f25cbead1c76b50426834dc
Merge: a301d7b cd83557
Author: 江天 <jt...@163.com>
AuthorDate: Wed Jul 17 19:38:49 2019 +0800
Merge branch 'master' into dev_merge
# Conflicts:
# iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutor.java
# iotdb/src/main/java/org/apache/iotdb/db/query/factory/SeriesReaderFactoryImpl.java
# iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/SequenceSeriesReader.java
# iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/SequenceSeriesReaderByTimestamp.java
# iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/UnSealedTsFileReader.java
# iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/UnSealedTsFileReaderByTimestamp.java
.../2-Data Type.md | 12 +-
.../3-Operation Manual/3-Data Import.md | 2 +-
.../3-Operation Manual/4-Data Query.md | 6 +-
.../4-Deployment and Management/1-Deployment.md | 4 +-
.../4-Deployment and Management/2-Configuration.md | 20 +-
.../3-System Monitor.md | 4 +-
docs/Documentation/UserGuideV0.7.0/0-Content.md | 7 +-
.../2-Data Type.md | 10 +-
.../3-Operation Manual/3-Data Import.md | 4 +-
.../3-Operation Manual/4-Data Query.md | 6 +-
.../4-Deployment and Management/2-Configuration.md | 18 +-
.../3-System Monitor.md | 6 +-
grafana/pom.xml | 2 +-
.../iotdb/db/conf/adapter/CompressionRatio.java | 6 +-
.../db/engine/merge/manage/MergeResource.java | 5 +-
.../db/query/aggregation/AggregateFunction.java | 4 +-
.../db/query/aggregation/impl/CountAggrFunc.java | 4 +-
.../db/query/aggregation/impl/FirstAggrFunc.java | 4 +-
.../db/query/aggregation/impl/LastAggrFunc.java | 4 +-
.../db/query/aggregation/impl/MaxTimeAggrFunc.java | 4 +-
.../query/aggregation/impl/MaxValueAggrFunc.java | 4 +-
.../db/query/aggregation/impl/MeanAggrFunc.java | 4 +-
.../db/query/aggregation/impl/MinTimeAggrFunc.java | 4 +-
.../query/aggregation/impl/MinValueAggrFunc.java | 4 +-
.../dataset/EngineDataSetWithValueFilter.java | 10 +-
.../groupby/GroupByWithValueFilterDataSet.java | 15 +-
.../groupby/GroupByWithoutValueFilterDataSet.java | 25 +-
.../db/query/executor/AggregateEngineExecutor.java | 40 +--
.../iotdb/db/query/executor/EngineExecutor.java | 27 +-
.../db/query/factory/ISeriesReaderFactory.java | 90 -------
.../db/query/factory/SeriesReaderFactoryImpl.java | 286 ---------------------
.../java/org/apache/iotdb/db/query/fill/IFill.java | 5 +-
.../iotdb/db/query/reader/IAggregateReader.java | 7 +-
.../apache/iotdb/db/query/reader/IBatchReader.java | 1 -
.../apache/iotdb/db/query/reader/IPointReader.java | 3 +-
...derByTimeStamp.java => IReaderByTimestamp.java} | 17 +-
.../reader/SeriesReaderWithoutValueFilter.java | 134 ----------
.../DiskChunkReader.java | 16 +-
.../DiskChunkReaderByTimestamp.java | 36 ++-
.../{mem => chunkRelated}/MemChunkReader.java | 19 +-
.../MemChunkReaderByTimestamp.java | 19 +-
.../FileSeriesReaderAdapter.java | 11 +-
.../FileSeriesReaderByTimestampAdapter.java | 15 +-
.../fileRelated/UnSealedTsFileIterateReader.java | 130 ++++++++++
.../UnSealedTsFileReaderByTimestamp.java | 96 +++++++
.../reader/merge/SeriesReaderByTimestamp.java | 72 ------
.../resourceRelated/SeqResourceIterateReader.java | 186 ++++++++++++++
.../SeqResourceReaderByTimestamp.java | 189 ++++++++++++++
.../resourceRelated/UnseqResourceMergeReader.java | 141 ++++++++++
.../UnseqResourceReaderByTimestamp.java | 97 +++++++
.../reader/sequence/SequenceSeriesReader.java | 155 -----------
.../sequence/SequenceSeriesReaderByTimestamp.java | 156 -----------
.../reader/sequence/UnSealedTsFileReader.java | 96 -------
.../sequence/UnSealedTsFileReaderByTimestamp.java | 93 -------
.../seriesRelated/SeriesReaderByTimestamp.java | 57 ++++
.../SeriesReaderWithValueFilter.java | 42 ++-
.../SeriesReaderWithoutValueFilter.java | 171 ++++++++++++
.../{sequence => universal}/IterateReader.java | 20 +-
.../PriorityMergeReader.java} | 26 +-
.../universal/PriorityMergeReaderByTimestamp.java | 75 ++++++
.../query/timegenerator/EngineNodeConstructor.java | 6 +-
.../db/conf/adapter/CompressionRatioTest.java | 14 +-
.../iotdb/db/engine/merge/MergeTaskTest.java | 14 +-
.../iotdb/db/integration/IoTDBAggregationIT.java | 14 +-
.../iotdb/db/integration/IoTDBDeletionIT.java | 10 +-
.../iotdb/db/query/reader/ReaderTestHelper.java | 1 +
.../UnSealedTsFileReaderTest.java} | 23 +-
.../SeqResourceReaderTest.java} | 46 ++--
.../resourceRelated/UnseqResourceReaderTest.java | 141 ++++++++++
.../{ => seriesRelated}/FakedIBatchPoint.java | 26 +-
.../{ => seriesRelated}/FakedIPointReader.java | 9 +-
.../SeriesReaderWithValueFilterTest.java | 61 +++++
.../SeriesReaderWithoutValueFilterTest.java | 14 +-
.../FakedSeriesReaderByTimestamp.java | 9 +-
.../PriorityMergeReaderByTimestampTest.java} | 30 +--
.../PriorityMergeReaderTest.java} | 15 +-
.../PriorityMergeReaderTest2.java} | 21 +-
.../UnseqSeriesReaderByTimestampTest.java | 97 -------
.../unsequence/UnsequenceSeriesReaderTest.java | 95 -------
.../writelog/recover/UnseqTsFileRecoverTest.java | 12 +-
jdbc/README.md | 157 +++++++++--
jdbc/pom.xml | 3 +-
pom.xml | 6 +-
service-rpc/pom.xml | 2 +-
.../iotdb/tsfile/file/header/ChunkHeader.java | 2 +-
.../iotdb/tsfile/utils/ReadWriteIOUtils.java | 4 +-
86 files changed, 1911 insertions(+), 1647 deletions(-)
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
index 7792828,0000000..4028102
mode 100644,000000..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
@@@ -1,188 -1,0 +1,187 @@@
+/**
+ * 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.db.engine.merge.manage;
+
+import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.query.context.QueryContext;
- import org.apache.iotdb.db.query.factory.SeriesReaderFactoryImpl;
+import org.apache.iotdb.db.query.reader.IPointReader;
++import org.apache.iotdb.db.query.reader.resourceRelated.UnseqResourceMergeReader;
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.read.controller.MetadataQuerier;
+import org.apache.iotdb.tsfile.read.controller.MetadataQuerierByFileImpl;
+import org.apache.iotdb.tsfile.write.chunk.ChunkBuffer;
+import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
+import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
+
+/**
+ * MergeResource manages files and caches of readers, writers, MeasurementSchemas and
+ * modifications to avoid unnecessary object creations and file openings.
+ */
+public class MergeResource {
+
+ private List<TsFileResource> seqFiles;
+ private List<TsFileResource> unseqFiles;
+
+ // future feature
+ // keeping ChunkMetadata in memory avoids reading them again when we need to move unmerged
+ // chunks to the merged file, but this may consume memory considerably
+ private boolean keepChunkMetadata = false;
+
+ private QueryContext mergeContext = new QueryContext();
+
+ private Map<TsFileResource, TsFileSequenceReader> fileReaderCache = new HashMap<>();
+ private Map<TsFileResource, RestorableTsFileIOWriter> fileWriterCache = new HashMap<>();
+ private Map<TsFileResource, List<Modification>> modificationCache = new HashMap<>();
+ private Map<TsFileResource, MetadataQuerier> metadataQuerierCache = new HashMap<>();
+ private Map<String, MeasurementSchema> measurementSchemaMap = new HashMap<>();
+ private Map<MeasurementSchema, IChunkWriter> chunkWriterCache = new HashMap<>();
+
+ public MergeResource(
+ List<TsFileResource> seqFiles,
+ List<TsFileResource> unseqFiles) {
+ this.seqFiles = seqFiles;
+ this.unseqFiles = unseqFiles;
+ }
+
+ public void clear() throws IOException {
+ for (TsFileSequenceReader sequenceReader : fileReaderCache.values()) {
+ sequenceReader.close();
+ }
+
+ fileReaderCache.clear();
+ fileWriterCache.clear();
+ modificationCache.clear();
+ metadataQuerierCache.clear();
+ measurementSchemaMap.clear();
+ chunkWriterCache.clear();
+ }
+
+ public MeasurementSchema getSchema(String measurementId) {
+ return measurementSchemaMap.get(measurementId);
+ }
+
+ public RestorableTsFileIOWriter getMergeFileWriter(TsFileResource resource) throws IOException {
+ RestorableTsFileIOWriter writer = fileWriterCache.get(resource);
+ if (writer == null) {
+ writer = new RestorableTsFileIOWriter(new File(resource.getFile().getPath() + MERGE_SUFFIX));
+ fileWriterCache.put(resource, writer);
+ }
+ return writer;
+ }
+
+ private MetadataQuerier getMetadataQuerier(TsFileResource seqFile) throws IOException {
+ MetadataQuerier metadataQuerier = metadataQuerierCache.get(seqFile);
+ if (metadataQuerier == null) {
+ metadataQuerier = new MetadataQuerierByFileImpl(getFileReader(seqFile));
+ metadataQuerierCache.put(seqFile, metadataQuerier);
+ }
+ return metadataQuerier;
+ }
+
+ public List<ChunkMetaData> queryChunkMetadata(Path path, TsFileResource seqFile)
+ throws IOException {
+ MetadataQuerier metadataQuerier = getMetadataQuerier(seqFile);
+ List<ChunkMetaData> chunkMetaDataList = metadataQuerier.getChunkMetaDataList(path);
+ if (!keepChunkMetadata) {
+ metadataQuerier.clear();
+ }
+ return chunkMetaDataList;
+ }
+
+ public IPointReader getUnseqReader(Path path) throws IOException {
- return SeriesReaderFactoryImpl.getInstance().createUnseqSeriesReader(path, unseqFiles,
- mergeContext, null);
++ return new UnseqResourceMergeReader(path, unseqFiles, mergeContext, null);
+ }
+
+ public IChunkWriter getChunkWriter(MeasurementSchema measurementSchema) {
+ return chunkWriterCache.computeIfAbsent(measurementSchema,
+ k -> new ChunkWriterImpl(k, new ChunkBuffer(k), TSFileConfig.pageCheckSizeThreshold));
+ }
+
+ public TsFileSequenceReader getFileReader(TsFileResource tsFileResource) throws IOException {
+ TsFileSequenceReader reader = fileReaderCache.get(tsFileResource);
+ if (reader == null) {
+ reader = new TsFileSequenceReader(tsFileResource.getFile().getPath());
+ fileReaderCache.put(tsFileResource, reader);
+ }
+ return reader;
+ }
+
+ public List<Modification> getModifications(TsFileResource tsFileResource, Path path) {
+ // copy from TsFileResource so queries are not affected
+ List<Modification> modifications = modificationCache.computeIfAbsent(tsFileResource,
+ tsFileResource1 -> new ArrayList<>(tsFileResource.getModFile().getModifications()));
+ List<Modification> pathModifications = new ArrayList<>();
+ Iterator<Modification> modificationIterator = modifications.iterator();
+ // each path is visited only once in a merge, so the modifications can be removed after visiting
+ while (modificationIterator.hasNext()) {
+ Modification modification = modificationIterator.next();
+ if (modification.getPath().equals(path)) {
+ pathModifications.add(modification);
+ modificationIterator.remove();
+ }
+ }
+ return pathModifications;
+ }
+
+ public void removeFileWriter(TsFileResource tsFileResource) throws IOException {
+ RestorableTsFileIOWriter newFileWriter = fileWriterCache.remove(tsFileResource);
+ if (newFileWriter != null) {
+ newFileWriter.close();
+ newFileWriter.getFile().delete();
+ fileWriterCache.remove(tsFileResource);
+ }
+ }
+
+ public List<TsFileResource> getSeqFiles() {
+ return seqFiles;
+ }
+
+ public List<TsFileResource> getUnseqFiles() {
+ return unseqFiles;
+ }
+
+ public void setSeqFiles(List<TsFileResource> seqFiles) {
+ this.seqFiles = seqFiles;
+ }
+
+ public void setUnseqFiles(
+ List<TsFileResource> unseqFiles) {
+ this.unseqFiles = unseqFiles;
+ }
+
+ public Map<String, MeasurementSchema> getMeasurementSchemaMap() {
+ return measurementSchemaMap;
+ }
+}
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
index 7281917,2993a7d..d5771f3
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
@@@ -73,9 -73,15 +73,12 @@@ public class GroupByWithValueFilterData
throws StorageEngineException, PathErrorException, ProcessorException, IOException {
initAggreFuction(aggres);
- QueryResourceManager.getInstance().beginQueryOfGivenExpression(context.getJobId(), expression);
- QueryResourceManager
- .getInstance().beginQueryOfGivenQueryPaths(context.getJobId(), selectedSeries);
this.timestampGenerator = new EngineTimeGenerator(expression, context);
- this.allDataReaderList = SeriesReaderFactoryImpl.getInstance()
- .createSeriesReadersByTimestamp(selectedSeries, context);
+ this.allDataReaderList = new ArrayList<>();
+ for (Path path : selectedSeries) {
+ SeriesReaderByTimestamp seriesReaderByTimestamp = new SeriesReaderByTimestamp(path, context);
+ allDataReaderList.add(seriesReaderByTimestamp);
+ }
}
@Override
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/executor/AggregateEngineExecutor.java
index 1a891ef,6b31fe9..f4eb378
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/executor/AggregateEngineExecutor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/executor/AggregateEngineExecutor.java
@@@ -257,10 -258,16 +256,13 @@@ public class AggregateEngineExecutor
*/
public QueryDataSet executeWithValueFilter(QueryContext context)
throws StorageEngineException, PathErrorException, IOException, ProcessorException {
- QueryResourceManager
- .getInstance().beginQueryOfGivenQueryPaths(context.getJobId(), selectedSeries);
- QueryResourceManager.getInstance().beginQueryOfGivenExpression(context.getJobId(), expression);
EngineTimeGenerator timestampGenerator = new EngineTimeGenerator(expression, context);
- List<IReaderByTimeStamp> readersOfSelectedSeries = SeriesReaderFactoryImpl.getInstance()
- .createSeriesReadersByTimestamp(selectedSeries, context);
+ List<IReaderByTimestamp> readersOfSelectedSeries = new ArrayList<>();
+ for (Path path : selectedSeries) {
+ SeriesReaderByTimestamp seriesReaderByTimestamp = new SeriesReaderByTimestamp(path, context);
+ readersOfSelectedSeries.add(seriesReaderByTimestamp);
+ }
List<AggregateFunction> aggregateFunctions = new ArrayList<>();
for (int i = 0; i < selectedSeries.size(); i++) {
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutor.java
index 004cb02,0279f9a..c2bd996
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/executor/EngineExecutor.java
@@@ -94,17 -94,19 +92,14 @@@ public class EngineExecutor
*/
public QueryDataSet executeWithValueFilter(QueryContext context) throws StorageEngineException, IOException {
- EngineTimeGenerator timestampGenerator;
- List<IReaderByTimeStamp> readersOfSelectedSeries;
- QueryResourceManager.getInstance()
- .beginQueryOfGivenQueryPaths(context.getJobId(), queryExpression.getSelectedSeries());
- QueryResourceManager.getInstance()
- .beginQueryOfGivenExpression(context.getJobId(), queryExpression.getExpression());
--
- timestampGenerator = new EngineTimeGenerator(queryExpression.getExpression(), context);
- readersOfSelectedSeries = SeriesReaderFactoryImpl.getInstance()
- .createSeriesReadersByTimestamp(queryExpression.getSelectedSeries(), context);
+ EngineTimeGenerator timestampGenerator = new EngineTimeGenerator(
+ queryExpression.getExpression(), context);
+ List<IReaderByTimestamp> readersOfSelectedSeries = new ArrayList<>();
List<TSDataType> dataTypes = new ArrayList<>();
-
for (Path path : queryExpression.getSelectedSeries()) {
try {
+ // add data type
dataTypes.add(MManager.getInstance().getSeriesType(path.getFullPath()));
} catch (PathErrorException e) {
throw new StorageEngineException(e);
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/reader/fileRelated/UnSealedTsFileIterateReader.java
index 0000000,b0a3f04..d60340a
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/reader/fileRelated/UnSealedTsFileIterateReader.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/reader/fileRelated/UnSealedTsFileIterateReader.java
@@@ -1,0 -1,130 +1,130 @@@
+ /**
+ * 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.db.query.reader.fileRelated;
+
+ import java.io.IOException;
+ import java.util.Collections;
+ import java.util.List;
+ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+ import org.apache.iotdb.db.query.control.FileReaderManager;
+ import org.apache.iotdb.db.query.reader.IAggregateReader;
+ import org.apache.iotdb.db.query.reader.chunkRelated.MemChunkReader;
+ import org.apache.iotdb.db.query.reader.universal.IterateReader;
+ import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+ import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+ import org.apache.iotdb.tsfile.read.controller.ChunkLoader;
+ import org.apache.iotdb.tsfile.read.controller.ChunkLoaderImpl;
+ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReader;
+ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReaderWithFilter;
+ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReaderWithoutFilter;
+
+ /**
+ * To read an unsealed sequence TsFile, this class extends {@link IterateReader} to implement {@link
+ * IAggregateReader} for the TsFile.
+ * <p>
+ * Note that an unsealed sequence TsFile consists of two parts of data in chronological order: 1)
+ * data that has been flushed to disk and 2) data in the flushing memtable list.
+ * <p>
+ * This class is used in {@link org.apache.iotdb.db.query.reader.resourceRelated.SeqResourceIterateReader}.
+ */
+ public class UnSealedTsFileIterateReader extends IterateReader {
+
+ private TsFileResource unsealedTsFile;
+ private Filter filter;
+
+ /**
+ * Whether the reverse order is enabled.
+ * <p>
+ * True to iterate over chunk data in reverse chronological order (from newest to oldest); False
+ * to iterate over chunk data in chronological order (from oldest to newest).
+ */
+ private boolean enableReverse;
+
+ /**
+ * Constructor function.
+ * <p>
+ * An unsealed sequence TsFile consists of two parts of data in chronological order: 1) data that
+ * has been flushed to disk and 2) data in the flushing memtable list. <code>IterateReader</code>
+ * is used to iterate over the two parts. Therefore, this method calls the parent class
+ * <code>IterateReader</code>'s constructor to set <code>readerSize</code> to be 2. Readers for
+ * the two parts of data are created in order later in the method <code>constructNextReader</code>.
+ *
+ * @param unsealedTsFile the TsFileResource corresponding to the unsealed TsFile
+ * @param filter filter condition
+ * @param isReverse True to iterate over chunk data in reverse chronological order (from newest to
+ * oldest); False to iterate over chunk data in chronological order (from oldest to newest).
+ */
+ public UnSealedTsFileIterateReader(TsFileResource unsealedTsFile, Filter filter,
+ boolean isReverse) {
+ super(2);
+ this.enableReverse = isReverse;
+ this.unsealedTsFile = unsealedTsFile;
+ this.filter = filter;
+ }
+
+ @Override
+ protected boolean constructNextReader(int idx) throws IOException {
+ if (idx == 0) {
+ if (enableReverse) {
+ // data in memory first if it is to iterate over chunk data in reverse chronological order
+ currentSeriesReader = new MemChunkReader(unsealedTsFile.getReadOnlyMemChunk(), filter);
+ } else {
+ // data on disk first if it is to iterate over chunk data in chronological order
+ currentSeriesReader = initUnSealedTsFileDiskReader(unsealedTsFile, filter);
+ }
+ } else { // idx=1
+ if (enableReverse) {
+ currentSeriesReader = initUnSealedTsFileDiskReader(unsealedTsFile, filter);
+ } else {
+ currentSeriesReader = new MemChunkReader(unsealedTsFile.getReadOnlyMemChunk(), filter);
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Creates <code>IAggregateReader</code> for an unsealed sequence TsFile's on-disk data.
+ */
+ private IAggregateReader initUnSealedTsFileDiskReader(TsFileResource unSealedTsFile,
+ Filter filter)
+ throws IOException {
+
+ // prepare metaDataList
+ List<ChunkMetaData> metaDataList = unSealedTsFile.getChunkMetaDatas();
+ if (enableReverse && metaDataList != null && !metaDataList.isEmpty()) {
+ Collections.reverse(metaDataList);
+ }
+
+ // prepare chunkLoader
+ TsFileSequenceReader unClosedTsFileReader = FileReaderManager.getInstance()
- .get(unSealedTsFile.getFile().getPath(), false);
++ .get(unSealedTsFile, false);
+ ChunkLoader chunkLoader = new ChunkLoaderImpl(unClosedTsFileReader);
+
+ // init fileSeriesReader
+ FileSeriesReader fileSeriesReader;
+ if (filter == null) {
+ fileSeriesReader = new FileSeriesReaderWithoutFilter(chunkLoader, metaDataList);
+ } else {
+ fileSeriesReader = new FileSeriesReaderWithFilter(chunkLoader, metaDataList, filter);
+ }
+
+ return new FileSeriesReaderAdapter(fileSeriesReader);
+ }
+ }
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/reader/fileRelated/UnSealedTsFileReaderByTimestamp.java
index 0000000,f9fdd27..5f28131
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/reader/fileRelated/UnSealedTsFileReaderByTimestamp.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/reader/fileRelated/UnSealedTsFileReaderByTimestamp.java
@@@ -1,0 -1,96 +1,96 @@@
+ /**
+ * 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.db.query.reader.fileRelated;
+
+ import java.io.IOException;
+ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+ import org.apache.iotdb.db.query.control.FileReaderManager;
+ import org.apache.iotdb.db.query.reader.IReaderByTimestamp;
+ import org.apache.iotdb.db.query.reader.chunkRelated.MemChunkReaderByTimestamp;
+ import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+ import org.apache.iotdb.tsfile.read.controller.ChunkLoader;
+ import org.apache.iotdb.tsfile.read.controller.ChunkLoaderImpl;
+ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReaderByTimestamp;
+
+ /**
+ * To read an unsealed sequence TsFile by timestamp, this class implements an interface {@link
+ * IReaderByTimestamp} for the TsFile.
+ * <p>
+ * Note that an unsealed sequence TsFile consists of two parts of data in chronological order: 1)
+ * data that has been flushed to disk and 2) data in the flushing memtable list.
+ * <p>
+ * This class is used in {@link org.apache.iotdb.db.query.reader.resourceRelated.SeqResourceReaderByTimestamp}.
+ */
+ public class UnSealedTsFileReaderByTimestamp implements IReaderByTimestamp {
+
+ /**
+ * <code>FileSeriesReaderByTimestamp</code> for data which has been flushed to disk.
+ */
+ private FileSeriesReaderByTimestamp unSealedTsFileDiskReaderByTs;
+
+ /**
+ * <code>IReaderByTimestamp</code> for data in the flushing memtable list.
+ */
+ private IReaderByTimestamp unSealedTsFileMemReaderByTs;
+
+ /**
+ * Whether unSealedTsFileDiskReaderByTs has been run out of.
+ * <p>
+ * True means the current reader is unSealedTsFileMemReaderByTs; False means the current reader is
+ * still unSealedTsFileDiskReaderByTs.
+ */
+ private boolean unSealedTsFileDiskReaderEnded;
+
+ public UnSealedTsFileReaderByTimestamp(TsFileResource unsealedTsFile) throws IOException {
+ // create IReaderByTimestamp for data in the flushing memtable list
+ unSealedTsFileMemReaderByTs = new MemChunkReaderByTimestamp(
+ unsealedTsFile.getReadOnlyMemChunk());
+
+ // create FileSeriesReaderByTimestamp for data which has been flushed to disk
+ TsFileSequenceReader unClosedTsFileReader = FileReaderManager.getInstance()
- .get(unsealedTsFile.getFile().getPath(), false);
++ .get(unsealedTsFile, false);
+ ChunkLoader chunkLoader = new ChunkLoaderImpl(unClosedTsFileReader);
+ unSealedTsFileDiskReaderByTs = new FileSeriesReaderByTimestamp(chunkLoader,
+ unsealedTsFile.getChunkMetaDatas());
+
+ unSealedTsFileDiskReaderEnded = false;
+ }
+
+ @Override
+ public Object getValueInTimestamp(long timestamp) throws IOException {
+ if (!unSealedTsFileDiskReaderEnded) {
+ Object value = unSealedTsFileDiskReaderByTs.getValueInTimestamp(timestamp);
+ if (value != null || unSealedTsFileDiskReaderByTs.hasNext()) {
+ return value;
+ } else {
+ unSealedTsFileDiskReaderEnded = true;
+ }
+ }
+ return unSealedTsFileMemReaderByTs.getValueInTimestamp(timestamp);
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (unSealedTsFileDiskReaderEnded) {
+ return unSealedTsFileMemReaderByTs.hasNext();
+ }
+ return (unSealedTsFileDiskReaderByTs.hasNext() || unSealedTsFileMemReaderByTs.hasNext());
+ }
+
+ }
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/SeqResourceIterateReader.java
index 0000000,8c08649..f48ff3e
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/SeqResourceIterateReader.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/SeqResourceIterateReader.java
@@@ -1,0 -1,186 +1,186 @@@
+ /**
+ * 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.db.query.reader.resourceRelated;
+
+ import java.io.IOException;
+ import java.util.Collections;
+ import java.util.List;
+ import org.apache.iotdb.db.engine.modification.Modification;
+ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+ import org.apache.iotdb.db.query.context.QueryContext;
+ import org.apache.iotdb.db.query.control.FileReaderManager;
+ import org.apache.iotdb.db.query.reader.IAggregateReader;
+ import org.apache.iotdb.db.query.reader.fileRelated.FileSeriesReaderAdapter;
+ import org.apache.iotdb.db.query.reader.fileRelated.UnSealedTsFileIterateReader;
+ import org.apache.iotdb.db.query.reader.universal.IterateReader;
+ import org.apache.iotdb.db.utils.QueryUtils;
+ import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+ import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+ import org.apache.iotdb.tsfile.read.common.Path;
+ import org.apache.iotdb.tsfile.read.controller.ChunkLoader;
+ import org.apache.iotdb.tsfile.read.controller.ChunkLoaderImpl;
+ import org.apache.iotdb.tsfile.read.controller.MetadataQuerierByFileImpl;
+ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReader;
+ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReaderWithFilter;
+ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReaderWithoutFilter;
+
+ /**
+ * To read a chronologically ordered list of sequence TsFiles, this class extends {@link
+ * IterateReader} to implements <code>IAggregateReader</code> for the TsFiles.
+ * <p>
+ * Notes: 1) The list of sequence TsFiles is in strict chronological order. 2) The data in a
+ * sequence TsFile is also organized in chronological order. 3) A sequence TsFile can be either
+ * sealed or unsealed. 4) An unsealed sequence TsFile consists of two parts of data in chronological
+ * order: data that has been flushed to disk and data in the flushing memtable list.
+ * <p>
+ * This class is used in {@link org.apache.iotdb.db.query.reader.seriesRelated.SeriesReaderWithoutValueFilter}.
+ */
+ public class SeqResourceIterateReader extends IterateReader {
+
+ private Path seriesPath;
+
+ /**
+ * Whether the reverse order is enabled.
+ * <p>
+ * True to iterate over the list of sequence TsFiles and chunks in TsFiles in reverse
+ * chronological order (from newest to oldest); False to iterate in chronological order (from
+ * oldest to newest).
+ */
+ private boolean enableReverse;
+
+ private List<TsFileResource> seqResources;
+ private Filter filter;
+ private QueryContext context;
+
+ /**
+ * Constructor function.
+ * <p>
+ * <code>IterateReader</code> is used to iterate over the chronologically ordered list of
+ * sequence TsFiles. Therefore, this method calls the parent class <code>IterateReader</code>'s
+ * constructor to set <code>readerSize</code> to be the size of the TsFile list. Readers for the
+ * TsFiles are created in order later in the method <code>constructNextReader</code>.
+ *
+ * @param seriesPath the path of the series data
+ * @param seqResources a list of sequence TsFile resources in chronological order
+ * @param filter filter condition. Null if no filter.
+ * @param context query context
+ * @param isReverse True to iterate over data in reverse chronological order (from newest to
+ * oldest); False to iterate over data in chronological order (from oldest to newest).
+ */
+ public SeqResourceIterateReader(Path seriesPath, List<TsFileResource> seqResources,
+ Filter filter, QueryContext context, boolean isReverse) {
+ super(seqResources.size());
+ this.seriesPath = seriesPath;
+ this.enableReverse = isReverse;
+ if (isReverse) {
+ Collections.reverse(seqResources);
+ }
+ this.seqResources = seqResources;
+ this.filter = filter;
+ this.context = context;
+ }
+
+ public SeqResourceIterateReader(Path seriesPath, List<TsFileResource> seqResources,
+ Filter timeFilter, QueryContext context) {
+ this(seriesPath, seqResources, timeFilter, context, false);
+ }
+
+ /**
+ * If the idx-th TsFile in the <code>seqResources</code> might satisfy this <code>filter</code>,
+ * then construct <code>IAggregateReader</code> for it, assign to <code>currentSeriesReader</code>
+ * and return true. Otherwise, return false.
+ *
+ * @param idx the index of the TsFile in the resource list
+ * @return True if the reader is constructed; False if not.
+ */
+ @Override
+ public boolean constructNextReader(int idx) throws IOException {
+ TsFileResource tsFileResource = seqResources.get(idx);
+ if (tsFileResource.isClosed()) {
+ if (isTsFileNotSatisfied(tsFileResource, filter)) {
+ return false;
+ }
+ currentSeriesReader = initSealedTsFileReader(tsFileResource, filter, context);
+ return true;
+ } else {
+ // an unsealed sequence TsFile's endTimeMap size may be equal to 0 or greater than 0
+ // If endTimeMap size is 0, conservatively assume that this TsFile might satisfy this filter.
+ // If endTimeMap size is not 0, call isTsFileNotSatisfied to check.
+ if (tsFileResource.getEndTimeMap().size() != 0) {
+ if (isTsFileNotSatisfied(tsFileResource, filter)) {
+ return false;
+ }
+ }
+ currentSeriesReader = new UnSealedTsFileIterateReader(tsFileResource, filter,
+ enableReverse);
+ return true;
+ }
+ }
+
+ /**
+ * Returns true if the start and end time of the series data in this sequence TsFile do not
+ * satisfy the filter condition. Returns false if satisfy.
+ * <p>
+ * This method is used to in <code>constructNextReader</code> to check whether this TsFile can be
+ * skipped.
+ *
+ * @param tsFile the TsFileResource corresponding to this TsFile
+ * @param filter filter condition. Null if no filter.
+ * @return True if the TsFile's start and end time do not satisfy the filter condition; False if
+ * satisfy.
+ */
+ private boolean isTsFileNotSatisfied(TsFileResource tsFile, Filter filter) {
+ if (filter == null) {
+ return false;
+ }
+ long startTime = tsFile.getStartTimeMap().get(seriesPath.getDevice());
+ long endTime = tsFile.getEndTimeMap().get(seriesPath.getDevice());
+ return !filter.satisfyStartEndTime(startTime, endTime);
+ }
+
+ private IAggregateReader initSealedTsFileReader(TsFileResource sealedTsFile, Filter filter,
+ QueryContext context) throws IOException {
+ // prepare metaDataList
+ TsFileSequenceReader tsFileReader = FileReaderManager.getInstance()
- .get(sealedTsFile.getFile().getPath(), true);
++ .get(sealedTsFile, true);
+ MetadataQuerierByFileImpl metadataQuerier = new MetadataQuerierByFileImpl(tsFileReader);
+ List<ChunkMetaData> metaDataList = metadataQuerier.getChunkMetaDataList(seriesPath);
+ List<Modification> pathModifications = context.getPathModifications(sealedTsFile.getModFile(),
+ seriesPath.getFullPath());
+ if (!pathModifications.isEmpty()) {
+ QueryUtils.modifyChunkMetaData(metaDataList, pathModifications);
+ }
+
+ if (enableReverse) {
+ Collections.reverse(metaDataList);
+ }
+ // prepare chunkLoader
+ ChunkLoader chunkLoader = new ChunkLoaderImpl(tsFileReader);
+
+ // init fileSeriesReader
+ FileSeriesReader fileSeriesReader;
+ if (filter == null) {
+ fileSeriesReader = new FileSeriesReaderWithoutFilter(chunkLoader, metaDataList);
+ } else {
+ fileSeriesReader = new FileSeriesReaderWithFilter(chunkLoader, metaDataList, filter);
+ }
+ return new FileSeriesReaderAdapter(fileSeriesReader);
+ }
+ }
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/SeqResourceReaderByTimestamp.java
index 0000000,f8a5984..e791d94
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/SeqResourceReaderByTimestamp.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/SeqResourceReaderByTimestamp.java
@@@ -1,0 -1,189 +1,189 @@@
+ /**
+ * 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.db.query.reader.resourceRelated;
+
+ import java.io.IOException;
+ import java.util.List;
+ import org.apache.iotdb.db.engine.modification.Modification;
+ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+ import org.apache.iotdb.db.query.context.QueryContext;
+ import org.apache.iotdb.db.query.control.FileReaderManager;
+ import org.apache.iotdb.db.query.reader.IReaderByTimestamp;
+ import org.apache.iotdb.db.query.reader.fileRelated.FileSeriesReaderByTimestampAdapter;
+ import org.apache.iotdb.db.query.reader.fileRelated.UnSealedTsFileReaderByTimestamp;
+ import org.apache.iotdb.db.utils.QueryUtils;
+ import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+ import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+ import org.apache.iotdb.tsfile.read.common.Path;
+ import org.apache.iotdb.tsfile.read.controller.ChunkLoader;
+ import org.apache.iotdb.tsfile.read.controller.ChunkLoaderImpl;
+ import org.apache.iotdb.tsfile.read.controller.MetadataQuerierByFileImpl;
+ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReaderByTimestamp;
+
+ /**
+ * To read a chronologically ordered list of sequence TsFiles by timestamp, this class implements
+ * <code>IReaderByTimestamp</code> for the TsFiles.
+ * <p>
+ * Notes: 1) The list of sequence TsFiles is in strict chronological order. 2) The data in a
+ * sequence TsFile is also organized in chronological order. 3) A sequence TsFile can be either
+ * sealed or unsealed. 4) An unsealed sequence TsFile consists of two parts of data in chronological
+ * order: data that has been flushed to disk and data in the flushing memtable list.
+ * <p>
+ * This class is used in {@link org.apache.iotdb.db.query.reader.seriesRelated.SeriesReaderByTimestamp}.
+ */
+
+ public class SeqResourceReaderByTimestamp implements IReaderByTimestamp {
+
+ protected Path seriesPath;
+ private List<TsFileResource> seqResources;
+ private QueryContext context;
+ private int nextIntervalFileIndex;
+ private IReaderByTimestamp seriesReader;
+
+ /**
+ * Constructor function.
+ * <p>
+ *
+ * @param seriesPath the path of the series data
+ * @param seqResources a list of sequence TsFile resources in chronological order
+ * @param context query context
+ */
+ public SeqResourceReaderByTimestamp(Path seriesPath, List<TsFileResource> seqResources,
+ QueryContext context) {
+ this.seriesPath = seriesPath;
+ this.seqResources = seqResources;
+ this.context = context;
+ this.nextIntervalFileIndex = 0;
+ this.seriesReader = null;
+ }
+
+ @Override
+ public Object getValueInTimestamp(long timestamp) throws IOException {
+ Object value = null;
+
+ if (seriesReader != null) {
+ value = seriesReader.getValueInTimestamp(timestamp);
+ // if get value or no value in this timestamp but has next, return.
+ if (value != null || seriesReader.hasNext()) {
+ return value;
+ }
+ }
+
+ // Because the sequence TsFile resources are chronologically globally ordered, there exists at
+ // most one TsFile resource that overlaps this timestamp.
+ while (nextIntervalFileIndex < seqResources.size()) {
+ boolean isConstructed = constructNextReader(nextIntervalFileIndex++, timestamp);
+ if (isConstructed) {
+ value = seriesReader.getValueInTimestamp(timestamp);
+ // if get value or no value in this timestamp but has next, return.
+ if (value != null || seriesReader.hasNext()) {
+ return value;
+ }
+ }
+ }
+ return value;
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (seriesReader != null && seriesReader.hasNext()) {
+ return true;
+ }
+
+ while (nextIntervalFileIndex < seqResources.size()) {
+ TsFileResource tsFileResource = seqResources.get(nextIntervalFileIndex++);
+ if (tsFileResource.isClosed()) {
+ seriesReader = initSealedTsFileReaderByTimestamp(tsFileResource, context);
+ } else {
+ seriesReader = new UnSealedTsFileReaderByTimestamp(tsFileResource);
+ }
+ if (seriesReader.hasNext()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * If the idx-th TsFile in the <code>seqResources</code> might overlap this
+ * <code>timestamp</code>, then construct <code>IReaderByTimestamp</code> for it, assign to the
+ * <code>currentSeriesReader</code> and return true. Otherwise, return false.
+ * <p>
+ * Note that the list of sequence TsFiles is chronologically ordered, so there will be at most one
+ * TsFile that overlaps this timestamp.
+ *
+ * @param idx the index of the TsFile in the resource list
+ * @param timestamp check whether or not to construct the reader according to this timestamp
+ * @return True if the reader is constructed; False if not.
+ */
+ private boolean constructNextReader(int idx, long timestamp) throws IOException {
+ TsFileResource tsFileResource = seqResources.get(idx);
+ if (tsFileResource.isClosed()) {
+ if (isTsFileNotSatisfied(tsFileResource, timestamp)) {
+ return false;
+ }
+ seriesReader = initSealedTsFileReaderByTimestamp(tsFileResource, context);
+ return true;
+ } else {
+ // an unsealed sequence TsFile's endTimeMap size may be equal to 0 or greater than 0
+ // If endTimeMap size is 0, conservatively assume that this TsFile might overlap this timestamp.
+ // If endTimeMap size is not 0, call isTsFileNotSatisfied to check.
+ if (tsFileResource.getEndTimeMap().size() != 0) {
+ if (isTsFileNotSatisfied(tsFileResource, timestamp)) {
+ return false;
+ }
+ }
+ seriesReader = new UnSealedTsFileReaderByTimestamp(tsFileResource);
+ return true;
+ }
+ }
+
+ /**
+ * Returns true if the end time of the series data in this sequence TsFile is smaller than this
+ * timestamp.
+ * <p>
+ * Note that <code>seqResources</code> is a list of chronologically ordered sequence TsFiles, so
+ * there will be at most one TsFile that overlaps this timestamp.
+ * <p>
+ * This method is used to in <code>constructNextReader</code> to check whether this TsFile can be
+ * skipped.
+ */
+ private boolean isTsFileNotSatisfied(TsFileResource tsFile, long timestamp) {
+ return tsFile.getEndTimeMap().get(seriesPath.getDevice()) < timestamp;
+ }
+
+ private IReaderByTimestamp initSealedTsFileReaderByTimestamp(TsFileResource sealedTsFile,
+ QueryContext context) throws IOException {
+ // prepare metaDataList
+ TsFileSequenceReader tsFileReader = FileReaderManager.getInstance()
- .get(sealedTsFile.getFile().getPath(), true);
++ .get(sealedTsFile, true);
+ MetadataQuerierByFileImpl metadataQuerier = new MetadataQuerierByFileImpl(tsFileReader);
+ List<ChunkMetaData> metaDataList = metadataQuerier.getChunkMetaDataList(seriesPath);
+ List<Modification> pathModifications = context.getPathModifications(sealedTsFile.getModFile(),
+ seriesPath.getFullPath());
+ if (!pathModifications.isEmpty()) {
+ QueryUtils.modifyChunkMetaData(metaDataList, pathModifications);
+ }
+ // prepare chunkLoader
+ ChunkLoader chunkLoader = new ChunkLoaderImpl(tsFileReader);
+
+ return new FileSeriesReaderByTimestampAdapter(
+ new FileSeriesReaderByTimestamp(chunkLoader, metaDataList));
+ }
+ }
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceMergeReader.java
index 0000000,44b14ed..994c73b
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceMergeReader.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceMergeReader.java
@@@ -1,0 -1,141 +1,141 @@@
+ /**
+ * 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.db.query.reader.resourceRelated;
+
+ import java.io.IOException;
+ import java.util.List;
+ import org.apache.iotdb.db.engine.modification.Modification;
+ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+ import org.apache.iotdb.db.query.context.QueryContext;
+ import org.apache.iotdb.db.query.control.FileReaderManager;
+ import org.apache.iotdb.db.query.reader.chunkRelated.DiskChunkReader;
+ import org.apache.iotdb.db.query.reader.chunkRelated.MemChunkReader;
+ import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader;
+ import org.apache.iotdb.db.utils.QueryUtils;
+ import org.apache.iotdb.tsfile.common.constant.StatisticConstant;
+ import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+ 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.controller.ChunkLoaderImpl;
+ import org.apache.iotdb.tsfile.read.controller.MetadataQuerierByFileImpl;
+ import org.apache.iotdb.tsfile.read.filter.DigestForFilter;
+ import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+ import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
+ import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithFilter;
+ import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithoutFilter;
+
+ /**
+ * To read a list of unsequence TsFiles, this class extends {@link PriorityMergeReader} to
+ * implement
+ * <code>IPointReader</code> for the TsFiles.
+ * <p>
+ * Note that an unsequence TsFile can be either closed or unclosed. An unclosed unsequence TsFile
+ * consists of data on disk and data in memtables that will be flushed to this unclosed TsFile.
+ * <p>
+ * This class is used in {@link org.apache.iotdb.db.query.reader.seriesRelated.SeriesReaderWithoutValueFilter}.
+ */
+ public class UnseqResourceMergeReader extends PriorityMergeReader {
+
+ private Path seriesPath;
+
+ public UnseqResourceMergeReader(Path seriesPath, List<TsFileResource> unseqResources,
+ QueryContext context, Filter filter) throws IOException {
+ this.seriesPath = seriesPath;
+
+ int priorityValue = 1;
+ for (TsFileResource tsFileResource : unseqResources) {
+ TsFileSequenceReader tsFileReader = FileReaderManager.getInstance()
- .get(tsFileResource.getFile().getPath(), tsFileResource.isClosed());
++ .get(tsFileResource, tsFileResource.isClosed());
+
+ // prepare metaDataList
+ List<ChunkMetaData> metaDataList;
+ if (tsFileResource.isClosed()) {
+ if (isTsFileNotSatisfied(tsFileResource, filter)) {
+ continue;
+ }
+ MetadataQuerierByFileImpl metadataQuerier = new MetadataQuerierByFileImpl(tsFileReader);
+ metaDataList = metadataQuerier.getChunkMetaDataList(seriesPath);
+ List<Modification> pathModifications = context
+ .getPathModifications(tsFileResource.getModFile(), seriesPath.getFullPath());
+ if (!pathModifications.isEmpty()) {
+ QueryUtils.modifyChunkMetaData(metaDataList, pathModifications);
+ }
+ } else {
+ if (tsFileResource.getEndTimeMap().size() != 0) {
+ if (isTsFileNotSatisfied(tsFileResource, filter)) {
+ continue;
+ }
+ }
+ metaDataList = tsFileResource.getChunkMetaDatas();
+ }
+
+ // create and add ChunkReader with priority
+ ChunkLoaderImpl chunkLoader = new ChunkLoaderImpl(tsFileReader);
+ for (ChunkMetaData chunkMetaData : metaDataList) {
+
+ DigestForFilter digest = new DigestForFilter(chunkMetaData.getStartTime(),
+ chunkMetaData.getEndTime(),
+ chunkMetaData.getDigest().getStatistics().get(StatisticConstant.MIN_VALUE),
+ chunkMetaData.getDigest().getStatistics().get(StatisticConstant.MAX_VALUE),
+ chunkMetaData.getTsDataType());
+
+ if (filter != null && !filter.satisfy(digest)) {
+ continue;
+ }
+
+ Chunk chunk = chunkLoader.getChunk(chunkMetaData);
+ ChunkReader chunkReader = filter != null ? new ChunkReaderWithFilter(chunk, filter)
+ : new ChunkReaderWithoutFilter(chunk);
+
+ addReaderWithPriority(new DiskChunkReader(chunkReader), priorityValue);
+
+ priorityValue++;
+ }
+
+ if (!tsFileResource.isClosed()) {
+ // create and add MemChunkReader with priority
+ addReaderWithPriority(
+ new MemChunkReader(tsFileResource.getReadOnlyMemChunk(), filter), priorityValue++);
+ }
+ }
+ }
+
+ /**
+ * Returns true if the start and end time of the series data in this unsequence TsFile do not
+ * satisfy the filter condition. Returns false if satisfy.
+ * <p>
+ * This method is used to in the constructor function to check whether this TsFile can be
+ * skipped.
+ *
+ * @param tsFile the TsFileResource corresponding to this TsFile
+ * @param filter filter condition. Null if no filter.
+ * @return True if the TsFile's start and end time do not satisfy the filter condition; False if
+ * satisfy.
+ */
+ // TODO future work: deduplicate code. See SeqResourceIterateReader.
+ private boolean isTsFileNotSatisfied(TsFileResource tsFile, Filter filter) {
+ if (filter == null) {
+ return false;
+ }
+ long startTime = tsFile.getStartTimeMap().get(seriesPath.getDevice());
+ long endTime = tsFile.getEndTimeMap().get(seriesPath.getDevice());
+ return !filter.satisfyStartEndTime(startTime, endTime);
+ }
+ }
diff --cc iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceReaderByTimestamp.java
index 0000000,d2a5047..a8b743d
mode 000000,100644..100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceReaderByTimestamp.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/reader/resourceRelated/UnseqResourceReaderByTimestamp.java
@@@ -1,0 -1,97 +1,97 @@@
+ /**
+ * 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.db.query.reader.resourceRelated;
+
+ import java.io.IOException;
+ import java.util.List;
+ import org.apache.iotdb.db.engine.modification.Modification;
+ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+ import org.apache.iotdb.db.query.context.QueryContext;
+ import org.apache.iotdb.db.query.control.FileReaderManager;
+ import org.apache.iotdb.db.query.reader.chunkRelated.DiskChunkReaderByTimestamp;
+ import org.apache.iotdb.db.query.reader.chunkRelated.MemChunkReaderByTimestamp;
+ import org.apache.iotdb.db.query.reader.universal.PriorityMergeReaderByTimestamp;
+ import org.apache.iotdb.db.utils.QueryUtils;
+ import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+ 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.controller.ChunkLoaderImpl;
+ import org.apache.iotdb.tsfile.read.controller.MetadataQuerierByFileImpl;
+ import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderByTimestamp;
+
+ /**
+ * To read a list of unsequence TsFiles by timestamp, this class extends {@link
+ * PriorityMergeReaderByTimestamp} to implement <code>IReaderByTimestamp</code> for the TsFiles.
+ * <p>
+ * Note that an unsequence TsFile can be either closed or unclosed. An unclosed unsequence TsFile
+ * consists of data on disk and data in memtables that will be flushed to this unclosed TsFile.
+ * <p>
+ * This class is used in {@link org.apache.iotdb.db.query.reader.seriesRelated.SeriesReaderByTimestamp}.
+ */
+ public class UnseqResourceReaderByTimestamp extends PriorityMergeReaderByTimestamp {
+
+ public UnseqResourceReaderByTimestamp(Path seriesPath,
+ List<TsFileResource> unseqResources, QueryContext context) throws IOException {
+ int priorityValue = 1;
+
+ for (TsFileResource tsFileResource : unseqResources) {
+ TsFileSequenceReader tsFileReader = FileReaderManager.getInstance()
- .get(tsFileResource.getFile().getPath(), tsFileResource.isClosed());
++ .get(tsFileResource, tsFileResource.isClosed());
+
+ // prepare metaDataList
+ List<ChunkMetaData> metaDataList;
+ if (tsFileResource.isClosed()) {
+ MetadataQuerierByFileImpl metadataQuerier = new MetadataQuerierByFileImpl(tsFileReader);
+ metaDataList = metadataQuerier.getChunkMetaDataList(seriesPath);
+ List<Modification> pathModifications = context
+ .getPathModifications(tsFileResource.getModFile(), seriesPath.getFullPath());
+ if (!pathModifications.isEmpty()) {
+ QueryUtils.modifyChunkMetaData(metaDataList, pathModifications);
+ }
+ } else {
+ metaDataList = tsFileResource.getChunkMetaDatas();
+ }
+
+ // create and add ChunkReaderByTimestamp with priority
+ ChunkLoaderImpl chunkLoader = new ChunkLoaderImpl(tsFileReader);
+ for (ChunkMetaData chunkMetaData : metaDataList) {
+
+ Chunk chunk = chunkLoader.getChunk(chunkMetaData);
+ ChunkReaderByTimestamp chunkReader = new ChunkReaderByTimestamp(chunk);
+
+ addReaderWithPriority(new DiskChunkReaderByTimestamp(chunkReader),
+ priorityValue);
+
+ priorityValue++;
+ }
+
+ if (!tsFileResource.isClosed()) {
+ // create and add MemChunkReader with priority
+ addReaderWithPriority(
+ new MemChunkReaderByTimestamp(tsFileResource.getReadOnlyMemChunk()), priorityValue++);
+ }
+ }
+
+ // TODO add external sort when needed
+
+ // TODO future work: create reader when getValueInTimestamp so that resources
+ // whose start and end time do not satisfy can be skipped.
+ }
+ }
diff --cc iotdb/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
index 96b6aae,0000000..8ecb26f
mode 100644,000000..100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
@@@ -1,195 -1,0 +1,195 @@@
+/**
+ * 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.db.engine.merge;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.query.context.QueryContext;
- import org.apache.iotdb.db.query.reader.sequence.SequenceSeriesReader;
++import org.apache.iotdb.db.query.reader.resourceRelated.SeqResourceIterateReader;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class MergeTaskTest extends MergeTest {
+
+ private File tempSGDir;
+
+ @Before
+ public void setUp() throws IOException, WriteProcessException {
+ super.setUp();
+ tempSGDir = new File("tempSG");
+ tempSGDir.mkdirs();
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ super.tearDown();
+ FileUtils.deleteDirectory(tempSGDir);
+ }
+
+ @Test
+ public void testMerge() throws Exception {
+ MergeTask mergeTask =
+ new MergeTask(seqResources, unseqResources, tempSGDir.getPath(), (k, v, l) -> {}, "test",
+ false);
+ mergeTask.call();
+
+ QueryContext context = new QueryContext();
+ Path path = new Path(deviceIds[0], measurementSchemas[0].getMeasurementId());
- SequenceSeriesReader tsFilesReader = new SequenceSeriesReader(path,
++ SeqResourceIterateReader tsFilesReader = new SeqResourceIterateReader(path,
+ Collections.singletonList(seqResources.get(0)),
+ null, context);
+ while (tsFilesReader.hasNext()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i) + 20000.0, batchData.getDoubleByIndex(i), 0.001);
+ }
+ }
+ tsFilesReader.close();
+ }
+
+ @Test
+ public void testFullMerge() throws Exception {
+ MergeTask mergeTask =
+ new MergeTask(seqResources, unseqResources, tempSGDir.getPath(), (k, v, l) -> {}, "test",
+ true);
+ mergeTask.call();
+
+ QueryContext context = new QueryContext();
+ Path path = new Path(deviceIds[0], measurementSchemas[0].getMeasurementId());
- SequenceSeriesReader tsFilesReader = new SequenceSeriesReader(path,
++ SeqResourceIterateReader tsFilesReader = new SeqResourceIterateReader(path,
+ Collections.singletonList(seqResources.get(0)),
+ null, context);
+ while (tsFilesReader.hasNext()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i) + 20000.0, batchData.getDoubleByIndex(i), 0.001);
+ }
+ }
+ tsFilesReader.close();
+ }
+
+ @Test
+ public void testChunkNumThreshold() throws Exception {
+ IoTDBDescriptor.getInstance().getConfig().setChunkMergePointThreshold(Integer.MAX_VALUE);
+ MergeTask mergeTask =
+ new MergeTask(seqResources, unseqResources, tempSGDir.getPath(), (k, v, l) -> {}, "test",
+ false);
+ mergeTask.call();
+
+ QueryContext context = new QueryContext();
+ Path path = new Path(deviceIds[0], measurementSchemas[0].getMeasurementId());
- SequenceSeriesReader tsFilesReader = new SequenceSeriesReader(path,
++ SeqResourceIterateReader tsFilesReader = new SeqResourceIterateReader(path,
+ Collections.singletonList(seqResources.get(0)),
+ null, context);
+ while (tsFilesReader.hasNext()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i) + 20000.0, batchData.getDoubleByIndex(i), 0.001);
+ }
+ }
+ tsFilesReader.close();
+ }
+
+ @Test
+ public void testPartialMerge1() throws Exception {
+ MergeTask mergeTask =
+ new MergeTask(seqResources, unseqResources.subList(0, 1), tempSGDir.getPath(),
+ (k, v, l) -> {}, "test", false);
+ mergeTask.call();
+
+ QueryContext context = new QueryContext();
+ Path path = new Path(deviceIds[0], measurementSchemas[0].getMeasurementId());
- SequenceSeriesReader tsFilesReader = new SequenceSeriesReader(path,
++ SeqResourceIterateReader tsFilesReader = new SeqResourceIterateReader(path,
+ Collections.singletonList(seqResources.get(0)),
+ null, context);
+ while (tsFilesReader.hasNext()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ if (batchData.getTimeByIndex(i) < 20) {
+ assertEquals(batchData.getTimeByIndex(i) + 10000.0, batchData.getDoubleByIndex(i), 0.001);
+ } else {
+ assertEquals(batchData.getTimeByIndex(i) + 0.0, batchData.getDoubleByIndex(i), 0.001);
+ }
+ }
+ }
+ tsFilesReader.close();
+ }
+
+ @Test
+ public void testPartialMerge2() throws Exception {
+ MergeTask mergeTask =
+ new MergeTask(seqResources, unseqResources.subList(5, 6), tempSGDir.getPath(),
+ (k, v, l) -> {}, "test", false);
+ mergeTask.call();
+
+ QueryContext context = new QueryContext();
+ Path path = new Path(deviceIds[0], measurementSchemas[0].getMeasurementId());
- SequenceSeriesReader tsFilesReader = new SequenceSeriesReader(path,
++ SeqResourceIterateReader tsFilesReader = new SeqResourceIterateReader(path,
+ Collections.singletonList(seqResources.get(0)),
+ null, context);
+ while (tsFilesReader.hasNext()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i) + 20000.0, batchData.getDoubleByIndex(i), 0.001);
+ }
+ }
+ tsFilesReader.close();
+ }
+
+ @Test
+ public void mergeWithDeletionTest() throws Exception {
+ seqResources.get(0).getModFile().write(new Deletion(new Path(deviceIds[0],
+ measurementSchemas[0].getMeasurementId()), 10000, 50));
+ seqResources.get(0).getModFile().close();
+
+
+ MergeTask mergeTask =
+ new MergeTask(seqResources, unseqResources, tempSGDir.getPath(), (k, v, l) -> {}, "test",
+ false);
+ mergeTask.call();
+
+ QueryContext context = new QueryContext();
+ Path path = new Path(deviceIds[0], measurementSchemas[0].getMeasurementId());
- SequenceSeriesReader tsFilesReader = new SequenceSeriesReader(path,
++ SeqResourceIterateReader tsFilesReader = new SeqResourceIterateReader(path,
+ Collections.singletonList(seqResources.get(0)),
+ null, context);
+ int count = 0;
+ while (tsFilesReader.hasNext()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i) + 20000.0, batchData.getDoubleByIndex(i), 0.001);
+ count ++;
+ }
+ }
+ assertEquals(49, count);
+ tsFilesReader.close();
+ }
+}
diff --cc iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationIT.java
index 6f5d209,be1eee8..a4677e0
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBAggregationIT.java
@@@ -592,9 -594,9 +594,9 @@@ public class IoTDBAggregationIT
}
statement.execute("flush");
for (int i = 3000; i < 6500; i++) {
- statement.execute(String.format(insertTemplate, i, i, i, (double) i, "\'" + i + "\'", "false"));
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'" + i + "\'", "false"));
}
-// statement.execute("merge");
+ statement.execute("merge");
// prepare BufferWrite cache
for (int i = 9000; i < 10000; i++) {
diff --cc iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBDeletionIT.java
index 7c92e22,eef24e3..6f68f79
--- a/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBDeletionIT.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/integration/IoTDBDeletionIT.java
@@@ -202,19 -204,19 +204,19 @@@ public class IoTDBDeletionIT
Statement statement = connection.createStatement();
// prepare BufferWrite file
for (int i = 201; i <= 300; i++) {
- statement.execute(String.format(insertTemplate, i, i, i, (double) i, "\'" + i + "\'",
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'" + i + "\'",
i % 2 == 0));
}
-// statement.execute("merge");
+ statement.execute("merge");
// prepare Unseq-File
for (int i = 1; i <= 100; i++) {
- statement.execute(String.format(insertTemplate, i, i, i, (double) i, "\'" + i + "\'",
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'" + i + "\'",
i % 2 == 0));
}
-// statement.execute("merge");
+ statement.execute("merge");
// prepare BufferWrite cache
for (int i = 301; i <= 400; i++) {
- statement.execute(String.format(insertTemplate, i, i, i, (double) i, "\'" + i + "\'",
+ statement.execute(String.format(Locale.ENGLISH, insertTemplate, i, i, i, (double) i, "\'" + i + "\'",
i % 2 == 0));
}
// prepare Overflow cache