Repository navigation
LOAD TsFile: use the Pair the resolver takes and cover the HA cases - #18901
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
left a comment
Caideyipi
left a comment
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.
| // 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); |
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.
| 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]); |
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.
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.
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
| if (page.needDecode || page.pageHeader == null || page.pageHeader.getStatistics() == null) { | ||
| return null; | ||
| } |
There was a problem hiding this comment.
So the only-one-page chunk must be decoded? Why?
There was a problem hiding this comment.
You are right, that branch does not decode: it returns null so the chunk falls back to AlignedChunkWriterImpl. The actual defect was the check order - a page without statistics was rejected even when it is an empty page (an all-null slice of the column), so every chunk containing one was re-encoded. Empty pages are now copied as the page slot they are, since the reader pairs value pages with time pages by position, and only a non-empty page without statistics falls back. Fixed in d025e1a, covered by ChunkDataDirectWriteTest#testEmptyUndecodedPageKeepsItsPageSlot and the all-null-page split scenario.
| final File file = locatePayloadFile(); | ||
| if (size == 0) { | ||
| return; | ||
| } |
There was a problem hiding this comment.
Check the size before locating.
There was a problem hiding this comment.
Done in d025e1a: a zero-length payload returns before locatePayloadFile().
| if (read < 0) { | ||
| throw new EOFException(); | ||
| } |
There was a problem hiding this comment.
Provide the path and the current position, etc.
There was a problem hiding this comment.
Done in d025e1a: newStagedFileEofException now reports the staged file path, the requested length, the offset and the position the file ended at.
…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.
consumeAllAlignedChunkData built a BatchedAlignedValueChunkData for every chunk data and then consumed the original objects (the map's keys), so the wrapper and its satisfied-length copy were allocated and discarded. A LinkedHashSet keeps the same insertion order and the same de-duplication without the wrapper; the copy constructors that only served it go away with it. writePageBuffersToWriter built a HashSet of the chunk indexes that carry data for every chunk it re-encodes. Only an empty page asks that question, so answer it by scanning the pages of that one chunk instead: no set, no boxing, and no work at all when the chunk has no empty page.
The splitter drains pageIndex2ChunkData to the consumer and keeps using the same map for the pages that follow. The last cleanup dropped the wrapper that detached a drained object from that map, so a chunk data that had been handed to the consumer stayed in place: every later flush drained it again and re-emitted the chunks it had already emitted. A file split at four flush points therefore ended up with four time chunks while s2 and s3 had only two chunks each. The reader pairs the two lists by index (AbstractAlignedTimeSeriesMetadata#getAlignedChunkMetadata), so it threw IndexOutOfBoundsException: Index 2 out of bounds for length 2, which is what BatchedCompactionWithTsFileSplitterTest reported. Store a fresh BatchedAlignedValueChunkData back into every drained entry so the pages that follow land in a new object, and keep the hasPageWithData scan that replaced the per-chunk index set.
The batched layout that compaction writes when compactionMaxAlignedSeriesNumInOneBatch is smaller than the series count keeps one time chunk per time window and appends the value chunks of the remaining series batches behind all time chunks, so the splitter flushes the chunk data of a time chunk several times. None of the splitter tests built such a file, which is why the duplicated chunks of the last fix went unnoticed. Add three cases that write the batched layout by hand: - two time windows in two time partitions, all series in three batches, which must be split into one chunk group per partition without duplicated chunks - the same layout with an empty value chunk and an all-null value page in a window that crosses a time partition boundary, so the copied time chunk and the decoded one are both flushed more than once - the same layout for two devices of one table The cases check the emitted chunk data, and the split TsFiles are read back to verify the chunk metadata and the values of every series.
…s-review-fix # Conflicts: # iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java # iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
3 participants
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