This is an automated email from the ASF dual-hosted git repository. sunzesong pushed a commit to branch hot_compaction_sonar in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit dbf029f1dcc09d4404ecea3df6450271a5435b3b Author: samperson1997 <[email protected]> AuthorDate: Tue Jul 7 13:40:26 2020 +0800 Make sonar happy --- .../iotdb/db/engine/flush/MemTableFlushTask.java | 5 +-- .../apache/iotdb/db/engine/flush/VmMergeTask.java | 27 ++++++++------- .../engine/storagegroup/StorageGroupProcessor.java | 19 +++++++---- .../db/engine/storagegroup/TsFileProcessor.java | 39 +++++++++++++--------- .../iotdb/db/writelog/recover/LogReplayer.java | 1 - .../writelog/recover/TsFileRecoverPerformer.java | 36 +++++++------------- 6 files changed, 66 insertions(+), 61 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java index 3f0484e..2fe84a1 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/MemTableFlushTask.java @@ -24,6 +24,7 @@ import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.VM_SUFFIX; import java.io.File; import java.io.IOException; +import java.nio.file.Files; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -176,7 +177,7 @@ public class MemTableFlushTask { } if (flushLogFile != null) { - flushLogFile.delete(); + Files.delete(flushLogFile.toPath()); } if (vmLogger != null) { @@ -334,7 +335,7 @@ public class MemTableFlushTask { currWriter.startChunkGroup(((StartFlushGroupIOTask) ioMessage).deviceId); } else if (ioMessage instanceof MergeVmIoTask) { if (flushLogFile != null) { - flushLogFile.delete(); + Files.delete(flushLogFile.toPath()); } RestorableTsFileIOWriter mergeWriter = ((MergeVmIoTask) ioMessage).mergeWriter; VmMergeTask vmMergeTask = new VmMergeTask(mergeWriter, vmWriters, diff --git a/server/src/main/java/org/apache/iotdb/db/engine/flush/VmMergeTask.java b/server/src/main/java/org/apache/iotdb/db/engine/flush/VmMergeTask.java index 3b333e1..fa16d22 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/flush/VmMergeTask.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/flush/VmMergeTask.java @@ -22,7 +22,6 @@ package org.apache.iotdb.db.engine.flush; import static org.apache.iotdb.db.utils.MergeUtils.writeTimeValuePair; import java.io.IOException; -import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -93,12 +92,14 @@ public class VmMergeTask { } } if (!sequence) { - for (String deviceId : deviceMeasurementMap.keySet()) { + for (Entry<String, Map<String, MeasurementSchema>> deviceMeasurementEntry : deviceMeasurementMap + .entrySet()) { + String deviceId = deviceMeasurementEntry.getKey(); writer.startChunkGroup(deviceId); long maxVersion = Long.MIN_VALUE; - for (String measurementId : deviceMeasurementMap.get(deviceId).keySet()) { - MeasurementSchema measurementSchema = deviceMeasurementMap.get(deviceId) - .get(measurementId); + for (Entry<String, MeasurementSchema> entry : deviceMeasurementEntry.getValue() + .entrySet()) { + String measurementId = entry.getKey(); Map<Long, TimeValuePair> timeValuePairMap = new TreeMap<>(); for (RestorableTsFileIOWriter vmWriter : vmWriters) { TsFileSequenceReader reader = tsFileSequenceReaderMap @@ -117,9 +118,8 @@ public class VmMergeTask { if (reader == null) { continue; } - List<ChunkMetadata> chunkMetadataList = vmWriter - .getVisibleMetadataList(deviceId, measurementId, - measurementSchema.getType()); + List<ChunkMetadata> chunkMetadataList = vmWriter.getVisibleMetadataList(deviceId, + measurementId, entry.getValue().getType()); for (ChunkMetadata chunkMetadata : chunkMetadataList) { maxVersion = Math.max(chunkMetadata.getVersion(), maxVersion); IChunkReader chunkReader = new ChunkReaderByTimestamp( @@ -134,8 +134,7 @@ public class VmMergeTask { } } } - IChunkWriter chunkWriter = new ChunkWriterImpl( - deviceMeasurementMap.get(deviceId).get(measurementId)); + IChunkWriter chunkWriter = new ChunkWriterImpl(entry.getValue()); for (TimeValuePair timeValuePair : timeValuePairMap.values()) { writeTimeValuePair(timeValuePair, chunkWriter); } @@ -148,9 +147,13 @@ public class VmMergeTask { } } } else { - for (String deviceId : deviceMeasurementMap.keySet()) { + for (Entry<String, Map<String, MeasurementSchema>> deviceMeasurementEntry : deviceMeasurementMap + .entrySet()) { + String deviceId = deviceMeasurementEntry.getKey(); writer.startChunkGroup(deviceId); - for (String measurementId : deviceMeasurementMap.get(deviceId).keySet()) { + for (Entry<String, MeasurementSchema> entry : deviceMeasurementEntry.getValue() + .entrySet()) { + String measurementId = entry.getKey(); ChunkMetadata newChunkMetadata = null; Chunk newChunk = null; for (RestorableTsFileIOWriter vmWriter : vmWriters) { diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java index 018892e..f8c7533 100755 --- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java @@ -28,6 +28,7 @@ import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.VM_SUFFIX; import java.io.File; import java.io.IOException; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -67,7 +68,13 @@ import org.apache.iotdb.db.engine.querycontext.QueryDataSource; import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk; import org.apache.iotdb.db.engine.version.SimpleFileVersionController; import org.apache.iotdb.db.engine.version.VersionController; -import org.apache.iotdb.db.exception.*; +import org.apache.iotdb.db.exception.BatchInsertionException; +import org.apache.iotdb.db.exception.DiskSpaceInsufficientException; +import org.apache.iotdb.db.exception.LoadFileException; +import org.apache.iotdb.db.exception.MergeException; +import org.apache.iotdb.db.exception.StorageGroupProcessorException; +import org.apache.iotdb.db.exception.TsFileProcessorException; +import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.query.OutOfTTLException; import org.apache.iotdb.db.exception.query.QueryProcessException; @@ -508,7 +515,7 @@ public class StorageGroupProcessor { return new Pair<>(ret, upgradeRet); } - private Map<String, List<TsFileResource>> getAllVms(List<String> folders) { + private Map<String, List<TsFileResource>> getAllVms(List<String> folders) throws IOException { List<File> vmFiles = new ArrayList<>(); for (String baseDir : folders) { File fileFolder = fsFactory.getFile(baseDir, storageGroupName); @@ -521,13 +528,13 @@ public class StorageGroupProcessor { if (partitionFolder.isDirectory()) { for (File tmpFile : fsFactory.listFilesBySuffix(partitionFolder.getAbsolutePath(), PATH_UPGRADE)) { - tmpFile.delete(); + Files.delete(tmpFile.toPath()); } for (File mergedFile : fsFactory.listFilesBySuffix(partitionFolder.getAbsolutePath(), MERGED_SUFFIX)) { for (File shouldRemove : fsFactory .listFilesBySuffix(partitionFolder.getAbsolutePath(), VM_SUFFIX)) { - shouldRemove.delete(); + Files.delete(shouldRemove.toPath()); } File newVMFile = FSFactoryProducer.getFSFactory().getFile(mergedFile.getParent(), mergedFile.getName().split(MERGED_SUFFIX)[0]); @@ -586,7 +593,7 @@ public class StorageGroupProcessor { .recover(); writer = pair.left; vmWriters = pair.right; - vmWriters.forEach(vmWriter -> vmWriter.makeMetadataVisible()); + vmWriters.forEach(RestorableTsFileIOWriter::makeMetadataVisible); } catch (StorageGroupProcessorException e) { logger.warn("Skip TsFile: {} because of error in recover: ", tsFileResource.getPath(), e); continue; @@ -633,7 +640,7 @@ public class StorageGroupProcessor { writer = pair.left; vmWriters = pair.right; - vmWriters.forEach(vmWriter -> vmWriter.makeMetadataVisible()); + vmWriters.forEach(RestorableTsFileIOWriter::makeMetadataVisible); } catch (StorageGroupProcessorException e) { logger.warn("Skip TsFile: {} because of error in recover: ", tsFileResource.getPath(), e); continue; diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java index 43de0f0..9dabee5 100644 --- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java +++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java @@ -26,6 +26,7 @@ import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.VM_SUFFIX; import java.io.File; import java.io.IOException; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -51,6 +52,7 @@ import org.apache.iotdb.db.engine.flush.VmLogAnalyzer; import org.apache.iotdb.db.engine.flush.VmLogger; import org.apache.iotdb.db.engine.flush.VmMergeTask; import org.apache.iotdb.db.engine.memtable.IMemTable; +import org.apache.iotdb.db.engine.memtable.IWritableMemChunk; import org.apache.iotdb.db.engine.modification.Deletion; import org.apache.iotdb.db.engine.modification.Modification; import org.apache.iotdb.db.engine.modification.ModificationFile; @@ -259,9 +261,8 @@ public class TsFileProcessor { /** * Delete data which belongs to the timeseries `deviceId.measurementId` and the timestamp of which - * <= 'timestamp' in the deletion. <br/> - * <p> - * Delete data in both working MemTable and flushing MemTables. + * <= 'timestamp' in the deletion. <br/> <p> Delete data in both working MemTable and flushing + * MemTables. */ public void deleteDataInMemory(Deletion deletion) { flushQueryLock.writeLock().lock(); @@ -597,7 +598,7 @@ public class TsFileProcessor { ChunkMetadataCache.getInstance().remove(seqFile); FileReaderManager.getInstance().closeFileAndRemoveReader(seqFile.getPath()); seqFile.setDeleted(true); - seqFile.getFile().delete(); + Files.delete(seqFile.getFile().toPath()); } catch (Exception e) { logger.error(e.getMessage(), e); } finally { @@ -632,13 +633,13 @@ public class TsFileProcessor { } vmWriters.clear(); vmTsFileResources.clear(); - logFile.delete(); + Files.delete(logFile.toPath()); } else { File[] tmpFiles = FSFactoryProducer.getFSFactory() .listFilesBySuffix(writer.getFile().getParent(), PATH_UPGRADE); if (tmpFiles.length > 0) { for (File file : tmpFiles) { - file.delete(); + Files.delete(file.toPath()); } } } @@ -681,25 +682,31 @@ public class TsFileProcessor { } // all flush to target file Map<Path, MeasurementSchema> pathMeasurementSchemaMap = new HashMap<>(); - for (String device : memTableToFlush.getMemTableMap().keySet()) { - for (String measurement : memTableToFlush.getMemTableMap().get(device).keySet()) { + for (Entry<String, Map<String, IWritableMemChunk>> memTableEntry : memTableToFlush + .getMemTableMap().entrySet()) { + String device = memTableEntry.getKey(); + for (Entry<String, IWritableMemChunk> entry : memTableEntry.getValue().entrySet()) { + String measurement = entry.getKey(); pathMeasurementSchemaMap.putIfAbsent(new Path(device, measurement), - memTableToFlush.getMemTableMap().get(device).get(measurement).getSchema()); + entry.getValue().getSchema()); } } for (RestorableTsFileIOWriter vmWriter : vmWriters) { Map<String, Map<String, List<ChunkMetadata>>> schemaMap = vmWriter .getMetadatasForQuery(); - for (String device : schemaMap.keySet()) { - for (String measurement : schemaMap.get(device).keySet()) { - List<ChunkMetadata> chunkMetadataList = schemaMap.get(device).get(measurement); + for (Entry<String, Map<String, List<ChunkMetadata>>> schemaMapEntry : schemaMap + .entrySet()) { + String device = schemaMapEntry.getKey(); + for (Entry<String, List<ChunkMetadata>> entry : schemaMapEntry.getValue() + .entrySet()) { + String measurement = entry.getKey(); pathMeasurementSchemaMap.putIfAbsent(new Path(device, measurement), - new MeasurementSchema(measurement, chunkMetadataList.get(0).getDataType())); + new MeasurementSchema(measurement, entry.getValue().get(0).getDataType())); } } } if ((vmPointNum + memTableToFlush.getTotalPointsNum()) / pathMeasurementSchemaMap.size() - > config.getMergeChunkPointNumberThreshold() + > config.getMergeChunkPointNumberThreshold() || mergeTimes >= config.getMaxMergeChunkNumInTsFile()) { isVm = false; isFull = false; @@ -765,7 +772,7 @@ public class TsFileProcessor { File logFile = FSFactoryProducer.getFSFactory() .getFile(tsFileResource.getFile().getParent(), tsFileResource.getFile().getName() + VM_LOG_NAME); - logFile.delete(); + Files.delete(logFile.toPath()); } } catch (Exception e) { logger.error("{}: {} meet error when flushing a memtable, change system mode to read-only", @@ -819,7 +826,7 @@ public class TsFileProcessor { File logFile = FSFactoryProducer.getFSFactory() .getFile(tsFileResource.getFile().getParent(), tsFileResource.getFile().getName() + VM_LOG_NAME); - logFile.delete(); + Files.delete(logFile.toPath()); } writer.mark(); diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java index 4ebfa28..c2684e8 100644 --- a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java +++ b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java @@ -27,7 +27,6 @@ import org.apache.iotdb.db.engine.version.VersionController; import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.exception.metadata.MetadataException; import org.apache.iotdb.db.exception.query.QueryProcessException; -import org.apache.iotdb.db.metadata.MManager; import org.apache.iotdb.db.qp.physical.PhysicalPlan; import org.apache.iotdb.db.qp.physical.crud.DeletePlan; import org.apache.iotdb.db.qp.physical.crud.InsertPlan; diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java b/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java index f5a0148..19c17bd 100644 --- a/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java +++ b/server/src/main/java/org/apache/iotdb/db/writelog/recover/TsFileRecoverPerformer.java @@ -28,6 +28,7 @@ import static org.apache.iotdb.db.engine.storagegroup.TsFileResource.RESOURCE_SU import java.io.File; import java.io.IOException; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -54,16 +55,6 @@ import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.File; -import java.io.IOException; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Map.Entry; -import java.util.concurrent.ExecutionException; - -import static org.apache.iotdb.db.engine.storagegroup.TsFileResource.RESOURCE_SUFFIX; - /** * TsFileRecoverPerformer recovers a SeqTsFile to correct status, redoes the WALs since last crash * and removes the redone logs. @@ -82,7 +73,7 @@ public class TsFileRecoverPerformer { private List<TsFileResource> vmTsFileResources; /** - * @param isLastFile whether this TsFile is the last file of its partition + * @param isLastFile whether this TsFile is the last file of its partition * @param vmTsFileResources only last file could have non-empty vmTsFileResources */ public TsFileRecoverPerformer(String logNodePrefix, VersionController versionController, @@ -162,7 +153,8 @@ public class TsFileRecoverPerformer { } else { if (!vmTsFileResources.isEmpty()) { for (int i = 0; i < vmTsFileResources.size(); i++) { - recoverResourceFromWriter(vmRestorableTsFileIOWriterList.get(i), vmTsFileResources.get(i)); + recoverResourceFromWriter(vmRestorableTsFileIOWriterList.get(i), + vmTsFileResources.get(i)); } recoverResourceFromWriter(restorableTsFileIOWriter, resource); boolean vmFileNotCrashed = !getFlushLogFile(restorableTsFileIOWriter).exists(); @@ -177,26 +169,22 @@ public class TsFileRecoverPerformer { File newVmFile = createNewVMFile(resource); TsFileResource newVmTsFileResource = new TsFileResource(newVmFile); RestorableTsFileIOWriter newVMWriter = new RestorableTsFileIOWriter(newVmFile); - if (redoLogs(newVMWriter,newVmTsFileResource)) { + if (redoLogs(newVMWriter, newVmTsFileResource)) { vmTsFileResources.add(newVmTsFileResource); vmRestorableTsFileIOWriterList.add(newVMWriter); } else { - newVmFile.delete(); + Files.delete(newVmFile.toPath()); } } else { IMemTable recoverMemTable = new PrimitiveMemTable(); recoverMemTable.setVersion(versionController.nextVersion()); - LogReplayer logReplayer = new LogReplayer(logNodePrefix, filePath, resource.getModFile(), - versionController, resource, recoverMemTable, sequence); + LogReplayer logReplayer = new LogReplayer(logNodePrefix, filePath, + resource.getModFile(), versionController, resource, recoverMemTable, sequence); logReplayer.replayLogs(); } // clean logs - try { - MultiFileLogNodeManager.getInstance().deleteNode( - logNodePrefix + SystemFileFactory.INSTANCE.getFile(filePath).getName()); - } catch (IOException e) { - throw new StorageGroupProcessorException(e); - } + MultiFileLogNodeManager.getInstance().deleteNode( + logNodePrefix + SystemFileFactory.INSTANCE.getFile(filePath).getName()); updateTsFileResource(); return new Pair<>(restorableTsFileIOWriter, vmRestorableTsFileIOWriterList); } catch (IOException e) { @@ -299,7 +287,7 @@ public class TsFileRecoverPerformer { } } long fileVersion = Long.parseLong( - tsFileResource.getFile().getName().split(IoTDBConstant.FILE_NAME_SEPARATOR)[1]); + tsFileResource.getFile().getName().split(IoTDBConstant.FILE_NAME_SEPARATOR)[1]); tsFileResource.setHistoricalVersions(Collections.singleton(fileVersion)); } @@ -326,7 +314,7 @@ public class TsFileRecoverPerformer { File logFile = FSFactoryProducer.getFSFactory() .getFile(tsFileResource.getFile().getParent(), tsFileResource.getFile().getName() + VM_LOG_NAME); - logFile.delete(); + Files.delete(logFile.toPath()); res = true; }
