Repository navigation
feat(load): route decoded tsfile pieces through consensus - #18698
Open
luoluoyuyu wants to merge 59 commits into
Open
luoluoyuyu wants to merge 59 commits into
luoluoyuyu wants to merge 59 commits into
Conversation
…irectories Stage every LOAD piece in its own directory under the configured LOAD directories, apply it through the consensus state machine and keep the staged bytes until no replica can read them back any more: - every progress entry records the consensus index of the command that brought it, and the tail of the progress file records the COMMIT or ABORT the task finished with plus that command's index - a staged directory is resumed from those records, including a file that still has a hole because its piece arrives after a restart - a DataNode level cleaner deletes a finished directory: an ABORT once every replica applied it, a COMMIT once they did and the staged files are complete, which falls back to deleting right away on consensus V2 and on regions without replication
Caideyipi
reviewed
Sep 22, 2026
The second phase used to send PREPARE and COMMIT per region in a single loop, so a region could import its files while a later region refused to prepare, leaving a partially applied load behind. Run the two rounds of the protocol instead: PREPARE is sent to every touched region first, and COMMIT only once every one of them agreed. - a failed PREPARE now rolls back every touched region, because nothing was imported yet - a failed COMMIT rolls nothing back: the regions that agreed must still commit, and the failure is left to the tablet fallback
Caideyipi
reviewed
Sep 22, 2026
Caideyipi
reviewed
Sep 22, 2026
Caideyipi
reviewed
Sep 22, 2026
| for (final File file : files) { | ||
| if (!file.isFile()) { | ||
| continue; | ||
| } |
Collaborator
There was a problem hiding this comment.
[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.
Collaborator
There was a problem hiding this comment.
The progress-first copy protects records already present from referring to bytes that were not copied, but the inverse race remains: after copyProgressLogs lists the task, a new piece can create its progress log and write its TsFile before copyStagedFiles lists files. The snapshot can then contain the new TsFile without its progress log. If the region snapshot point already includes that piece, WAL replay will not recreate it. Please tie the copied per-task files to one apply/checkpoint boundary, and cover a piece that starts between the progress and TsFile scans.
Collaborator
There was a problem hiding this comment.
[P1] The snapshot repair still cannot repair the race described here
copyContentFiles() scans before copyProgressLogs(). A new staged file and its progress log can both be created between those scans, so the second scan copies the log while the first scan never copied its staged file; repairStagedFilesAgainstProgressLogs() then explicitly continues when the copied file is missing. Also, copyRange() opens the snapshot target itself as the source, so it cannot grow that copy from the live staged file when a copied log records a longer prefix. Please keep the live source file/path through repair (or capture one consistent file set) and add a test that creates a new partition/file between the two scans.
Caideyipi
reviewed
Sep 22, 2026
Caideyipi
reviewed
Sep 22, 2026
A COMMIT or an ABORT is the last thing the coordinator sends about a task, so a submission that failed transiently used to leave a region with staged data that only a scan of the staging directories could reclaim. Both commands now go through the same bounded retry as a piece (3 attempts, 100ms x attempt backoff). - ABORT is retried whatever the failure kind, because dropping staged data is idempotent: an ABORT of a task this region no longer holds succeeds - COMMIT is retried on transient failures only: a permanent rejection either means the task was already imported, or that the import failed half way and repeating it could import a partition of the staged file twice
Caideyipi
reviewed
Sep 22, 2026
| commitStatus.getMessage()); | ||
| // This region may have imported its files before the failure was reported, so it is left | ||
| // alone as well; the regions behind it have not committed anything yet. | ||
| committedReplicaSets.add(replicaSet); |
Collaborator
There was a problem hiding this comment.
[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.
Collaborator
There was a problem hiding this comment.
COMMIT now retries transient statuses, but only up to the same three-attempt limit. If all attempts fail, or the final status is classified as permanent, commitAllRegions returns failure without a terminal record or a status/recovery check. The staged task can remain indefinitely while the caller falls back, even though the region may have committed. Please persist and drive the presumed-commit decision to completion, or resolve the outcome before returning.
Collaborator
There was a problem hiding this comment.
[P1] ABORT still does not resolve an unacknowledged COMMIT
resolveTheCommitOutcomeOf() treats ABORT as a status check, but LoadTsFileManager.deleteAll() returns false whenever the writer manager is already absent. That is also the state after a successful COMMIT, so if every COMMIT response is lost, the following ABORT cannot distinguish committed from unknown and the coordinator still returns failure; the caller may then run fallback after data was imported. Please make terminal commands return an explicit persisted outcome (or make repeated ABORT report the committed outcome) and decide success/fallback from that result.
Caideyipi
reviewed
Sep 22, 2026
…he review A consensus LOAD had several ways of losing staged data without saying so, and the metadata of a task that survived a restart was filed under a device no reader asks for. This commit addresses the findings one by one. Routing: - a Ratis leader is reported as a node id and no endpoint, so resolving the write peer by endpoint threw before anything was sent; match it by node id instead - a route that does not hold the leader is stale: fail the dispatch rather than send the command to a replica that cannot commit it; the first location is kept only while the leader is unknown (IoTConsensus, or a coordinator that holds no replica) Completeness of a staged file: - isReady accepted a file as long as its last recorded chunk while a piece was missing in the middle, so PREPARE sealed the hole as zeros and the cleaner deleted bytes a replica still had to read back. One predicate now gates both, and a piece that arrives after a later one is written at its own offset instead of at the end of the file, which is what makes a hole mean a piece that never arrived Resuming, releasing and repairing a staged file: - a progress record kept its device as a string and rebuilt it as a segmented id, which splits a three segment device into two; metadata restored from such a record was filed under a device nobody queries, and the chunks staged before a restart were unreachable - the defect the ignored test reproduced - a progress log that ends in the middle of an entry, which is what a snapshot taken while pieces are still being applied holds, is cut back to its last complete entry instead of making the whole staged file unresumable - a piece of a staged file that no writer could resume now fails instead of being dropped silently, which used to import a file that missed those chunks Snapshot of the staging directories: - an enumeration failure aborts the snapshot instead of shortening it quietly - a copied file is published under its final name only once it is complete, so the transfer layer cannot pick up a partial one; the progress logs are copied before the staged files they describe, which is what the removed lock used to guarantee - chunk payload references and inline payload lengths are validated before they size a buffer, and a malformed reference fails as input instead of as an unchecked exception
A staged file was referenced by its absolute path, so the reference stopped describing the file as soon as the task left the root it was staged in: a node with several staging roots, or a node whose roots are named differently, could not read those bytes back, and a restored snapshot placed every task in the first root while the recorded references still pointed at the old one. - a piece and a chunk payload now record the path of their file relative to the staging root, which the reader resolves under its own roots; an absolute path is still accepted, so references written by an older build keep working - the snapshot records the root every task was staged in, and a restore puts the task back there when this node has that root and on a root it does have otherwise, which is all the reference needs The tests stage two tasks in two roots, restore them, and read a payload back through a reference that survived the round trip, including a restore onto a node that has a single, differently named root.
Every command of a transaction re-resolved the partition route and fell back to the stale replica set when that lookup failed, so a migration between two pieces could stage the rest of a task under a route whose regions hold a different plan, while PREPARE, COMMIT and ABORT kept iterating the set the splitter had resolved. A refreshed route also says nothing about whether the bytes staged under the pinned one may be committed. - the replica set the splitter resolved is pinned for the whole transaction and carried by every command of it - a write-node switch inside that set is still followed, because the write peer is resolved from the set itself (the current Ratis leader, or the first location for IoTConsensus) - a pinned route that has become obsolete fails the dispatch instead of being silently replaced; the load then fails and is retried from the start, which is what the migration check of the scheduler asks for as well
A region that was asked to commit a task it had already imported answered with a failure, because it only recognized the tasks it still holds a writer for. The coordinator cannot tell a lost answer from a lost command, so it re-sends the COMMIT that ended a task; a retry that follows a lost answer therefore reported a load that committed everywhere as failed, and left the region in the "unknown outcome" state the review asked to resolve. - a region remembers the tasks it committed, and answers a repeated COMMIT of one of them with success; the record is bounded, and a task whose staged directory is still there is recognized from the terminal record of its progress files as well - a repeated COMMIT of a task that committed before a restart, and whose staged directory was released, stays unknown, which is what it is
The review asked for focused fault-injection tests around the invariants the recovery of a load depends on. - a repeated PREPARE seals the staged files once, and a repeated ABORT leaves the region as the first one left it - a snapshot taken while pieces are being applied is restorable: every range a copied progress log describes is inside the bytes copied with it, which is the invariant the copy order exists for, and the test checks it on snapshots taken while a writer thread is still appending - a staged file is only read back from a configured staging root: a directory whose name merely starts with the name of one is not inside it, which closes the path-boundary finding of the review as a non-issue with a test
Only a consensus protocol that transfers the staging directory has to carry it in a snapshot: a replica of IoTConsensus rebuilds the staged files from the entries it applies as a member of the region, so a replica that joins in the middle of a task inherits the staged state - and the progress its recovery resumes - from the migration snapshot, while Ratis replicates the payload of every piece to every replica itself. - the snapshot of the staging directory is taken for IoTConsensus V1 and the V2 aliases, and skipped for the protocols that do not transfer it, which also keeps those files out of the transfer list
A command whose route is not the one that holds the region failed without ever learning that the region had moved: the refresh was removed when the transaction started pinning its route, which left a switched write node or a finished migration to fail the whole load. Look the route up again instead, and let the transaction adopt the one the region has now. - on a transient failure the route is looked up again, with what this DataNode has cached dropped first, exactly as the query path does before it re-analyzes a statement it was redirected for; the attempts are bounded - the route that changed is adopted by the transaction, so the commands that follow use it too instead of leaving the pieces staged under one route and the terminal commands under another - a route that cannot be resolved is no route at all: the command is repeated on the route it had and its answer is what the caller decides from, so the stale route is never replaced by an empty one silently - PREPARE goes through the bounded submission as well: it is idempotent, which the replay test of the previous commit pins - the two constants the earlier commits left unused are removed from both locales
The English and Chinese design documents were lost when the branch was reset before the review fixes; they are restored from the commit that added them and brought back in line with what the code does now. - the failure matrix and the retry policy are per command: PIECE, PREPARE and COMMIT repeat transient failures, ABORT repeats every kind of failure, and a repeated COMMIT of a task the region already imported answers success, which is what makes the outcome of a lost answer decidable - the route of a transaction: pinned per region, looked up again with the partition cache dropped first when a command fails, adopted by the whole transaction when it changed - the staging side: the gap-free completeness check that PREPARE and the cleaner share, a late piece written at its own planned offset, a torn trailing progress entry dropped on recovery, and a staged file that cannot be resumed failing its piece instead of dropping it - references recorded relative to the staging root, and the snapshot of the staging directory taken for IoTConsensus only, with the roots manifest, the progress-before-files copy order and the atomic publication of copied files
…c them" This reverts commit 471a9df.
Caideyipi
reviewed
Sep 24, 2026
Caideyipi
reviewed
Sep 24, 2026
Caideyipi
reviewed
Sep 24, 2026
| final int read = stream.read(bytes, offset, len - offset); | ||
| if (read < 0) { | ||
| throw new IOException( | ||
| DataNodeQueryMessages.EXCEPTION_UNKNOWN_LOADTSFILECONSENSUSOP_ORDINAL_ARG_62848FC2 |
Collaborator
There was a problem hiding this comment.
[P2] Localize the deserialization context
The raw progressIndex literal becomes part of an exception message, so the zh-locale build still emits English. Please represent this context in the matching EN and ZH DataNodeQueryMessages template.
Collaborator
There was a problem hiding this comment.
[P2] The context is still not localized
The call site still formats EXCEPTION_UNKNOWN_LOADTSFILECONSENSUSOP_ORDINAL_ARG_EC07D9BE with the raw progressIndex argument. The zh template therefore emits the English token progressIndex, and the message also describes an unknown op ordinal rather than a truncated progress-index payload. Please add a dedicated EN/ZH template for this deserialization failure and remove the raw English context argument.
… block MemoryBoundedBuffer kept the memory of the pieces of one TsFile twice: its own dataSize and the usage of the shared LoadTsFileDataCacheMemoryBlock the manager handed out. Both had to be moved by hand, and clear() was the last chance to return a leftover, so a release that was missed left the shared pool thinking the load still holds the bytes. LoadTsFileAbstractMemoryBlock#createSubBlock gives a block a sub-block that caps what one part of a task may hold and accounts every byte on the block itself, so the two views cannot drift apart: the sub-block keeps its parent in step, close() gives the leftover back, and hasEnoughMemory answers for both the cap and what is left of the parent. - MemoryBoundedBuffer keeps only the per-file cap (a piece has to stay transmittable in one frame, thriftMaxFrameSize >> 2) and delegates the accounting; clear() closes the sub-block, which stays safe to call again - nothing is reserved for a sub-block, unlike for a child of MemoryManager: it only adds a second cap on top of the shared one - LoadTsFileSubMemoryBlockTest covers the parent accounting, both caps, the release on close, a cap that is not positive, and resizing
stopLoadTsFileManager cleared the manager of a region and stopped it, but nothing kept the region from publishing a new manager right after: a piece that met the region just before it was removed could still create one and stage into the directory that was being deleted. The manager kept taking writes after stop() too, and a writer could be closed from under the piece writing it. Follow the rule the region already uses for its teardown, the one loadNewTsFile uses to refuse a deleted region: - markDeleted sets the deleted flag first and stops the LOAD manager with the rest of the region teardown, under the write lock of the region - getLoadTsFileManager refuses to publish a manager for a deleted region, so a LOAD command fails loudly and its leader retries it, instead of staging into a region that is going away; a plain TsFile load can be skipped, a consensus command cannot - the storage engine no longer stops the manager before marking the region deleted: markDeleted owns that now, which is what makes the order right Inside the manager one fair read-write lock guards the writers and the map of them, and nothing else: the pieces hold the read side while they create and write their writer, stop() and closing a writer hold the write side. Fairness keeps a stream of pieces from delaying a stop, and the stop waits for the pieces in flight instead of closing a writer under them. Staging into one staged file stays single-piece because a region applies the commands of a load in order, so the writer needs no monitor of its own.
stopLoadTsFileManager cleared the field after stopping the manager, which left a window for a piece to publish a second manager for a region that was being deleted, and it had to read the field under a monitor for that reason. A region keeps the one manager it publishes now, and stop() is terminal for it, so there is nothing to overwrite. The ordering lives in the volatile field and the deleted flag instead: a piece that publishes the manager while the region is being deleted sees the flag afterwards and stops its own manager, while the one the deleter saw is stopped by stopLoadTsFileManager, which needs no lock of its own. - getLoadTsFileManager checks deleted after it obtains the manager and stops that manager before the command fails - getLoadTsFileManagerIfPresent hands out the stopped manager; the cleaner and the snapshot already treat it as nothing to do - the message of a deleted region says it takes no LOAD work, since that check is no longer only about creating the manager
…aligned wire format NonAlignedChunkData wrote a constant false for needDecodeChunk and ignored it when reading the byte back: serialize() encodes the pages with endChunk() before serializeAttr, so the flag cannot be true there. It was only kept to mirror the aligned layout, where the flag is real - serializeAttr runs before the endChunk() fallback there, and the flag decides whether the page numbers follow. The format is not released yet, and the same bytes go into the WAL, so this is the moment to drop it rather than freeze it: the write, the read and the Byte.BYTES of serializedSize go together. Also fixes the comment above the alignment marker of the aligned layout, which claimed "always false" while writing true.
…f a marker A finished LOAD task that still has to be expanded for a replica kept its staged bytes but had its progress files deleted, so it needed a second file, terminal.marker, to tell a restart that the directory was finished - and that file carried only the search index, leaving the command to be guessed as COMMIT. Delete the progress files only when the staged files go too, like the other artifacts of a writer: a retained directory keeps them, and their tail already records the command the task ended with, because the writer records it before it closes. Recovery reads that record, so the marker file, its write and its read are gone, and an aborted task keeps reading as ABORT instead of falling back to COMMIT. The cleaner judges completeness by scanning the progress files, which was vacuous while they were deleted: keeping them makes that check real again for a committed task, while a task that never reached a terminal stays the only one the idle timeout has to judge.
The progress file is read back by a build that may be older than the one that wrote it, so an entry is now stepped over by its declared length: a version this build does not know, or a known one that gained a field, is skipped instead of being misread as the next entry, while a length that the parsed fields already overrun is still an error. A step that cannot advance fails instead of looping. Queueing a record costs no file access at all now: a chunk is built and kept in memory, and the file is opened, appended and forced once per piece rather than once per chunk (and once per task for the terminal record, which a restart reads to release the directory). Every stream an instance opens is closed in the method that opened it, so no file handle outlives a call. Writing is cheaper and more durable at the same time: the entry is built in one buffer the instance reuses instead of allocating one per chunk, the search index rides in the record it belongs to (which removes the offset map, the pending wrapper and the cached header size that a chunk header had to be deserialized for), and the chunk bodies are exact-sized arrays rather than views over a larger one. A piece forces its staged bytes and then the records that describe them, in that order, so a record never reaches the disk before the bytes it names.
The release path ran a full sweep, disk scan included, every time the consensus watermark moved, and the scan walked every staged directory of every region: for a task that is simply in progress that walk opened every progress file of the directory to find no terminal record at all. Only the registered tasks can become releasable when the watermark moves, so the release path sweeps those and leaves the disk scan to the sweep period it already has. The scan also asks the region first: a task the region still holds is in progress, owned by the command path and the terminal path, so its records are not read - it only goes through the idle timeout, which decides without touching the disk.
The list holds exactly the task directories of the writers the manager tracks, so it can be created with that size instead of growing.
The callers listed a directory and then skipped everything that was not a progress file themselves. A FilenameFilter does that inside the listing, where a rejected name never becomes a File at all, and the filter can live next to the suffix it matches instead of every caller re-deriving it.
When the staged bytes have to stay for a replica that has not applied the command yet, the import cannot move them, so it copied the whole TsFile into the region while the staged one stayed behind: two copies of the same bytes for as long as the watermark takes to pass. A sealed TsFile is never written again, so the two paths can name the same file: the import now asks LoadUtil for a hard link and falls back to a copy when the file system cannot make one (a different device, for instance). That helper already existed for the local transfer path and is public now. The resource file is still copied, because the load updates it, and the branch that does not retain the staged bytes keeps moving the file.
The mark looks like an unpaired leftover next to the length read, but removing it makes the repair miss the tear: without a mark, BufferedInputStream.skip delegates to the file stream, which reports a skip past the end of the file as a full one, so the half-written entry looks complete and the file is not truncated. With the mark the skip stays inside the buffer and stops at the end of the file, which is what detects the tear.
serializeChunkHeader and serializeStatistics built the bytes in a PublicBAOS and then copied them into an exact-sized array, because the record held a byte[]. The record keeps the buffer the bytes were built in again, which is what a ByteBuffer view gives without touching them at all: the copy was an allocation and a memcpy per body, for every chunk of every piece, while the array it saved was only marginally smaller. The rest of that change stands: the search index travels in the record, and the size of the chunk no longer deserializes the header to be known.
…piece Both bodies know their serialized size before they are built, so the PublicBAOS they are built in is created with that size: the buffer neither grows nor keeps spare room, and the record still holds a view over it rather than a copy of it. A piece is no longer fsynced either. The staged bytes and the progress records went to the disk together on every piece, which is what kept a record from ever being on the disk without the bytes it describes. Dropping only one of the two would be worse than keeping both: a record that reached the disk without its bytes makes the staged file shorter than its records, and recovery calls that unresumable instead of truncating the extra tail and staging it again. So both are flushed and left to the page cache, and a tail a crash keeps off the disk is truncated and staged again. The terminal record of a finished task is still synced, because that one is per task and decides whether a restart releases the directory or resumes it.
…hold Flushing the staged file of a piece looked harmless because the explicit force had gone, but the channel output's flush() is itself channel.force(false): every piece was still an fsync, and so was every chunk that carried a deletion. The bytes of a piece go straight to the file through the channel, with no buffer in between, so there is nothing to flush and both calls are gone. The records of a piece are still flushed into their file, just not fsynced, and the terminal record keeps its fsync because that one is per task. The two serialized bodies are created with the size their header and statistics announce, which is the size they serialize to: the buffer neither grows nor keeps spare room, and a test now asserts that its capacity is exactly that size.
… before Nothing is flushed while a piece is written: its bytes go straight to the file through the channel, so there is no buffer to drain per chunk, and the flush that every chunk carrying a deletion used to do is gone. When the whole piece is there, its files are forced once, before the records that describe them are appended. That order is the point: the bytes are durable before a record can name them, so a record may lag the bytes it describes, but it can never lead them, which is what keeps a staged file resumable after a crash. The record side needs no fsync of its own for that reason, and the terminal record keeps the one it has, because that one is per task.
…escribe A committed task's staged files were already checked to be complete twice on the way to the terminal - PREPARE refuses a file that is not continuous, and the import refuses one that was never prepared - but the terminal was written unconditionally. An aborted task's files were not checked at all, although a piece that never arrived leaves a hole in them. The terminal is now recorded only when every staged file can carry it: complete (no hole and at least as long as its records) for a commit, continuous for an abort, whose bytes are never imported. A directory that does not qualify is left without the record, so a restart keeps treating it as unfinished and the idle timeout releases it instead of the watermark doing it right away. The contract of the terminal record, which fields it needs and why the index cannot be derived from the records, is documented where it is written and where it is retained.
…ecords Continuity alone was asked of an aborted task, which accepts a file whose last recorded chunk is only half written: no hole is visible between the records, because the missing bytes are past the last one. An aborted file may be short, since it was never imported, but not shorter than the chunks it records.
Whether a staged file was written completely was decided from the replica's own records and the length of the file. That cannot see a piece that never arrived: the file is simply shorter and still looks continuous, because nothing recorded what was expected of it. The coordinator can tell, since it computed the offset of every chunk it dispatched, so PREPARE, COMMIT and ABORT now carry the end offset of the staged file of each time partition: the maximum over the chunk groups of the pieces whose dispatch succeeded, so a piece that could not be sent is not expected of anyone and an aborted task can still reach its expectation. A replica checks the file against that offset at PREPARE, refusing a file that did not get there, and when it marks a task ended, withholding the terminal record when the file is short so that the directory waits for the idle timeout instead of being released as finished. The records still decide that the file is continuous and, for an aborted task, that no recorded chunk was left half written.
IoTConsensusV2 replicates through pipe and has no search index to wait for, so a staged directory is of no use to a replica once the task ended: it is dropped right away and no terminal record is written for it, which also means no watermark is waited for. What does not change is that the coordinator still tells every replica how far each staged file has to be, since that expectation has nothing to do with the index. V2 also never leaves a hole behind. In V1 a chunk may be written beyond the end of the file, and the gap is filled when the piece that was missing arrives again; in V2 that cannot happen, so such a chunk is refused where it would be written, with the size of the gap it would have left. The choice follows the consensus implementation the node runs, and a test can pin either behaviour.
The long explanations of the new fields are shortened to what a reader needs, and the lookup of the end a time partition is expected to reach is one helper instead of the same expression in two places.
Measuring it by serializing the header into a stream, for every chunk of a piece, allocated and threw the bytes away to learn a number the header already knows.
The stream of a transfer was collected through its fully qualified name, the last of its kind in the load package.
The snapshot listed the four spellings the factory may hold for the protocols that stage through consensus, two of which are the same protocol before and after it was renamed. Asking the factory for the canonical name keeps that knowledge where it belongs, and the set is then the two protocols it actually means: a configuration written before the rename still snapshots its staging directory.
A snapshot copies its files into a temp target and renames it into place; when the file system can link the source instead, that is one directory entry rather than the bytes of a staged file. The copy stays as the fallback for a file system that cannot link or for a source on another device.
Writing it where the stream already is produced a file that does not match the layout the pieces were built from, and only a warning said so.
…rmat a comment The test built its expectations and asserted nothing, so the list was dead; it now writes both pieces, imports the file and checks every series. A comment of the backfill test is reflowed into two readable lines.
…eir refs in memory A writer manager belongs to one region, so the region in the key of its four maps was the same everywhere: the maps are keyed by the time partition alone now, the deletion path no longer checks the region it is already in, and the staged file name, which the key used to carry, is built where the file is created. The refs of a piece are that piece's growth of the staged files, so instead of listing the task directory twice per piece the manager remembers the length it last published for each of its files, seeded when the file is created or recovered, and refs what the piece added.
…ormat a comment The test built its expectations and asserted nothing, so the list was dead; it now writes both pieces, checks that the staged file is continuous and holds every chunk, imports it and reads every point of every series back. The shared reader used to compare only the ends of a series, which a middle staged elsewhere still satisfies, so it now checks every value and its timestamp. A comment of the backfill test is reflowed.
…s-ha1 SubscriptionQueueRegistry keeps master's detached retention and the guard that leaves a deferred request un-serialized while it waits in a queue, and getSerializedRequests() stays a plain getter as master has it: the bytes of such a request are materialized where the batch that carries them is built. StorageEngine keeps the per-region stop of the staged writers and master's stopIoTConsensusWrites().
…load-tsfile-consensus-ha1
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.
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:
LOAD bypassed the consensus protocol, so consistency was at risk.
The coordinator fanned the serialized
LoadTsFilePieceNodeout to every replica over a dedicated RPC (sendTsFilePieceNode), and then told each node to persist its own copy throughEXECUTE/ROLLBACKcommands. 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.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:
EXECUTE/ROLLBACKcommand protocol with two-phase commit (2PC);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.
2.2 Two-Phase Commit (2PC) Protocol
2.2.1 Roles and Transaction Boundary
LoadTsFileSchedulertoTwoPhaseConsensusLoadStrategy).loadIdper region (regionLoadIds.computeIfAbsent(regionId, UUID)), not a single global id.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
loadIdand applies the command.BEGINDataExecutionVisitorreturnsOKdirectly (a no-op). The staged writer is created lazily by the firstPIECE. The phase is kept only for cross-version compatibility.PIECEwritePiece: locate or create the staged TsFile for(device, time partition), write the chunks and deletions, returnPieceRef, then write the node (with references) to the WAL.PREPAREprepare: close the modification file, verify that the staged file has no hole, write the metadata zone to seal it, update the TimeIndex, record theProgressIndex.isSealed()(idempotent).COMMITloadAll: evaluatemustRetainfirst, import the TsFile, finish the task, then write the WAL entry and the separator.loadIdalready finished and its writer was removed, a no-op success.ABORTdeleteAll, finish the task, then write the WAL entry and the separator (discard, or hand the directory to the retention mechanism).PULL2.2.3 Protocol Timeline
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 thePIECE. Before writing, the server asksprogress.hasChunkAt(offset):ChunkPayloadReffrom where the chunk already sits; without one, keep the reference that arrived);..._ARRIVED_WITHOUT_ITS_CHUNK_PAYLOAD, so data can never be dropped silently; a mismatch betweenincomingRefs.size()andchunks.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.
pieceIndexis only an identity and a log-correlation aid; it carries no ordering constraint.Mechanism 2: decide before acting (must retain before import)
loadAllevaluatesretention.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 samesearchIndex, so "the decision allowed deletion while the import moved the file" cannot happen.Mechanism 3: a terminal record that makes the transaction recoverable
PREPARE,COMMITandABORTare all written to the WAL (logLoadNodeToWALplusinsertSeparatorToWAL). Before the progress files are dropped, the terminal marker (terminal op plus its consensus index) is appended to their tail.loadIdin 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.
PREPAREfailed on any regionPREPAREis not retried.PREPARE@1, PREPARE@2, ABORT@1 x3, ABORT@2 x3PREPAREfailed on the first regionPREPARE@1, ABORT@1 x3, ABORT@2 x3PREPAREsucceededCOMMITis sent to every region in turn. The commit point has passed and cannot be rolled back.PREPARE@1, PREPARE@2, COMMIT@1, COMMIT@2COMMITfailed with a transient errorPREPARE@1, PREPARE@2, COMMIT@1 x3, COMMIT@2COMMITfailed with a permanent errorPREPARE@1, PREPARE@2, COMMIT@1, COMMIT@2ABORTfailedPREPARE@1, ABORT@1 x3, ABORT@2 x32.2.6 Retry Policy and Its Interaction with Consensus
LOAD_CONSENSUS_SUBMIT_MAX_RETRIES) with a100ms x attemptbackoff, 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.pieceMarker: index, checksum and byte size, a few dozen bytes). The bytes themselves are read back on demand from the staged file throughChunkPayloadRef(deferred materialization).2.3 Splitting and Streaming Dispatch (Phase 1)
TsFileSplitteremitsTsFileData(CHUNK / DELETION) ->TsFileSplitConsumerbuffers and routes it throughDataPartitionRouter->PieceDispatcherdispatches it.MemoryBoundedBuffer(budget =thriftMaxFrameSize >> 2) stays in sync with the cluster-wide LOAD data cache; when the budget is exceeded,PieceDispatcherevicts the largest buffered piece first (largest-first).ProgressIndexis produced per time partition and later carried byPREPARE/COMMIT.2.4 Staging Directory Layout and Hole-Tolerant Resume
<load dir>/<database-region>/<load id>/(LoadStagingDirs), deliberately independent of the DataRegionsequence/unsequencedata layout.LoadTsFileProgressis an append-only progress file that records chunk offsets, the chunk header and statistics metadata, the exact physical data range, and the consensus index.TsFilePrecalculatedChunkWritermetadata 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:
LoadTaskRetentiondecides when the directory may be released, based on the WAL safe-deletion watermark ->LoadTsFileCleanerdeletes it. The release is triggered by the watermark callback, not by polling.replicasReached). ACOMMITadditionally requiresisComplete(no hole, progress covering the end of the file), because otherwise a replica could not read the payloads back. AnABORTed task is plain garbage once the watermark passed: it was never imported.ConsensusReqReader.DEFAULT_SAFELY_DELETED_SEARCH_INDEX(Long.MIN_VALUE, meaning "never reported"). That sentinel value is what marks "no follower to wait for", so aCOMMITorABORTmay be released as soon as it was applied.2.6 Zero-Copy and Snapshot Isolation
ChunkPayloadRefreferences. The reservation accounting ofIndexedConsensusRequest,LogDispatcher,SubscriptionQueueRegistryandIoTConsensusMemoryManagerwas adjusted accordingly (the reserved amount is snapshotted when the entry is queued and returned unchanged when it is released), so accounting cannot drift.LoadTsFileSnapshotowns the snapshot and restore of the LOAD staging tree. The genericSnapshotTaker/SnapshotLoaderexplicitly skip the LOAD directories, so a.progressfile 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
setLoadTsFileDirswas added (it refreshes the canonical paths as well), and the lower bound ofloadTsFileSpiltPartitionMaxSizeis now explicitly>= 1, since a value of 0 would fail every LOAD whose source file spans at least one time partition.LoadFallbackHandlerconverts a failed TsFile into tablets and retries. The source file is deleted physically only after the whole LOAD batch has ended, which keeps thedeleteAfterLoadsemantics safe.3. Core Design Philosophy
mustRetainis evaluated before the import and passed into it, so the decision and the action can never disagree.scheduler.loadandstorageengine.load.4. Compatibility, Degradation and Feature Overview
4.1 Failure Degradation and Fallback
LoadFallbackHandlerconverts the failed TsFile into tablets and retries (table model throughconvertForTableModel, tree model throughconvertForTreeModel). If the fallback succeeds the state machine ends inFINISHED, otherwise inFAILED. The source file is deleted physically only after the whole batch ends.4.2 Compatibility and Degradation
PULLprotocol is no longer supported; an old node callingsendLoadCommandreceives an explicit "protocol removed" error; slice-carrying piece dispatch remains compatible (the Fix oversized Load TsFile piece dispatch #18627 path was kept and adapted).LocalLoadStrategyand never crosses the network.4.3 Monitoring and Configuration
setLoadTsFileDirswas added, andloadTsFileSpiltPartitionMaxSize >= 1is enforced, so a value of 0 cannot fail every LOAD.LoadPointCountMetrics, plus per-phase cost metrics (LoadTsFileCostMetricsSet:FIRST_PHASE,SECOND_PHASE,SCHEDULER_CAST_TABLETS, and so on).4.4 Feature List
Scheduling and splitting
needDecodeTsFile: local direct load (no decode) and two-phase consensus load (decode);RegionReplicaSetChangedExceptionwhen the replica set changes);FINISHED/FAILED.2PC protocol
loadIdper region;PREPARE/COMMITcarryingpieceCount/totalBytes/checksumand the per-time-partition ProgressIndex;RegionWriteExecutorlocally, internal RPC (sendBatchPlanNode) remotely;isSealedreplay skip, and no-op COMMIT / ABORT for a finishedloadId;mustRetain(a retained task is imported from a copy);Staging writes
TsFilePrecalculatedChunkWriterwriting directly with pre-calculated metadata;PieceRef;preparesealing,loadAllimporting,closediscarding;ChunkPayloadRef) and the "payload unavailable" exception (ChunkPayloadUnavailableException).Progress and resume
isReadycompleteness check, and a terminal marker that prevents a finished task from being resumed.Reclamation and retention
LoadTaskRetentionkeeping a directory until the WAL watermark passes;LoadTsFileCleanerdeleting through both its registry and a directory scan;ABORTdeleted as soon as the watermark passed,COMMITadditionally requiring "no hole"; immediate deletion on V2 or on regions without replication; missed directories reclaimed after a restart.WAL / consensus / memory
LOAD_TSFILE_CONSENSUS_NODE(13)(isUserData classification, serialization and deserialization,IWALNode.log(memTableId, node));LoadPieceConsensusRequestthat reads the payloads back on demand;IWALNode.setSafeDeletedSearchIndexListenerandgetSafelyDeletedSearchIndex();getQueueReservedMemorySize,hasDeferredRequests).Snapshot
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