This is an automated email from the ASF dual-hosted git repository. xiangweiwei pushed a commit to branch clusterDescBug in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8dbcd2f05e303055219609636fd24e8df021966d Author: lta <[email protected]> AuthorDate: Wed Jun 23 10:59:46 2021 +0800 Fix commit conflict --- .../iotdb/cluster/query/reader/ClusterReaderFactory.java | 12 ++++++++++-- ...dMergeReader.java => ManagedDescPriorityMergeReader.java} | 8 +++++--- ...nagedMergeReader.java => ManagedPriorityMergeReader.java} | 4 ++-- .../iotdb/cluster/query/reader/mult/MultBatchReader.java | 5 +++-- 4 files changed, 20 insertions(+), 9 deletions(-) diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactory.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactory.java index 64a2e3b..3555af2 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactory.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ClusterReaderFactory.java @@ -62,6 +62,7 @@ import org.apache.iotdb.db.query.reader.series.SeriesRawDataBatchReader; import org.apache.iotdb.db.query.reader.series.SeriesRawDataPointReader; import org.apache.iotdb.db.query.reader.series.SeriesReader; import org.apache.iotdb.db.query.reader.series.SeriesReaderByTimestamp; +import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader; import org.apache.iotdb.db.utils.SerializeUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; @@ -364,7 +365,12 @@ public class ClusterReaderFactory { metaGroupMember.getName(), path, partitionGroups.size()); - ManagedMergeReader mergeReader = new ManagedMergeReader(dataType); + PriorityMergeReader mergeReader; + if (ascending) { + mergeReader = new ManagedPriorityMergeReader(dataType); + } else { + mergeReader = new ManagedDescPriorityMergeReader(dataType); + } try { // build a reader for each group and merge them for (PartitionGroup partitionGroup : partitionGroups) { @@ -383,7 +389,9 @@ public class ClusterReaderFactory { } catch (IOException | QueryProcessException e) { throw new StorageEngineException(e); } - return mergeReader; + // The instance of merge reader is either ManagedPriorityMergeReader or + // ManagedDescPriorityMergeReader, which is safe to cast type. + return (ManagedSeriesReader) mergeReader; } /** diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedMergeReader.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedDescPriorityMergeReader.java similarity index 88% copy from cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedMergeReader.java copy to cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedDescPriorityMergeReader.java index e54dede..653dea4 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedMergeReader.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedDescPriorityMergeReader.java @@ -20,16 +20,18 @@ package org.apache.iotdb.cluster.query.reader; import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader; -import org.apache.iotdb.db.query.reader.universal.PriorityMergeReader; +import org.apache.iotdb.db.query.reader.universal.DescPriorityMergeReader; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.read.TimeValuePair; import org.apache.iotdb.tsfile.read.common.BatchData; +import org.apache.iotdb.tsfile.read.reader.IPointReader; import java.io.IOException; import java.util.NoSuchElementException; @SuppressWarnings("common-java:DuplicatedBlocks") -public class ManagedMergeReader extends PriorityMergeReader implements ManagedSeriesReader { +public class ManagedDescPriorityMergeReader extends DescPriorityMergeReader + implements ManagedSeriesReader, IPointReader { private static final int BATCH_SIZE = 4096; @@ -39,7 +41,7 @@ public class ManagedMergeReader extends PriorityMergeReader implements ManagedSe private BatchData batchData; private TSDataType dataType; - public ManagedMergeReader(TSDataType dataType) { + public ManagedDescPriorityMergeReader(TSDataType dataType) { this.dataType = dataType; } diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedMergeReader.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedPriorityMergeReader.java similarity index 94% rename from cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedMergeReader.java rename to cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedPriorityMergeReader.java index e54dede..e57f4d9 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedMergeReader.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/ManagedPriorityMergeReader.java @@ -29,7 +29,7 @@ import java.io.IOException; import java.util.NoSuchElementException; @SuppressWarnings("common-java:DuplicatedBlocks") -public class ManagedMergeReader extends PriorityMergeReader implements ManagedSeriesReader { +public class ManagedPriorityMergeReader extends PriorityMergeReader implements ManagedSeriesReader { private static final int BATCH_SIZE = 4096; @@ -39,7 +39,7 @@ public class ManagedMergeReader extends PriorityMergeReader implements ManagedSe private BatchData batchData; private TSDataType dataType; - public ManagedMergeReader(TSDataType dataType) { + public ManagedPriorityMergeReader(TSDataType dataType) { this.dataType = dataType; } diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultBatchReader.java b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultBatchReader.java index dd7f21b..dc47deb 100644 --- a/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultBatchReader.java +++ b/cluster/src/main/java/org/apache/iotdb/cluster/query/reader/mult/MultBatchReader.java @@ -41,7 +41,7 @@ public class MultBatchReader implements IMultBatchReader { @Override public boolean hasNextBatch() throws IOException { for (IBatchReader reader : pathBatchReaders.values()) { - if (reader.hasNextBatch()) { + if (reader != null && reader.hasNextBatch()) { return true; } } @@ -50,7 +50,8 @@ public class MultBatchReader implements IMultBatchReader { @Override public boolean hasNextBatch(String fullPath) throws IOException { - return pathBatchReaders.get(fullPath).hasNextBatch(); + IBatchReader reader = pathBatchReaders.get(fullPath); + return reader != null && reader.hasNextBatch(); } @Override
