diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java index d47953abe9bbe..56511e8acb20e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java @@ -229,7 +229,11 @@ public Analysis analyzeFileByFile(Analysis analysis) { return analysis; } - LOGGER.info("Load - Analysis Stage: all tsfiles have been analyzed."); + if (isGeneratedByPipe) { + LOGGER.debug("Load - Analysis Stage: all tsfiles have been analyzed."); + } else { + LOGGER.info("Load - Analysis Stage: all tsfiles have been analyzed."); + } if (reconstructStatementIfMiniFileConverted()) { // All mini tsfiles are converted to tablets, so the analysis is finished. @@ -287,22 +291,14 @@ private boolean doAnalyzeFileByFile(Analysis analysis) { if (LOGGER.isWarnEnabled()) { LOGGER.warn("TsFile {} is empty.", tsFile.getPath()); } - if (LOGGER.isInfoEnabled()) { - LOGGER.info( - "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%", - i + 1, tsfileNum, String.format("%.3f", (i + 1) * 100.00 / tsfileNum)); - } + logAnalyzeProgress(i + 1, tsfileNum); continue; } final long startTime = System.nanoTime(); try { analyzeSingleTsFile(tsFile, i); - if (LOGGER.isInfoEnabled()) { - LOGGER.info( - "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%", - i + 1, tsfileNum, String.format("%.3f", (i + 1) * 100.00 / tsfileNum)); - } + logAnalyzeProgress(i + 1, tsfileNum); } catch (AuthException e) { setFailAnalysisForAuthException(analysis, e); return false; @@ -339,6 +335,26 @@ private boolean doAnalyzeFileByFile(Analysis analysis) { return true; } + private void logAnalyzeProgress(final int analyzedTsFileNum, final int totalTsFileNum) { + if (isGeneratedByPipe && !LOGGER.isDebugEnabled()) { + return; + } + if (!isGeneratedByPipe && !LOGGER.isInfoEnabled()) { + return; + } + + final String progress = String.format("%.3f", analyzedTsFileNum * 100.00 / totalTsFileNum); + if (isGeneratedByPipe) { + LOGGER.debug( + "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%", + analyzedTsFileNum, totalTsFileNum, progress); + } else { + LOGGER.info( + "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%", + analyzedTsFileNum, totalTsFileNum, progress); + } + } + private void analyzeSingleTsFile(final File tsFile, int index) throws Exception { try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) { // can be reused when constructing tsfile resource diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java index 37f2fba832d4c..b734412ca9cc8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java @@ -69,7 +69,7 @@ import static com.google.common.util.concurrent.Futures.immediateFuture; -public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { +public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher, AutoCloseable { private static final Logger LOGGER = LoggerFactory.getLogger(LoadTsFileDispatcherImpl.class); @@ -83,7 +83,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { private final int localhostInternalPort; private final IClientManager internalServiceClientManager; - private final ExecutorService executor; + private ExecutorService executor; private final boolean isGeneratedByPipe; public LoadTsFileDispatcherImpl( @@ -92,11 +92,17 @@ public LoadTsFileDispatcherImpl( this.internalServiceClientManager = internalServiceClientManager; this.localhostIpAddr = IoTDBDescriptor.getInstance().getConfig().getInternalAddress(); this.localhostInternalPort = IoTDBDescriptor.getInstance().getConfig().getInternalPort(); - this.executor = - IoTDBThreadPoolFactory.newCachedThreadPool(LoadTsFileDispatcherImpl.class.getName()); this.isGeneratedByPipe = isGeneratedByPipe; } + private synchronized ExecutorService getOrCreateExecutor() { + if (executor == null || executor.isShutdown()) { + executor = + IoTDBThreadPoolFactory.newCachedThreadPool(LoadTsFileDispatcherImpl.class.getName()); + } + return executor; + } + public void setUuid(String uuid) { this.uuid = uuid; } @@ -104,24 +110,26 @@ public void setUuid(String uuid) { @Override public Future dispatch( SubPlan root, List instances) { - return executor.submit( - () -> { - for (FragmentInstance instance : instances) { - try (SetThreadName threadName = - new SetThreadName( - "load-dispatcher" + "-" + instance.getId().getFullId() + "-" + uuid)) { - dispatchOneInstance(instance); - } catch (FragmentInstanceDispatchException e) { - return new FragInstanceDispatchResult(e.getFailureStatus()); - } catch (Exception t) { - LOGGER.warn("cannot dispatch FI for load operation", t); - return new FragInstanceDispatchResult( - RpcUtils.getStatus( - TSStatusCode.INTERNAL_SERVER_ERROR, "Unexpected errors: " + t.getMessage())); - } - } - return new FragInstanceDispatchResult(true); - }); + return getOrCreateExecutor() + .submit( + () -> { + for (FragmentInstance instance : instances) { + try (SetThreadName threadName = + new SetThreadName( + "load-dispatcher" + "-" + instance.getId().getFullId() + "-" + uuid)) { + dispatchOneInstance(instance); + } catch (FragmentInstanceDispatchException e) { + return new FragInstanceDispatchResult(e.getFailureStatus()); + } catch (Exception t) { + LOGGER.warn("cannot dispatch FI for load operation", t); + return new FragInstanceDispatchResult( + RpcUtils.getStatus( + TSStatusCode.INTERNAL_SERVER_ERROR, + "Unexpected errors: " + t.getMessage())); + } + } + return new FragInstanceDispatchResult(true); + }); } private void dispatchOneInstance(FragmentInstance instance) @@ -147,7 +155,11 @@ private void dispatchOneInstance(FragmentInstance instance) } public void dispatchLocally(FragmentInstance instance) throws FragmentInstanceDispatchException { - LOGGER.info("Receive load node from uuid {}.", uuid); + if (isGeneratedByPipe) { + LOGGER.debug("Receive load node from uuid {}.", uuid); + } else { + LOGGER.info("Receive load node from uuid {}.", uuid); + } ConsensusGroupId groupId = ConsensusGroupId.Factory.createFromTConsensusGroupId( @@ -352,6 +364,14 @@ private static void adjustTimeoutIfNecessary(Throwable e) { @Override public void abort() { - // Do nothing + close(); + } + + @Override + public synchronized void close() { + if (executor != null) { + executor.shutdownNow(); + executor = null; + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index e61367ed896a0..8381e44f75335 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -242,11 +242,19 @@ public void start() { if (isLoadSingleTsFileSuccess) { node.clean(); - LOGGER.info( - "Load TsFile {} Successfully, load process [{}/{}]", - filePath, - i + 1, - tsFileNodeListSize); + if (isGeneratedByPipe) { + LOGGER.debug( + "Load TsFile {} Successfully, load process [{}/{}]", + filePath, + i + 1, + tsFileNodeListSize); + } else { + LOGGER.info( + "Load TsFile {} Successfully, load process [{}/{}]", + filePath, + i + 1, + tsFileNodeListSize); + } } else { isLoadSuccess = false; failedTsFileNodeIndexes.add(i); @@ -299,6 +307,7 @@ public void start() { } } } finally { + dispatcher.close(); LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock(); } } @@ -387,7 +396,11 @@ private boolean dispatchOnePieceNode( private boolean secondPhase( boolean isFirstPhaseSuccess, String uuid, TsFileResource tsFileResource) { - LOGGER.info("Start dispatching Load command for uuid {}", uuid); + if (isGeneratedByPipe) { + LOGGER.debug("Start dispatching Load command for uuid {}", uuid); + } else { + LOGGER.info("Start dispatching Load command for uuid {}", uuid); + } final File tsFile = tsFileResource.getTsFile(); final TLoadCommandReq loadCommandReq = new TLoadCommandReq( @@ -594,7 +607,7 @@ private void convertFailedTsFilesToTabletsAndRetry() { @Override public void stop(Throwable t) { - // Do nothing + dispatcher.abort(); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index d06362ffd3794..5568aa4015c6d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -3244,10 +3244,17 @@ public void loadNewTsFile( 0); if (!newFileName.equals(tsfileToBeInserted.getName())) { - logger.info( - "TsFile {} must be renamed to {} for loading into the unsequence list.", - tsfileToBeInserted.getName(), - newFileName); + if (isGeneratedByPipe) { + logger.debug( + "TsFile {} must be renamed to {} for loading into the unsequence list.", + tsfileToBeInserted.getName(), + newFileName); + } else { + logger.info( + "TsFile {} must be renamed to {} for loading into the unsequence list.", + tsfileToBeInserted.getName(), + newFileName); + } newTsFileResource.setFile( fsFactory.getFile(tsfileToBeInserted.getParentFile(), newFileName)); } @@ -3282,7 +3289,11 @@ public void loadNewTsFile( } onTsFileLoaded(newTsFileResource, isFromConsensus, lastReader); - logger.info("TsFile {} is successfully loaded in unsequence list.", newFileName); + if (isGeneratedByPipe) { + logger.debug("TsFile {} is successfully loaded in unsequence list.", newFileName); + } else { + logger.info("TsFile {} is successfully loaded in unsequence list.", newFileName); + } } catch (final DiskSpaceInsufficientException e) { logger.error( "Failed to append the tsfile {} to database processor {} because the disk space is insufficient.", @@ -3451,10 +3462,17 @@ private boolean loadTsFileToUnSequence( return false; } - logger.info( - "Load tsfile in unsequence list, move file from {} to {}", - tsFileToLoad.getAbsolutePath(), - targetFile.getAbsolutePath()); + if (isGeneratedByPipe) { + logger.debug( + "Load tsfile in unsequence list, move file from {} to {}", + tsFileToLoad.getAbsolutePath(), + targetFile.getAbsolutePath()); + } else { + logger.info( + "Load tsfile in unsequence list, move file from {} to {}", + tsFileToLoad.getAbsolutePath(), + targetFile.getAbsolutePath()); + } LoadTsFileRateLimiter.getInstance().acquire(tsFileResource.getTsFile().length()); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java index 3223305dc80f3..b1e29c8563d34 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java @@ -523,7 +523,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool return; } - LOGGER.info( + LOGGER.debug( "Receiver id = {}: Writing file {} is not existed or name is not correct, try to create it. " + "Current writing file is {}.", receiverId.get(), @@ -541,7 +541,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool // This may be useless, because receiver file dir is created when handshake. just in case. if (!receiverFileDirWithIdSuffix.get().exists()) { if (receiverFileDirWithIdSuffix.get().mkdirs()) { - LOGGER.info( + LOGGER.debug( "Receiver id = {}: Receiver file dir {} was created.", receiverId.get(), receiverFileDirWithIdSuffix.get().getPath()); @@ -556,7 +556,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool writingFile = targetPath.toFile(); writingFileWriter = new RandomAccessFile(writingFile, "rw"); - LOGGER.info( + LOGGER.debug( "Receiver id = {}: Writing file {} was created. Ready to write file pieces.", receiverId.get(), writingFile.getPath()); @@ -730,7 +730,7 @@ protected final TPipeTransferResp handleTransferFileSealV1(final PipeTransferFil final TSStatus status = loadFileV1(req, fileAbsolutePath); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { shouldDeleteSealedFile = false; - LOGGER.info( + LOGGER.debug( "Receiver id = {}: Seal file {} successfully.", receiverId.get(), fileAbsolutePath); } else { PipeLogger.log( @@ -835,7 +835,7 @@ protected final TPipeTransferResp handleTransferFileSealV2(final PipeTransferFil final TSStatus status = loadFileV2(req, fileAbsolutePaths); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.info( + LOGGER.debug( "Receiver id = {}: Seal file {} successfully.", receiverId.get(), fileAbsolutePaths); } else { PipeLogger.log( diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java index 7aaa66a8b752a..ebf0898c019af 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java @@ -120,14 +120,14 @@ public static void cleanPipeReceiverDir(final File receiverFileDir) { FileUtils.deleteDirectory(receiverFileDir); return null; }); - LOGGER.info("Clean pipe receiver dir {} successfully.", receiverFileDir); + LOGGER.debug("Clean pipe receiver dir {} successfully.", receiverFileDir); } catch (final Exception e) { LOGGER.warn("Clean pipe receiver dir {} failed.", receiverFileDir, e); } try { FileUtils.forceMkdir(receiverFileDir); - LOGGER.info("Create pipe receiver dir {} successfully.", receiverFileDir); + LOGGER.debug("Create pipe receiver dir {} successfully.", receiverFileDir); } catch (final IOException e) { LOGGER.warn("Create pipe receiver dir {} failed.", receiverFileDir, e); }