This is an automated email from the ASF dual-hosted git repository.
hxd pushed a commit to branch rel/0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.11 by this push:
new 9e37145 [To rel/0.11] Fix: high CPU usage during the compraction
process (#2921)
9e37145 is described below
commit 9e371454b05f9a0e24fabb2c84f7ce690f14a882
Author: Steve Yurong Su <[email protected]>
AuthorDate: Wed Mar 31 16:21:18 2021 +0800
[To rel/0.11] Fix: high CPU usage during the compraction process (#2921)
* add the method getMeasurementChunkMetadataListMapIterator for compaction
module
* reduce CPU load in Compaction process
---
.../engine/compaction/utils/CompactionUtils.java | 334 ++++++++-----
.../db/engine/merge/task/MergeMultiChunkTask.java | 102 +++-
.../iotdb/db/engine/merge/MergeOverLapTest.java | 92 ++--
.../apache/iotdb/db/engine/merge/MergeTest.java | 71 ++-
.../tsfile/exception/TsFileRuntimeException.java | 2 +-
.../iotdb/tsfile/read/TsFileSequenceReader.java | 546 +++++++++++++--------
...easurementChunkMetadataListMapIteratorTest.java | 197 ++++++++
7 files changed, 969 insertions(+), 375 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
index 73bfcbe..3ca13fe 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
@@ -23,18 +23,22 @@ import static
org.apache.iotdb.db.utils.MergeUtils.writeTVPair;
import static org.apache.iotdb.db.utils.QueryUtils.modifyChunkMetaData;
import com.google.common.util.concurrent.RateLimiter;
+import java.io.File;
import java.io.IOException;
-import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.LinkedHashMap;
+import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
+import org.apache.commons.collections4.keyvalue.DefaultMapEntry;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.compaction.TsFileManagement;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.modification.ModificationFile;
@@ -61,8 +65,8 @@ import org.slf4j.LoggerFactory;
public class CompactionUtils {
private static final Logger logger =
LoggerFactory.getLogger(CompactionUtils.class);
- private static final int MERGE_PAGE_POINT_NUM =
IoTDBDescriptor.getInstance().getConfig()
- .getMergePagePointNumberThreshold();
+ private static final int MERGE_PAGE_POINT_NUM =
+
IoTDBDescriptor.getInstance().getConfig().getMergePagePointNumberThreshold();
private CompactionUtils() {
throw new IllegalStateException("Utility class");
@@ -70,17 +74,18 @@ public class CompactionUtils {
private static Pair<ChunkMetadata, Chunk> readByAppendMerge(
Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap,
- Map<String, List<Modification>> modificationCache, PartialPath
seriesPath,
+ Map<String, List<Modification>> modificationCache,
+ PartialPath seriesPath,
List<Modification> modifications)
throws IOException {
ChunkMetadata newChunkMetadata = null;
Chunk newChunk = null;
- for (Entry<TsFileSequenceReader, List<ChunkMetadata>> entry :
readerChunkMetadataMap
- .entrySet()) {
+ for (Entry<TsFileSequenceReader, List<ChunkMetadata>> entry :
+ readerChunkMetadataMap.entrySet()) {
TsFileSequenceReader reader = entry.getKey();
List<ChunkMetadata> chunkMetadataList = entry.getValue();
- modifyChunkMetaDataWithCache(reader, chunkMetadataList,
modificationCache, seriesPath,
- modifications);
+ modifyChunkMetaDataWithCache(
+ reader, chunkMetadataList, modificationCache, seriesPath,
modifications);
for (ChunkMetadata chunkMetadata : chunkMetadataList) {
Chunk chunk = reader.readMemChunk(chunkMetadata);
if (newChunkMetadata == null) {
@@ -96,22 +101,24 @@ public class CompactionUtils {
}
private static long readByDeserializeMerge(
- Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap,
long maxVersion,
- Map<Long, TimeValuePair> timeValuePairMap, Map<String,
List<Modification>> modificationCache,
- PartialPath seriesPath, List<Modification> modifications) throws
IOException {
- for (Entry<TsFileSequenceReader, List<ChunkMetadata>> entry :
readerChunkMetadataMap
- .entrySet()) {
+ Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap,
+ long maxVersion,
+ Map<Long, TimeValuePair> timeValuePairMap,
+ Map<String, List<Modification>> modificationCache,
+ PartialPath seriesPath,
+ List<Modification> modifications)
+ throws IOException {
+ for (Entry<TsFileSequenceReader, List<ChunkMetadata>> entry :
+ readerChunkMetadataMap.entrySet()) {
TsFileSequenceReader reader = entry.getKey();
List<ChunkMetadata> chunkMetadataList = entry.getValue();
- modifyChunkMetaDataWithCache(reader, chunkMetadataList,
modificationCache, seriesPath,
- modifications);
+ modifyChunkMetaDataWithCache(
+ reader, chunkMetadataList, modificationCache, seriesPath,
modifications);
for (ChunkMetadata chunkMetadata : chunkMetadataList) {
maxVersion = Math.max(chunkMetadata.getVersion(), maxVersion);
- IChunkReader chunkReader = new ChunkReaderByTimestamp(
- reader.readMemChunk(chunkMetadata));
+ IChunkReader chunkReader = new
ChunkReaderByTimestamp(reader.readMemChunk(chunkMetadata));
while (chunkReader.hasNextSatisfiedPage()) {
- IPointReader iPointReader = new BatchDataIterator(
- chunkReader.nextPageData());
+ IPointReader iPointReader = new
BatchDataIterator(chunkReader.nextPageData());
while (iPointReader.hasNextTimeValuePair()) {
TimeValuePair timeValuePair = iPointReader.nextTimeValuePair();
timeValuePairMap.put(timeValuePair.getTimestamp(), timeValuePair);
@@ -122,20 +129,29 @@ public class CompactionUtils {
return maxVersion;
}
- public static long writeByAppendMerge(long maxVersion, String device,
+ public static long writeByAppendMerge(
+ long maxVersion,
+ String device,
RateLimiter compactionWriteRateLimiter,
Entry<String, Map<TsFileSequenceReader, List<ChunkMetadata>>> entry,
- TsFileResource targetResource, RestorableTsFileIOWriter writer,
- Map<String, List<Modification>> modificationCache, List<Modification>
modifications)
+ TsFileResource targetResource,
+ RestorableTsFileIOWriter writer,
+ Map<String, List<Modification>> modificationCache,
+ List<Modification> modifications)
throws IOException, IllegalPathException {
- Pair<ChunkMetadata, Chunk> chunkPair = readByAppendMerge(entry.getValue(),
- modificationCache, new PartialPath(device, entry.getKey()),
modifications);
+ Pair<ChunkMetadata, Chunk> chunkPair =
+ readByAppendMerge(
+ entry.getValue(),
+ modificationCache,
+ new PartialPath(device, entry.getKey()),
+ modifications);
ChunkMetadata newChunkMetadata = chunkPair.left;
Chunk newChunk = chunkPair.right;
if (newChunkMetadata != null && newChunk != null) {
maxVersion = Math.max(newChunkMetadata.getVersion(), maxVersion);
// wait for limit write
- MergeManager.mergeRateLimiterAcquire(compactionWriteRateLimiter,
+ MergeManager.mergeRateLimiterAcquire(
+ compactionWriteRateLimiter,
(long) newChunk.getHeader().getDataSize() +
newChunk.getData().position());
writer.writeChunk(newChunk, newChunkMetadata);
targetResource.updateStartTime(device, newChunkMetadata.getStartTime());
@@ -144,16 +160,26 @@ public class CompactionUtils {
return maxVersion;
}
- public static long writeByDeserializeMerge(long maxVersion, String device,
+ public static long writeByDeserializeMerge(
+ long maxVersion,
+ String device,
RateLimiter compactionRateLimiter,
Entry<String, Map<TsFileSequenceReader, List<ChunkMetadata>>> entry,
- TsFileResource targetResource, RestorableTsFileIOWriter writer,
- Map<String, List<Modification>> modificationCache, List<Modification>
modifications)
+ TsFileResource targetResource,
+ RestorableTsFileIOWriter writer,
+ Map<String, List<Modification>> modificationCache,
+ List<Modification> modifications)
throws IOException, IllegalPathException {
Map<Long, TimeValuePair> timeValuePairMap = new TreeMap<>();
Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap =
entry.getValue();
- maxVersion = readByDeserializeMerge(readerChunkMetadataMap, maxVersion,
timeValuePairMap,
- modificationCache, new PartialPath(device, entry.getKey()),
modifications);
+ maxVersion =
+ readByDeserializeMerge(
+ readerChunkMetadataMap,
+ maxVersion,
+ timeValuePairMap,
+ modificationCache,
+ new PartialPath(device, entry.getKey()),
+ modifications);
boolean isChunkMetadataEmpty = true;
for (List<ChunkMetadata> chunkMetadataList :
readerChunkMetadataMap.values()) {
if (!chunkMetadataList.isEmpty()) {
@@ -166,8 +192,9 @@ public class CompactionUtils {
}
IChunkWriter chunkWriter;
try {
- chunkWriter = new ChunkWriterImpl(
- IoTDB.metaManager.getSeriesSchema(new PartialPath(device),
entry.getKey()));
+ chunkWriter =
+ new ChunkWriterImpl(
+ IoTDB.metaManager.getSeriesSchema(new PartialPath(device),
entry.getKey()));
} catch (MetadataException e) {
// this may caused in IT by restart
logger.error("{} get schema {} error,skip this sensor", device,
entry.getKey());
@@ -179,20 +206,20 @@ public class CompactionUtils {
targetResource.updateEndTime(device, timeValuePair.getTimestamp());
}
// wait for limit write
- MergeManager
- .mergeRateLimiterAcquire(compactionRateLimiter,
chunkWriter.getCurrentChunkSize());
+ MergeManager.mergeRateLimiterAcquire(compactionRateLimiter,
chunkWriter.getCurrentChunkSize());
chunkWriter.writeToFileWriter(writer);
return maxVersion;
}
- private static Set<String> getTsFileDevicesSet(List<TsFileResource>
subLevelResources,
- Map<String, TsFileSequenceReader> tsFileSequenceReaderMap, String
storageGroup)
+ private static Set<String> getTsFileDevicesSet(
+ List<TsFileResource> subLevelResources,
+ Map<String, TsFileSequenceReader> tsFileSequenceReaderMap,
+ String storageGroup)
throws IOException {
Set<String> tsFileDevicesSet = new HashSet<>();
for (TsFileResource levelResource : subLevelResources) {
- TsFileSequenceReader reader =
buildReaderFromTsFileResource(levelResource,
- tsFileSequenceReaderMap,
- storageGroup);
+ TsFileSequenceReader reader =
+ buildReaderFromTsFileResource(levelResource,
tsFileSequenceReaderMap, storageGroup);
if (reader == null) {
continue;
}
@@ -201,6 +228,15 @@ public class CompactionUtils {
return tsFileDevicesSet;
}
+ private static boolean hasNextChunkMetadataList(
+ Collection<Iterator<Map<String, List<ChunkMetadata>>>> iteratorSet) {
+ boolean hasNextChunkMetadataList = false;
+ for (Iterator<Map<String, List<ChunkMetadata>>> iterator : iteratorSet) {
+ hasNextChunkMetadataList = hasNextChunkMetadataList ||
iterator.hasNext();
+ }
+ return hasNextChunkMetadataList;
+ }
+
/**
* @param targetResource the target resource to be merged to
* @param tsFileResources the source resource to be merged
@@ -209,93 +245,150 @@ public class CompactionUtils {
* @param devices the devices to be skipped(used by recover)
*/
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
- public static void merge(TsFileResource targetResource,
- List<TsFileResource> tsFileResources, String storageGroup,
+ public static void merge(
+ TsFileResource targetResource,
+ List<TsFileResource> tsFileResources,
+ String storageGroup,
CompactionLogger compactionLogger,
- Set<String> devices, boolean sequence, List<Modification> modifications)
+ Set<String> devices,
+ boolean sequence,
+ List<Modification> modifications)
throws IOException, IllegalPathException {
RestorableTsFileIOWriter writer = new
RestorableTsFileIOWriter(targetResource.getTsFile());
Map<String, TsFileSequenceReader> tsFileSequenceReaderMap = new
HashMap<>();
Map<String, List<Modification>> modificationCache = new HashMap<>();
RateLimiter compactionWriteRateLimiter =
MergeManager.getINSTANCE().getMergeWriteRateLimiter();
- Set<String> tsFileDevicesMap = getTsFileDevicesSet(tsFileResources,
tsFileSequenceReaderMap,
- storageGroup);
+ Set<String> tsFileDevicesMap =
+ getTsFileDevicesSet(tsFileResources, tsFileSequenceReaderMap,
storageGroup);
for (String device : tsFileDevicesMap) {
if (devices.contains(device)) {
continue;
}
+ long maxVersion = Long.MIN_VALUE;
writer.startChunkGroup(device);
- // sort chunkMeta by measurement
- Map<String, Map<TsFileSequenceReader, List<ChunkMetadata>>>
measurementChunkMetadataMap = new HashMap<>();
- for (TsFileResource levelResource : tsFileResources) {
- TsFileSequenceReader reader =
buildReaderFromTsFileResource(levelResource,
- tsFileSequenceReaderMap, storageGroup);
+ Map<TsFileSequenceReader, Map<String, List<ChunkMetadata>>>
chunkMetadataListCacheForMerge =
+ new TreeMap<>(
+ (o1, o2) ->
+ TsFileManagement.compareFileName(
+ new File(o1.getFileName()), new File(o2.getFileName())));
+ Map<TsFileSequenceReader, Iterator<Map<String, List<ChunkMetadata>>>>
+ chunkMetadataListIteratorCache =
+ new TreeMap<>(
+ (o1, o2) ->
+ TsFileManagement.compareFileName(
+ new File(o1.getFileName()), new
File(o2.getFileName())));
+ for (TsFileResource tsFileResource : tsFileResources) {
+ TsFileSequenceReader reader =
+ buildReaderFromTsFileResource(tsFileResource,
tsFileSequenceReaderMap, storageGroup);
if (reader == null) {
- continue;
+ throw new IOException();
}
- Map<String, List<ChunkMetadata>> chunkMetadataMap = reader
- .readChunkMetadataInDevice(device);
- for (Entry<String, List<ChunkMetadata>> entry :
chunkMetadataMap.entrySet()) {
- for (ChunkMetadata chunkMetadata : entry.getValue()) {
- Map<TsFileSequenceReader, List<ChunkMetadata>>
readerChunkMetadataMap;
- String measurementUid = chunkMetadata.getMeasurementUid();
- if (measurementChunkMetadataMap.containsKey(measurementUid)) {
- readerChunkMetadataMap =
measurementChunkMetadataMap.get(measurementUid);
+ Iterator<Map<String, List<ChunkMetadata>>> iterator =
+ reader.getMeasurementChunkMetadataListMapIterator(device);
+ chunkMetadataListIteratorCache.put(reader, iterator);
+ chunkMetadataListCacheForMerge.put(reader, new TreeMap<>());
+ }
+ while
(hasNextChunkMetadataList(chunkMetadataListIteratorCache.values())) {
+ String lastSensor = null;
+ Set<String> allSensors = new HashSet<>();
+ for (Entry<TsFileSequenceReader, Map<String, List<ChunkMetadata>>>
+ chunkMetadataListCacheForMergeEntry :
chunkMetadataListCacheForMerge.entrySet()) {
+ TsFileSequenceReader reader =
chunkMetadataListCacheForMergeEntry.getKey();
+ Map<String, List<ChunkMetadata>> sensorChunkMetadataListMap =
+ chunkMetadataListCacheForMergeEntry.getValue();
+ if (sensorChunkMetadataListMap.size() <= 0) {
+ if (chunkMetadataListIteratorCache.get(reader).hasNext()) {
+ sensorChunkMetadataListMap =
chunkMetadataListIteratorCache.get(reader).next();
+ chunkMetadataListCacheForMerge.put(reader,
sensorChunkMetadataListMap);
} else {
- readerChunkMetadataMap = new LinkedHashMap<>();
+ continue;
}
- List<ChunkMetadata> chunkMetadataList;
- if (readerChunkMetadataMap.containsKey(reader)) {
- chunkMetadataList = readerChunkMetadataMap.get(reader);
- } else {
- chunkMetadataList = new ArrayList<>();
+ }
+ // get the min last sensor in the current chunkMetadata cache list
for merge
+ String maxSensor =
Collections.max(sensorChunkMetadataListMap.keySet());
+ if (lastSensor == null) {
+ lastSensor = maxSensor;
+ } else {
+ if (maxSensor.compareTo(lastSensor) < 0) {
+ lastSensor = maxSensor;
}
- chunkMetadataList.add(chunkMetadata);
- readerChunkMetadataMap.put(reader, chunkMetadataList);
- measurementChunkMetadataMap
- .put(chunkMetadata.getMeasurementUid(),
readerChunkMetadataMap);
}
+ // get all sensor used later
+ allSensors.addAll(sensorChunkMetadataListMap.keySet());
}
- }
- if (!sequence) {
- long maxVersion = Long.MIN_VALUE;
- for (Entry<String, Map<TsFileSequenceReader, List<ChunkMetadata>>>
entry : measurementChunkMetadataMap
- .entrySet()) {
- maxVersion = writeByDeserializeMerge(maxVersion, device,
compactionWriteRateLimiter,
- entry, targetResource, writer, modificationCache, modifications);
- }
- writer.endChunkGroup();
- writer.writeVersion(maxVersion);
- } else {
- long maxVersion = Long.MIN_VALUE;
- for (Entry<String, Map<TsFileSequenceReader, List<ChunkMetadata>>>
entry : measurementChunkMetadataMap
- .entrySet()) {
- Map<TsFileSequenceReader, List<ChunkMetadata>>
readerChunkMetadatasMap = entry.getValue();
- boolean isPageEnoughLarge = true;
- for (List<ChunkMetadata> chunkMetadatas :
readerChunkMetadatasMap.values()) {
- for (ChunkMetadata chunkMetadata : chunkMetadatas) {
- if (chunkMetadata.getNumOfPoints() < MERGE_PAGE_POINT_NUM) {
- isPageEnoughLarge = false;
- break;
+
+ for (String sensor : allSensors) {
+ if (sensor.compareTo(lastSensor) <= 0) {
+ Map<TsFileSequenceReader, List<ChunkMetadata>>
readerChunkMetadataListMap =
+ new TreeMap<>(
+ (o1, o2) ->
+ TsFileManagement.compareFileName(
+ new File(o1.getFileName()), new
File(o2.getFileName())));
+ // find all chunkMetadata of a sensor
+ for (Entry<TsFileSequenceReader, Map<String, List<ChunkMetadata>>>
+ chunkMetadataListCacheForMergeEntry :
chunkMetadataListCacheForMerge.entrySet()) {
+ TsFileSequenceReader reader =
chunkMetadataListCacheForMergeEntry.getKey();
+ Map<String, List<ChunkMetadata>> sensorChunkMetadataListMap =
+ chunkMetadataListCacheForMergeEntry.getValue();
+ if (sensorChunkMetadataListMap.containsKey(sensor)) {
+ readerChunkMetadataListMap.put(reader,
sensorChunkMetadataListMap.get(sensor));
+ sensorChunkMetadataListMap.remove(sensor);
+ }
+ }
+ Entry<String, Map<TsFileSequenceReader, List<ChunkMetadata>>>
+ sensorReaderChunkMetadataListEntry =
+ new DefaultMapEntry<>(sensor, readerChunkMetadataListMap);
+ if (!sequence) {
+ writeByDeserializeMerge(
+ maxVersion,
+ device,
+ compactionWriteRateLimiter,
+ sensorReaderChunkMetadataListEntry,
+ targetResource,
+ writer,
+ modificationCache,
+ modifications);
+ } else {
+ boolean isPageEnoughLarge = true;
+ for (List<ChunkMetadata> chunkMetadatas :
readerChunkMetadataListMap.values()) {
+ for (ChunkMetadata chunkMetadata : chunkMetadatas) {
+ if (chunkMetadata.getNumOfPoints() < MERGE_PAGE_POINT_NUM) {
+ isPageEnoughLarge = false;
+ break;
+ }
+ }
+ }
+ if (isPageEnoughLarge) {
+ logger.debug("{} [Compaction] page enough large, use append
merge", storageGroup);
+ // append page in chunks, so we do not have to deserialize a
chunk
+ writeByAppendMerge(
+ maxVersion,
+ device,
+ compactionWriteRateLimiter,
+ sensorReaderChunkMetadataListEntry,
+ targetResource,
+ writer,
+ modificationCache,
+ modifications);
+ } else {
+ logger.debug("{} [Compaction] page too small, use deserialize
merge", storageGroup);
+ // we have to deserialize chunks to merge pages
+ writeByDeserializeMerge(
+ maxVersion,
+ device,
+ compactionWriteRateLimiter,
+ sensorReaderChunkMetadataListEntry,
+ targetResource,
+ writer,
+ modificationCache,
+ modifications);
}
}
- }
- if (isPageEnoughLarge) {
- logger.debug("{} [Compaction] page enough large, use append
merge", storageGroup);
- // append page in chunks, so we do not have to deserialize a chunk
- maxVersion = writeByAppendMerge(maxVersion, device,
compactionWriteRateLimiter,
- entry, targetResource, writer, modificationCache,
modifications);
- } else {
- logger
- .debug("{} [Compaction] page too small, use deserialize
merge", storageGroup);
- // we have to deserialize chunks to merge pages
- maxVersion = writeByDeserializeMerge(maxVersion, device,
compactionWriteRateLimiter,
- entry, targetResource, writer, modificationCache,
modifications);
}
}
- writer.endChunkGroup();
- writer.writeVersion(maxVersion);
}
+ writer.endChunkGroup();
+ writer.writeVersion(maxVersion);
if (compactionLogger != null) {
compactionLogger.logDevice(device, writer.getPos());
}
@@ -314,9 +407,12 @@ public class CompactionUtils {
targetResource.close();
}
- private static TsFileSequenceReader
buildReaderFromTsFileResource(TsFileResource levelResource,
- Map<String, TsFileSequenceReader> tsFileSequenceReaderMap, String
storageGroup) {
- return
tsFileSequenceReaderMap.computeIfAbsent(levelResource.getTsFile().getAbsolutePath(),
+ private static TsFileSequenceReader buildReaderFromTsFileResource(
+ TsFileResource levelResource,
+ Map<String, TsFileSequenceReader> tsFileSequenceReaderMap,
+ String storageGroup) {
+ return tsFileSequenceReaderMap.computeIfAbsent(
+ levelResource.getTsFile().getAbsolutePath(),
path -> {
try {
if (levelResource.getTsFile().exists()) {
@@ -328,20 +424,26 @@ public class CompactionUtils {
} catch (IOException e) {
logger.error(
"Storage group {}, flush recover meets error. reader create
failed.",
- storageGroup, e);
+ storageGroup,
+ e);
return null;
}
});
}
- private static void modifyChunkMetaDataWithCache(TsFileSequenceReader reader,
- List<ChunkMetadata> chunkMetadataList, Map<String, List<Modification>>
modificationCache,
- PartialPath seriesPath, List<Modification> usedModifications) {
+ private static void modifyChunkMetaDataWithCache(
+ TsFileSequenceReader reader,
+ List<ChunkMetadata> chunkMetadataList,
+ Map<String, List<Modification>> modificationCache,
+ PartialPath seriesPath,
+ List<Modification> usedModifications) {
List<Modification> modifications =
- modificationCache.computeIfAbsent(reader.getFileName(),
- fileName -> new LinkedList<>(
- new ModificationFile(fileName + ModificationFile.FILE_SUFFIX)
- .getModifications()));
+ modificationCache.computeIfAbsent(
+ reader.getFileName(),
+ fileName ->
+ new LinkedList<>(
+ new ModificationFile(fileName +
ModificationFile.FILE_SUFFIX)
+ .getModifications()));
List<Modification> seriesModifications = new LinkedList<>();
for (Modification modification : modifications) {
if (modification.getPath().matchFullPath(seriesPath)) {
@@ -351,4 +453,4 @@ public class CompactionUtils {
}
modifyChunkMetaData(chunkMetadataList, seriesModifications);
}
-}
\ No newline at end of file
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java
b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java
index 8be0bfe..c5fd3fd 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java
@@ -23,16 +23,22 @@ import static
org.apache.iotdb.db.utils.MergeUtils.writeBatchPoint;
import static org.apache.iotdb.db.utils.MergeUtils.writeTVPair;
import static org.apache.iotdb.db.utils.QueryUtils.modifyChunkMetaData;
+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 java.util.Map.Entry;
import java.util.PriorityQueue;
+import java.util.TreeMap;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.compaction.TsFileManagement;
import org.apache.iotdb.db.engine.merge.manage.MergeContext;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
@@ -81,6 +87,20 @@ public class MergeMultiChunkTask {
private int concurrentMergeSeriesNum;
private List<PartialPath> currMergingPaths = new ArrayList<>();
+ // need to be cleared every device
+ private final Map<TsFileSequenceReader, Iterator<Map<String,
List<ChunkMetadata>>>>
+ measurementChunkMetadataListMapIteratorCache =
+ new TreeMap<>(
+ (o1, o2) ->
+ TsFileManagement.compareFileName(
+ new File(o1.getFileName()), new File(o2.getFileName())));
+ // need to be cleared every device
+ private final Map<TsFileSequenceReader, Map<String, List<ChunkMetadata>>>
+ chunkMetadataListCacheForMerge =
+ new TreeMap<>(
+ (o1, o2) ->
+ TsFileManagement.compareFileName(
+ new File(o1.getFileName()), new File(o2.getFileName())));
private String storageGroupName;
@@ -127,6 +147,8 @@ public class MergeMultiChunkTask {
}
mergedSeriesCnt += currMergingPaths.size();
logMergeProgress();
+ measurementChunkMetadataListMapIteratorCache.clear();
+ chunkMetadataListCacheForMerge.clear();
}
}
if (logger.isInfoEnabled()) {
@@ -197,20 +219,78 @@ public class MergeMultiChunkTask {
TsFileSequenceReader fileSequenceReader =
resource.getFileReader(currTsFile);
List<Modification>[] modifications = new List[currMergingPaths.size()];
List<ChunkMetadata>[] seqChunkMeta = new List[currMergingPaths.size()];
- for (int i = 0; i < currMergingPaths.size(); i++) {
- modifications[i] = resource.getModifications(currTsFile,
currMergingPaths.get(i));
- seqChunkMeta[i] = resource.queryChunkMetadata(currMergingPaths.get(i),
currTsFile);
- modifyChunkMetaData(seqChunkMeta[i], modifications[i]);
- for (ChunkMetadata chunkMetadata : seqChunkMeta[i]) {
- resource.updateStartTime(currTsFile, deviceId,
chunkMetadata.getStartTime());
- resource.updateEndTime(currTsFile, deviceId,
chunkMetadata.getEndTime());
+ Iterator<Map<String, List<ChunkMetadata>>>
measurementChunkMetadataListMapIterator =
+ measurementChunkMetadataListMapIteratorCache.computeIfAbsent(
+ fileSequenceReader,
+ (tsFileSequenceReader -> {
+ try {
+ return
tsFileSequenceReader.getMeasurementChunkMetadataListMapIterator(deviceId);
+ } catch (IOException e) {
+ logger.error(
+ "unseq compaction task {},
getMeasurementChunkMetadataListMapIterator meets error. iterator create
failed.",
+ taskName,
+ e);
+ return null;
+ }
+ }));
+ if (measurementChunkMetadataListMapIterator == null) {
+ return;
+ }
+
+ String lastSensor = currMergingPaths.get(currMergingPaths.size() -
1).getMeasurement();
+ String currSensor = null;
+ Map<String, List<ChunkMetadata>> measurementChunkMetadataListMap = new
TreeMap<>();
+ // find all sensor to merge in order, if exceed, then break
+ while (currSensor == null || currSensor.compareTo(lastSensor) < 0) {
+ measurementChunkMetadataListMap =
+ chunkMetadataListCacheForMerge.computeIfAbsent(
+ fileSequenceReader, tsFileSequenceReader -> new TreeMap<>());
+ // if empty, get measurementChunkMetadataList block to use later
+ if (measurementChunkMetadataListMap.isEmpty()) {
+ // if do not have more sensor, just break
+ if (measurementChunkMetadataListMapIterator.hasNext()) {
+
measurementChunkMetadataListMap.putAll(measurementChunkMetadataListMapIterator.next());
+ } else {
+ break;
+ }
}
- if (Thread.interrupted()) {
- Thread.currentThread().interrupt();
- return;
+ Iterator<Entry<String, List<ChunkMetadata>>>
measurementChunkMetadataListEntryIterator =
+ measurementChunkMetadataListMap.entrySet().iterator();
+ while (measurementChunkMetadataListEntryIterator.hasNext()) {
+ Entry<String, List<ChunkMetadata>> measurementChunkMetadataListEntry =
+ measurementChunkMetadataListEntryIterator.next();
+ currSensor = measurementChunkMetadataListEntry.getKey();
+
+ // fill modifications and seqChunkMetas to be used later
+ for (int i = 0; i < currMergingPaths.size(); i++) {
+ if (currMergingPaths.get(i).getMeasurement().equals(currSensor)) {
+ modifications[i] = resource.getModifications(currTsFile,
currMergingPaths.get(i));
+ seqChunkMeta[i] = measurementChunkMetadataListEntry.getValue();
+ modifyChunkMetaData(seqChunkMeta[i], modifications[i]);
+ for (ChunkMetadata chunkMetadata : seqChunkMeta[i]) {
+ resource.updateStartTime(currTsFile, deviceId,
chunkMetadata.getStartTime());
+ resource.updateEndTime(currTsFile, deviceId,
chunkMetadata.getEndTime());
+ }
+
+ if (Thread.interrupted()) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ break;
+ }
+ }
+
+ // current sensor larger than last needed sensor, just break out to
outer loop
+ if (currSensor.compareTo(lastSensor) > 0) {
+ break;
+ } else {
+ measurementChunkMetadataListEntryIterator.remove();
+ }
}
}
+ // update measurementChunkMetadataListMap
+ chunkMetadataListCacheForMerge.put(fileSequenceReader,
measurementChunkMetadataListMap);
List<Integer> unskippedPathIndices = filterNoDataPaths(seqChunkMeta,
seqFileIdx);
if (unskippedPathIndices.isEmpty()) {
@@ -276,7 +356,7 @@ public class MergeMultiChunkTask {
int idx = 0;
for (int i = 0; i < currMergingPaths.size(); i++) {
chunkIdxHeaps[idx % mergeChunkSubTaskNum].add(i);
- if (seqChunkMeta[i].isEmpty()) {
+ if (seqChunkMeta[i] == null || seqChunkMeta[i].isEmpty()) {
continue;
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeOverLapTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeOverLapTest.java
index d841fd0..ef28256 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeOverLapTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeOverLapTest.java
@@ -56,8 +56,7 @@ public class MergeOverLapTest extends MergeTest {
private File tempSGDir;
@Before
- public void setUp()
- throws IOException, WriteProcessException, MetadataException {
+ public void setUp() throws IOException, WriteProcessException,
MetadataException {
ptNum = 1000;
super.setUp();
tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
@@ -73,10 +72,15 @@ public class MergeOverLapTest extends MergeTest {
@Override
void prepareFiles(int seqFileNum, int unseqFileNum) throws IOException,
WriteProcessException {
for (int i = 0; i < seqFileNum; i++) {
- File file = new File(TestConstant.BASE_OUTPUT_PATH.concat(
- i + "seq" + IoTDBConstant.FILE_NAME_SEPARATOR + i +
IoTDBConstant.FILE_NAME_SEPARATOR
- + i + IoTDBConstant.FILE_NAME_SEPARATOR + 0
- + ".tsfile"));
+ File file =
+ new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ i
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + i
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 0
+ + ".tsfile"));
TsFileResource tsFileResource = new TsFileResource(file);
tsFileResource.setClosed(true);
tsFileResource.setHistoricalVersions(Collections.singleton((long) i));
@@ -84,22 +88,30 @@ public class MergeOverLapTest extends MergeTest {
prepareFile(tsFileResource, i * ptNum, ptNum, 0);
}
for (int i = 0; i < unseqFileNum; i++) {
- File file = new File(TestConstant.BASE_OUTPUT_PATH.concat(
- i + "unseq" + IoTDBConstant.FILE_NAME_SEPARATOR + i
- + IoTDBConstant.FILE_NAME_SEPARATOR
- + i + IoTDBConstant.FILE_NAME_SEPARATOR + 0
- + ".tsfile"));
+ File file =
+ new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ (10000 + i)
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + (10000 + i)
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 0
+ + ".tsfile"));
TsFileResource tsFileResource = new TsFileResource(file);
tsFileResource.setClosed(true);
tsFileResource.setHistoricalVersions(Collections.singleton((long) (i +
seqFileNum)));
unseqResources.add(tsFileResource);
prepareUnseqFile(tsFileResource, i * ptNum, ptNum * (i + 1) /
unseqFileNum, 10000);
}
- File file = new File(TestConstant.BASE_OUTPUT_PATH.concat(
- unseqFileNum + "unseq" + IoTDBConstant.FILE_NAME_SEPARATOR +
unseqFileNum
- + IoTDBConstant.FILE_NAME_SEPARATOR + unseqFileNum
- + IoTDBConstant.FILE_NAME_SEPARATOR + 0
- + ".tsfile"));
+ File file =
+ new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ unseqFileNum
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + unseqFileNum
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 0
+ + ".tsfile"));
TsFileResource tsFileResource = new TsFileResource(file);
tsFileResource.setClosed(true);
tsFileResource.setHistoricalVersions(Collections.singleton((long)
(seqFileNum + unseqFileNum)));
@@ -107,8 +119,8 @@ public class MergeOverLapTest extends MergeTest {
prepareUnseqFile(tsFileResource, 0, ptNum * unseqFileNum, 20000);
}
- private void prepareUnseqFile(TsFileResource tsFileResource, long
timeOffset, long ptNum,
- long valueOffset)
+ private void prepareUnseqFile(
+ TsFileResource tsFileResource, long timeOffset, long ptNum, long
valueOffset)
throws IOException, WriteProcessException {
TsFileWriter fileWriter = new TsFileWriter(tsFileResource.getTsFile());
for (String deviceId : deviceIds) {
@@ -121,8 +133,11 @@ public class MergeOverLapTest extends MergeTest {
for (int j = 0; j < deviceNum; j++) {
TSRecord record = new TSRecord(i, deviceIds[j]);
for (int k = 0; k < measurementNum; k++) {
-
record.addTuple(DataPoint.getDataPoint(measurementSchemas[k].getType(),
- measurementSchemas[k].getMeasurementId(), String.valueOf(i +
valueOffset)));
+ record.addTuple(
+ DataPoint.getDataPoint(
+ measurementSchemas[k].getType(),
+ measurementSchemas[k].getMeasurementId(),
+ String.valueOf(i + valueOffset)));
}
fileWriter.write(record);
tsFileResource.updateStartTime(deviceIds[j], i);
@@ -133,8 +148,11 @@ public class MergeOverLapTest extends MergeTest {
for (int j = 0; j < deviceNum; j++) {
TSRecord record = new TSRecord(i, deviceIds[j]);
for (int k = 0; k < measurementNum; k++) {
-
record.addTuple(DataPoint.getDataPoint(measurementSchemas[k].getType(),
- measurementSchemas[k].getMeasurementId(), String.valueOf(i +
valueOffset)));
+ record.addTuple(
+ DataPoint.getDataPoint(
+ measurementSchemas[k].getType(),
+ measurementSchemas[k].getMeasurementId(),
+ String.valueOf(i + valueOffset)));
}
fileWriter.write(record);
tsFileResource.updateStartTime(deviceIds[j], i);
@@ -151,18 +169,34 @@ public class MergeOverLapTest extends MergeTest {
@Test
public void testFullMerge() throws Exception {
MergeTask mergeTask =
- new MergeTask(new MergeResource(seqResources, unseqResources),
tempSGDir.getPath(),
- (k, v, l) -> {
- }, "test",
- true, 1, MERGE_TEST_SG);
+ new MergeTask(
+ new MergeResource(seqResources, unseqResources),
+ tempSGDir.getPath(),
+ (k, v, l) -> {},
+ "test",
+ true,
+ 1,
+ MERGE_TEST_SG);
mergeTask.call();
QueryContext context = new QueryContext();
- PartialPath path = new PartialPath(deviceIds[0] +
TsFileConstant.PATH_SEPARATOR + measurementSchemas[0].getMeasurementId());
+ PartialPath path =
+ new PartialPath(
+ deviceIds[0]
+ + TsFileConstant.PATH_SEPARATOR
+ + measurementSchemas[0].getMeasurementId());
List<TsFileResource> resources = new ArrayList<>();
resources.add(seqResources.get(0));
- IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(), context,
- resources, new ArrayList<>(), null, null, true);
+ IBatchReader tsFilesReader =
+ new SeriesRawDataBatchReader(
+ path,
+ measurementSchemas[0].getType(),
+ context,
+ resources,
+ new ArrayList<>(),
+ null,
+ null,
+ true);
int cnt = 0;
try {
while (tsFilesReader.hasNextBatch()) {
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
index 4416022..123674a 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTest.java
@@ -88,7 +88,9 @@ abstract class MergeTest {
removeFiles();
seqResources.clear();
unseqResources.clear();
-
IoTDBDescriptor.getInstance().getConfig().setMergeChunkPointNumberThreshold(prevMergeChunkThreshold);
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setMergeChunkPointNumberThreshold(prevMergeChunkThreshold);
ChunkCache.getInstance().clear();
ChunkMetadataCache.getInstance().clear();
TimeSeriesMetadataCache.getInstance().clear();
@@ -100,8 +102,9 @@ abstract class MergeTest {
private void prepareSeries() throws MetadataException, MetadataException {
measurementSchemas = new MeasurementSchema[measurementNum];
for (int i = 0; i < measurementNum; i++) {
- measurementSchemas[i] = new MeasurementSchema("sensor" + i,
TSDataType.DOUBLE,
- encoding, CompressionType.UNCOMPRESSED);
+ measurementSchemas[i] =
+ new MeasurementSchema(
+ "sensor" + i, TSDataType.DOUBLE, encoding,
CompressionType.UNCOMPRESSED);
}
deviceIds = new String[deviceNum];
for (int i = 0; i < deviceNum; i++) {
@@ -112,20 +115,26 @@ abstract class MergeTest {
for (MeasurementSchema measurementSchema : measurementSchemas) {
PartialPath devicePath = new PartialPath(device);
IoTDB.metaManager.createTimeseries(
- devicePath.concatNode(measurementSchema.getMeasurementId()),
measurementSchema
- .getType(), measurementSchema.getEncodingType(),
measurementSchema.getCompressor(),
+ devicePath.concatNode(measurementSchema.getMeasurementId()),
+ measurementSchema.getType(),
+ measurementSchema.getEncodingType(),
+ measurementSchema.getCompressor(),
Collections.emptyMap());
}
}
}
- void prepareFiles(int seqFileNum, int unseqFileNum)
- throws IOException, WriteProcessException {
+ void prepareFiles(int seqFileNum, int unseqFileNum) throws IOException,
WriteProcessException {
for (int i = 0; i < seqFileNum; i++) {
- File file = new File(TestConstant.BASE_OUTPUT_PATH.concat(
- i + "seq" + IoTDBConstant.FILE_NAME_SEPARATOR + i +
IoTDBConstant.FILE_NAME_SEPARATOR
- + i + IoTDBConstant.FILE_NAME_SEPARATOR + 0
- + ".tsfile"));
+ File file =
+ new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ i
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + i
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 0
+ + ".tsfile"));
TsFileResource tsFileResource = new TsFileResource(file);
tsFileResource.setClosed(true);
tsFileResource.setHistoricalVersions(Collections.singleton((long) i));
@@ -133,11 +142,15 @@ abstract class MergeTest {
prepareFile(tsFileResource, i * ptNum, ptNum, 0);
}
for (int i = 0; i < unseqFileNum; i++) {
- File file = new File(TestConstant.BASE_OUTPUT_PATH.concat(
- i + "unseq" + IoTDBConstant.FILE_NAME_SEPARATOR
- + i + IoTDBConstant.FILE_NAME_SEPARATOR
- + i + IoTDBConstant.FILE_NAME_SEPARATOR + 0
- + ".tsfile"));
+ File file =
+ new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ (10000 + i)
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + (10000 + i)
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 0
+ + ".tsfile"));
TsFileResource tsFileResource = new TsFileResource(file);
tsFileResource.setClosed(true);
tsFileResource.setHistoricalVersions(Collections.singleton((long) (i +
seqFileNum)));
@@ -145,10 +158,15 @@ abstract class MergeTest {
prepareFile(tsFileResource, i * ptNum, ptNum * (i + 1) / unseqFileNum,
10000);
}
- File file = new File(TestConstant.BASE_OUTPUT_PATH
- .concat(unseqFileNum + "unseq" + IoTDBConstant.FILE_NAME_SEPARATOR +
unseqFileNum
- + IoTDBConstant.FILE_NAME_SEPARATOR + unseqFileNum
- + IoTDBConstant.FILE_NAME_SEPARATOR + 0 + ".tsfile"));
+ File file =
+ new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ unseqFileNum
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + unseqFileNum
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 0
+ + ".tsfile"));
TsFileResource tsFileResource = new TsFileResource(file);
tsFileResource.setClosed(true);
tsFileResource.setHistoricalVersions(Collections.singleton((long)
(seqFileNum + unseqFileNum)));
@@ -168,8 +186,7 @@ abstract class MergeTest {
FileReaderManager.getInstance().stop();
}
- void prepareFile(TsFileResource tsFileResource, long timeOffset, long ptNum,
- long valueOffset)
+ void prepareFile(TsFileResource tsFileResource, long timeOffset, long ptNum,
long valueOffset)
throws IOException, WriteProcessException {
TsFileWriter fileWriter = new TsFileWriter(tsFileResource.getTsFile());
for (String deviceId : deviceIds) {
@@ -182,8 +199,11 @@ abstract class MergeTest {
for (int j = 0; j < deviceNum; j++) {
TSRecord record = new TSRecord(i, deviceIds[j]);
for (int k = 0; k < measurementNum; k++) {
-
record.addTuple(DataPoint.getDataPoint(measurementSchemas[k].getType(),
- measurementSchemas[k].getMeasurementId(), String.valueOf(i +
valueOffset)));
+ record.addTuple(
+ DataPoint.getDataPoint(
+ measurementSchemas[k].getType(),
+ measurementSchemas[k].getMeasurementId(),
+ String.valueOf(i + valueOffset)));
}
fileWriter.write(record);
tsFileResource.updateStartTime(deviceIds[j], i);
@@ -195,5 +215,4 @@ abstract class MergeTest {
}
fileWriter.close();
}
-
-}
\ No newline at end of file
+}
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/exception/TsFileRuntimeException.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/exception/TsFileRuntimeException.java
index 812ea87..666d156 100644
---
a/tsfile/src/main/java/org/apache/iotdb/tsfile/exception/TsFileRuntimeException.java
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/exception/TsFileRuntimeException.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.tsfile.exception;
* This Exception is the parent class for all runtime exceptions.<br>
* This Exception extends super class {@link java.lang.RuntimeException}
*/
-public abstract class TsFileRuntimeException extends RuntimeException {
+public class TsFileRuntimeException extends RuntimeException {
private static final long serialVersionUID = 6455048223316780984L;
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
index 764a78f..8bb293e 100644
---
a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
@@ -18,27 +18,10 @@
*/
package org.apache.iotdb.tsfile.read;
-import java.io.File;
-import java.io.IOException;
-import java.nio.BufferOverflowException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.stream.Collectors;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.compress.IUnCompressor;
+import org.apache.iotdb.tsfile.exception.TsFileRuntimeException;
import org.apache.iotdb.tsfile.file.MetaMarker;
import org.apache.iotdb.tsfile.file.footer.ChunkGroupFooter;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
@@ -63,9 +46,36 @@ import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.apache.iotdb.tsfile.utils.VersionUtils;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.NoSuchElementException;
+import java.util.Queue;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static
org.apache.iotdb.tsfile.file.metadata.enums.MetadataIndexNodeType.INTERNAL_DEVICE;
+
public class TsFileSequenceReader implements AutoCloseable {
private static final Logger logger =
LoggerFactory.getLogger(TsFileSequenceReader.class);
@@ -79,15 +89,16 @@ public class TsFileSequenceReader implements AutoCloseable {
private int totalChunkNum;
private TsFileMetadata tsFileMetaData;
// device -> measurement -> TimeseriesMetadata
- private Map<String, Map<String, TimeseriesMetadata>> cachedDeviceMetadata =
new ConcurrentHashMap<>();
+ private Map<String, Map<String, TimeseriesMetadata>> cachedDeviceMetadata =
+ new ConcurrentHashMap<>();
private static final ReadWriteLock cacheLock = new ReentrantReadWriteLock();
private boolean cacheDeviceMetadata;
/**
* Create a file reader of the given file. The reader will read the tail of
the file to get the
- * file metadata size.Then the reader will skip the first
TSFileConfig.MAGIC_STRING.getBytes().length
- * + TSFileConfig.NUMBER_VERSION.getBytes().length bytes of the file for
preparing reading real
- * data.
+ * file metadata size.Then the reader will skip the first
+ * TSFileConfig.MAGIC_STRING.getBytes().length +
TSFileConfig.NUMBER_VERSION.getBytes().length
+ * bytes of the file for preparing reading real data.
*
* @param file the data file
* @throws IOException If some I/O error occurs
@@ -99,7 +110,7 @@ public class TsFileSequenceReader implements AutoCloseable {
/**
* construct function for TsFileSequenceReader.
*
- * @param file -given file name
+ * @param file -given file name
* @param loadMetadataSize -whether load meta data size
*/
public TsFileSequenceReader(String file, boolean loadMetadataSize) throws
IOException {
@@ -127,9 +138,9 @@ public class TsFileSequenceReader implements AutoCloseable {
/**
* Create a file reader of the given file. The reader will read the tail of
the file to get the
- * file metadata size.Then the reader will skip the first
TSFileConfig.MAGIC_STRING.getBytes().length
- * + TSFileConfig.NUMBER_VERSION.getBytes().length bytes of the file for
preparing reading real
- * data.
+ * file metadata size.Then the reader will skip the first
+ * TSFileConfig.MAGIC_STRING.getBytes().length +
TSFileConfig.NUMBER_VERSION.getBytes().length
+ * bytes of the file for preparing reading real data.
*
* @param input given input
*/
@@ -140,7 +151,7 @@ public class TsFileSequenceReader implements AutoCloseable {
/**
* construct function for TsFileSequenceReader.
*
- * @param input -given input
+ * @param input -given input
* @param loadMetadataSize -load meta data size
*/
public TsFileSequenceReader(TsFileInput input, boolean loadMetadataSize)
throws IOException {
@@ -158,10 +169,10 @@ public class TsFileSequenceReader implements
AutoCloseable {
/**
* construct function for TsFileSequenceReader.
*
- * @param input the input of a tsfile. The current position
should be a markder and
- * then a chunk Header, rather than the magic number
- * @param fileMetadataPos the position of the file metadata in the
TsFileInput from the beginning
- * of the input to the current position
+ * @param input the input of a tsfile. The current position should be a
markder and then a chunk
+ * Header, rather than the magic number
+ * @param fileMetadataPos the position of the file metadata in the
TsFileInput from the beginning
+ * of the input to the current position
* @param fileMetadataSize the byte size of the file metadata in the input
*/
public TsFileSequenceReader(TsFileInput input, long fileMetadataPos, int
fileMetadataSize) {
@@ -173,13 +184,17 @@ public class TsFileSequenceReader implements
AutoCloseable {
public void loadMetadataSize() throws IOException {
ByteBuffer metadataSize = ByteBuffer.allocate(Integer.BYTES);
if (readTailMagic().equals(TSFileConfig.MAGIC_STRING)) {
- tsFileInput.read(metadataSize,
+ tsFileInput.read(
+ metadataSize,
tsFileInput.size() - TSFileConfig.MAGIC_STRING.getBytes().length -
Integer.BYTES);
metadataSize.flip();
// read file metadata size and position
fileMetadataSize = ReadWriteIOUtils.readInt(metadataSize);
- fileMetadataPos = tsFileInput.size() -
TSFileConfig.MAGIC_STRING.getBytes().length
- - Integer.BYTES - fileMetadataSize;
+ fileMetadataPos =
+ tsFileInput.size()
+ - TSFileConfig.MAGIC_STRING.getBytes().length
+ - Integer.BYTES
+ - fileMetadataSize;
}
}
@@ -191,45 +206,36 @@ public class TsFileSequenceReader implements
AutoCloseable {
return fileMetadataSize;
}
- /**
- * this function does not modify the position of the file reader.
- */
+ /** this function does not modify the position of the file reader. */
public String readTailMagic() throws IOException {
long totalSize = tsFileInput.size();
- ByteBuffer magicStringBytes = ByteBuffer
- .allocate(TSFileConfig.MAGIC_STRING.getBytes().length);
+ ByteBuffer magicStringBytes =
ByteBuffer.allocate(TSFileConfig.MAGIC_STRING.getBytes().length);
tsFileInput.read(magicStringBytes, totalSize -
TSFileConfig.MAGIC_STRING.getBytes().length);
magicStringBytes.flip();
return new String(magicStringBytes.array());
}
- /**
- * whether the file is a complete TsFile: only if the head magic and tail
magic string exists.
- */
+ /** whether the file is a complete TsFile: only if the head magic and tail
magic string exists. */
public boolean isComplete() throws IOException {
- return tsFileInput.size() >= TSFileConfig.MAGIC_STRING.getBytes().length *
2
- + TSFileConfig.VERSION_NUMBER.getBytes().length
- && (readTailMagic().equals(readHeadMagic()) || readTailMagic()
- .equals(TSFileConfig.VERSION_NUMBER_V1));
+ return tsFileInput.size()
+ >= TSFileConfig.MAGIC_STRING.getBytes().length * 2
+ + TSFileConfig.VERSION_NUMBER.getBytes().length
+ && (readTailMagic().equals(readHeadMagic())
+ || readTailMagic().equals(TSFileConfig.VERSION_NUMBER_V1));
}
- /**
- * this function does not modify the position of the file reader.
- */
+ /** this function does not modify the position of the file reader. */
public String readHeadMagic() throws IOException {
- ByteBuffer magicStringBytes = ByteBuffer
- .allocate(TSFileConfig.MAGIC_STRING.getBytes().length);
+ ByteBuffer magicStringBytes =
ByteBuffer.allocate(TSFileConfig.MAGIC_STRING.getBytes().length);
tsFileInput.read(magicStringBytes, 0);
magicStringBytes.flip();
return new String(magicStringBytes.array());
}
- /**
- * this function reads version number and checks compatibility of TsFile.
- */
+ /** this function reads version number and checks compatibility of TsFile. */
public String readVersionNumber() throws IOException {
- ByteBuffer versionNumberBytes = ByteBuffer
- .allocate(TSFileConfig.VERSION_NUMBER.getBytes().length);
+ ByteBuffer versionNumberBytes =
+ ByteBuffer.allocate(TSFileConfig.VERSION_NUMBER.getBytes().length);
tsFileInput.read(versionNumberBytes,
TSFileConfig.MAGIC_STRING.getBytes().length);
versionNumberBytes.flip();
return new String(versionNumberBytes.array());
@@ -243,8 +249,8 @@ public class TsFileSequenceReader implements AutoCloseable {
public TsFileMetadata readFileMetadata() throws IOException {
try {
if (tsFileMetaData == null) {
- tsFileMetaData = TsFileMetadata
- .deserializeFrom(readData(fileMetadataPos, fileMetadataSize));
+ tsFileMetaData =
+ TsFileMetadata.deserializeFrom(readData(fileMetadataPos,
fileMetadataSize));
}
} catch (BufferOverflowException e) {
logger.error("Something error happened while reading file metadata of
file {}", file, e);
@@ -312,8 +318,8 @@ public class TsFileSequenceReader implements AutoCloseable {
public TimeseriesMetadata readTimeseriesMetadata(Path path) throws
IOException {
readFileMetadata();
MetadataIndexNode deviceMetadataIndexNode =
tsFileMetaData.getMetadataIndex();
- Pair<MetadataIndexEntry, Long> metadataIndexPair = getMetadataAndEndOffset(
- deviceMetadataIndexNode, path.getDevice(),
MetadataIndexNodeType.INTERNAL_DEVICE, true);
+ Pair<MetadataIndexEntry, Long> metadataIndexPair =
+ getMetadataAndEndOffset(deviceMetadataIndexNode, path.getDevice(),
INTERNAL_DEVICE, true);
if (metadataIndexPair == null) {
throw new IOException("Device {" + path.getDevice() + "} is not in
tsFileMetaData");
}
@@ -323,12 +329,16 @@ public class TsFileSequenceReader implements
AutoCloseable {
try {
metadataIndexNode = MetadataIndexNode.deserializeFrom(buffer);
} catch (BufferOverflowException e) {
- logger.error("Something error happened while deserializing
MetadataIndexNode of file {}",
- file, e);
+ logger.error(
+ "Something error happened while deserializing MetadataIndexNode of
file {}", file, e);
throw e;
}
- metadataIndexPair = getMetadataAndEndOffset(metadataIndexNode,
- path.getMeasurement(), MetadataIndexNodeType.INTERNAL_MEASUREMENT,
false);
+ metadataIndexPair =
+ getMetadataAndEndOffset(
+ metadataIndexNode,
+ path.getMeasurement(),
+ MetadataIndexNodeType.INTERNAL_MEASUREMENT,
+ false);
}
if (metadataIndexPair == null) {
return null;
@@ -339,14 +349,14 @@ public class TsFileSequenceReader implements
AutoCloseable {
try {
timeseriesMetadataList.add(TimeseriesMetadata.deserializeFrom(buffer));
} catch (BufferOverflowException e) {
- logger.error("Something error happened while deserializing
TimeseriesMetadata of file {}",
- file, e);
+ logger.error(
+ "Something error happened while deserializing TimeseriesMetadata
of file {}", file, e);
throw e;
}
}
// return null if path does not exist in the TsFile
- int searchResult =
binarySearchInTimeseriesMetadataList(timeseriesMetadataList,
- path.getMeasurement());
+ int searchResult =
+ binarySearchInTimeseriesMetadataList(timeseriesMetadataList,
path.getMeasurement());
return searchResult >= 0 ? timeseriesMetadataList.get(searchResult) : null;
}
@@ -358,8 +368,8 @@ public class TsFileSequenceReader implements AutoCloseable {
throws IOException {
readFileMetadata();
MetadataIndexNode deviceMetadataIndexNode =
tsFileMetaData.getMetadataIndex();
- Pair<MetadataIndexEntry, Long> metadataIndexPair = getMetadataAndEndOffset(
- deviceMetadataIndexNode, path.getDevice(),
MetadataIndexNodeType.INTERNAL_DEVICE, true);
+ Pair<MetadataIndexEntry, Long> metadataIndexPair =
+ getMetadataAndEndOffset(deviceMetadataIndexNode, path.getDevice(),
INTERNAL_DEVICE, true);
if (metadataIndexPair == null) {
return null;
}
@@ -369,12 +379,16 @@ public class TsFileSequenceReader implements
AutoCloseable {
try {
metadataIndexNode = MetadataIndexNode.deserializeFrom(buffer);
} catch (BufferOverflowException e) {
- logger.error("Something error happened while deserializing
MetadataIndexNode of file {}",
- file, e);
+ logger.error(
+ "Something error happened while deserializing MetadataIndexNode of
file {}", file, e);
throw e;
}
- metadataIndexPair = getMetadataAndEndOffset(metadataIndexNode,
- path.getMeasurement(), MetadataIndexNodeType.INTERNAL_MEASUREMENT,
false);
+ metadataIndexPair =
+ getMetadataAndEndOffset(
+ metadataIndexNode,
+ path.getMeasurement(),
+ MetadataIndexNodeType.INTERNAL_MEASUREMENT,
+ false);
}
if (metadataIndexPair == null) {
return null;
@@ -386,8 +400,8 @@ public class TsFileSequenceReader implements AutoCloseable {
try {
timeseriesMetadata = TimeseriesMetadata.deserializeFrom(buffer);
} catch (BufferOverflowException e) {
- logger.error("Something error happened while deserializing
TimeseriesMetadata of file {}",
- file, e);
+ logger.error(
+ "Something error happened while deserializing TimeseriesMetadata
of file {}", file, e);
throw e;
}
if (allSensors.contains(timeseriesMetadata.getMeasurementId())) {
@@ -401,8 +415,8 @@ public class TsFileSequenceReader implements AutoCloseable {
throws IOException {
readFileMetadata();
MetadataIndexNode deviceMetadataIndexNode =
tsFileMetaData.getMetadataIndex();
- Pair<MetadataIndexEntry, Long> metadataIndexPair = getMetadataAndEndOffset(
- deviceMetadataIndexNode, device,
MetadataIndexNodeType.INTERNAL_DEVICE, false);
+ Pair<MetadataIndexEntry, Long> metadataIndexPair =
+ getMetadataAndEndOffset(deviceMetadataIndexNode, device,
INTERNAL_DEVICE, false);
if (metadataIndexPair == null) {
return Collections.emptyList();
}
@@ -421,24 +435,31 @@ public class TsFileSequenceReader implements
AutoCloseable {
try {
metadataIndexNode = MetadataIndexNode.deserializeFrom(buffer);
} catch (BufferOverflowException e) {
- logger.error("Something error happened while deserializing
MetadataIndexNode of file {}",
- file, e);
+ logger.error(
+ "Something error happened while deserializing MetadataIndexNode
of file {}", file, e);
throw e;
}
- measurementMetadataIndexPair =
getMetadataAndEndOffset(metadataIndexNode,
- measurementList.get(i),
MetadataIndexNodeType.INTERNAL_MEASUREMENT, false);
+ measurementMetadataIndexPair =
+ getMetadataAndEndOffset(
+ metadataIndexNode,
+ measurementList.get(i),
+ MetadataIndexNodeType.INTERNAL_MEASUREMENT,
+ false);
}
if (measurementMetadataIndexPair == null) {
return Collections.emptyList();
}
- buffer = readData(measurementMetadataIndexPair.left.getOffset(),
- measurementMetadataIndexPair.right);
+ buffer =
+ readData(
+ measurementMetadataIndexPair.left.getOffset(),
measurementMetadataIndexPair.right);
while (buffer.hasRemaining()) {
try {
timeseriesMetadataList.add(TimeseriesMetadata.deserializeFrom(buffer));
} catch (BufferOverflowException e) {
- logger.error("Something error happened while deserializing
TimeseriesMetadata of file {}",
- file, e);
+ logger.error(
+ "Something error happened while deserializing TimeseriesMetadata
of file {}",
+ file,
+ e);
throw e;
}
}
@@ -465,14 +486,16 @@ public class TsFileSequenceReader implements
AutoCloseable {
* number of queried measurements is too large. Attention: This method is
not used currently
*
* @param timeseriesMetadataList TimeseriesMetadata list, to store the result
- * @param type MetadataIndexNode type
- * @param metadataIndexPair <MetadataIndexEntry, offset> pair
- * @param measurements measurements to be queried
+ * @param type MetadataIndexNode type
+ * @param metadataIndexPair <MetadataIndexEntry, offset> pair
+ * @param measurements measurements to be queried
* @throws IOException io error
*/
private void traverseAndReadTimeseriesMetadataInOneDevice(
- List<TimeseriesMetadata> timeseriesMetadataList, MetadataIndexNodeType
type,
- Pair<MetadataIndexEntry, Long> metadataIndexPair, Set<String>
measurements)
+ List<TimeseriesMetadata> timeseriesMetadataList,
+ MetadataIndexNodeType type,
+ Pair<MetadataIndexEntry, Long> metadataIndexPair,
+ Set<String> measurements)
throws IOException {
ByteBuffer buffer = readData(metadataIndexPair.left.getOffset(),
metadataIndexPair.right);
switch (type) {
@@ -485,9 +508,11 @@ public class TsFileSequenceReader implements AutoCloseable
{
if (i != metadataIndexListSize - 1) {
endOffset = metadataIndexNode.getChildren().get(i + 1).getOffset();
}
- traverseAndReadTimeseriesMetadataInOneDevice(timeseriesMetadataList,
+ traverseAndReadTimeseriesMetadataInOneDevice(
+ timeseriesMetadataList,
metadataIndexNode.getNodeType(),
- new Pair<>(metadataIndexNode.getChildren().get(i), endOffset),
measurements);
+ new Pair<>(metadataIndexNode.getChildren().get(i), endOffset),
+ measurements);
}
break;
case LEAF_MEASUREMENT:
@@ -499,13 +524,15 @@ public class TsFileSequenceReader implements
AutoCloseable {
}
break;
default:
- throw new IOException("Failed to traverse and read TimeseriesMetadata
in device: " +
- metadataIndexPair.left.getName() + ". Wrong MetadataIndexEntry
type.");
+ throw new IOException(
+ "Failed to traverse and read TimeseriesMetadata in device: "
+ + metadataIndexPair.left.getName()
+ + ". Wrong MetadataIndexEntry type.");
}
}
- private int binarySearchInTimeseriesMetadataList(List<TimeseriesMetadata>
timeseriesMetadataList,
- String key) {
+ private int binarySearchInTimeseriesMetadataList(
+ List<TimeseriesMetadata> timeseriesMetadataList, String key) {
int low = 0;
int high = timeseriesMetadataList.size() - 1;
@@ -522,7 +549,7 @@ public class TsFileSequenceReader implements AutoCloseable {
return mid; // key found
}
}
- return -1; // key not found
+ return -1; // key not found
}
public List<String> getAllDevices() throws IOException {
@@ -549,8 +576,10 @@ public class TsFileSequenceReader implements AutoCloseable
{
MetadataIndexNode node = MetadataIndexNode.deserializeFrom(buffer);
if (node.getNodeType().equals(MetadataIndexNodeType.LEAF_DEVICE)) {
// if node in next level is LEAF_DEVICE, put all devices in node
entry into the set
-
deviceList.addAll(node.getChildren().stream().map(MetadataIndexEntry::getName).collect(
- Collectors.toList()));
+ deviceList.addAll(
+ node.getChildren().stream()
+ .map(MetadataIndexEntry::getName)
+ .collect(Collectors.toList()));
} else {
// keep traversing
deviceList.addAll(getAllDevices(node));
@@ -587,7 +616,8 @@ public class TsFileSequenceReader implements AutoCloseable {
Map<String, List<ChunkMetadata>> seriesMetadata = new HashMap<>();
while (buffer.hasRemaining()) {
ChunkMetadata chunkMetadata = ChunkMetadata.deserializeFrom(buffer);
- seriesMetadata.computeIfAbsent(chunkMetadata.getMeasurementUid(), key ->
new ArrayList<>())
+ seriesMetadata
+ .computeIfAbsent(chunkMetadata.getMeasurementUid(), key -> new
ArrayList<>())
.add(chunkMetadata);
}
@@ -619,14 +649,18 @@ public class TsFileSequenceReader implements
AutoCloseable {
/**
* Traverse the metadata index from MetadataIndexEntry to get
TimeseriesMetadatas
*
- * @param metadataIndex MetadataIndexEntry
- * @param buffer byte buffer
- * @param deviceId String
+ * @param metadataIndex MetadataIndexEntry
+ * @param buffer byte buffer
+ * @param deviceId String
* @param timeseriesMetadataMap map: deviceId -> timeseriesMetadata list
*/
- private void generateMetadataIndex(MetadataIndexEntry metadataIndex,
ByteBuffer buffer,
- String deviceId, MetadataIndexNodeType type,
- Map<String, List<TimeseriesMetadata>> timeseriesMetadataMap) throws
IOException {
+ private void generateMetadataIndex(
+ MetadataIndexEntry metadataIndex,
+ ByteBuffer buffer,
+ String deviceId,
+ MetadataIndexNodeType type,
+ Map<String, List<TimeseriesMetadata>> timeseriesMetadataMap)
+ throws IOException {
try {
switch (type) {
case INTERNAL_DEVICE:
@@ -640,10 +674,14 @@ public class TsFileSequenceReader implements
AutoCloseable {
if (i != metadataIndexListSize - 1) {
endOffset = metadataIndexNode.getChildren().get(i +
1).getOffset();
}
- ByteBuffer nextBuffer =
readData(metadataIndexNode.getChildren().get(i).getOffset(),
- endOffset);
- generateMetadataIndex(metadataIndexNode.getChildren().get(i),
nextBuffer, deviceId,
- metadataIndexNode.getNodeType(), timeseriesMetadataMap);
+ ByteBuffer nextBuffer =
+ readData(metadataIndexNode.getChildren().get(i).getOffset(),
endOffset);
+ generateMetadataIndex(
+ metadataIndexNode.getChildren().get(i),
+ nextBuffer,
+ deviceId,
+ metadataIndexNode.getNodeType(),
+ timeseriesMetadataMap);
}
break;
case LEAF_MEASUREMENT:
@@ -651,7 +689,8 @@ public class TsFileSequenceReader implements AutoCloseable {
while (buffer.hasRemaining()) {
timeseriesMetadataList.add(TimeseriesMetadata.deserializeFrom(buffer));
}
- timeseriesMetadataMap.computeIfAbsent(deviceId, k -> new
ArrayList<>())
+ timeseriesMetadataMap
+ .computeIfAbsent(deviceId, k -> new ArrayList<>())
.addAll(timeseriesMetadataList);
break;
}
@@ -675,23 +714,27 @@ public class TsFileSequenceReader implements
AutoCloseable {
endOffset = metadataIndexEntryList.get(i + 1).getOffset();
}
ByteBuffer buffer = readData(metadataIndexEntry.getOffset(), endOffset);
- generateMetadataIndex(metadataIndexEntry, buffer, null,
- metadataIndexNode.getNodeType(), timeseriesMetadataMap);
+ generateMetadataIndex(
+ metadataIndexEntry, buffer, null, metadataIndexNode.getNodeType(),
timeseriesMetadataMap);
}
return timeseriesMetadataMap;
}
private List<TimeseriesMetadata> getDeviceTimeseriesMetadata(String device)
throws IOException {
MetadataIndexNode metadataIndexNode = tsFileMetaData.getMetadataIndex();
- Pair<MetadataIndexEntry, Long> metadataIndexPair = getMetadataAndEndOffset(
- metadataIndexNode, device, MetadataIndexNodeType.INTERNAL_DEVICE,
true);
+ Pair<MetadataIndexEntry, Long> metadataIndexPair =
+ getMetadataAndEndOffset(metadataIndexNode, device, INTERNAL_DEVICE,
true);
if (metadataIndexPair == null) {
return Collections.emptyList();
}
ByteBuffer buffer = readData(metadataIndexPair.left.getOffset(),
metadataIndexPair.right);
Map<String, List<TimeseriesMetadata>> timeseriesMetadataMap = new
TreeMap<>();
- generateMetadataIndex(metadataIndexPair.left, buffer, device,
- MetadataIndexNodeType.INTERNAL_MEASUREMENT, timeseriesMetadataMap);
+ generateMetadataIndex(
+ metadataIndexPair.left,
+ buffer,
+ device,
+ MetadataIndexNodeType.INTERNAL_MEASUREMENT,
+ timeseriesMetadataMap);
List<TimeseriesMetadata> deviceTimeseriesMetadata = new ArrayList<>();
for (List<TimeseriesMetadata> timeseriesMetadataList :
timeseriesMetadataMap.values()) {
deviceTimeseriesMetadata.addAll(timeseriesMetadataList);
@@ -703,33 +746,32 @@ public class TsFileSequenceReader implements
AutoCloseable {
* Get target MetadataIndexEntry and its end offset
*
* @param metadataIndex given MetadataIndexNode
- * @param name target device / measurement name
- * @param type target MetadataIndexNodeType, either INTERNAL_DEVICE
or
- * INTERNAL_MEASUREMENT. When searching for a device
node, return when it is
- * not INTERNAL_DEVICE. Likewise, when searching for a
measurement node,
- * return when it is not INTERNAL_MEASUREMENT. This
works for the situation
- * when the index tree does NOT have the device level
and ONLY has the
- * measurement level.
- * @param exactSearch if is in exact search mode, return null when there
is no entry with name;
- * or else return the nearest MetadataIndexEntry before
it (for deeper
- * search)
+ * @param name target device / measurement name
+ * @param type target MetadataIndexNodeType, either INTERNAL_DEVICE or
INTERNAL_MEASUREMENT. When
+ * searching for a device node, return when it is not INTERNAL_DEVICE.
Likewise, when
+ * searching for a measurement node, return when it is not
INTERNAL_MEASUREMENT. This works
+ * for the situation when the index tree does NOT have the device level
and ONLY has the
+ * measurement level.
+ * @param exactSearch if is in exact search mode, return null when there is
no entry with name; or
+ * else return the nearest MetadataIndexEntry before it (for deeper
search)
* @return target MetadataIndexEntry, endOffset pair
*/
- private Pair<MetadataIndexEntry, Long>
getMetadataAndEndOffset(MetadataIndexNode metadataIndex,
- String name, MetadataIndexNodeType type, boolean exactSearch) throws
IOException {
+ private Pair<MetadataIndexEntry, Long> getMetadataAndEndOffset(
+ MetadataIndexNode metadataIndex, String name, MetadataIndexNodeType
type, boolean exactSearch)
+ throws IOException {
try {
if (!metadataIndex.getNodeType().equals(type)) {
return metadataIndex.getChildIndexEntry(name, exactSearch);
} else {
- Pair<MetadataIndexEntry, Long> childIndexEntry = metadataIndex
- .getChildIndexEntry(name, false);
+ Pair<MetadataIndexEntry, Long> childIndexEntry =
+ metadataIndex.getChildIndexEntry(name, false);
ByteBuffer buffer = readData(childIndexEntry.left.getOffset(),
childIndexEntry.right);
- return
getMetadataAndEndOffset(MetadataIndexNode.deserializeFrom(buffer), name, type,
- false);
+ return getMetadataAndEndOffset(
+ MetadataIndexNode.deserializeFrom(buffer), name, type, false);
}
} catch (BufferOverflowException e) {
- logger
- .error("Something error happened while deserializing MetadataIndex
of file {}", file, e);
+ logger.error(
+ "Something error happened while deserializing MetadataIndex of file
{}", file, e);
throw e;
}
}
@@ -748,7 +790,7 @@ public class TsFileSequenceReader implements AutoCloseable {
/**
* read data from current position of the input, and deserialize it to a
CHUNK_GROUP_FOOTER.
*
- * @param position the offset of the chunk group footer in the file
+ * @param position the offset of the chunk group footer in the file
* @param markerRead true if the offset does not contains the marker ,
otherwise false
* @return a CHUNK_GROUP_FOOTER
* @throws IOException io error
@@ -768,8 +810,8 @@ public class TsFileSequenceReader implements AutoCloseable {
}
/**
- * read data from current position of the input, and deserialize it to a
CHUNK_HEADER. <br> This
- * method is not threadsafe.
+ * read data from current position of the input, and deserialize it to a
CHUNK_HEADER. <br>
+ * This method is not threadsafe.
*
* @return a CHUNK_HEADER
* @throws IOException io error
@@ -781,9 +823,9 @@ public class TsFileSequenceReader implements AutoCloseable {
/**
* read the chunk's header.
*
- * @param position the file offset of this chunk's header
+ * @param position the file offset of this chunk's header
* @param chunkHeaderSize the size of chunk's header
- * @param markerRead true if the offset does not contains the marker ,
otherwise false
+ * @param markerRead true if the offset does not contains the marker ,
otherwise false
*/
private ChunkHeader readChunkHeader(long position, int chunkHeaderSize,
boolean markerRead)
throws IOException {
@@ -810,15 +852,16 @@ public class TsFileSequenceReader implements
AutoCloseable {
public Chunk readMemChunk(ChunkMetadata metaData) throws IOException {
int chunkHeadSize =
ChunkHeader.getSerializedSize(metaData.getMeasurementUid());
ChunkHeader header = readChunkHeader(metaData.getOffsetOfChunkHeader(),
chunkHeadSize, false);
- ByteBuffer buffer = readChunk(metaData.getOffsetOfChunkHeader() +
header.getSerializedSize(),
- header.getDataSize());
+ ByteBuffer buffer =
+ readChunk(
+ metaData.getOffsetOfChunkHeader() + header.getSerializedSize(),
header.getDataSize());
return new Chunk(header, buffer, metaData.getDeleteIntervalList());
}
/**
* read all Chunks of given device.
- * <p>
- * note that this method loads all the chunks into memory, so it needs to be
invoked carefully.
+ *
+ * <p>note that this method loads all the chunks into memory, so it needs to
be invoked carefully.
*
* @param device name
* @return measurement -> chunks list
@@ -875,15 +918,15 @@ public class TsFileSequenceReader implements
AutoCloseable {
ByteBuffer uncompressedBuffer =
ByteBuffer.allocate(header.getUncompressedSize());
if (type == CompressionType.UNCOMPRESSED) {
return buffer;
- }// FIXME if the buffer is not array-implemented.
- unCompressor.uncompress(buffer.array(), buffer.position(),
buffer.remaining(),
- uncompressedBuffer.array(),
- 0);
+ } // FIXME if the buffer is not array-implemented.
+ unCompressor.uncompress(
+ buffer.array(), buffer.position(), buffer.remaining(),
uncompressedBuffer.array(), 0);
return uncompressedBuffer;
}
/**
- * read one byte from the input. <br> this method is not thread safe
+ * read one byte from the input. <br>
+ * this method is not thread safe
*/
public byte readMarker() throws IOException {
markerBuffer.clear();
@@ -911,13 +954,13 @@ public class TsFileSequenceReader implements
AutoCloseable {
/**
* read data from tsFileInput, from the current position (if position = -1),
or the given
- * position. <br> if position = -1, the tsFileInput's position will be
changed to the current
- * position + real data size that been read. Other wise, the tsFileInput's
position is not
- * changed.
+ * position. <br>
+ * if position = -1, the tsFileInput's position will be changed to the
current position + real
+ * data size that been read. Other wise, the tsFileInput's position is not
changed.
*
* @param position the start position of data in the tsFileInput, or the
current position if
- * position = -1
- * @param size the size of data that want to read
+ * position = -1
+ * @param size the size of data that want to read
* @return data that been read.
*/
private ByteBuffer readData(long position, int size) throws IOException {
@@ -930,8 +973,10 @@ public class TsFileSequenceReader implements AutoCloseable
{
long actualReadSize = ReadWriteIOUtils.readAsPossible(tsFileInput,
buffer, position, size);
if (actualReadSize != size) {
throw new IOException(
- String.format("reach the end of the data. Size of data that want
to read: %s,"
- + "actual read size: %s, posiotion: %s", size, actualReadSize,
position));
+ String.format(
+ "reach the end of the data. Size of data that want to read:
%s,"
+ + "actual read size: %s, posiotion: %s",
+ size, actualReadSize, position));
}
}
buffer.flip();
@@ -943,17 +988,15 @@ public class TsFileSequenceReader implements
AutoCloseable {
* position.
*
* @param start the start position of data in the tsFileInput, or the
current position if position
- * = -1
- * @param end the end position of data that want to read
+ * = -1
+ * @param end the end position of data that want to read
* @return data that been read.
*/
private ByteBuffer readData(long start, long end) throws IOException {
return readData(start, (int) (end - start));
}
- /**
- * notice, the target bytebuffer are not flipped.
- */
+ /** notice, the target bytebuffer are not flipped. */
public int readRaw(long position, int length, ByteBuffer target) throws
IOException {
return ReadWriteIOUtils.readAsPossible(tsFileInput, target, position,
length);
}
@@ -961,19 +1004,21 @@ public class TsFileSequenceReader implements
AutoCloseable {
/**
* Self Check the file and return the position before where the data is safe.
*
- * @param newSchema the schema on each time series in the file
+ * @param newSchema the schema on each time series in the file
* @param chunkGroupMetadataList ChunkGroupMetadata List
- * @param versionInfo version pair List
- * @param fastFinish if true and the file is complete, then
newSchema and
- * chunkGroupMetadataList parameter will be
not modified.
+ * @param versionInfo version pair List
+ * @param fastFinish if true and the file is complete, then newSchema and
chunkGroupMetadataList
+ * parameter will be not modified.
* @return the position of the file that is fine. All data after the
position in the file should
- * be truncated.
+ * be truncated.
*/
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
- public long selfCheck(Map<Path, MeasurementSchema> newSchema,
+ public long selfCheck(
+ Map<Path, MeasurementSchema> newSchema,
List<ChunkGroupMetadata> chunkGroupMetadataList,
List<Pair<Long, Long>> versionInfo,
- boolean fastFinish) throws IOException {
+ boolean fastFinish)
+ throws IOException {
File checkFile = FSFactoryProducer.getFSFactory().getFile(this.file);
long fileSize;
if (!checkFile.exists()) {
@@ -990,13 +1035,13 @@ public class TsFileSequenceReader implements
AutoCloseable {
List<ChunkMetadata> chunkMetadataList = null;
String deviceID;
- int headerLength = TSFileConfig.MAGIC_STRING.getBytes().length +
TSFileConfig.VERSION_NUMBER
- .getBytes().length;
+ int headerLength =
+ TSFileConfig.MAGIC_STRING.getBytes().length +
TSFileConfig.VERSION_NUMBER.getBytes().length;
if (fileSize < headerLength) {
return TsFileCheckStatus.INCOMPATIBLE_FILE;
}
- if (!TSFileConfig.MAGIC_STRING.equals(readHeadMagic()) ||
!TSFileConfig.VERSION_NUMBER
- .equals(readVersionNumber())) {
+ if (!TSFileConfig.MAGIC_STRING.equals(readHeadMagic())
+ || !TSFileConfig.VERSION_NUMBER.equals(readVersionNumber())) {
return TsFileCheckStatus.INCOMPATIBLE_FILE;
}
@@ -1030,9 +1075,12 @@ public class TsFileSequenceReader implements
AutoCloseable {
// insertion is not tolerable
ChunkHeader chunkHeader = this.readChunkHeader();
measurementID = chunkHeader.getMeasurementID();
- MeasurementSchema measurementSchema = new
MeasurementSchema(measurementID,
- chunkHeader.getDataType(),
- chunkHeader.getEncodingType(),
chunkHeader.getCompressionType());
+ MeasurementSchema measurementSchema =
+ new MeasurementSchema(
+ measurementID,
+ chunkHeader.getDataType(),
+ chunkHeader.getEncodingType(),
+ chunkHeader.getCompressionType());
measurementSchemaList.add(measurementSchema);
dataType = chunkHeader.getDataType();
Statistics<?> chunkStatistics =
Statistics.getStatsByType(dataType);
@@ -1042,8 +1090,8 @@ public class TsFileSequenceReader implements
AutoCloseable {
chunkStatistics.mergeStatistics(pageHeader.getStatistics());
this.skipPageData(pageHeader);
}
- currentChunk = new ChunkMetadata(measurementID, dataType,
fileOffsetOfChunk,
- chunkStatistics);
+ currentChunk =
+ new ChunkMetadata(measurementID, dataType, fileOffsetOfChunk,
chunkStatistics);
chunkMetadataList.add(currentChunk);
chunkCnt++;
break;
@@ -1080,8 +1128,11 @@ public class TsFileSequenceReader implements
AutoCloseable {
// ChunkGroupFooter is complete.
truncatedSize = this.position() - 1;
} catch (Exception e) {
- logger.info("TsFile {} self-check cannot proceed at position {} " +
"recovered, because : {}",
- file, this.position(), e.getMessage());
+ logger.info(
+ "TsFile {} self-check cannot proceed at position {} " + "recovered,
because : {}",
+ file,
+ this.position(),
+ e.getMessage());
}
// Despite the completeness of the data section, we will discard current
FileMetadata
// so that we can continue to write data into this tsfile.
@@ -1154,11 +1205,24 @@ public class TsFileSequenceReader implements
AutoCloseable {
return result;
}
+ public Map<String, List<String>> getDeviceMeasurementsMap() throws
IOException {
+ Map<String, List<String>> result = new HashMap<>();
+ for (String device : getAllDevices()) {
+ Map<String, TimeseriesMetadata> timeseriesMetadataMap =
readDeviceMetadata(device);
+ for (TimeseriesMetadata timeseriesMetadata :
timeseriesMetadataMap.values()) {
+ result
+ .computeIfAbsent(device, d -> new ArrayList<>())
+ .add(timeseriesMetadata.getMeasurementId());
+ }
+ }
+ return result;
+ }
+
/**
* get device names which has valid chunks in [start, end)
*
* @param start start of the partition
- * @param end end of the partition
+ * @param end end of the partition
* @return device names in range
*/
public List<String> getDeviceNameInRange(long start, long end) throws
IOException {
@@ -1176,15 +1240,15 @@ public class TsFileSequenceReader implements
AutoCloseable {
* Check if the device has at least one Chunk in this partition
*
* @param seriesMetadataMap chunkMetaDataList of each measurement
- * @param start the start position of the space partition
- * @param end the end position of the space partition
+ * @param start the start position of the space partition
+ * @param end the end position of the space partition
*/
- private boolean hasDataInPartition(Map<String, List<ChunkMetadata>>
seriesMetadataMap,
- long start, long end) {
+ private boolean hasDataInPartition(
+ Map<String, List<ChunkMetadata>> seriesMetadataMap, long start, long
end) {
for (List<ChunkMetadata> chunkMetadataList : seriesMetadataMap.values()) {
for (ChunkMetadata chunkMetadata : chunkMetadataList) {
- LocateStatus location = MetadataQuerierByFileImpl
- .checkLocateStatus(chunkMetadata, start, end);
+ LocateStatus location =
+ MetadataQuerierByFileImpl.checkLocateStatus(chunkMetadata, start,
end);
if (location == LocateStatus.in) {
return true;
}
@@ -1194,12 +1258,110 @@ public class TsFileSequenceReader implements
AutoCloseable {
}
/**
- * The location of a chunkGroupMetaData with respect to a space partition
constraint. <p> in - the
- * middle point of the chunkGroupMetaData is located in the current space
partition. before - the
- * middle point of the chunkGroupMetaData is located before the current
space partition. after -
- * the middle point of the chunkGroupMetaData is located after the current
space partition.
+ * The location of a chunkGroupMetaData with respect to a space partition
constraint.
+ *
+ * <p>in - the middle point of the chunkGroupMetaData is located in the
current space partition.
+ * before - the middle point of the chunkGroupMetaData is located before the
current space
+ * partition. after - the middle point of the chunkGroupMetaData is located
after the current
+ * space partition.
*/
public enum LocateStatus {
- in, before, after
+ in,
+ before,
+ after
+ }
+
+ /**
+ * @return An iterator of linked hashmaps ( measurement -> chunk metadata
list ). When traversing
+ * the linked hashmap, you will get chunk metadata lists according to
the lexicographic order
+ * of the measurements. The first measurement of the linked hashmap of
each iteration is
+ * always larger than the last measurement of the linked hashmap of the
previous iteration in
+ * lexicographic order.
+ */
+ public Iterator<Map<String, List<ChunkMetadata>>>
getMeasurementChunkMetadataListMapIterator(
+ String device) throws IOException {
+ readFileMetadata();
+
+ MetadataIndexNode metadataIndexNode = tsFileMetaData.getMetadataIndex();
+ Pair<MetadataIndexEntry, Long> metadataIndexPair =
+ getMetadataAndEndOffset(metadataIndexNode, device, INTERNAL_DEVICE,
true);
+
+ if (metadataIndexPair == null) {
+ return new Iterator<Map<String, List<ChunkMetadata>>>() {
+
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public LinkedHashMap<String, List<ChunkMetadata>> next() {
+ throw new NoSuchElementException();
+ }
+ };
+ }
+
+ Queue<Pair<Long, Long>> queue = new LinkedList<>();
+ ByteBuffer buffer = readData(metadataIndexPair.left.getOffset(),
metadataIndexPair.right);
+ collectEachLeafMeasurementNodeOffsetRange(buffer, queue);
+
+ return new Iterator<Map<String, List<ChunkMetadata>>>() {
+
+ @Override
+ public boolean hasNext() {
+ return !queue.isEmpty();
+ }
+
+ @Override
+ public LinkedHashMap<String, List<ChunkMetadata>> next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ Pair<Long, Long> startEndPair = queue.remove();
+ LinkedHashMap<String, List<ChunkMetadata>>
measurementChunkMetadataList =
+ new LinkedHashMap<>();
+ try {
+ List<TimeseriesMetadata> timeseriesMetadataList = new ArrayList<>();
+ ByteBuffer nextBuffer = readData(startEndPair.left,
startEndPair.right);
+ while (nextBuffer.hasRemaining()) {
+
timeseriesMetadataList.add(TimeseriesMetadata.deserializeFrom(nextBuffer));
+ }
+ for (TimeseriesMetadata timeseriesMetadata : timeseriesMetadataList)
{
+ measurementChunkMetadataList
+ .computeIfAbsent(timeseriesMetadata.getMeasurementId(), m ->
new ArrayList<>())
+ .addAll(readChunkMetaDataList(timeseriesMetadata));
+ }
+ return measurementChunkMetadataList;
+ } catch (IOException e) {
+ throw new TsFileRuntimeException(
+ "Error occurred while reading a time series metadata block.");
+ }
+ }
+ };
+ }
+
+ private void collectEachLeafMeasurementNodeOffsetRange(
+ ByteBuffer buffer, Queue<Pair<Long, Long>> queue) throws IOException {
+ try {
+ final MetadataIndexNode metadataIndexNode =
MetadataIndexNode.deserializeFrom(buffer);
+ final MetadataIndexNodeType metadataIndexNodeType =
metadataIndexNode.getNodeType();
+ final int metadataIndexListSize = metadataIndexNode.getChildren().size();
+ for (int i = 0; i < metadataIndexListSize; ++i) {
+ long startOffset = metadataIndexNode.getChildren().get(i).getOffset();
+ long endOffset = metadataIndexNode.getEndOffset();
+ if (i != metadataIndexListSize - 1) {
+ endOffset = metadataIndexNode.getChildren().get(i + 1).getOffset();
+ }
+ if
(metadataIndexNodeType.equals(MetadataIndexNodeType.LEAF_MEASUREMENT)) {
+ queue.add(new Pair<>(startOffset, endOffset));
+ continue;
+ }
+ collectEachLeafMeasurementNodeOffsetRange(readData(startOffset,
endOffset), queue);
+ }
+ } catch (BufferOverflowException e) {
+ logger.error(
+ "Error occurred while collecting offset ranges of measurement nodes
of file {}", file);
+ throw e;
+ }
}
}
diff --git
a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/MeasurementChunkMetadataListMapIteratorTest.java
b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/MeasurementChunkMetadataListMapIteratorTest.java
new file mode 100644
index 0000000..e4a1313
--- /dev/null
+++
b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/MeasurementChunkMetadataListMapIteratorTest.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.tsfile.read;
+
+import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.utils.FileGenerator;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+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 java.util.Map.Entry;
+
+public class MeasurementChunkMetadataListMapIteratorTest {
+
+ private static final String FILE_PATH = FileGenerator.outputDataFile;
+ private final TSFileConfig conf = TSFileDescriptor.getInstance().getConfig();
+
+ private int maxDegreeOfIndexNode;
+
+ @Before
+ public void before() {
+ maxDegreeOfIndexNode = conf.getMaxDegreeOfIndexNode();
+ conf.setMaxDegreeOfIndexNode(3);
+ }
+
+ @After
+ public void after() {
+ FileGenerator.after();
+ conf.setMaxDegreeOfIndexNode(maxDegreeOfIndexNode);
+ }
+
+ @Test
+ public void test0() throws IOException {
+ testCorrectness(1, 1);
+ testSequentiality(1, 1);
+ }
+
+ @Test
+ public void test1() throws IOException {
+ testCorrectness(1, 10);
+ testSequentiality(1, 10);
+ }
+
+ @Test
+ public void test2() throws IOException {
+ testCorrectness(2, 1);
+ testSequentiality(2, 1);
+ }
+
+ @Test
+ public void test3() throws IOException {
+ testCorrectness(2, 2);
+ testSequentiality(2, 2);
+ }
+
+ @Test
+ public void test4() throws IOException {
+ testCorrectness(2, 100);
+ testSequentiality(2, 100);
+ }
+
+ @Test
+ public void test5() throws IOException {
+ testCorrectness(50, 2);
+ testSequentiality(50, 2);
+ }
+
+ @Test
+ public void test6() throws IOException {
+ testCorrectness(50, 50);
+ testSequentiality(50, 50);
+ }
+
+ @Test
+ public void test7() throws IOException {
+ testCorrectness(50, 100);
+ testSequentiality(50, 100);
+ }
+
+ @Test
+ public void test8() throws IOException {
+ testCorrectness(33, 733);
+ testSequentiality(33, 733);
+ }
+
+ @Test
+ public void test9() throws IOException {
+ testCorrectness(733, 33);
+ testSequentiality(733, 33);
+ }
+
+ public void testCorrectness(int deviceNum, int measurementNum) throws
IOException {
+ FileGenerator.generateFile(10000, deviceNum, measurementNum);
+
+ try (TsFileSequenceReader fileReader = new
TsFileSequenceReader(FILE_PATH)) {
+ Map<String, List<String>> deviceMeasurementListMap =
fileReader.getDeviceMeasurementsMap();
+
+ List<String> devices = fileReader.getAllDevices();
+
+ Map<String, Map<String, List<ChunkMetadata>>>
expectedDeviceMeasurementChunkMetadataListMap =
+ new HashMap<>();
+ for (String device : devices) {
+ for (String measurement : deviceMeasurementListMap.get(device)) {
+ expectedDeviceMeasurementChunkMetadataListMap
+ .computeIfAbsent(device, d -> new HashMap<>())
+ .computeIfAbsent(measurement, m -> new ArrayList<>())
+ .addAll(fileReader.getChunkMetadataList(new Path(device,
measurement)));
+ }
+ }
+
+ for (String device : devices) {
+ Map<String, List<ChunkMetadata>> expected =
+ expectedDeviceMeasurementChunkMetadataListMap.get(device);
+
+ Map<String, List<ChunkMetadata>> actual = new HashMap<>();
+ Iterator<Map<String, List<ChunkMetadata>>> iterator =
+ fileReader.getMeasurementChunkMetadataListMapIterator(device);
+ while (iterator.hasNext()) {
+ Map<String, List<ChunkMetadata>> next = iterator.next();
+ for (Entry<String, List<ChunkMetadata>> entry : next.entrySet()) {
+ actual.computeIfAbsent(entry.getKey(), m -> new
ArrayList<>()).addAll(entry.getValue());
+ }
+ }
+
+ checkCorrectness(expected, actual);
+ }
+ }
+
+ FileGenerator.after();
+ }
+
+ private void checkCorrectness(
+ Map<String, List<ChunkMetadata>> expected, Map<String,
List<ChunkMetadata>> actual) {
+ Assert.assertEquals(expected.keySet(), actual.keySet());
+ for (String measurement : expected.keySet()) {
+ List<ChunkMetadata> expectedChunkMetadataList =
expected.get(measurement);
+ List<ChunkMetadata> actualChunkMetadataList = actual.get(measurement);
+ Assert.assertEquals(expectedChunkMetadataList.size(),
actualChunkMetadataList.size());
+ final int size = expectedChunkMetadataList.size();
+ for (int i = 0; i < size; ++i) {
+ Assert.assertEquals(
+ expectedChunkMetadataList.get(i).toString(),
actualChunkMetadataList.get(i).toString());
+ }
+ }
+ }
+
+ public void testSequentiality(int deviceNum, int measurementNum) throws
IOException {
+ FileGenerator.generateFile(10000, deviceNum, measurementNum);
+
+ try (TsFileSequenceReader fileReader = new
TsFileSequenceReader(FILE_PATH)) {
+ for (String device : fileReader.getAllDevices()) {
+ Iterator<Map<String, List<ChunkMetadata>>> iterator =
+ fileReader.getMeasurementChunkMetadataListMapIterator(device);
+
+ String lastMeasurement = null;
+ while (iterator.hasNext()) {
+ for (String measurement : iterator.next().keySet()) {
+ if (lastMeasurement != null) {
+ Assert.assertTrue(lastMeasurement.compareTo(measurement) < 0);
+ }
+ lastMeasurement = measurement;
+ }
+ }
+ }
+ }
+
+ FileGenerator.after();
+ }
+}
