feat(load): route decoded tsfile pieces through consensus

[luoluoyuyu]

LOAD TsFile over the Consensus Protocol and 2PC-based Staging Directory Lifecycle Management

1. Background and Problems to Solve

The previous LOAD TsFile implementation had three structural defects:

1. LOAD bypassed the consensus protocol, so consistency was at risk.
   The coordinator fanned the serialized LoadTsFilePieceNode out to every replica over a dedicated RPC (sendTsFilePieceNode), and then told each node to persist its own copy through EXECUTE/ROLLBACK commands. That path bypassed the existing consensus state machine, so consistency across replicas rested entirely on the coordinator. A crash or a restart could easily leave an inconsistent intermediate state in which some replicas lagged behind while others ran ahead.

2. The staging data lifecycle was a guess rather than a decision.
   There was no reliable basis for deciding when a staged file could be deleted. Deleting too early left a catching-up replica unable to read the bytes back; deleting too late kept the files of finished tasks on disk (a disk leak).

Goals of this design:

• Fold the LOAD write path into the existing consensus write path;
• Replace the old EXECUTE/ROLLBACK command protocol with two-phase commit (2PC);
• Manage the staging directories through a deterministic reclamation driven by the consensus watermark.

---

2. Overall Design

2.1 Layered Architecture

The coordinator submits once, to the write peer of the partition; replicas receive the command through ordinary consensus log replication, exactly like a normal write. The old mode, in which the coordinator sent one RPC per replica, is gone.

[Coordinator DataNode]  LoadTsFileScheduler
        │ (choose the strategy by needDecodeTsFile)
        ├── LocalLoadStrategy          (no decode needed: hand the whole file to the local region)
        └── TwoPhaseConsensusLoadStrategy(decode needed: split -> stream -> two-phase commit)
                 │
                 │ LoadConsensusSubmitter (resolve the write peer as the normal write path does:
                 │                          Ratis leader / IoTConsensus write node)
                 ▼
[Consensus]  LoadTsFileConsensusNode (5 phase commands: BEGIN / PIECE / PREPARE / COMMIT / ABORT)
             (via DataRegionConsensusImpl.write -> WAL replication -> replica state machines)
                 ▼
[Storage: DataRegion]  DataExecutionVisitor
        └── writeLoadTsFile{Begin,Piece,Prepare,Commit,Abort}
                 └── LoadTsFileManager (one per DataRegion)
                          └── TsFileWriterManager (one per task: staging dir + staged TsFile + progress)

2.2 Two-Phase Commit (2PC) Protocol

2.2.1 Roles and Transaction Boundary

Concept	Design
Coordinator	The DataNode that receives the LOAD statement (LoadTsFileScheduler to TwoPhaseConsensusLoadStrategy).
Participant	Every DataRegion touched by the TsFile (the write peer executes; replicas take part through consensus log replication).
Transaction id	One loadId per region (regionLoadIds.computeIfAbsent(regionId, UUID)), not a single global id.
Transaction boundary	One source TsFile x one DataRegion. A TsFile spanning several regions is several independent 2PC transactions, which avoids a global transaction coordinator across consensus groups.
Submission channel	LoadConsensusSubmitter -> write peer of the partition -> DataRegionConsensusImpl.write -> WAL replication -> DataExecutionVisitor.

2.2.2 Phase Protocol and Server-side Semantics

The server keeps no in-memory transaction state per load: phase ordering and idempotency are the client's responsibility. The server only locates the staging directory by loadId and applies the command.

Command	Server behaviour	Replay / duplicate delivery
BEGIN	DataExecutionVisitor returns OK directly (a no-op). The staged writer is created lazily by the first PIECE. The phase is kept only for cross-version compatibility.	Idempotent by nature.
PIECE	writePiece: locate or create the staged TsFile for (device, time partition), write the chunks and deletions, return PieceRef, then write the node (with references) to the WAL.	Idempotent by physical offset (see 2.2.4).
PREPARE	prepare: close the modification file, verify that the staged file has no hole, write the metadata zone to seal it, update the TimeIndex, record the ProgressIndex.	Skipped when the writer isSealed() (idempotent).
COMMIT	loadAll: evaluate mustRetain first, import the TsFile, finish the task, then write the WAL entry and the separator.	If the loadId already finished and its writer was removed, a no-op success.
ABORT	deleteAll, finish the task, then write the WAL entry and the separator (discard, or hand the directory to the retention mechanism).	Same as above, a no-op success.
PULL	Only the command definition is kept, for compatibility with older nodes; the logic is no longer supported.	-

PREPARE and COMMIT carry the accumulated pieceCount, totalBytes and checksum, plus the serialized ProgressIndex of every time partition, so the pipe/subscription progress semantics are preserved instead of degrading to MinimumProgressIndex.

2.2.3 Protocol Timeline

Coordinator                                  Region Write Peer (and its replicas)
    │                                                │
    │---- PIECE(0, chunks, chunkLayout) ────────────>│ create the staged writer lazily, append chunks by offset
    │---- PIECE(1, chunks, chunkLayout) ────────────>│ append chunks (out-of-order arrival is fine)
    │---- ...                                        │
    │---- PREPARE(pieceCount, bytes, progress) ─────>│ verify no hole, seal the staged TsFile
    │---- COMMIT ───────────────────────────────────>│ import the staged TsFile (mustRetain first)
    │                                                │
 on failure:                                         │
    │---- ABORT ────────────────────────────────────>│ drop the staged data (best-effort, no retry)

2.2.4 The Three Mechanisms that Make 2PC Work

Mechanism 1: idempotency by offset (instead of a server-side dedup table)

While splitting, the coordinator computes the physical layout of every chunk up front with ChunkOffsetCalculator (ChunkLayout: chunk-group-header offset, chunk offset, whether the chunk is the first of its group) and sends it along with the PIECE. Before writing, the server asks progress.hasChunkAt(offset):

• already recorded: skip the write and reuse the original reference (with a payload present, build a ChunkPayloadRef from where the chunk already sits; without one, keep the reference that arrived);
• not recorded and no payload: refuse with ..._ARRIVED_WITHOUT_ITS_CHUNK_PAYLOAD, so data can never be dropped silently; a mismatch between incomingRefs.size() and chunks.size() is refused as well.

Why this matters: out-of-order pieces, network retries and WAL replays all converge, and the server never has to maintain a dedup table. pieceIndex is only an identity and a log-correlation aid; it carries no ordering constraint.

Mechanism 2: decide before acting (must retain before import)

loadAll evaluates retention.mustRetain(searchIndex) before it imports, and passes the outcome into the import. The reason is that the WAL PIECE entries hold references only, so moving the staged file away would leave a catching-up replica unable to read those bytes; therefore a retained task is imported from a copy. Decision and import are both driven by the same searchIndex, so "the decision allowed deletion while the import moved the file" cannot happen.

Mechanism 3: a terminal record that makes the transaction recoverable

PREPARE, COMMIT and ABORT are all written to the WAL (logLoadNodeToWAL plus insertSeparatorToWAL). Before the progress files are dropped, the terminal marker (terminal op plus its consensus index) is appended to their tail.

• Crash and restart: a scan that sees the marker deletes the directory instead of mistaking a finished task for one that must be resumed;
• Command replay: no loadId in memory (the writer was removed) means a safe no-op success.

2.2.5 Commit Point and Failure-Handling Matrix

Core rule: roll back only the regions that are known not to have committed.

Failure	Behaviour	Test assertion
Phase 1 (splitting failure, or any piece failed to submit)	Full ABORT of every touched region; no region has voted yet. Each ABORT is retried up to 3 times.	all ABORT
PREPARE failed on any region	The transaction is undecided: every touched region is aborted, and a region that already prepared successfully does not commit. PREPARE is not retried.	PREPARE@1, PREPARE@2, ABORT@1 x3, ABORT@2 x3
PREPARE failed on the first region	Same: every region is aborted.	PREPARE@1, ABORT@1 x3, ABORT@2 x3
Every PREPARE succeeded	The commit round runs: COMMIT is sent to every region in turn. The commit point has passed and cannot be rolled back.	PREPARE@1, PREPARE@2, COMMIT@1, COMMIT@2
COMMIT failed with a transient error	Retried 3 times; when it still fails, no region is rolled back: that region may have imported already, and the regions that agreed must still commit (presumed commit). The failure is reported and the file goes to the tablet fallback.	PREPARE@1, PREPARE@2, COMMIT@1 x3, COMMIT@2
COMMIT failed with a permanent error	Not retried: a repetition is either refused again, or could import a time partition of the staged file twice. No region is rolled back, and the remaining regions still commit.	PREPARE@1, PREPARE@2, COMMIT@1, COMMIT@2
ABORT failed	Retried 3 times whatever the failure kind, because dropping staged data is idempotent. What is still left after the last attempt is logged as a warning and reclaimed by the cleaner.	PREPARE@1, ABORT@1 x3, ABORT@2 x3

Any failure of the second phase must reach an explicit ABORT: staging directories are named by loadId and never expire, so a task nobody finishes would hold disk space forever, and replaying the same pieces later would run into a half-filled staged file.

2.2.6 Retry Policy and Its Interaction with Consensus

• Bounded retry: at most 3 attempts (LOAD_CONSENSUS_SUBMIT_MAX_RETRIES) with a 100ms x attempt backoff, and only for transient failures (DISPATCH_ERROR, INTERNAL_SERVER_ERROR, NO_AVAILABLE_REGION_GROUP, EXECUTE_STATEMENT_ERROR). A permanent rejection is returned immediately so the coordinator aborts. Retry safety rests entirely on the offset idempotency of mechanism 1.
• Ratis: the complete command, chunk data included, travels through the Ratis log, and every replica writes its own staging directory.
• IoTConsensus: after the write peer applies the piece, followers receive a marker-only entry (pieceMarker: index, checksum and byte size, a few dozen bytes). The bytes themselves are read back on demand from the staged file through ChunkPayloadRef (deferred materialization).

2.3 Splitting and Streaming Dispatch (Phase 1)

• Pipeline: TsFileSplitter emits TsFileData (CHUNK / DELETION) -> TsFileSplitConsumer buffers and routes it through DataPartitionRouter -> PieceDispatcher dispatches it.
• Memory budget: MemoryBoundedBuffer (budget = thriftMaxFrameSize >> 2) stays in sync with the cluster-wide LOAD data cache; when the budget is exceeded, PieceDispatcher evicts the largest buffered piece first (largest-first).
• DELETION semantics: a deletion is copied into every buffered piece, and chunks are routed before the deletion is written, so a deletion never overtakes its data.
• Progress index: while splitting, a ProgressIndex is produced per time partition and later carried by PREPARE / COMMIT.

2.4 Staging Directory Layout and Hole-Tolerant Resume

• Layout: <load dir>/<database-region>/<load id>/ (LoadStagingDirs), deliberately independent of the DataRegion sequence/unsequence data layout.
• Progress bitmap: LoadTsFileProgress is an append-only progress file that records chunk offsets, the chunk header and statistics metadata, the exact physical data range, and the consensus index.
• Resume and verification: after a restart the TsFilePrecalculatedChunkWriter metadata is rebuilt from those records; comparing the physical length with the recorded total length precisely identifies a hole (a piece that only arrives after the restart) and allows the file to be completed; isReady(fileLength) verifies that a staged file is complete.

2.5 Deterministic Staging Directory Reclamation (Watermark-Driven)

Timeouts are gone; the lifetime is driven by what the system actually did:

• Chain: a finished task writes its terminal marker -> LoadTaskRetention decides when the directory may be released, based on the WAL safe-deletion watermark -> LoadTsFileCleaner deletes it. The release is triggered by the watermark callback, not by polling.
• Reclamation condition: every replica must have passed that index (replicasReached). A COMMIT additionally requires isComplete (no hole, progress covering the end of the file), because otherwise a replica could not read the payloads back. An ABORTed task is plain garbage once the watermark passed: it was never imported.
• Degradation: consensus V2 and every protocol without a WAL watermark report ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX (Long.MIN_VALUE, meaning "never reported"). That sentinel value is what marks "no follower to wait for", so a COMMIT or ABORT may be released as soon as it was applied.
• Restart fallback: on startup the staged directories are scanned, their terminal records are read, and directories that a previous run did not get to delete are reclaimed.

2.6 Zero-Copy and Snapshot Isolation

• Zero-copy references in the WAL: there is exactly one copy of every payload, and it lives in the staged file. The WAL, the replication queues and the memory accounting all hold ChunkPayloadRef references. The reservation accounting of IndexedConsensusRequest, LogDispatcher, SubscriptionQueueRegistry and IoTConsensusMemoryManager was adjusted accordingly (the reserved amount is snapshotted when the entry is queued and returned unchanged when it is released), so accounting cannot drift.
• Snapshot isolation: LoadTsFileSnapshot owns the snapshot and restore of the LOAD staging tree. The generic SnapshotTaker / SnapshotLoader explicitly skip the LOAD directories, so a .progress file is never mistaken for a data file; a recovering replica inherits the partial physical state plus the progress bitmaps and can continue an unfinished 2PC.

2.7 Configuration and Failure Fallback

• Configuration: setLoadTsFileDirs was added (it refreshes the canonical paths as well), and the lower bound of loadTsFileSpiltPartitionMaxSize is now explicitly >= 1, since a value of 0 would fail every LOAD whose source file spans at least one time partition.
• Failure fallback: LoadFallbackHandler converts a failed TsFile into tablets and retries. The source file is deleted physically only after the whole LOAD batch has ended, which keeps the deleteAfterLoad semantics safe.

---

3. Core Design Philosophy

1. Reuse the existing consensus stack instead of building another one. The LOAD write path is the ordinary write path, so consistency, crash recovery and catch-up semantics come for free.
2. Self-describing commands, stateless server. A phase command carries everything it needs (totals, checksum, physical layout). The server keeps no volatile state machine, which is what makes duplicate delivery and out-of-order restarts robust.
3. Physical facts before in-memory state. Idempotency by offset replaces a dedup table: writing the same piece twice is harmless by construction. The progress file is append-only, and both resume and reclamation rest on what is durably on disk.
4. Lifetime driven by the real watermark. Liveness is decided by the consensus watermark, not by a timeout or a reference count.
5. Delete conservatively, recover aggressively. Deletion requires no hole plus an advanced watermark; recovery reads as much as it can and supports completing a file that still has a hole.
6. Rollback must know when to stop. A participant that already committed is never rolled back, because a fake rollback would only create an illusion of consistency; a cross-region failure that cannot be made atomic is degraded to the tablet retry path instead.
7. Decide before acting. mustRetain is evaluated before the import and passed into it, so the decision and the action can never disagree.
8. Single responsibility and separated concerns. Splitting, routing, buffering, dispatching, submitting, rolling back and falling back are separate, individually testable classes; the cleaner is a single DataNode-level service; the LOAD staging snapshot logic does not pollute the generic DataRegion snapshot; all LOAD classes live in scheduler.load and storageengine.load.

---

4. Compatibility, Degradation and Feature Overview

4.1 Failure Degradation and Fallback

• Tablet fallback: whenever the 2PC path leaves a cross-region inconsistency or fails overall, LoadFallbackHandler converts the failed TsFile into tablets and retries (table model through convertForTableModel, tree model through convertForTreeModel). If the fallback succeeds the state machine ends in FINISHED, otherwise in FAILED. The source file is deleted physically only after the whole batch ends.
• Not strictly atomic: this is an optimistic two-phase commit whose channel is the consensus log, whose idempotency is a physical offset, and whose liveness is a consensus watermark. A non-atomic cross-region outcome (A committed, B rolled back) is covered by the tablet retry described above.

4.2 Compatibility and Degradation

• Protocol degradation: consensus V2 and regions without replication degrade to immediate release.
• Version compatibility: the old PULL protocol is no longer supported; an old node calling sendLoadCommand receives an explicit "protocol removed" error; slice-carrying piece dispatch remains compatible (the Fix oversized Load TsFile piece dispatch #18627 path was kept and adapted).
• Local loading: a single-node or local load goes through LocalLoadStrategy and never crosses the network.

4.3 Monitoring and Configuration

• Configuration validation: setLoadTsFileDirs was added, and loadTsFileSpiltPartitionMaxSize >= 1 is enforced, so a value of 0 cannot fail every LOAD.
• Metrics: the point-count metric LoadPointCountMetrics, plus per-phase cost metrics (LoadTsFileCostMetricsSet: FIRST_PHASE, SECOND_PHASE, SCHEDULER_CAST_TABLETS, and so on).

4.4 Feature List

Scheduling and splitting

• Two strategies chosen by needDecodeTsFile: local direct load (no decode) and two-phase consensus load (decode);
• splitting into CHUNK / DELETION, batched partition lookup and region routing (including the table-model and pipe database hint);
• memory budget with largest-first eviction, and a flush of the remaining pieces at end of file;
• region migration detection (RegionReplicaSetChangedException when the replica set changes);
• failure fallback: TsFile to tablet retry, with the state machine ending in FINISHED / FAILED.

2PC protocol

• Phase commands BEGIN / PIECE / PREPARE / COMMIT / ABORT applied through the consensus state machine; BEGIN is a no-op and the writer is created lazily by the first PIECE; the old PULL is no longer supported;
• one loadId per region; PREPARE / COMMIT carrying pieceCount / totalBytes / checksum and the per-time-partition ProgressIndex;
• submission to the write peer of the partition: Ratis leader, IoTConsensus write node; RegionWriteExecutor locally, internal RPC (sendBatchPlanNode) remotely;
• offset idempotency, the no-hole check on PREPARE with isSealed replay skip, and no-op COMMIT / ABORT for a finished loadId;
• the failure matrix (full ABORT on phase-1 failure, rollback of uncommitted regions on PREPARE failure, the current region left alone on COMMIT failure, best-effort ABORT);
• the decide-before-import ordering of mustRetain (a retained task is imported from a copy);
• bounded retry of transient failures (3 attempts, 100ms backoff, 4 status codes).

Staging writes

• one directory per task; one staged TsFile per time partition; TsFilePrecalculatedChunkWriter writing directly with pre-calculated metadata;
• PIECE writes recording the physical landing point of every chunk and returning a PieceRef; prepare sealing, loadAll importing, close discarding;
• chunk references (ChunkPayloadRef) and the "payload unavailable" exception (ChunkPayloadUnavailableException).

Progress and resume

• an append-only progress file (magic, uuid, the physical range and metadata of every chunk, the consensus index);
• restart recovery: writer metadata rebuilt from the records, completion of a file that still has a hole, isReady completeness check, and a terminal marker that prevents a finished task from being resumed.

Reclamation and retention

• the terminal record in the progress tail (op plus index); LoadTaskRetention keeping a directory until the WAL watermark passes; LoadTsFileCleaner deleting through both its registry and a directory scan;
• ABORT deleted as soon as the watermark passed, COMMIT additionally requiring "no hole"; immediate deletion on V2 or on regions without replication; missed directories reclaimed after a restart.

WAL / consensus / memory

• the new WAL entry type LOAD_TSFILE_CONSENSUS_NODE(13) (isUserData classification, serialization and deserialization, IWALNode.log(memTableId, node));
• the WAL holding references only: downstream V1 sync forwards a LoadPieceConsensusRequest that reads the payloads back on demand;
• the watermark callback chain IWALNode.setSafeDeletedSearchIndexListener and getSafelyDeletedSearchIndex();
• the queue-accounting fix for deferred-serialization requests (getQueueReservedMemorySize, hasDeferredRequests).

Snapshot

• snapshot / restore / clear of the LOAD staging tree; the generic snapshot path skipping the LOAD directories; the staged files and progress bitmaps of in-flight tasks restored together.

---

This PR has:

• been self-reviewed.
  • concurrent read
  • concurrent write
  • concurrent read and write
• added documentation for new or modified features or behaviors.
• added Javadocs for most classes and all non-trivial methods.
• added or updated version, license, or notice information
• added comments explaining the "why" and the intent of the code wherever would not be obvious
  for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold
  for code coverage.
• added integration tests.
• been tested in a test IoTDB cluster.

---

Key changed/added classes (or packages if there are too many classes) in this PR

[Caideyipi]

[P1] PREPARE/cleanup does not actually verify gap-free coverage

isReady() only compares the file length with the maximum physicalEnd recorded in the progress file. If a later piece arrives while an earlier piece is missing, the file length can still reach that maximum offset and the hole can be treated as complete. The design requires merging the recorded physical ranges and verifying continuous coverage from the TsFile header to the end before PREPARE/COMMIT. getContiguousPrefix() already exists, but it is not used here. Please make isReady(), TsFileWriterManager.prepare(), and LoadTsFileCleaner.isComplete() share a gap-free completeness check, and add a test where an out-of-order missing piece makes PREPARE fail.

[Caideyipi]

[P2] Compare the full endpoint when selecting the Ratis leader

The leader is matched only by IP address. If a deployment has more than one internal endpoint on the same host, this can select the first replica with that IP even when its port is not the current Ratis leader. Please compare both IP and port (or the complete endpoint) and add a test with identical IPs and different ports.

[Caideyipi]

[P1] Do not swallow snapshot file traversal failures

visitFileFailed() returns CONTINUE without surfacing the IOException. A snapshot can therefore report success and be transferred without one or more LOAD staged files or progress files. Restoring such a snapshot can resume a task with missing physical ranges, which conflicts with the HA requirement. Please propagate the failure (or return an explicit failure) so snapshot creation/transfer is aborted when any staged file cannot be enumerated.

[Caideyipi]

[P1] Snapshot restore only writes to the first configured staging directory

restore() selects LoadStagingDirs.baseDirs()[0], while LOAD writers can be allocated across multiple configured base directories by the folder manager. A snapshot containing tasks/files from other staging roots is therefore restored into a different root, and the subsequent recovery/retention scan can miss it or use paths that no longer match the recorded PieceRefs. Please preserve the source base-directory mapping in the snapshot (or restore through the same folder manager) and add a multi-root snapshot/restore test.

[Caideyipi]

[P1] Snapshotting active LOAD tasks is not atomic with piece writes

LoadTsFileSnapshot.snapshot() copies each staged file and its .progress file independently while the writer can append and update them concurrently. A snapshot may therefore contain a newer file length with an older progress log (or vice versa). After restore, recovery can either truncate valid bytes or accept inconsistent metadata, and the missing range may not be recoverable from consensus. Please coordinate snapshotting with the writer/flush lock or capture a consistent per-task checkpoint, and add a concurrent snapshot/write test.

[Caideyipi]

[P1] A failed PREPARE/COMMIT cleanup can permanently leave staged data

abortRegions() sends ABORT only once and does not use the retry helper used for PIECE/PREPARE/COMMIT. A transient dispatch failure can therefore leave a task directory and its WAL-referenced bytes behind indefinitely; the caller returns failure without a guaranteed rollback. Please retry ABORT on transient failures (or persist a cleanup task) and cover a transient ABORT failure followed by recovery/restart.

[Caideyipi]

[P1] Do not fall back to a stale replica set after route refresh failure

refreshReplicaSet() catches every exception and returns the replica set captured before splitting. During region migration or a write-node switch, this can route PREPARE/COMMIT/ABORT to an obsolete node. The retry loop then retries the same stale route, while the operation may already have partially succeeded on the new route. Please fail the dispatch or explicitly re-resolve with bounded retries instead of silently using stale metadata.

[Caideyipi]

[P1] Do not mark a failed COMMIT region as committed before its outcome is known

When COMMIT returns a non-success status, this code adds the current region to committedReplicaSets and excludes it from ABORT. A non-success RPC can mean the request timed out after the consensus entry was actually committed, but it can also mean the entry was never applied. In the latter case the region is left with prepared/staged data and no ABORT. Please distinguish an acknowledged COMMIT from an unknown outcome, or issue an idempotent recovery/status check before excluding the region from rollback.

[Caideyipi]

[P2] Snapshot restore should preserve all staged base directories

restore() always targets baseDirs()[0], but the writer manager allocates task directories through the configured folder manager. With multiple data directories, a task copied from another root can be restored to the wrong root and its PieceRef paths no longer describe the restored files. Please preserve the source root in the snapshot layout or restore via the folder manager, and add a multi-root test.

[Caideyipi]

[P2] Refreshing the partition route on every piece can change the target replica set mid-load

allReplicaSets and regionLoadIds are keyed from the replica set returned by the splitter, but submit() may send individual pieces to a newly refreshed replica set. If migration happens between pieces, PREPARE/COMMIT/ABORT still iterate the original set while the pieces may have been staged under the new route. Please either pin the route for one load transaction or update the transaction state atomically when migration is detected, and add a mid-load migration test.

[Caideyipi]

[P2] Snapshot restore currently ignores the configured storage allocation policy

The snapshot format stores only loadId directories and filenames. It does not retain which configured load base directory owned the task, while LoadStagingDirs can distribute tasks across roots. Restoring everything to the first root can cause disk imbalance and, more importantly, make the absolute paths embedded in PieceRefs invalid. Please preserve the originating root or rewrite all references consistently during restore.

[Caideyipi]

[P2] Snapshot copy can expose partially copied files to the receiver

The snapshot code copies directly to the final snapshot path with REPLACE_EXISTING. If snapshot transfer starts while a large staged TsFile is still being copied, a receiver may observe a truncated file and a complete-looking filename. Please copy to a temporary file and atomically rename it, or ensure the snapshot manifest is published only after every task file has been copied successfully.

[Caideyipi]

[P1] Validate ChunkPayloadRef bounds before allocating the payload buffer

The reference constructor accepts arbitrary negative or oversized offset/size values. readPayload() then calls Math.toIntExact(size) and allocates
ew byte[...]; a malformed consensus/WAL entry can trigger egativeArraySizeException, OutOfMemoryError, or invalid seeks outside the intended staged range. PieceRef has explicit bounds checks, but ChunkPayloadRef does not. Please validate non-negative values, an upper bound, and offset + size overflow at deserialization/construction, and convert malformed input into a controlled LOAD error.

[Caideyipi]

[P2] Use a path-boundary check that distinguishes a directory from its prefix

locateFile() accepts a canonical path when payloadPath.startsWith(baseDirPath) is true. A configured root such as /data/load would also accept /data/load-evil/..., because the comparison is string/path-prefix based rather than a child-path check. Please normalize the base path and require payloadPath.startsWith(basePath.resolve(...)) with a path-boundary/relativize check, or use Path#startsWith on a normalized directory path with the separator semantics verified.

[Caideyipi]

[P2] Add focused fault-injection tests without expanding the IT suite

The current tests cover several command-ordering cases, but the critical recovery invariants would be better protected by small unit tests rather than more cluster IT cases: (1) an out-of-order later range with a missing earlier range must make PREPARE fail, then succeed after the missing range arrives; (2) replaying the same PIECE/PREPARE/COMMIT must be idempotent; (3) a transient ABORT/COMMIT transport failure must leave a recoverable cleanup state; (4) snapshot/restore must preserve staged files and progress metadata, including multiple configured staging roots. These can use temporary directories and mocked submitters/consensus responses, so they should not materially increase integration-test runtime.