This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/reduce-pipe-load-log-noise in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ea1545e572cbfb6fbd5aceb565132c8424ca50a9 Author: Caideyipi <[email protected]> AuthorDate: Fri Jul 3 13:00:05 2026 +0800 Reduce pipe load success log noise --- .../plan/analyze/load/LoadTsFileAnalyzer.java | 49 ++++++++++++----- .../scheduler/load/LoadTsFileDispatcherImpl.java | 64 ++++++++++++++-------- .../plan/scheduler/load/LoadTsFileScheduler.java | 27 ++++++--- .../db/storageengine/dataregion/DataRegion.java | 6 +- .../compaction/execute/utils/CompactionUtils.java | 2 +- .../db/storageengine/load/LoadTsFileManager.java | 2 +- .../commons/pipe/receiver/IoTDBFileReceiver.java | 10 ++-- .../commons/pipe/receiver/IoTDBReceiverAgent.java | 4 +- 8 files changed, 108 insertions(+), 56 deletions(-) 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 b1bdd077dc4..812b62ff721 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 @@ -237,7 +237,11 @@ public class LoadTsFileAnalyzer implements AutoCloseable { return analysis; } - LOGGER.info(DataNodeQueryMessages.LOAD_ANALYSIS_STAGE_ALL_TSFILES_HAVE_BEEN_ANALYZED); + if (isGeneratedByPipe) { + LOGGER.debug(DataNodeQueryMessages.LOAD_ANALYSIS_STAGE_ALL_TSFILES_HAVE_BEEN_ANALYZED); + } else { + LOGGER.info(DataNodeQueryMessages.LOAD_ANALYSIS_STAGE_ALL_TSFILES_HAVE_BEEN_ANALYZED); + } setTsFileModelInfoToStatement(); if (reconstructStatementIfMiniFileConverted()) { @@ -318,22 +322,16 @@ public class LoadTsFileAnalyzer implements AutoCloseable { if (LOGGER.isWarnEnabled()) { LOGGER.warn(DataNodeQueryMessages.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; @@ -370,14 +368,39 @@ public class LoadTsFileAnalyzer implements AutoCloseable { 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 i) throws Exception { final SessionInfo sessionInfo = context.getSession(); try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) { // check whether the tsfile is tree-model or not final Map<String, TableSchema> tableSchemaMap = reader.getTableSchemaMap(); final boolean isTableModelFile = Objects.nonNull(tableSchemaMap) && !tableSchemaMap.isEmpty(); - LOGGER.info( - "TsFile {} is a {}-model file.", tsFile.getPath(), isTableModelFile ? "table" : "tree"); + if (isGeneratedByPipe) { + LOGGER.debug( + "TsFile {} is a {}-model file.", tsFile.getPath(), isTableModelFile ? "table" : "tree"); + } else { + LOGGER.info( + "TsFile {} is a {}-model file.", tsFile.getPath(), isTableModelFile ? "table" : "tree"); + } // can be reused when constructing tsfile resource final TsFileSequenceReaderTimeseriesMetadataIterator timeseriesMetadataIterator = 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 bd5356ccff4..39d1d79ec82 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 @@ -74,7 +74,7 @@ import java.util.concurrent.atomic.AtomicInteger; 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); @@ -88,7 +88,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { private final int localhostInternalPort; private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> internalServiceClientManager; - private final ExecutorService executor; + private ExecutorService executor; private final boolean isGeneratedByPipe; public LoadTsFileDispatcherImpl( @@ -97,11 +97,17 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { 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; } @@ -109,24 +115,26 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { @Override public Future<FragInstanceDispatchResult> dispatch( SubPlan root, List<FragmentInstance> 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(DataNodeQueryMessages.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(DataNodeQueryMessages.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) @@ -152,7 +160,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { } public void dispatchLocally(FragmentInstance instance) throws FragmentInstanceDispatchException { - LOGGER.info(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid); + LOGGER.debug(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid); ConsensusGroupId groupId = ConsensusGroupId.Factory.createFromTConsensusGroupId( @@ -360,6 +368,14 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { @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 e2db158475a..205b9e8ec3f 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 @@ -251,11 +251,19 @@ public class LoadTsFileScheduler implements IScheduler { 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); @@ -308,6 +316,7 @@ public class LoadTsFileScheduler implements IScheduler { } } } finally { + dispatcher.close(); LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock(); } } @@ -398,7 +407,11 @@ public class LoadTsFileScheduler implements IScheduler { private boolean secondPhase( boolean isFirstPhaseSuccess, String uuid, TsFileResource tsFileResource) { - LOGGER.info(DataNodeQueryMessages.START_DISPATCHING_LOAD_COMMAND_FOR_UUID, uuid); + if (isGeneratedByPipe) { + LOGGER.debug(DataNodeQueryMessages.START_DISPATCHING_LOAD_COMMAND_FOR_UUID, uuid); + } else { + LOGGER.info(DataNodeQueryMessages.START_DISPATCHING_LOAD_COMMAND_FOR_UUID, uuid); + } final File tsFile = tsFileResource.getTsFile(); final TLoadCommandReq loadCommandReq = new TLoadCommandReq( @@ -662,7 +675,7 @@ public class LoadTsFileScheduler implements IScheduler { @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 03807c60d74..06e0876e3a6 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 @@ -4144,7 +4144,7 @@ public class DataRegion implements IDataRegionForQuery { 0); if (!newFileName.equals(tsfileToBeInserted.getName())) { - logger.info( + logger.debug( "TsFile {} must be renamed to {} for loading into the unsequence list.", tsfileToBeInserted.getName(), newFileName); @@ -4196,7 +4196,7 @@ public class DataRegion implements IDataRegionForQuery { } onTsFileLoaded(newTsFileResource, isFromConsensus, lastReader); - logger.info(StorageEngineMessages.TSFILE_LOADED_IN_UNSEQ_LIST, newFileName); + logger.debug(StorageEngineMessages.TSFILE_LOADED_IN_UNSEQ_LIST, newFileName); } catch (final DiskSpaceInsufficientException e) { logger.error( "Failed to append the tsfile {} to database processor {} because the disk space is insufficient.", @@ -4350,7 +4350,7 @@ public class DataRegion implements IDataRegionForQuery { return false; } - logger.info( + logger.debug( "Load tsfile in unsequence list, move file from {} to {}", tsFileToLoad.getAbsolutePath(), targetFile.getAbsolutePath()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java index 7bdd492fa74..cd0204ae754 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java @@ -411,7 +411,7 @@ public class CompactionUtils { "[Compaction] delete file failed, file path is {}", resource.getTsFile().getAbsolutePath()); } else { - logger.info( + logger.debug( StorageEngineMessages.COMPACTION_DELETE_FILE, resource.getTsFile().getAbsolutePath()); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java index 05e53a90763..fb4e4412f61 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/LoadTsFileManager.java @@ -463,7 +463,7 @@ public class LoadTsFileManager { FileUtils.deleteFileOrDirectoryWithRetry(dir); } if (dir.mkdirs()) { - LOGGER.info(StorageEngineMessages.LOAD_TSFILE_DIR_CREATED, dir.getPath()); + LOGGER.debug(StorageEngineMessages.LOAD_TSFILE_DIR_CREATED, dir.getPath()); } } 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 8304161c9d0..591d9d19a55 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 @@ -441,7 +441,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver { return; } - LOGGER.info( + LOGGER.debug( PipeMessages.RECEIVER_WRITING_FILE_NOT_EXIST, receiverId.get(), fileName, @@ -458,7 +458,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver { // 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( PipeMessages.RECEIVER_FILE_DIR_CREATED, receiverId.get(), receiverFileDirWithIdSuffix.get().getPath()); @@ -473,7 +473,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver { writingFile = targetPath.toFile(); writingFileWriter = new RandomAccessFile(writingFile, "rw"); - LOGGER.info( + LOGGER.debug( PipeMessages.RECEIVER_WRITING_FILE_CREATED, receiverId.get(), writingFile.getPath()); } @@ -630,7 +630,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver { final TSStatus status = loadFileV1(req, fileAbsolutePath); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { shouldDeleteSealedFile = false; - LOGGER.info(PipeMessages.RECEIVER_SEAL_FILE_SUCCESS, receiverId.get(), fileAbsolutePath); + LOGGER.debug(PipeMessages.RECEIVER_SEAL_FILE_SUCCESS, receiverId.get(), fileAbsolutePath); } else { PipeLogger.log( LOGGER::warn, @@ -738,7 +738,7 @@ public abstract class IoTDBFileReceiver implements IoTDBReceiver { final TSStatus status = loadFileV2(req, fileAbsolutePaths); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.info(PipeMessages.RECEIVER_SEAL_FILE_SUCCESS, receiverId.get(), fileAbsolutePaths); + LOGGER.debug(PipeMessages.RECEIVER_SEAL_FILE_SUCCESS, receiverId.get(), fileAbsolutePaths); } else { PipeLogger.log( LOGGER::warn, 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 c51f36d78f0..e718300978b 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 @@ -117,14 +117,14 @@ public abstract class IoTDBReceiverAgent { FileUtils.deleteDirectory(receiverFileDir); return null; }); - LOGGER.info(PipeMessages.CLEAN_RECEIVER_DIR_SUCCESS, receiverFileDir); + LOGGER.debug(PipeMessages.CLEAN_RECEIVER_DIR_SUCCESS, receiverFileDir); } catch (final Exception e) { LOGGER.warn(PipeMessages.CLEAN_RECEIVER_DIR_FAILED, receiverFileDir, e); } try { FileUtils.forceMkdir(receiverFileDir); - LOGGER.info(PipeMessages.CREATE_RECEIVER_DIR_SUCCESS, receiverFileDir); + LOGGER.debug(PipeMessages.CREATE_RECEIVER_DIR_SUCCESS, receiverFileDir); } catch (final IOException e) { LOGGER.warn(PipeMessages.CREATE_RECEIVER_DIR_FAILED, receiverFileDir, e); }
