apache · jt2594838 · Jul 10, 2026 · Jul 3, 2026 · Jul 7, 2026 · Jul 9, 2026
diff --git a/...est/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java b/...est/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java
@@ -41,6 +41,7 @@
 import java.nio.channels.FileChannel;
 import java.nio.file.StandardOpenOption;
 import java.sql.Connection;
+import java.sql.ResultSet;
 import java.sql.Statement;
 import java.util.HashMap;
 import java.util.Map;
@@ -101,11 +102,13 @@ public void showConfigurationDisplaysEffectiveValue() throws Exception {
             + "conf"
             + File.separator
             + "iotdb-system.properties";
+    int dataNodeId = findDataNodeIdByRpcPort(dataNodeWrapper.getPort());
 
     // Case 1: non-positive guard params (`if (x > 0) setX(x)`) must display the effective
     // default value, not the raw file value (0 / -1).
     Map<String, String> afterGuardReload =
         appendLinesLoadAndShow(
+            dataNodeId,
             confPath,
             "cte_buffer_size_in_bytes=0",
             "max_rows_in_cte_buffer=-1",
@@ -116,20 +119,20 @@ public void showConfigurationDisplaysEffectiveValue() throws Exception {
 
     // Case 2: a valid value is not rewritten, so display must equal the file value (regression).
     Map<String, String> afterValidReload =
-        appendLinesLoadAndShow(confPath, "cte_buffer_size_in_bytes=262144");
+        appendLinesLoadAndShow(dataNodeId, confPath, "cte_buffer_size_in_bytes=262144");
     Assert.assertEquals("262144", afterValidReload.get("cte_buffer_size_in_bytes"));
 
     // Case 3: params whose template default is 0 are always rewritten to a computed default,
     // so they must never display 0 (this was always wrong, even without a bad file value).
-    Map<String, String> defaultsReload = appendLinesLoadAndShow(confPath);
+    Map<String, String> defaultsReload = appendLinesLoadAndShow(dataNodeId, confPath);
     assertPositiveNonZero(defaultsReload, "sort_buffer_size_in_bytes");
     assertPositiveNonZero(defaultsReload, "mods_cache_size_limit_per_fi_in_bytes");
   }
 
   // Appends the given lines to the config file, runs `load configuration`, fetches the
   // `show configuration` result, and restores the file to its original length.
-  private Map<String, String> appendLinesLoadAndShow(String confPath, String... lines)
-      throws Exception {
+  private Map<String, String> appendLinesLoadAndShow(
+      int dataNodeId, String confPath, String... lines) throws Exception {
     long length = new File(confPath).length();
     try {
       if (lines.length > 0) {
@@ -145,7 +148,7 @@ private Map<String, String> appendLinesLoadAndShow(String confPath, String... li
           Statement statement = connection.createStatement()) {
         statement.execute("LOAD CONFIGURATION");
       }
-      return fetchShowConfiguration();
+      return fetchShowConfiguration(dataNodeId);
     } finally {
       try (FileChannel fileChannel =
           FileChannel.open(new File(confPath).toPath(), StandardOpenOption.WRITE)) {
@@ -154,11 +157,24 @@ private Map<String, String> appendLinesLoadAndShow(String confPath, String... li
     }
   }
 
-  private Map<String, String> fetchShowConfiguration() throws Exception {
+  private int findDataNodeIdByRpcPort(int rpcPort) throws Exception {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement();
+        ResultSet resultSet = statement.executeQuery("SHOW DATANODES")) {
+      while (resultSet.next()) {
+        if (resultSet.getInt(4) == rpcPort) {
+          return resultSet.getInt(1);
+        }
+      }
+    }
+    throw new AssertionError("Cannot find DataNode id for RPC port: " + rpcPort);
+  }
+
+  private Map<String, String> fetchShowConfiguration(int dataNodeId) throws Exception {
     Map<String, String> result = new HashMap<>();
     try (ITableSession tableSessionConnection = EnvFactory.getEnv().getTableSessionConnection()) {
       SessionDataSet sessionDataSet =
-          tableSessionConnection.executeQueryStatement("show configuration");
+          tableSessionConnection.executeQueryStatement("show configuration on " + dataNodeId);
       SessionDataSet.DataIterator iterator = sessionDataSet.iterator();
       while (iterator.next()) {
         String name = iterator.getString(1);

diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -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 =

diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -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 =

diff --git a/...e/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/...e/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
@@ -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()) {
@@ -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;
@@ -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 {
     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 =

diff --git a/...org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/SchemaPredicateUtil.java b/...org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/SchemaPredicateUtil.java
@@ -226,7 +226,8 @@ static List<Integer> extractTagSingleMatchExpressionCases(
     final int tagCount = tableInstance.getTagNum();
     for (int i = 0; i < index2FilterMapList.size(); i++) {
       final Map<Integer, List<SchemaFilter>> filterMap = index2FilterMapList.get(i);
-      if (filterMap.size() == tagCount
+      if (tagCount > 0
+          && filterMap.size() == tagCount
           && filterMap.values().stream()
               .allMatch(
                   filterList ->

diff --git a/...in/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java b/...in/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
@@ -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);
 
@@ -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,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())));
+                }
+              }
+              return new FragInstanceDispatchResult(true);
+            });
   }
 
   private void dispatchOneInstance(FragmentInstance instance)
@@ -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(
@@ -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));
@@ -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;
+    }
   }
 }
diff --git a/...rc/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/...rc/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
@@ -171,7 +171,7 @@
  }

  @Override
  public void start() {
    try {
      stateMachine.transitionToRunning();
      int tsFileNodeListSize = tsFileNodeList.size();
@@ -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);
@@ -313,6 +321,7 @@
         }
       }
     } finally {
+      dispatcher.close();
       LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock();
     }
   }
@@ -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(
@@ -686,7 +699,7 @@
 
   @Override
   public void stop(Throwable t) {
-    // Do nothing
+    dispatcher.abort();
   }
 
   @Override