Repository navigation
LOAD TsFile: use the Pair the resolver takes and cover the HA cases - #18901
Open
luoluoyuyu wants to merge 17 commits into
Open
luoluoyuyu wants to merge 17 commits into
luoluoyuyu wants to merge 17 commits into
Conversation
The router and the resolver each wrapped a (device, time partition) pair into a private record only to unwrap it again on the way in or out. Pair already carries the value equality the deduplication maps need, so keep the pair itself and drop both SlotKey records, the record and its conversion. Cover the two cluster cases the consensus LOAD path has to survive: - a LOAD that runs while one DataNode is down: restarting that DataNode must replay the LOAD commands it missed and serve the loaded file. - a LOAD whose region is migrated mid-flight: the LOAD has to follow the refreshed route or fall back to the tablet path, but every point of the file must be imported and the migration must finish.
Caideyipi
reviewed
Oct 19, 2026
Caideyipi
left a comment
Caideyipi
left a comment
Collaborator
There was a problem hiding this comment.
The Pair refactor passes all four existing DataPartitionRouterTest cases when the PR sources are compiled against the cached dependencies. I also checked the successful Cluster IT - 1C3D run, which executed both added HA tests. I found two issues with how the new tests establish replica recovery and concurrent migration coverage; details are inline.
Comment on lines
+302
to
+304
| // The connection is pinned to the restarted DataNode, so this waits for the replicas it | ||
| // replayed the LOAD into and fails if the restart left them behind. | ||
| awaitPointCount(restartedDataNode, expectedPointCount); |
Collaborator
There was a problem hiding this comment.
[P2] Verify recovery using the restarted node's local replicas
getConnection(dataNode) fixes the JDBC endpoint, but the restarted node still plans a distributed query. Under the default strong read consistency, AbstractFragmentParallelPlanner.selectTargetDataNode() chooses the first available replica in the region route, which can be DN1 or DN2. Those nodes already returned expectedPointCount before the restart. Consequently, this assertion can pass as soon as DN0 accepts connections even if its replicas never replay the missed LOAD commands. Please force the storage reads onto the restarted node, or verify its local replica data/applied state, so a broken catch-up path makes this test fail.
Comment on lines
+164
to
+169
| final int[] migration = awaitMigratableRegion(); | ||
| Assert.assertFalse("the LOAD finished before the migration could overlap it", load.isDone()); | ||
| try (final Connection connection = EnvFactory.getEnv().getConnection(); | ||
| final Statement statement = connection.createStatement()) { | ||
| statement.execute( | ||
| "migrate region " + migration[0] + " from " + migration[1] + " to " + migration[2]); |
Collaborator
There was a problem hiding this comment.
[P2] Synchronize the LOAD with the migration notification
load.isDone() observes the future before opening another connection and submitting MIGRATE REGION. The LOAD can finish during that gap. Also, ProcedureManager.submitRegionOperationProcedures() returns after enqueueing the migration, and the DataNodes receive the migration notification later. In either case, the remaining checks can pass with the LOAD and migration running sequentially, without exercising route refresh or migration fallback. A fast LOAD that finishes before the 200 ms region poll instead fails the assertion despite correct behavior. Please add a deterministic barrier/test hook that pauses the LOAD after dispatching a piece, waits for the migration to become active, and then resumes it.
jt2594838
approved these changes
Oct 19, 2026
A connection pinned to the restarted DataNode can still read from another replica, so the point-count check alone can pass before that node has replayed the LOAD. Compare each local replica's on-disk size with the other replicas of its region instead, and disable compaction so the sizes stay comparable.
A region may migrate while the LOAD is between two pieces. The piece dispatched after it still goes to the pinned route first, fails there, and must be retried on the route the region has now; when that route is gone, the piece must fail instead of being sent to a stale route. Both are covered with a scripted submitter, so no test hook is needed in production code.
The writer used to hand every small write to the FileChannel, which costs one syscall per chunk header, per chunk and per metadata field. Collect the writes in memory and flush them once the buffer (1 MiB, allocated at half and grown on demand) is full, so a staged TsFile reaches the channel in large blocks. getPosition() counts the buffered bytes without flushing, so the chunk offsets recorded by the writer stay exact, and a seek flushes the buffer first. The size is configurable through the internal load_tsfile_write_buffer_size (startup and hot reload, must be positive); it is deliberately not part of iotdb-system.properties.template.
jt2594838
reviewed
Oct 20, 2026
Resolve the bounds of the time partition of an AlignedChunkData once instead of once per buffered point. TimePartitionUtils derives the bounds through BigInteger arithmetic, so a chunk group whose points sit one time partition apart otherwise runs O(points * partitions) bound computations and allocates the matching temporary objects while the pages are encoded. Splitting a TsFile of 2048 points, one per time partition, drops from 1.5 GB to under 128 MB of allocation. The splitter already decodes a value page once and shares the array with every partition that it spans; the new TsFileSplitterTest case covers the whole split and reports the 1.5 GB without this change. Verified with TsFileSplitterTest of iotdb-core/datanode: 5 tests, 0 failures.
- Reuse a ByteBuffer view for buffered flushes and a per-thread payload buffer. - Group buffered aligned pages once instead of scanning them per chunk. - Harden staged payload bounds and cover payloads larger than the scratch buffer. - Throttle the migration IT LOAD so it cannot finish before migration submission.
The buffered output is used by a single writer, and the per-call 8 KiB payload buffer does not justify a thread-local allocation in the I/O path.
Keep the LOAD paused until show regions reports the migrated region as Adding or Removing, and assert it is still running at that point, so the migration IT cannot pass with the two operations serialized.
Set the DataNode property through the MPP config before the cluster starts, so the regression test adds no test-framework API.
…s-review-fix # Conflicts: # iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/load/TsFileSplitterTest.java
jt2594838
reviewed
Oct 21, 2026
Comment on lines
+410
to
+412
| if (page.needDecode || page.pageHeader == null || page.pageHeader.getStatistics() == null) { | ||
| return null; | ||
| } |
Contributor
There was a problem hiding this comment.
So the only-one-page chunk must be decoded? Why?
Comment on lines
+144
to
+147
| final File file = locatePayloadFile(); | ||
| if (size == 0) { | ||
| return; | ||
| } |
Contributor
There was a problem hiding this comment.
Check the size before locating.
Comment on lines
+154
to
+156
| if (read < 0) { | ||
| throw new EOFException(); | ||
| } |
Contributor
There was a problem hiding this comment.
Provide the path and the current position, etc.
…om node config * Check every copied page size and the assembled chunk length before handing out a chunk, and fall back to re-encoding when they do not match * Pass the buffer size to BufferedFileChannelOutput instead of letting it read the DataNode configuration, and drop its unreachable default size * Document why the splitter may binary-search the points of a partition: chunk group writers reject out-of-order timestamps, so a page is always ascending
Address the review comments on the TsFile piece splitter and fix the page alignment defects they exposed. ChunkPayloadRef returns before locating the staged file when the payload is empty, and reports the staged file path, the requested length, the offset and the position reached when a read runs past the end of the file. An empty page (an all-null column, serialized as a single 0 varint) has to keep its page slot in both the copied and the re-encoded chunk: the aligned reader pairs the pages of a value chunk with the pages of its time chunk by position, so dropping one shifts every later value of that column onto the wrong timestamps. Only a chunk without any page that carries data keeps the zero-page layout of an all-null column. A page that is not empty but has no statistics belongs to a single-page chunk, where the statistics live in the chunk metadata; the previous check rejected both cases and re-encoded chunks that merely contained an empty page. decodeValuePage tested the serialized size of a page against 0, which is never true for an empty page (it serializes to one byte) and made such pages decode to an empty value array, failing the split with an ArrayIndexOutOfBoundsException when the page had to be split at a time partition boundary. An empty page now expands to one null per time of the page.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
The router and the resolver each wrapped a (device, time partition) pair into a private record only to unwrap it again on the way in or out. Pair already carries the value equality the deduplication maps need, so keep the pair itself and drop both SlotKey records, the record and its conversion.
Cover the two cluster cases the consensus LOAD path has to survive:
This PR has:
for an unfamiliar reader.
for code coverage.
Key changed/added classes (or packages if there are too many classes) in this PR