When a new plan type is introduced, the top-level dispatch and the corresponding visitor + * method must be updated together. Default visitor methods should document whether they delegate + * to the top-level handler, return a default value, or reject the plan. + */ public abstract R visitPlan(final ConfigPhysicalPlan plan, final C context); public R visitCreateDatabase(final DatabaseSchemaPlan createDatabasePlan, final C context) { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java index 790f09c08b9f..b9b92e443da2 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java @@ -84,6 +84,12 @@ public String getClusterId() { return clusterInfo.getClusterId(); } + /** + * Gets the cluster ID with bounded retries when the local ConfigRegion is not ready. + * + * @return the cluster ID + * @throws ConsensusException if the cluster ID cannot be read after the retry policy is exhausted + */ public String getClusterIdWithRetry(long maxWaitTime) { long startTime = System.currentTimeMillis(); while (clusterInfo.getClusterId() == null @@ -110,6 +116,12 @@ private void generateClusterId() { } // TODO: Parallel test ConfigNode and DataNode + /** + * Submits connectivity checks to all known ConfigNodes and DataNodes. + * + *
The method is asynchronous; callers must use the returned or associated result mechanism
+ * to determine which nodes succeeded or failed.
+ */
public TTestConnectionResp submitTestConnectionTaskToEveryNode() {
TTestConnectionResp resp = new TTestConnectionResp();
resp.resultList = new ArrayList<>();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index f86681f2bbec..af3e6a2cf3e3 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -310,7 +310,10 @@
import static org.apache.iotdb.commons.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCARD;
import static org.apache.iotdb.commons.schema.table.Audit.TREE_MODEL_AUDIT_DATABASE;
-/** Entry of all management, AssignPartitionManager, AssignRegionManager. */
+/**
+ * Coordinates ConfigNode control-plane managers and routes requests related to nodes, schema,
+ * partitions, procedures, load management, plugins, quotas, TTL, and subscriptions.
+ */
public class ConfigManager implements IManager {
private static final Logger LOGGER = LoggerFactory.getLogger(ConfigManager.class);
@@ -318,7 +321,7 @@ public class ConfigManager implements IManager {
private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
private static final CommonConfig COMMON_CONF = CommonDescriptor.getInstance().getConfig();
- /** Manage PartitionTable read/write requests through the ConsensusLayer. */
+ /** Manages replicated ConfigRegion plans and reads through the consensus layer. */
private final AtomicReference Notice: The result will be an empty TDataNodeConfiguration if the specified DataNode doesn't
* register
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
index 28655923cca9..903964c855b3 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
@@ -151,8 +151,7 @@ public class PartitionManager {
// Monitor for leadership change
private final Object scheduleMonitor = new Object();
- /** Region cleaner. */
- // Try to delete Regions in every 10s
+ /** Period, in seconds, at which the region maintainer performs maintenance and cleanup. */
private static final int REGION_MAINTAINER_WORK_INTERVAL = 10;
private final ScheduledExecutorService regionMaintainer;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
index 12f6a2baaa18..78867e30b9b1 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
@@ -102,8 +102,7 @@
import static org.apache.iotdb.commons.schema.SchemaConstant.ROOT;
import static org.apache.iotdb.commons.schema.SchemaConstant.TABLE_MNODE_TYPE;
-// Since the ConfigMTree is all stored in memory, thus it is not restricted to manage MNode through
-// MTreeStore.
+// ConfigMTree is fully memory-resident. MTreeStore is retained only for traverser-related APIs.
public class ConfigMTree {
private static final String TABLE_ERROR_MSG =
@@ -136,7 +135,8 @@ public void clear() {
// region database Management
/**
- * CREATE DATABASE. Make sure check seriesPath before setting database
+ * Create the database after validating the path;
+ * intermediate nodes are created when necessary.
*
* @param path path
*/
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
index 74141fa73f7f..5506620f0155 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
@@ -90,7 +90,7 @@ public final boolean hasLock() {
* The code to undo what was done by the execute() code. It is called when the procedure or one of
* the sub-procedures failed or an abort was requested. It should cleanup all the resources
* created by the execute() call. The implementation must be idempotent since rollback() may be
- * called multiple time in case of machine failure in the middle of the execution.
+ * called multiple times in case of machine failure in the middle of the execution.
*
* @param env the environment passed to the ProcedureExecutor
* @throws IOException temporary failure, the rollback will retry later
@@ -277,7 +277,7 @@ final void releaseExecution() {
}
/**
- * Internal method called by the ProcedureExecutor that starts the user-level code execute().
+ * Internal method called by the ProcedureExecutor that invokes the user-level execute().
*
* @param env execute environment
* @return sub procedures
@@ -292,7 +292,7 @@ protected Procedure The method must preserve the distinction between seed and non-seed ConfigNodes and must
+ * start the RPC service only after the local services required to handle requests are ready.
+ */
public void active() {
LOGGER.info(ConfigNodeMessages.ACTIVATING, ConfigNodeConstant.GLOBAL_NAME);
try {
+ // Process pid file, register deleteOnExit
processPid();
// Add shutdown hook
addShutDownHook();
@@ -503,6 +510,12 @@ public void deactivate() throws IOException {
LOGGER.info(ConfigNodeMessages.IS_DEACTIVATED, ConfigNodeConstant.GLOBAL_NAME);
}
+ /**
+ * Stops ConfigNode services and releases their resources in reverse dependency order.
+ *
+ * The operation should be safe to invoke during partial startup and should not leave
+ * background scheduling, RPC, or consensus resources running.
+ */
public void stop() {
try {
deactivate();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index b14f2e1b6222..7d8fb336c651 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -628,7 +628,7 @@ public class IoTDBConfig {
private long cacheFileReaderClearPeriod = 100000;
/** the max executing time of query in ms. Unit: millisecond */
- private long queryTimeoutThreshold = 60000;
+ private long queryTimeoutThreshold = 3600000; // 1 hour
/** the max time to live of a session in ms. Unit: millisecond */
private int sessionTimeoutThreshold = 0;
From c6f12212dc4b6930756edaf27368206ad2716389 Mon Sep 17 00:00:00 2001
From: WishArdently <3022210040@tju.edu.cn>
Date: Fri, 4 Sep 2026 20:34:35 +0800
Subject: [PATCH 2/4] update comments on DataNode
---
.../dataregion/DataRegionStateMachine.java | 10 ++++++-
.../SchemaRegionStateMachine.java | 7 ++++-
.../thrift/impl/ClientRPCServiceImpl.java | 7 +++++
.../impl/DataNodeInternalRPCServiceImpl.java | 7 +++++
.../exchange/MPPDataExchangeManager.java | 14 ++++++---
.../fragment/FragmentInstanceManager.java | 7 +++++
.../execution/schedule/DriverScheduler.java | 8 ++++-
.../db/queryengine/plan/Coordinator.java | 7 +++--
.../plan/execution/QueryExecution.java | 17 ++++++-----
.../plan/statement/StatementVisitor.java | 7 ++++-
.../iotdb/db/schemaengine/SchemaEngine.java | 14 +++++++--
.../org/apache/iotdb/db/service/DataNode.java | 26 +++++++++++++++--
.../db/service/DataNodeShutdownHook.java | 18 ++++++++----
.../iotdb/db/storageengine/StorageEngine.java | 29 +++++++++++++------
.../storageengine/dataregion/DataRegion.java | 2 +-
.../CompactionScheduleTaskManager.java | 7 +++++
.../schedule/CompactionTaskManager.java | 7 ++++-
.../dataregion/memtable/TsFileProcessor.java | 15 +++++++---
.../dataregion/wal/WALManager.java | 19 +++++++-----
.../dataregion/wal/buffer/WALBuffer.java | 6 ++--
.../wal/recover/WALRecoverManager.java | 6 +++-
.../subscription/agent/SubscriptionAgent.java | 6 ++++
.../receiver/SubscriptionReceiverV1.java | 7 +++++
23 files changed, 198 insertions(+), 55 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
index 051dcbb27719..17262086129d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
@@ -55,6 +55,13 @@
import java.util.List;
import java.util.function.Supplier;
+/**
+ * Applies replicated write and query fragment operations to one DataRegion and exposes its
+ * snapshot and region-resource lifecycle to the consensus layer.
+ *
+ * Write-process rejection is retried here to preserve the atomicity expected by the consensus
+ * apply path; other statuses are delegated to the consensus retry mechanism.
+ */
public class DataRegionStateMachine extends BaseStateMachine {
private static final Logger logger = LoggerFactory.getLogger(DataRegionStateMachine.class);
@@ -74,7 +81,8 @@ public DataRegionStateMachine(DataRegion region) {
@Override
public void start() {
- // do nothing
+ // The consensus implementation owns the start lifecycle; this state machine has no additional
+ // start action.
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java
index 9396d8cc3543..479231d8143d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java
@@ -46,6 +46,10 @@
import java.util.List;
import java.util.Objects;
+/**
+ * Applies replicated schema operations to one SchemaRegion and coordinates schema-region
+ * snapshot, Pipe-leader, and attribute-security lifecycle callbacks.
+ */
public class SchemaRegionStateMachine extends BaseStateMachine {
private static final Logger logger = LoggerFactory.getLogger(SchemaRegionStateMachine.class);
@@ -60,7 +64,8 @@ public SchemaRegionStateMachine(final ISchemaRegion schemaRegion) {
@Override
public void start() {
- // Do nothing
+ // The consensus implementation owns the start lifecycle; this state machine has no additional
+ // start action.
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
index be0a00bddf82..76c42642938a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java
@@ -254,6 +254,13 @@
import static org.apache.iotdb.rpc.RpcUtils.TIME_PRECISION;
import static org.apache.iotdb.rpc.TSStatusCode.QUERY_WAS_KILLED;
+/**
+ * Implements the client-facing RPC surface for sessions, SQL execution, writes, query results,
+ * metadata operations, authentication, and resource control.
+ *
+ * This class adapts protocol requests to the Coordinator and DataNode managers; it must preserve
+ * session ownership, query cleanup, authorization, timeout, and status-conversion semantics.
+ */
public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(ClientRPCServiceImpl.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 34ef951296ce..2bfeedb20bb0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -410,6 +410,13 @@
import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onIoTDBException;
import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onQueryException;
+/**
+ * Implements the internal DataNode RPC surface used for fragment execution, region lifecycle,
+ * schema/data operations, load, consensus-related coordination, and cluster maintenance.
+ *
+ * Internal requests may arrive during startup, shutdown, migration, or recovery, so methods
+ * must document their readiness checks and idempotency behavior where it is not obvious.
+ */
public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface {
private static final Logger LOGGER =
LoggerFactory.getLogger(DataNodeInternalRPCServiceImpl.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java
index ceba3880122a..50cc50dc1101 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java
@@ -88,6 +88,13 @@
import static org.apache.iotdb.db.queryengine.metric.DataExchangeCountMetricSet.ON_ACKNOWLEDGE_DATA_BLOCK_NUM_SERVER;
import static org.apache.iotdb.db.queryengine.metric.DataExchangeCountMetricSet.SEND_NEW_DATA_BLOCK_NUM_SERVER;
+/**
+ * Manages local and remote source/sink handles used to exchange TsBlocks between MPP fragments.
+ *
+ * The manager processes data-block fetch, acknowledgement, close, and end-of-stream events.
+ * Late events are expected after downstream cancellation and must be ignored without leaking
+ * handles or corrupting completion state.
+ */
public class MPPDataExchangeManager implements IMPPDataExchangeManager {
private static final Logger LOGGER = LoggerFactory.getLogger(MPPDataExchangeManager.class);
@@ -299,10 +306,9 @@ public void onNewDataBlockEvent(TNewDataBlockEvent e) throws TException {
: (SourceHandle) sourceHandleMap.get(e.getTargetPlanNodeId());
if (sourceHandle == null || sourceHandle.isAborted() || sourceHandle.isFinished()) {
- // In some scenario, when the SourceHandle sends the data block ACK event, its upstream
- // may
- // have already been stopped. For example, in the read whit LimitOperator, the downstream
- // FragmentInstance may be finished, although the upstream is still working.
+ // A downstream fragment may finish early, for example when a LimitOperator has produced enough
+ // rows, while its upstream fragment is still sending events. Ignore late events for the finished
+ // or aborted SourceHandle.
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
DataNodeQueryMessages
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java
index fa6a45e83749..103f338e7c34 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceManager.java
@@ -73,6 +73,13 @@
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceExecution.createFragmentInstanceExecution;
import static org.apache.iotdb.rpc.TSStatusCode.TOO_MANY_CONCURRENT_QUERIES_ERROR;
+/**
+ * Creates, tracks, executes, and cleans up fragment instances on this DataNode.
+ *
+ * The manager owns fragment contexts and executions, constructs local pipeline drivers, rejects
+ * repeated dispatches, schedules timeout cleanup, and releases query-level resources when an
+ * instance reaches a terminal state.
+ */
@SuppressWarnings("squid:S6548")
public class FragmentInstanceManager {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java
index 0485de570fe5..c52dbf8187ff 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverScheduler.java
@@ -67,7 +67,13 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-/** The manager of fragment instances scheduling. */
+/**
+ * Schedules DataNode DriverTasks across ready, blocked, and timeout queues.
+ *
+ * The scheduler enforces query and task capacity, accounts for CPU and memory quotas, moves
+ * blocked tasks back to the ready queue, and aborts tasks when their query or fragment instance is
+ * cancelled or timed out.
+ */
public class DriverScheduler implements IDriverScheduler, IService {
private static final Logger logger = LoggerFactory.getLogger(DriverScheduler.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
index 09a35803595f..9bfd5204b0f1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java
@@ -194,9 +194,10 @@
import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfCharArray;
/**
- * The coordinator for MPP. It manages all the queries which are executed in current Node. And it
- * will be responsible for the lifecycle of a query. A query request will be represented as a
- * QueryExecution.
+ * Coordinates the lifecycle of queries executed on this DataNode.
+ *
+ * The coordinator creates query contexts, analyzes and plans statements, dispatches fragment
+ * instances, tracks query state, handles retries, and releases query resources after completion.
*/
public class Coordinator {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
index 40f5231a81fb..2563b20128e2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
@@ -79,10 +79,11 @@
import static org.apache.iotdb.rpc.TSStatusCode.DATE_OUT_OF_RANGE;
/**
- * QueryExecution stores all the status of a query which is being prepared or running inside the MPP
- * frame. It takes three main responsibilities: 1. Prepare a query. Transform a query from statement
- * to DistributedQueryPlan with fragment instances. 2. Dispatch all the fragment instances to
- * corresponding physical nodes. 3. Collect and monitor the progress/states of this query.
+ * Represents the lifecycle and execution state of one MPP query or write operation.
+ *
+ * It analyzes the statement, builds logical and distributed plans, dispatches fragment
+ * instances, exposes results through the local or remote exchange layer, monitors state changes,
+ * retries eligible failures, and releases resources exactly once.
*/
public class QueryExecution implements IQueryExecution {
private static final Logger LOGGER = LoggerFactory.getLogger(QueryExecution.class);
@@ -101,11 +102,10 @@ public class QueryExecution implements IQueryExecution {
private LogicalQueryPlan logicalPlan;
private DistributedQueryPlan distributedPlan;
- // The result of QueryExecution will be written to the MPPDataExchangeManager in current Node.
- // We use this SourceHandle to fetch the TsBlock from it.
+ /** Result blocks are published to the local exchange manager and read through this source handle. */
private ISourceHandle resultHandle;
- // used for cleaning resultHandle up exactly once
+ /** Guards exactly-once cleanup of the result source handle. */
private final AtomicBoolean resultHandleCleanUp;
private final AtomicBoolean stopped;
@@ -199,7 +199,8 @@ private void startInternal() {
return;
}
- // check timeout for query first
+ // Apply the timeout only to query operations. Write operations use the write-path retry and
+ // backpressure rules instead of a query execution deadline.
checkTimeOutForQuery();
doLogicalPlan();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java
index 5f2f72f43139..2c91a9e14086 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java
@@ -171,7 +171,12 @@ public R process(StatementNode node, C context) {
return node.accept(this, context);
}
- /** Top Level Description */
+ /**
+ * Visits the root of the statement hierarchy.
+ *
+ * Default visitor methods delegate to this method, allowing subclasses to handle only the
+ * statement types they need.
+ */
public abstract R visitNode(StatementNode node, C context);
public R visitStatement(Statement statement, C context) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
index ad8042ec7030..77e655dc409c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
@@ -73,7 +73,14 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-// manage all the schemaRegion in this dataNode
+/**
+ * Owns the schema regions hosted by this DataNode and coordinates their loading, recovery,
+ * metrics, schema-resource management, and lifecycle operations.
+ *
+ * The selected schema-engine mode determines whether schema state is memory-resident or
+ * cached. Schema metrics and shared schema resources must be initialized before schema regions,
+ * and cleared only after all schema regions have been cleared.
+ */
public class SchemaEngine {
private static final Logger logger = LoggerFactory.getLogger(SchemaEngine.class);
@@ -118,8 +125,9 @@ public void init() {
initSchemaEngineStatistics();
SchemaResourceManager.initSchemaResource(schemaEngineStatistics);
- // CachedSchemaEngineMetric depend on CacheMemoryManager, so it should be initialized after
- // CacheMemoryManager
+
+ // Cached schema metrics depend on CacheMemoryManager, so initialize them only after the cache
+ // memory manager is ready.
schemaMetricManager = new SchemaMetricManager(schemaEngineStatistics);
initSchemaRegion();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 5ce282db7a70..b2174648fd30 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -156,6 +156,14 @@
import static org.apache.iotdb.commons.utils.StatusUtils.retrieveExitStatusCode;
import static org.apache.iotdb.db.conf.IoTDBStartCheck.PROPERTIES_FILE_NAME;
+/**
+ * The process-level service for a DataNode.
+ *
+ * DataNode registers with the ConfigNode cluster, restores local schema and data regions,
+ * starts consensus and query services, and exposes client and internal RPC endpoints. Startup
+ * order is significant because query, write, and region-management services must not serve
+ * requests before local recovery and runtime configuration have completed.
+ */
public class DataNode extends ServerCommandLine implements DataNodeMBean {
private static final Logger logger = LoggerFactory.getLogger(DataNode.class);
@@ -252,13 +260,20 @@ public static void main(final String[] args) {
}
}
+ /**
+ * Starts the DataNode by preparing local state, synchronizing cluster configuration, registering
+ * or restarting the node, recovering regions, and starting the remaining services in dependency
+ * order.
+ *
+ * The first-start and restart paths intentionally perform different registration and security
+ * checks.
+ */
@Override
protected void start() {
logger.info(DataNodeMiscMessages.STARTING_DATANODE);
boolean isFirstStart;
try {
- IoTDBDescriptor.getInstance().getMemoryConfig().activateAutoResizingBufferMemoryControl();
- // Check if this DataNode is start for the first time and do other pre-checks
+ // Check whether this is the first DataNode startup and run the remaining startup checks.
isFirstStart = prepareDataNode();
if (isFirstStart) {
@@ -1368,6 +1383,13 @@ public void deleteDataNodeSystemProperties() {
DataNodeSystemPropertiesHandler.getInstance().delete();
}
+ /**
+ * Stops DataNode services and releases resources in an order that preserves WAL, TsFile,
+ * consensus, query, and RPC shutdown dependencies.
+ *
+ * The method must remain safe during partial startup because startup failures invoke it as
+ * cleanup.
+ */
public void stop() {
stopTriggerRelatedServices();
registerManager.deregisterAll();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
index 6dd082c0efa4..7e337e055ab1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
@@ -56,6 +56,14 @@
import java.util.Map;
+/**
+ * Performs an orderly DataNode shutdown.
+ *
+ * The hook first prevents new writes and drains write-related resources, then closes or
+ * snapshots storage according to the configured consensus protocol, persists Pipe progress,
+ * stops DataNode services, reports shutdown to the ConfigNode leader, and finally releases the
+ * directory lock.
+ */
public class DataNodeShutdownHook extends Thread {
private static final Logger logger = LoggerFactory.getLogger(DataNodeShutdownHook.class);
@@ -134,9 +142,8 @@ public void run() {
// We did this work because the RatisConsensus recovery mechanism is different from other
// consensus algorithms, which will replace the underlying storage engine based on its
- // own
- // latest snapshot, while other consensus algorithms will not. This judgement ensures that
- // compaction work is not discarded even if there are frequent restarts
+ // own latest snapshot, while other consensus algorithms will not. This judgement ensures
+ // that compaction work is not discarded even if there are frequent restarts
if (IoTDBDescriptor.getInstance()
.getConfig()
.getDataRegionConsensusProtocolClass()
@@ -189,9 +196,8 @@ public void run() {
// set encryption key to 16-byte zero.
TSFileDescriptor.getInstance().getConfig().setEncryptKey(new byte[16]);
- // Actually stop all services started by the DataNode.
- // If we don't call this, services like the RestService are not stopped and I can't re-start
- // it.
+ // Stop every service started by DataNode. Otherwise services such as RestService may retain
+ // resources and prevent a subsequent restart in the same JVM.
DataNode.getInstance().stop();
// Set and report shutdown to cluster ConfigNode-leader
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index 1ac6c15488b1..0b8d4b67a924 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -149,11 +149,13 @@ public class StorageEngine implements IService {
private final ConcurrentHashMap we don't use computeIfAbsent because we don't want to create a new region if the region is
- * absent, we just want to run the runnable in a synchronized way.
+ * computeIfAbsent is intentionally not used because the action may perform initialization that
+ * must not be started while another deletion or replacement is in progress.
*
* @return true if the region is absent and the runnable is run. false if the region is present.
*/
@@ -912,9 +918,9 @@ public boolean runIfAbsent(DataRegionId regionId, Runnable runnable) {
}
/**
- * run the consumer if the region is present. if the region is absent, do nothing.
+ * Run the consumer if the region is present. if the region is absent, do nothing.
*
- * we don't use computeIfPresent because we don't want to remove the region if the consumer
+ * We don't use computeIfPresent because we don't want to remove the region if the consumer
* returns null, we just want to run the consumer in a synchronized way.
*
* @return true if the region is present and the consumer is run. false if the region is absent.
@@ -947,7 +953,12 @@ public int getDataRegionNumber() {
return dataRegionMap.size();
}
- /** This method is not thread-safe */
+ /**
+ * Replaces a local DataRegion while loading a snapshot.
+ *
+ * This method is not thread-safe and may be called only while external region access is
+ * quiesced by the snapshot-loading protocol.
+ */
public DataRegion setDataRegionForSnapshotLoad(
DataRegionId regionId, Supplier Schedule tasks are protected by the manager lock. Configuration changes are applied only
+ * after active tasks have stopped, and repair tasks temporarily prevent incompatible schedule
+ * changes.
+ */
public class CompactionScheduleTaskManager implements IService {
private int compactionSelectorNum =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java
index 169d10137d8b..91a1c4c5833a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java
@@ -53,7 +53,12 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-/** CompactionMergeTaskPoolManager provides a ThreadPool tPro queue and run all compaction tasks. */
+/**
+ * Owns the worker pools and candidate queue used to execute DataRegion compaction tasks.
+ *
+ * The manager tracks task futures by database and DataRegion, applies compaction rate limits,
+ * and coordinates graceful or immediate shutdown of compaction workers.
+ */
@SuppressWarnings("squid:S6548")
public class CompactionTaskManager implements IService {
private static final Logger logger =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index de1a5a09b7dd..67103c9b13f3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -127,10 +127,17 @@
import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.FLUSHING_MEMTABLE;
import static org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet.WORKING_MEMTABLE;
+/**
+ * Manages one writable TsFile, its working and flushing MemTables, WAL entries, flush lifecycle,
+ * and resource metadata.
+ *
+ * The flush/query lock coordinates reads, asynchronous flush, synchronous close, deletion, and
+ * resource publication. A processor is closed only after its pending MemTables have been flushed.
+ */
@SuppressWarnings("java:S1135") // ignore todos
public class TsFileProcessor {
- /** Logger fot this class. */
+ /** Logger for this class. */
private static final Logger logger = LoggerFactory.getLogger(TsFileProcessor.class);
private static final int NUM_MEM_TO_ESTIMATE = 3;
@@ -167,12 +174,12 @@ public class TsFileProcessor {
*/
private volatile boolean managedByFlushManager;
- /** A lock to mutual exclude read and read */
+ /** Read/write lock coordinating query access with flush, close, and deletion operations. */
private final ReadWriteLock flushQueryLock = new ReentrantReadWriteLock();
/**
- * It is set by the StorageGroupProcessor and checked by flush threads. (If shouldClose == true
- * and its flushingMemTables are all flushed, then the flush thread will close this file.)
+ * Set by DataRegion when this processor must close after all MemTables currently being flushed
+ * have completed.
*/
private volatile boolean shouldClose;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
index 8fafee304b2b..e000b791633f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
@@ -56,18 +56,23 @@
import static org.apache.iotdb.commons.conf.IoTDBConstant.FILE_NAME_SEPARATOR;
-/** This class is used to manage and allocate wal nodes. */
+/**
+ * Allocates, tracks, flushes, and removes write-ahead-log nodes used by DataRegion write paths.
+ *
+ * The allocation and deletion behavior depends on the configured consensus protocol and WAL
+ * mode. Disabled WAL mode must remain a no-op for lifecycle and allocation operations.
+ */
public class WALManager implements IService {
private static final Logger logger = LoggerFactory.getLogger(WALManager.class);
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- // manage all wal nodes and decide how to allocate them
+ /** Allocates WAL nodes and applies the strategy selected for the configured consensus protocol. */
private final NodeAllocationStrategy walNodesManager;
- // single thread to delete old .wal files
+ /** Single-thread scheduler that deletes expired WAL files. */
private ScheduledExecutorService walDeleteThread;
- // total disk usage of wal files
+ /** Aggregate disk usage of all WAL nodes. */
private final AtomicLong totalDiskUsage = new AtomicLong();
- // total number of wal files
+ /** Aggregate number of WAL files across all WAL nodes. */
private final AtomicLong totalFileNum = new AtomicLong();
private WALManager() {
@@ -101,7 +106,7 @@ public IWALNode applyForWALNode(String applicantUniqueId) {
return walNodesManager.applyForWALNode(applicantUniqueId);
}
- /** WAL node will be registered only when using iot series consensus protocol. */
+ /** Registers a WAL node only for IoTConsensus and IoTConsensusV2. */
public void registerWALNode(
String applicantUniqueId, String logDirectory, long startFileVersion, long startSearchIndex) {
if (config.getWalMode() == WALMode.DISABLE
@@ -117,7 +122,7 @@ public void registerWALNode(
WritingMetrics.getInstance().createWALNodeInfoMetrics(applicantUniqueId);
}
- /** WAL node will be deleted only when using iot series consensus protocol. */
+ /** Deletes a WAL node only for IoTConsensus and IoTConsensusV2. */
public void deleteWALNode(String applicantUniqueId) {
if (config.getWalMode() == WALMode.DISABLE
|| (!config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
index d67cd88a4c48..496b812f8c5b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
@@ -71,8 +71,10 @@
import static org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode.DEFAULT_SEARCH_INDEX;
/**
- * This buffer guarantees the concurrent safety and uses double buffers mechanism to accelerate
- * writes and avoid waiting for buffer syncing to disk.
+ * Buffers WAL entries with a lock-protected working, syncing, and idle buffer rotation.
+ *
+ * The rotation allows serialization to continue while another buffer is written to disk. All
+ * buffer-state transitions must follow {@code buffersLock} and its conditions.
*/
public class WALBuffer extends AbstractWALBuffer {
private static final Logger logger = LoggerFactory.getLogger(WALBuffer.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java
index 54a45626a3d6..846d518f3cd5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java
@@ -51,7 +51,11 @@
import static org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils.getTsFileRelativePath;
-/** First set allVsgScannedLatch, then call recover method. */
+/**
+ * Coordinates WAL recovery after every DataRegion has scanned its local unsealed TsFiles.
+ *
+ * Callers must install the all-data-region-scanned latch before invoking {@link #recover()}.
+ */
public class WALRecoverManager {
private static final Logger logger = LoggerFactory.getLogger(WALRecoverManager.class);
private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgent.java
index 9118891a79ea..653b81c830d8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionAgent.java
@@ -19,6 +19,12 @@
package org.apache.iotdb.db.subscription.agent;
+/**
+ * Entry point for DataNode subscription receiver, runtime, consumer, broker, and topic agents.
+ *
+ * The singleton exposes the agents that own subscription protocol handling and runtime state;
+ * lifecycle and ownership rules are implemented by those agents.
+ */
public class SubscriptionAgent {
private final SubscriptionReceiverAgent receiverAgent;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
index b0eec505a5c4..769ea8dd1c97 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
@@ -103,6 +103,13 @@
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
+/**
+ * Handles version-1 subscription requests, including handshake, heartbeat, subscribe, poll,
+ * commit, seek, unsubscribe, and consumer close operations.
+ *
+ * Consumer state is shared across request threads and is fenced by consumer ownership. Poll,
+ * commit, seek, timeout, and exit paths must preserve in-flight request and progress invariants.
+ */
public class SubscriptionReceiverV1 implements SubscriptionReceiver {
private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionReceiverV1.class);
From 760a980bf5e1c509e01e1dd1cb6ce31daebc3912 Mon Sep 17 00:00:00 2001
From: WishArdently <3022210040@tju.edu.cn>
Date: Sat, 5 Sep 2026 22:02:21 +0800
Subject: [PATCH 3/4] fix ConfigNode/DataNode comments after review
---
.../consensus/request/ConfigPhysicalPlan.java | 4 ++--
.../iotdb/confignode/manager/ClusterManager.java | 14 +++++++-------
.../manager/load/service/EventService.java | 4 +---
.../manager/load/service/HeartbeatService.java | 5 +++--
.../iotdb/confignode/manager/node/NodeManager.java | 3 ++-
.../confignode/persistence/schema/ConfigMTree.java | 6 +++---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +-
.../dataregion/DataRegionStateMachine.java | 4 ++--
.../schemaregion/SchemaRegionStateMachine.java | 4 ++--
.../impl/DataNodeInternalRPCServiceImpl.java | 4 ++--
.../execution/exchange/MPPDataExchangeManager.java | 12 +++++++-----
.../queryengine/plan/execution/QueryExecution.java | 4 +++-
.../apache/iotdb/db/schemaengine/SchemaEngine.java | 12 ++++++------
.../java/org/apache/iotdb/db/service/DataNode.java | 8 ++++----
.../iotdb/db/service/DataNodeShutdownHook.java | 8 ++++----
.../iotdb/db/storageengine/StorageEngine.java | 8 ++++----
.../schedule/CompactionScheduleTaskManager.java | 5 ++---
.../storageengine/dataregion/wal/WALManager.java | 7 ++++++-
.../receiver/SubscriptionReceiverV1.java | 4 ++--
19 files changed, 63 insertions(+), 55 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
index ee23012bd5db..cf359bec5f55 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
@@ -198,8 +198,8 @@ public static class Factory {
* Deserializes a plan from the buffer using the encoded type discriminator.
*
* @param buffer the buffer containing one serialized plan
- * @return the deserialized plan, or null when the type is unknown if that is the supported
- * compatibility behavior
+ * @return the deserialized plan
+ * @throws IOException if the encoded plan type or payload cannot be read
*/
public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOException {
final short planType = buffer.getShort();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java
index b9b92e443da2..e085b4ae7906 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java
@@ -85,10 +85,11 @@ public String getClusterId() {
}
/**
- * Gets the cluster ID with bounded retries when the local ConfigRegion is not ready.
+ * Waits up to the specified time for the cluster ID to become available.
*
- * @return the cluster ID
- * @throws ConsensusException if the cluster ID cannot be read after the retry policy is exhausted
+ * @param maxWaitTime maximum wait time in milliseconds
+ * @return the cluster ID, or null if it is unavailable after the timeout or the wait is
+ * interrupted
*/
public String getClusterIdWithRetry(long maxWaitTime) {
long startTime = System.currentTimeMillis();
@@ -115,12 +116,11 @@ private void generateClusterId() {
}
}
- // TODO: Parallel test ConfigNode and DataNode
/**
- * Submits connectivity checks to all known ConfigNodes and DataNodes.
+ * Tests connectivity from this ConfigNode to all registered ConfigNodes and DataNodes and
+ * aggregates the results.
*
- * The method is asynchronous; callers must use the returned or associated result mechanism
- * to determine which nodes succeeded or failed.
+ * @return aggregated connection-test results
*/
public TTestConnectionResp submitTestConnectionTaskToEveryNode() {
TTestConnectionResp resp = new TTestConnectionResp();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java
index dca939ba8917..14591a8da763 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/EventService.java
@@ -50,9 +50,7 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-/**
- * Periodically checks cluster events that require ConfigNode-side handling.
- */
+/** Periodically checks cluster events that require ConfigNode-side handling. */
public class EventService {
private static final Logger LOGGER = LoggerFactory.getLogger(EventService.class);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java
index e71b40cf9257..8ea35e5acdff 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java
@@ -59,8 +59,9 @@
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
-/**
- * Guards heartbeat scheduling across start, stop, reload, and leadership transitions.
+/**
+ * Periodically sends heartbeat requests from the ConfigNode leader to registered ConfigNodes,
+ * DataNodes, and AINodes, and updates the corresponding load information.
*/
public class HeartbeatService {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index c50a9c49ba19..85715cf23c05 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -521,7 +521,8 @@ public TAINodeConfiguration getRegisteredAINode(int aiNodeId) {
}
/**
- * Serialize AINode registration so concurrent requests cannot violate the single-AINode registration constraint.
+ * Serialize AINode registration so concurrent requests cannot violate the single-AINode
+ * registration constraint.
*
* @param req TAINodeRegisterReq
* @return AINodeConfigurationDataSet. The {@link TSStatus} will be set to {@link
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
index 78867e30b9b1..f7c5e055d92a 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/schema/ConfigMTree.java
@@ -102,7 +102,8 @@
import static org.apache.iotdb.commons.schema.SchemaConstant.ROOT;
import static org.apache.iotdb.commons.schema.SchemaConstant.TABLE_MNODE_TYPE;
-// ConfigMTree is fully memory-resident. MTreeStore is retained only for traverser-related APIs.
+// ConfigMTree stores nodes in memory. ConfigMTreeStore provides in-memory node access and traversal
+// helpers.
public class ConfigMTree {
private static final String TABLE_ERROR_MSG =
@@ -135,8 +136,7 @@ public void clear() {
// region database Management
/**
- * Create the database after validating the path;
- * intermediate nodes are created when necessary.
+ * Create the database after validating the path; intermediate nodes are created when necessary.
*
* @param path path
*/
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 7d8fb336c651..b14f2e1b6222 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -628,7 +628,7 @@ public class IoTDBConfig {
private long cacheFileReaderClearPeriod = 100000;
/** the max executing time of query in ms. Unit: millisecond */
- private long queryTimeoutThreshold = 3600000; // 1 hour
+ private long queryTimeoutThreshold = 60000;
/** the max time to live of a session in ms. Unit: millisecond */
private int sessionTimeoutThreshold = 0;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
index 17262086129d..81ecf02f31f4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
@@ -56,8 +56,8 @@
import java.util.function.Supplier;
/**
- * Applies replicated write and query fragment operations to one DataRegion and exposes its
- * snapshot and region-resource lifecycle to the consensus layer.
+ * Applies replicated write and query fragment operations to one DataRegion and exposes its snapshot
+ * and region-resource lifecycle to the consensus layer.
*
* Write-process rejection is retried here to preserve the atomicity expected by the consensus
* apply path; other statuses are delegated to the consensus retry mechanism.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java
index 479231d8143d..757b8e8714d0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java
@@ -47,8 +47,8 @@
import java.util.Objects;
/**
- * Applies replicated schema operations to one SchemaRegion and coordinates schema-region
- * snapshot, Pipe-leader, and attribute-security lifecycle callbacks.
+ * Applies replicated schema operations to one SchemaRegion and coordinates schema-region snapshot,
+ * Pipe-leader, and attribute-security lifecycle callbacks.
*/
public class SchemaRegionStateMachine extends BaseStateMachine {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 2bfeedb20bb0..ac41fe5a7793 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -414,8 +414,8 @@
* Implements the internal DataNode RPC surface used for fragment execution, region lifecycle,
* schema/data operations, load, consensus-related coordination, and cluster maintenance.
*
- * Internal requests may arrive during startup, shutdown, migration, or recovery, so methods
- * must document their readiness checks and idempotency behavior where it is not obvious.
+ * Internal requests may arrive during startup, shutdown, migration, or recovery, so methods must
+ * document their readiness checks and idempotency behavior where it is not obvious.
*/
public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface {
private static final Logger LOGGER =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java
index 50cc50dc1101..5b6a5f178ad2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/MPPDataExchangeManager.java
@@ -91,9 +91,9 @@
/**
* Manages local and remote source/sink handles used to exchange TsBlocks between MPP fragments.
*
- * The manager processes data-block fetch, acknowledgement, close, and end-of-stream events.
- * Late events are expected after downstream cancellation and must be ignored without leaking
- * handles or corrupting completion state.
+ * The manager processes data-block fetch, acknowledgement, close, and end-of-stream events. Late
+ * events are expected after downstream cancellation and must be ignored without leaking handles or
+ * corrupting completion state.
*/
public class MPPDataExchangeManager implements IMPPDataExchangeManager {
@@ -306,8 +306,10 @@ public void onNewDataBlockEvent(TNewDataBlockEvent e) throws TException {
: (SourceHandle) sourceHandleMap.get(e.getTargetPlanNodeId());
if (sourceHandle == null || sourceHandle.isAborted() || sourceHandle.isFinished()) {
- // A downstream fragment may finish early, for example when a LimitOperator has produced enough
- // rows, while its upstream fragment is still sending events. Ignore late events for the finished
+ // A downstream fragment may finish early, for example when a LimitOperator has produced
+ // enough
+ // rows, while its upstream fragment is still sending events. Ignore late events for the
+ // finished
// or aborted SourceHandle.
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
index 2563b20128e2..39fa4f4d7023 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java
@@ -102,7 +102,9 @@ public class QueryExecution implements IQueryExecution {
private LogicalQueryPlan logicalPlan;
private DistributedQueryPlan distributedPlan;
- /** Result blocks are published to the local exchange manager and read through this source handle. */
+ /**
+ * Result blocks are published to the local exchange manager and read through this source handle.
+ */
private ISourceHandle resultHandle;
/** Guards exactly-once cleanup of the result source handle. */
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
index 77e655dc409c..c5e41e573be6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
@@ -74,12 +74,12 @@
import java.util.concurrent.TimeUnit;
/**
- * Owns the schema regions hosted by this DataNode and coordinates their loading, recovery,
- * metrics, schema-resource management, and lifecycle operations.
+ * Owns the schema regions hosted by this DataNode and coordinates their loading, recovery, metrics,
+ * schema-resource management, and lifecycle operations.
*
- * The selected schema-engine mode determines whether schema state is memory-resident or
- * cached. Schema metrics and shared schema resources must be initialized before schema regions,
- * and cleared only after all schema regions have been cleared.
+ * The selected schema-engine mode determines whether schema state is memory-resident or cached.
+ * Schema metrics and shared schema resources must be initialized before schema regions, and cleared
+ * only after all schema regions have been cleared.
*/
public class SchemaEngine {
@@ -125,7 +125,7 @@ public void init() {
initSchemaEngineStatistics();
SchemaResourceManager.initSchemaResource(schemaEngineStatistics);
-
+
// Cached schema metrics depend on CacheMemoryManager, so initialize them only after the cache
// memory manager is ready.
schemaMetricManager = new SchemaMetricManager(schemaEngineStatistics);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index b2174648fd30..cc1316a98046 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -159,10 +159,10 @@
/**
* The process-level service for a DataNode.
*
- * DataNode registers with the ConfigNode cluster, restores local schema and data regions,
- * starts consensus and query services, and exposes client and internal RPC endpoints. Startup
- * order is significant because query, write, and region-management services must not serve
- * requests before local recovery and runtime configuration have completed.
+ * DataNode registers with the ConfigNode cluster, restores local schema and data regions, starts
+ * consensus and query services, and exposes client and internal RPC endpoints. Startup order is
+ * significant because query, write, and region-management services must not serve requests before
+ * local recovery and runtime configuration have completed.
*/
public class DataNode extends ServerCommandLine implements DataNodeMBean {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
index 7e337e055ab1..0956deed827d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
@@ -60,9 +60,9 @@
* Performs an orderly DataNode shutdown.
*
* The hook first prevents new writes and drains write-related resources, then closes or
- * snapshots storage according to the configured consensus protocol, persists Pipe progress,
- * stops DataNode services, reports shutdown to the ConfigNode leader, and finally releases the
- * directory lock.
+ * snapshots storage according to the configured consensus protocol, persists Pipe progress, stops
+ * DataNode services, reports shutdown to the ConfigNode leader, and finally releases the directory
+ * lock.
*/
public class DataNodeShutdownHook extends Thread {
@@ -142,7 +142,7 @@ public void run() {
// We did this work because the RatisConsensus recovery mechanism is different from other
// consensus algorithms, which will replace the underlying storage engine based on its
- // own latest snapshot, while other consensus algorithms will not. This judgement ensures
+ // own latest snapshot, while other consensus algorithms will not. This judgement ensures
// that compaction work is not discarded even if there are frequent restarts
if (IoTDBDescriptor.getInstance()
.getConfig()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index 0b8d4b67a924..cf7ea4040425 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -900,8 +900,8 @@ public TSStatus deleteDataRegion(DataRegionId regionId) {
/**
* Runs the action only when the specified DataRegion is absent.
*
- * computeIfAbsent is intentionally not used because the action may perform initialization that
- * must not be started while another deletion or replacement is in progress.
+ * Uses computeIfAbsent to serialize the absence check and action; returning null keeps the map
+ * unchanged.
*
* @return true if the region is absent and the runnable is run. false if the region is present.
*/
@@ -920,8 +920,8 @@ public boolean runIfAbsent(DataRegionId regionId, Runnable runnable) {
/**
* Run the consumer if the region is present. if the region is absent, do nothing.
*
- * We don't use computeIfPresent because we don't want to remove the region if the consumer
- * returns null, we just want to run the consumer in a synchronized way.
+ * Uses computeIfPresent to serialize consumer invocation and returns the existing region so it
+ * remains in the map.
*
* @return true if the region is present and the consumer is run. false if the region is absent.
*/
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java
index aa2090ac1eb6..623a43f2dd92 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java
@@ -52,9 +52,8 @@
/**
* Schedules compaction-selection and TTL-check tasks for all local DataRegions.
*
- * Schedule tasks are protected by the manager lock. Configuration changes are applied only
- * after active tasks have stopped, and repair tasks temporarily prevent incompatible schedule
- * changes.
+ * Schedule tasks are protected by the manager lock. Configuration changes are applied only after
+ * active tasks have stopped, and repair tasks temporarily prevent incompatible schedule changes.
*/
public class CompactionScheduleTaskManager implements IService {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
index e000b791633f..576858b51a4d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java
@@ -66,12 +66,17 @@ public class WALManager implements IService {
private static final Logger logger = LoggerFactory.getLogger(WALManager.class);
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- /** Allocates WAL nodes and applies the strategy selected for the configured consensus protocol. */
+ /**
+ * Allocates WAL nodes and applies the strategy selected for the configured consensus protocol.
+ */
private final NodeAllocationStrategy walNodesManager;
+
/** Single-thread scheduler that deletes expired WAL files. */
private ScheduledExecutorService walDeleteThread;
+
/** Aggregate disk usage of all WAL nodes. */
private final AtomicLong totalDiskUsage = new AtomicLong();
+
/** Aggregate number of WAL files across all WAL nodes. */
private final AtomicLong totalFileNum = new AtomicLong();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
index 769ea8dd1c97..ded035b1d546 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
@@ -104,8 +104,8 @@
import java.util.stream.Collectors;
/**
- * Handles version-1 subscription requests, including handshake, heartbeat, subscribe, poll,
- * commit, seek, unsubscribe, and consumer close operations.
+ * Handles version-1 subscription requests, including handshake, heartbeat, subscribe, poll, commit,
+ * seek, unsubscribe, and consumer close operations.
*
* Consumer state is shared across request threads and is fenced by consumer ownership. Poll,
* commit, seek, timeout, and exit paths must preserve in-flight request and progress invariants.
From 5e86934b388575bdfa2720532e1e395d596c769e Mon Sep 17 00:00:00 2001
From: Wang Xiang