You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2019/07/12 03:05:17 UTC
[incubator-iotdb] branch dev_merge updated: add more logs
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch dev_merge
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/dev_merge by this push:
new ca9e047 add more logs
ca9e047 is described below
commit ca9e0473e8a6044dc4a2e0daf025e135459f8e28
Author: 江天 <jt...@163.com>
AuthorDate: Fri Jul 12 11:02:58 2019 +0800
add more logs
---
.../iotdb/db/engine/merge/MergeFileSelector.java | 97 +++++++++++++++-------
.../apache/iotdb/db/engine/merge/MergeTask.java | 26 +++++-
.../iotdb/db/engine/merge/RecoverMergeTask.java | 40 +++++++--
.../db/engine/storagegroup/TsFileResource.java | 5 ++
.../iotdb/tsfile/read/TsFileSequenceReader.java | 6 +-
5 files changed, 136 insertions(+), 38 deletions(-)
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/MergeFileSelector.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/MergeFileSelector.java
index f6d3636..b920862 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/MergeFileSelector.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/MergeFileSelector.java
@@ -38,6 +38,8 @@ import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.controller.MetadataQuerier;
import org.apache.iotdb.tsfile.read.controller.MetadataQuerierByFileImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* MergeFileSelector selects the most files from given seqFiles and unseqFiles which can be
@@ -45,8 +47,13 @@ import org.apache.iotdb.tsfile.read.controller.MetadataQuerierByFileImpl;
*/
public class MergeFileSelector {
+ private static final Logger logger = LoggerFactory.getLogger(MergeFileSelector.class);
+ private static final String LOG_FILE_COST = "Memory cost of file {} is {}";
+
private List<TsFileResource> seqFiles;
private List<TsFileResource> unseqFiles;
+
+ private long totalCost;
private long memoryBudget;
private Map<TsFileResource, Long> costMap = new HashMap<>();
@@ -55,6 +62,8 @@ public class MergeFileSelector {
private List<TsFileResource> selectedUnseqFiles = new ArrayList<>();
private List<TsFileResource> selectedSeqFiles = new ArrayList<>();
+ private Map<TsFileResource, TsFileSequenceReader> fileReaderCache = new HashMap<>();
+
private List<Integer> tmpSelectedSeqFiles;
private boolean[] seqSelected;
@@ -70,26 +79,38 @@ public class MergeFileSelector {
public List[] doSelect() throws MergeException {
try {
+ logger.info("Selecting merge candidates from {} seqFile, {} unseqFiles", seqFiles.size(),
+ unseqFiles.size());
doSelect(false);
if (selectedUnseqFiles.isEmpty()) {
doSelect(true);
}
+ clean();
if (selectedUnseqFiles.isEmpty()) {
return new List[0];
}
} catch (IOException | MetadataErrorException e) {
throw new MergeException(e);
}
+ logger.info("Selected merge candidates, {} seqFiles, {} unseqFiles, total memory cost {}",
+ selectedSeqFiles.size(), selectedUnseqFiles.size(), totalCost);
return new List[]{selectedSeqFiles, selectedUnseqFiles};
}
+ private void clean() throws IOException {
+ for (TsFileSequenceReader reader : fileReaderCache.values()) {
+ reader.close();
+ }
+ fileReaderCache.clear();
+ }
+
private void doSelect(boolean useTightBound) throws IOException, MetadataErrorException {
tmpSelectedSeqFiles = new ArrayList<>();
seqSelected = new boolean[seqFiles.size()];
unseqSelected = new boolean[unseqFiles.size()];
- long totalCost = 0;
+ totalCost = 0;
int unseqIndex = 0;
while (unseqIndex < unseqFiles.size()) {
@@ -113,6 +134,10 @@ public class MergeFileSelector {
seqSelected[seqIdx] = true;
selectedSeqFiles.add(seqFiles.get(seqIdx));
}
+ totalCost += newCost;
+ logger.debug("Adding a new unseqFile {} and seqFiles {} as candidates, new cost {}, total"
+ + " cost {}",
+ unseqFile, tmpSelectedSeqFiles, newCost, totalCost);
}
tmpSelectedSeqFiles.clear();
unseqIndex++;
@@ -177,18 +202,18 @@ public class MergeFileSelector {
// (when the file contains only one series), and writing those chunks to a new file creating
// new metadata, so it is doubled in the worst case
private long calculateSeqMemoryCost(TsFileResource seqFile) throws IOException {
- return 2 * calculateMetadataSize(seqFile);
+ long cost = 2 * calculateMetadataSize(seqFile);
+ logger.debug(LOG_FILE_COST, seqFile, cost);
+ return cost;
}
private long calculateMetadataSize(TsFileResource seqFile) throws IOException {
long minPos = Long.MAX_VALUE;
- try (TsFileSequenceReader sequenceReader =
- new TsFileSequenceReader(seqFile.getFile().getPath())) {
- TsFileMetaData fileMetaData = sequenceReader.readFileMetadata();
- Map<String, TsDeviceMetadataIndex> deviceMap = fileMetaData.getDeviceMap();
- for (TsDeviceMetadataIndex metadataIndex : deviceMap.values()) {
- minPos = metadataIndex.getOffset() < minPos ? metadataIndex.getOffset() : minPos;
- }
+ TsFileSequenceReader sequenceReader = getReader(seqFile);
+ TsFileMetaData fileMetaData = sequenceReader.readFileMetadata();
+ Map<String, TsDeviceMetadataIndex> deviceMap = fileMetaData.getDeviceMap();
+ for (TsDeviceMetadataIndex metadataIndex : deviceMap.values()) {
+ minPos = metadataIndex.getOffset() < minPos ? metadataIndex.getOffset() : minPos;
}
return seqFile.getFileSize() - minPos;
}
@@ -198,7 +223,9 @@ public class MergeFileSelector {
// memory and writing those chunks to a new file creating new metadata, so the metadata is doubled
// in the worst case
private long calculateUnseqMemoryCost(TsFileResource unseqFile) throws IOException {
- return unseqFile.getFileSize() + calculateMetadataSize(unseqFile);
+ long cost = unseqFile.getFileSize() + calculateMetadataSize(unseqFile);
+ logger.debug(LOG_FILE_COST, unseqFile, cost);
+ return cost;
}
// this method traverses all ChunkMetadata to find out which series has the most chunks and uses
@@ -209,7 +236,9 @@ public class MergeFileSelector {
long[] chunkNums = findLargestSeriesChunkNum(seqFile);
long totalChunkNum = chunkNums[0];
long maxChunkNum = chunkNums[1];
- return 2 * calculateMetadataSize(seqFile) * maxChunkNum / totalChunkNum;
+ long cost = 2 * calculateMetadataSize(seqFile) * maxChunkNum / totalChunkNum;
+ logger.debug(LOG_FILE_COST, seqFile, cost);
+ return cost;
}
// this method traverses all ChunkMetadata to find out which series has the most chunks and uses
@@ -219,7 +248,9 @@ public class MergeFileSelector {
long[] chunkNums = findLargestSeriesChunkNum(unseqFile);
long totalChunkNum = chunkNums[0];
long maxChunkNum = chunkNums[1];
- return calculateUnseqMemoryCost(unseqFile) * maxChunkNum / totalChunkNum;
+ long cost = calculateUnseqMemoryCost(unseqFile) * maxChunkNum / totalChunkNum;
+ logger.debug(LOG_FILE_COST, unseqFile, cost);
+ return cost;
}
// returns totalChunkNum of a file and the max number of chunks of a series
@@ -227,25 +258,25 @@ public class MergeFileSelector {
throws IOException, MetadataErrorException {
long totalChunkNum = 0;
long maxChunkNum = Long.MIN_VALUE;
- try (TsFileSequenceReader sequenceReader =
- new TsFileSequenceReader(tsFileResource.getFile().getPath())) {
- List<Path> paths = new ArrayList<>();
- for (String deviceId : tsFileResource.getEndTimeMap().keySet()) {
- List<String> strPaths = MManager.getInstance().getPaths(deviceId + ".*");
- for (String strPath : strPaths) {
- paths.add(new Path(strPath));
- }
+ TsFileSequenceReader sequenceReader = getReader(tsFileResource);
+ List<Path> paths = new ArrayList<>();
+ for (String deviceId : tsFileResource.getEndTimeMap().keySet()) {
+ List<String> strPaths = MManager.getInstance().getPaths(deviceId + ".*");
+ for (String strPath : strPaths) {
+ paths.add(new Path(strPath));
}
+ }
- MetadataQuerier metadataQuerier = new MetadataQuerierByFileImpl(sequenceReader);
- for (Path path : paths) {
- List<ChunkMetaData> chunkMetaDataList = metadataQuerier.getChunkMetaDataList(path);
- metadataQuerier.clear();
- totalChunkNum += chunkMetaDataList.size();
- maxChunkNum = chunkMetaDataList.size() > maxChunkNum ? chunkMetaDataList.size() :
- maxChunkNum;
- }
+ MetadataQuerier metadataQuerier = new MetadataQuerierByFileImpl(sequenceReader);
+ for (Path path : paths) {
+ List<ChunkMetaData> chunkMetaDataList = metadataQuerier.getChunkMetaDataList(path);
+ metadataQuerier.clear();
+ totalChunkNum += chunkMetaDataList.size();
+ maxChunkNum = chunkMetaDataList.size() > maxChunkNum ? chunkMetaDataList.size() :
+ maxChunkNum;
}
+ logger.debug("In file {}, total chunk num {}, series max chunk num {}", tsFileResource,
+ totalChunkNum, maxChunkNum);
return new long[] {totalChunkNum, maxChunkNum};
}
@@ -278,5 +309,13 @@ public class MergeFileSelector {
(l2 <= l1 && l1 <= r2) ||
(l2 <= r1 && r1 <= r2);
}
-
+
+ private TsFileSequenceReader getReader(TsFileResource tsFileResource) throws IOException {
+ TsFileSequenceReader reader = fileReaderCache.get(tsFileResource);
+ if (reader == null) {
+ reader = new TsFileSequenceReader(tsFileResource.getFile().getPath());
+ fileReaderCache.put(tsFileResource, reader);
+ }
+ return reader;
+ }
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/MergeTask.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/MergeTask.java
index 8f5b860..ba76111 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/MergeTask.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/MergeTask.java
@@ -129,6 +129,8 @@ public class MergeTask implements Callable<Void> {
}
private void doMerge() throws MetadataErrorException, IOException {
+ logger.info("{} starts", taskName);
+ long startTime = System.currentTimeMillis();
this.mergeLogger = new MergeLogger(storageGroupDir);
logFiles();
@@ -139,6 +141,9 @@ public class MergeTask implements Callable<Void> {
mergeFiles(unmergedFiles);
cleanUp(true);
+ if (logger.isInfoEnabled()) {
+ logger.info("{} ends after {}ms", taskName, System.currentTimeMillis() - startTime);
+ }
}
void mergeFiles(List<TsFileResource> unmergedFiles) throws IOException {
@@ -147,6 +152,7 @@ public class MergeTask implements Callable<Void> {
if (logger.isInfoEnabled()) {
logger.info("{} starts to merge {} files", taskName, unmergedFiles.size());
}
+ long startTime = System.currentTimeMillis();
int cnt = 0;
for (TsFileResource seqFile : unmergedFiles) {
int mergedChunkNum = mergedChunkCnt.getOrDefault(seqFile, 0);
@@ -171,7 +177,9 @@ public class MergeTask implements Callable<Void> {
logger.debug("{} has merged {}/{} files", taskName, cnt, unmergedFiles.size());
}
}
- logger.info("{} has merged all files", taskName);
+ if (logger.isInfoEnabled()) {
+ logger.info("{} has merged all files after {}ms", taskName, System.currentTimeMillis() - startTime);
+ }
mergeLogger.logMergeEnd();
}
@@ -185,6 +193,7 @@ public class MergeTask implements Callable<Void> {
if (logger.isInfoEnabled()) {
logger.info("{} starts to merge {} series", taskName, unmergedSeries.size());
}
+ long startTime = System.currentTimeMillis();
for (TsFileResource seqFile : seqFiles) {
unmergedChunkStartTimes.put(seqFile, new HashMap<>());
}
@@ -204,7 +213,9 @@ public class MergeTask implements Callable<Void> {
}
}
}
- logger.info("{} all series are merged", taskName);
+ if (logger.isInfoEnabled()) {
+ logger.info("{} all series are merged after {}ms", taskName, System.currentTimeMillis() - startTime);
+ }
mergeLogger.logAllTsEnd();
}
@@ -252,6 +263,7 @@ public class MergeTask implements Callable<Void> {
try {
oldFileWriter = new ForceAppendTsFileWriter(seqFile.getFile());
mergeLogger.logFileMergeStart(seqFile.getFile(), ((ForceAppendTsFileWriter) oldFileWriter).getTruncatePosition());
+ logger.debug("{} moving merged chunks of {} to the old file", taskName, seqFile);
((ForceAppendTsFileWriter) oldFileWriter).doTruncate();
} catch (TsFileNotCompleteException e) {
// this file may already be truncated if this merge is a system reboot merge
@@ -266,6 +278,9 @@ public class MergeTask implements Callable<Void> {
new TsFileSequenceReader(newFileWriter.getFile().getPath())) {
ChunkLoader chunkLoader = new ChunkLoaderImpl(newFileReader);
List<ChunkGroupMetaData> chunkGroupMetaDatas = newFileWriter.getChunkGroupMetaDatas();
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} find {} merged chunk groups", taskName, chunkGroupMetaDatas.size());
+ }
for (ChunkGroupMetaData chunkGroupMetaData : chunkGroupMetaDatas) {
writeMergedChunkGroup(chunkGroupMetaData, chunkLoader, oldFileWriter);
}
@@ -274,6 +289,7 @@ public class MergeTask implements Callable<Void> {
seqFile.serialize();
mergeLogger.logFileMergeEnd(seqFile.getFile());
+ logger.debug("{} moved merged chunks of {} to the old file", taskName, seqFile);
newFileWriter.getFile().delete();
} finally {
@@ -305,6 +321,7 @@ public class MergeTask implements Callable<Void> {
ChunkLoader chunkLoader = new ChunkLoaderImpl(getFileReader(seqFile));
mergeLogger.logFileMergeStart(fileWriter.getFile(), fileWriter.getFile().length());
+ logger.debug("{} moving unmerged chunks of {} to the new file", taskName, seqFile);
int unmergedChunkNum = unmergedChunkCnt.getOrDefault(seqFile, 0);
@@ -320,6 +337,10 @@ public class MergeTask implements Callable<Void> {
MeasurementSchema measurementSchema = getSchema(path);
IChunkWriter chunkWriter = getChunkWriter(measurementSchema);
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} find {} unmerged chunks", taskName, chunkMetaDataList.size());
+ }
+
fileWriter.startChunkGroup(path.getDevice());
long maxVersion = writeUnmergedChunks(chunkStartTimes, chunkMetaDataList, chunkLoader,
chunkWriter, measurementSchema, fileWriter);
@@ -331,6 +352,7 @@ public class MergeTask implements Callable<Void> {
seqFile.serialize();
mergeLogger.logFileMergeEnd(fileWriter.getFile());
+ logger.debug("{} moved unmerged chunks of {} to the new file", taskName, seqFile);
seqFile.getMergeQueryLock().writeLock().lock();
try {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/RecoverMergeTask.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/RecoverMergeTask.java
index a4b9c09..0132308 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/RecoverMergeTask.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/merge/RecoverMergeTask.java
@@ -69,9 +69,13 @@ public class RecoverMergeTask extends MergeTask {
}
mergeSeqFiles = new ArrayList<>();
mergeUnseqFiles = new ArrayList<>();
+ long startTime = System.currentTimeMillis();
Status status = determineStatus(logFile);
- logger.info("{} merge recovery status determined: {}", taskName, status);
+ if (logger.isInfoEnabled()) {
+ logger.info("{} merge recovery status determined: {} after {}ms", taskName, status,
+ (System.currentTimeMillis() - startTime));
+ }
switch (status) {
case NONE:
logFile.delete();
@@ -104,7 +108,9 @@ public class RecoverMergeTask extends MergeTask {
cleanUp(continueMerge);
break;
}
- logger.info("{} merge recovery ends", taskName);
+ if (logger.isInfoEnabled()) {
+ logger.info("{} merge recovery ends after", taskName, (System.currentTimeMillis() - startTime));
+ }
}
@@ -209,9 +215,10 @@ public class RecoverMergeTask extends MergeTask {
}
private void analyzeSeqFiles(BufferedReader bufferedReader) throws IOException {
+ long startTime = System.currentTimeMillis();
while ((currLine = bufferedReader.readLine()) != null) {
if (currLine.equals(MergeLogger.STR_UNSEQ_FILES)) {
- return;
+ break;
}
Iterator<TsFileResource> iterator = seqFiles.iterator();
while (iterator.hasNext()) {
@@ -223,12 +230,17 @@ public class RecoverMergeTask extends MergeTask {
}
}
}
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} found {} seq files after {}ms", taskName, mergeSeqFiles.size(),
+ (System.currentTimeMillis() - startTime));
+ }
}
private void analyzeUnseqFiles(BufferedReader bufferedReader) throws IOException {
+ long startTime = System.currentTimeMillis();
while ((currLine = bufferedReader.readLine()) != null) {
if (currLine.equals(MergeLogger.STR_MERGE_START)) {
- return;
+ break;
}
Iterator<TsFileResource> iterator = unseqFiles.iterator();
while (iterator.hasNext()) {
@@ -240,13 +252,18 @@ public class RecoverMergeTask extends MergeTask {
}
}
}
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} found {} seq files after {}ms", taskName, mergeUnseqFiles.size(),
+ (System.currentTimeMillis() - startTime));
+ }
}
private void analyzeMergedSeries(BufferedReader bufferedReader, List<Path> unmergedPaths) throws IOException {
Path currTS = null;
+ long startTime = System.currentTimeMillis();
while ((currLine = bufferedReader.readLine()) != null) {
if (currLine.equals(MergeLogger.STR_ALL_TS_END)) {
- return;
+ break;
}
if (currLine.contains(MergeLogger.STR_START)) {
// a TS starts to merge
@@ -268,13 +285,19 @@ public class RecoverMergeTask extends MergeTask {
mergedPaths.add(currTS);
}
}
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} found {} series have already been merged after {}ms", taskName,
+ mergeSeqFiles.size(), (System.currentTimeMillis() - startTime));
+ }
}
private void analyzeMergedFiles(BufferedReader bufferedReader) throws IOException {
File currFile = null;
+ long startTime = System.currentTimeMillis();
+ int mergedCnt = 0;
while ((currLine = bufferedReader.readLine()) != null) {
if (currLine.equals(MergeLogger.STR_MERGE_END)) {
- return;
+ break;
}
if (currLine.contains(MergeLogger.STR_START)) {
String[] splits = currLine.split(" ");
@@ -288,12 +311,17 @@ public class RecoverMergeTask extends MergeTask {
while (unmergedFileIter.hasNext()) {
TsFileResource seqFile = unmergedFileIter.next();
if (seqFile.getFile().getAbsolutePath().equals(seqFilePath)) {
+ mergedCnt ++;
unmergedFileIter.remove();
break;
}
}
}
}
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} found {} files have already been merged after {}ms", taskName,
+ mergedCnt, (System.currentTimeMillis() - startTime));
+ }
}
enum Status {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index f2484c1..ad922d9 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -247,4 +247,9 @@ public class TsFileResource {
file.delete();
new File(file.getPath() + RESOURCE_SUFFIX).delete();
}
+
+ @Override
+ public String toString() {
+ return file.toString();
+ }
}
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 50c6892..8ba77e4 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
@@ -62,6 +62,7 @@ public class TsFileSequenceReader implements AutoCloseable {
protected String file;
private int totalChunkNum;
+ private TsFileMetaData tsFileMetaData;
/**
* Create a file reader of the given file. The reader will read the tail of the file to get the
@@ -210,7 +211,10 @@ public class TsFileSequenceReader implements AutoCloseable {
* this function does not modify the position of the file reader.
*/
public TsFileMetaData readFileMetadata() throws IOException {
- return TsFileMetaData.deserializeFrom(readData(fileMetadataPos, fileMetadataSize));
+ if (tsFileMetaData == null) {
+ tsFileMetaData = TsFileMetaData.deserializeFrom(readData(fileMetadataPos, fileMetadataSize));
+ }
+ return tsFileMetaData;
}
/**