Skip to content

Reduce pipe load success log noise #18106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Reduce pipe load success log noise #18106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e5091cb
Reduce pipe load success log noise
Caideyipi Jul 3, 2026
4828508
Merge branch 'master' of https://github.com/apache/iotdb into fix/red…
Caideyipi Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1112,6 +1112,8 @@ public final class DataNodeQueryMessages {
"cancel query {} on node {} failed.";
public static final String CANNOT_DISPATCH_FI_FOR_LOAD_OPERATION =
"cannot dispatch FI for load operation";
public static final String MESSAGE_UNEXPECTED_ERRORS_ARG_78EE0800 =
"Unexpected errors: %s";
public static final String RECEIVE_LOAD_NODE_FROM_UUID =
"Receive load node from uuid {}.";
public static final String LOAD_TSFILE_NODE_ERROR =
Expand Down
2 changes: 2 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1096,6 +1096,8 @@ public final class DataNodeQueryMessages {
"在节点 {} 上取消查询 {} 失败。";
public static final String CANNOT_DISPATCH_FI_FOR_LOAD_OPERATION =
"无法为加载操作分发 FI";
public static final String MESSAGE_UNEXPECTED_ERRORS_ARG_78EE0800 =
"意外错误：%s";
public static final String RECEIVE_LOAD_NODE_FROM_UUID =
"接收来自 uuid {} 的加载节点。";
public static final String LOAD_TSFILE_NODE_ERROR =
Expand Down
65 changes: 44 additions & 21 deletions ...e/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,7 +237,11 @@
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()) {
Expand Down Expand Up @@ -319,28 +323,14 @@
if (LOGGER.isWarnEnabled()) {
LOGGER.warn(DataNodeQueryMessages.TSFILE_IS_EMPTY, tsFile.getPath());
}
if (LOGGER.isInfoEnabled()) {
LOGGER.info(
DataNodeQueryMessages
.LOAD_ANALYSIS_STAGE_ARG_ARG_TSFILES_HAVE_BEEN_ANALYZED_PROGRESS_ARG_PERCENT,
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(
DataNodeQueryMessages
.LOAD_ANALYSIS_STAGE_ARG_ARG_TSFILES_HAVE_BEEN_ANALYZED_PROGRESS_ARG_PERCENT,
i + 1,
tsfileNum,
String.format("%.3f", (i + 1) * 100.00 / tsfileNum));
}
logAnalyzeProgress(i + 1, tsfileNum);
} catch (AuthException e) {
setFailAnalysisForAuthException(analysis, e);
return false;
Expand Down Expand Up @@ -379,16 +369,49 @@
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(
DataNodeQueryMessages
.LOAD_ANALYSIS_STAGE_ARG_ARG_TSFILES_HAVE_BEEN_ANALYZED_PROGRESS_ARG_PERCENT,
analyzedTsFileNum,
totalTsFileNum,
progress);
} else {
LOGGER.info(
DataNodeQueryMessages
.LOAD_ANALYSIS_STAGE_ARG_ARG_TSFILES_HAVE_BEEN_ANALYZED_PROGRESS_ARG_PERCENT,
analyzedTsFileNum,
totalTsFileNum,
progress);
}
}

private void analyzeSingleTsFile(final File tsFile, int i) throws Exception {

Check failure on line 398 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this method to reduce its Cognitive Complexity from 17 to the 15 allowed. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ8mYNpxq2C1CtAez8Ni&open=AZ8mYNpxq2C1CtAez8Ni&pullRequest=18106
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(
DataNodeQueryMessages.TSFILE_ARG_IS_A_ARG_MODEL_FILE,
tsFile.getPath(),
isTableModelFile ? "table" : "tree");
if (isGeneratedByPipe) {
LOGGER.debug(
DataNodeQueryMessages.TSFILE_ARG_IS_A_ARG_MODEL_FILE,
tsFile.getPath(),
isTableModelFile ? "table" : "tree");
} else {
LOGGER.info(
DataNodeQueryMessages.TSFILE_ARG_IS_A_ARG_MODEL_FILE,
tsFile.getPath(),
isTableModelFile ? "table" : "tree");
}

// can be reused when constructing tsfile resource
final TsFileSequenceReaderTimeseriesMetadataIterator timeseriesMetadataIterator =
Expand Down
75 changes: 50 additions & 25 deletions ...in/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,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);

Expand All @@ -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(
Expand All @@ -97,36 +97,46 @@ 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;
}

@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,
String.format(
DataNodeQueryMessages.MESSAGE_UNEXPECTED_ERRORS_ARG_78EE0800,
t.getMessage())));
}
Comment on lines +131 to +136

@jt2594838 jt2594838 Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i18n

@Caideyipi Caideyipi Jul 7, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. Added en/zh i18n constants for the unexpected-error status message and switched both dispatch paths to use them.

}
return new FragInstanceDispatchResult(true);
});
}

private void dispatchOneInstance(FragmentInstance instance)
Expand All @@ -152,7 +162,11 @@ private void dispatchOneInstance(FragmentInstance instance)
}

public void dispatchLocally(FragmentInstance instance) throws FragmentInstanceDispatchException {
LOGGER.info(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid);
if (isGeneratedByPipe) {
LOGGER.debug(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid);
} else {
LOGGER.info(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid);
}

ConsensusGroupId groupId =
ConsensusGroupId.Factory.createFromTConsensusGroupId(
Expand Down Expand Up @@ -270,7 +284,10 @@ public Future<FragInstanceDispatchResult> dispatchCommand(
return immediateFuture(
new FragInstanceDispatchResult(
RpcUtils.getStatus(
TSStatusCode.INTERNAL_SERVER_ERROR, "Unexpected errors: " + t.getMessage())));
TSStatusCode.INTERNAL_SERVER_ERROR,
String.format(
DataNodeQueryMessages.MESSAGE_UNEXPECTED_ERRORS_ARG_78EE0800,
t.getMessage()))));
}
}
return immediateFuture(new FragInstanceDispatchResult(true));
Expand Down Expand Up @@ -365,6 +382,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;
}
}
}
27 changes: 20 additions & 7 deletions ...rc/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,7 @@
}

@Override
public void start() {

Check warning on line 174 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

A "Brain Method" was detected. Refactor it to reduce at least one of the following metrics: LOC from 140 to 64, Complexity from 15 to 14, Nesting Level from 5 to 2, Number of Variables from 18 to 6. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ8mYNn3q2C1CtAez8Nh&open=AZ8mYNn3q2C1CtAez8Nh&pullRequest=18106
try {
stateMachine.transitionToRunning();
int tsFileNodeListSize = tsFileNodeList.size();
Expand Down Expand Up @@ -255,11 +255,19 @@

if (isLoadSingleTsFileSuccess) {
node.clean();
LOGGER.info(
DataNodeQueryMessages.LOAD_TSFILE_ARG_SUCCESSFULLY_LOAD_PROCESS_ARG_ARG,
filePath,
i + 1,
tsFileNodeListSize);
if (isGeneratedByPipe) {
LOGGER.debug(
DataNodeQueryMessages.LOAD_TSFILE_ARG_SUCCESSFULLY_LOAD_PROCESS_ARG_ARG,
filePath,
i + 1,
tsFileNodeListSize);
} else {
LOGGER.info(
DataNodeQueryMessages.LOAD_TSFILE_ARG_SUCCESSFULLY_LOAD_PROCESS_ARG_ARG,
filePath,
i + 1,
tsFileNodeListSize);
}
} else {
isLoadSuccess = false;
failedTsFileNodeIndexes.add(i);
Expand Down Expand Up @@ -313,6 +321,7 @@
}
}
} finally {
dispatcher.close();
LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock();
}
}
Expand Down Expand Up @@ -412,7 +421,11 @@

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(
Expand Down Expand Up @@ -686,7 +699,7 @@

@Override
public void stop(Throwable t) {
// Do nothing
dispatcher.abort();
}

@Override
Expand Down
41 changes: 31 additions & 10 deletions ...-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2668,7 +2668,7 @@
* lock if return true, the caller is responsible for unlocking all the already-acquiring lock
* in needToUnLockList
*/
private boolean tryGetFLushLock(

Check warning on line 2671 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

A "Brain Method" was detected. Refactor it to reduce at least one of the following metrics: LOC from 81 to 64, Complexity from 19 to 14, Nesting Level from 5 to 2, Number of Variables from 17 to 6. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ8mYNiTq2C1CtAez8Nd&open=AZ8mYNiTq2C1CtAez8Nd&pullRequest=18106
long waitTimeInMs,
IDeviceID singleDeviceId,
Filter globalTimeFilter,
Expand Down Expand Up @@ -3107,7 +3107,7 @@
}
}

public void deleteByTable(RelationalDeleteDataNode node) throws IOException {

Check warning on line 3110 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

A "Brain Method" was detected. Refactor it to reduce at least one of the following metrics: LOC from 98 to 64, Complexity from 17 to 14, Nesting Level from 7 to 2, Number of Variables from 24 to 6. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ8mYNiTq2C1CtAez8Ne&open=AZ8mYNiTq2C1CtAez8Ne&pullRequest=18106
if (node.getDatabaseName() != null && !node.getDatabaseName().equals(databaseName)) {
// not targeted on this database, return
return;
Expand Down Expand Up @@ -3522,7 +3522,7 @@
}
}

private void deleteDataInSealedFiles(Collection<TsFileResource> sealedTsFiles, ModEntry deletion)

Check warning on line 3525 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

A "Brain Method" was detected. Refactor it to reduce at least one of the following metrics: LOC from 155 to 64, Complexity from 29 to 14, Nesting Level from 5 to 2, Number of Variables from 26 to 6. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ8mYNiTq2C1CtAez8Nf&open=AZ8mYNiTq2C1CtAez8Nf&pullRequest=18106
throws IOException {
Set<ModificationFile> involvedModificationFiles = new HashSet<>();
List<TsFileResource> deletedByMods = new ArrayList<>();
Expand Down Expand Up @@ -4131,7 +4131,7 @@
* @param deleteOriginFile whether to delete origin tsfile
* @param isGeneratedByPipe whether the load tsfile request is generated by pipe
*/
public void loadNewTsFile(

Check warning on line 4134 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

A "Brain Method" was detected. Refactor it to reduce at least one of the following metrics: LOC from 136 to 64, Complexity from 16 to 14, Nesting Level from 3 to 2, Number of Variables from 18 to 6. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AZ8mr9gfoBc5PoNg8qNt&open=AZ8mr9gfoBc5PoNg8qNt&pullRequest=18106
final TsFileResource newTsFileResource,
final boolean deleteOriginFile,
final boolean isGeneratedByPipe,
Expand Down Expand Up @@ -4196,11 +4196,19 @@
0);

if (!newFileName.equals(tsfileToBeInserted.getName())) {
logger.info(
StorageEngineMessages
.STORAGE_LOG_TSFILE_MUST_BE_RENAMED_TO_FOR_LOADING_INTO_THE_UNSEQUENCE_70321619,
tsfileToBeInserted.getName(),
newFileName);
if (isGeneratedByPipe) {
logger.debug(
StorageEngineMessages
.STORAGE_LOG_TSFILE_MUST_BE_RENAMED_TO_FOR_LOADING_INTO_THE_UNSEQUENCE_70321619,
tsfileToBeInserted.getName(),
newFileName);
} else {
logger.info(
StorageEngineMessages
.STORAGE_LOG_TSFILE_MUST_BE_RENAMED_TO_FOR_LOADING_INTO_THE_UNSEQUENCE_70321619,
tsfileToBeInserted.getName(),
newFileName);
}
newTsFileResource.setFile(
fsFactory.getFile(tsfileToBeInserted.getParentFile(), newFileName));
}
Expand Down Expand Up @@ -4249,7 +4257,11 @@
}

onTsFileLoaded(newTsFileResource, isFromConsensus, lastReader);
logger.info(StorageEngineMessages.TSFILE_LOADED_IN_UNSEQ_LIST, newFileName);
if (isGeneratedByPipe) {
logger.debug(StorageEngineMessages.TSFILE_LOADED_IN_UNSEQ_LIST, newFileName);
} else {
logger.info(StorageEngineMessages.TSFILE_LOADED_IN_UNSEQ_LIST, newFileName);
}
} catch (final DiskSpaceInsufficientException e) {
logger.error(
StorageEngineMessages
Expand Down Expand Up @@ -4405,10 +4417,19 @@
return false;
}

logger.info(
StorageEngineMessages.STORAGE_LOG_LOAD_TSFILE_IN_UNSEQUENCE_LIST_MOVE_FILE_FROM_TO_21E11AEB,
tsFileToLoad.getAbsolutePath(),
targetFile.getAbsolutePath());
if (isGeneratedByPipe) {
logger.debug(
StorageEngineMessages
.STORAGE_LOG_LOAD_TSFILE_IN_UNSEQUENCE_LIST_MOVE_FILE_FROM_TO_21E11AEB,
tsFileToLoad.getAbsolutePath(),
targetFile.getAbsolutePath());
} else {
logger.info(
StorageEngineMessages
.STORAGE_LOG_LOAD_TSFILE_IN_UNSEQUENCE_LIST_MOVE_FILE_FROM_TO_21E11AEB,
tsFileToLoad.getAbsolutePath(),
targetFile.getAbsolutePath());
}

LoadTsFileRateLimiter.getInstance().acquire(tsFileResource.getTsFile().length());

Expand Down
Loading
Loading