Repository navigation
Conversation
OneSizeFitsQuorum
force-pushed
the
master
branch
2 times, most recently
from
December 5, 2024 05:18
8ee8eb6 to
502f54a
Compare
* Subscription: improve WAL replay gap observability * [Subscription] Fix WAL iterator initialization
* Fix DataNode disk status recovery across multiple disks * Keep Pipe receiver disk failures out of node status * Address review comments on disk status handling
* [Subscription] Disable subscription by default * [Subscription] Enable subscription in DataNode unit tests * [Subscription] Refine subscription configuration * Update pom.xml
…18542) Signed-off-by: Zhu yizhang <95731595+godarrenw@users.noreply.github.com>
* Pipe: Fix concurrent TsFile reference increases * Test: show concise result set diffs * Pipe: Roll back failed TsFile reference increases * Clarify nullable pipe resource map lookup
#18533) * Filter all-null FIELD columns before tablet insert in Java/C++/Python clients Skip or shrink tablets that only contain null FIELD values so insert paths avoid shipping unused measurement columns across languages. * Keep TAG/ATTRIBUTE columns when all FIELD values are null in table model Table-model tablet inserts may carry only time and tag/attribute columns. Do not treat all-null FIELD columns as an empty tablet when non-FIELD columns remain, and clarify buildInsertTabletReq skip semantics. * Refactor C++ Tablet init and use metadata-only factory in filterNullColumns. Add a private constructor and createWithoutValueColumns to build tablets without pre-allocating value columns, fix bitmap null-check bounds, and update the Java table-model test constructor. * Check only active tablet rows when filtering all-null columns. BitMap is sized to maxRowNumber, so a full-map isAllMarked() can miss all-null FIELD columns when rowSize is smaller. Use isRangeAllMarked on [0, rowSize) in the Java, C++, and Python clients. * Keep null measurement names when filtering all-null tablet columns. A never-written column is treated as all-null; dropping it skipped the existing measurement-name check and omitted timeseries that tests expect.
* fix(load): validate TsFile before async move * fix(load): report invalid tsfile in async pipe path
Allow finite internal pipes to complete after an explicit empty DataRegion report while preserving realtime and external-source pipes.
* [Pipe] Add SHOW PIPE MEMORY * [Pipe] Add hierarchical memory diagnostics to SHOW PIPE MEMORY * [Pipe] Fix memory diagnostic allocation behavior * [Pipe] Address SHOW PIPE MEMORY review feedback * [Pipe] Separate peak and max memory diagnostics * [Pipe] Share SHOW PIPE MEMORY across models and report pipe ownership * [Test] Calculate information schema table count dynamically
* Support retaining subscription progress after unsubscribe * [Subscription] Address WAL retention review comments
* [Feature] Expose DROP TABLE mod file progress * [Fix] Wait for in-flight DROP TABLE mod progress refresh
* [Pipe] Add reliable DataNode completion metric * [Pipe] Address completion metric review feedback
* [Bugfix] Reclaim idle RPC buffers when the shared pool is exhausted * [Bugfix] Evict idle RPC clients to release buffers
…18876) * [Subscription] Support per-topic uncommitted event recycle timeouts * [Subscription] Allow disabling uncommitted event recycling
* Fix deletion WAL logging and recovery during memtable flush * Keep deletion fixes within single-memtable file lifecycle
* Fix queued IoTConsensus writes during shutdown * Test queued IoTConsensus writes across storage shutdown * Reproduce WAL checkpoint loss from writes after shutdown * Fence IoTConsensus applies before storage shutdown
This reverts commit 264053f.
#18887) * [Test] Retry session database cleanup after asynchronous table deletion * [Test] Add a field to invalid timestamp insert test
* [Fix] Preserve table insert errors for rejected fields * [Fix] Retain metadata errors for fully rejected table row batches * [Fix] Preserve table timestamps when all fields are rejected * [Test] Align rejected-field assertions and fix controlled flush fixtures
…mer closes (#18894) * [Bugfix] Preserve retained subscription progress after the last consumer closes * Fix detached retention queue capture
* [Subscription] Fix WAL replay progress after realtime gaps * Remove subscription WAL replay documentation * [Subscription] Initialize WAL position inside iterator factory
* LOAD TsFile: replicate pieces through consensus and clean up staged directories 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 * LOAD TsFile: commit only after every region agreed to prepare 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 * LOAD TsFile: retry the terminal commands of the 2PC protocol 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 * LOAD TsFile: fix the routing, completeness and snapshot findings of the 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 * LOAD TsFile: keep staged references valid across roots and nodes 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. * LOAD TsFile: pin the route of a load transaction 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 * LOAD TsFile: answer a repeated COMMIT of a committed task with success 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 * LOAD TsFile: cover the replayed commands and the concurrent snapshot 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 * LOAD TsFile: snapshot the staging directory only where it is transferred 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 * LOAD TsFile: follow the route a region has when a command fails 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 * Restore the LOAD TsFile consensus and 2PC design docs and sync them 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 * Revert "Restore the LOAD TsFile consensus and 2PC design docs and sync them" This reverts commit 471a9df. * LOAD TsFile: schedule the cleaner as a periodical job of the DataNode The cleaner used to own a daemon thread that slept for the configured interval and then swept the staged directories of every DataRegion. The node already provides that scheduling: the periodical job executor runs the low frequency background jobs of a DataNode on one shared thread and is started and stopped with the node, which is how the load cleanup and the pipe startup cleaner are scheduled on master as well. Register the sweep once, guarded by an AtomicBoolean, keeping the configured interval (load_cleanup_task_execution_delay_time_seconds). The shared executor cannot drop a single job, so stop() pauses the sweep through a flag instead of interrupting a thread. * LOAD TsFile: harden the staged copy, the piece references and the commit round - LoadTsFileSnapshot copies the staged files before the progress logs that describe them, repairs a copy whose log records bytes past its end, and fails a task directory that cannot be listed instead of transferring it as if it held no file. - ChunkPayloadRef and LoadTsFileConsensusNode.PieceRef bound only the end of the referenced range, so a staged TsFile larger than 2 GiB is no longer rejected. - TwoPhaseConsensusLoadStrategy stops repeating a command on a route that cannot be resolved again, and ends a task whose COMMIT outcome stayed unknown with the ABORT that the region acknowledges as well. - The log messages of these paths move into StorageEngineMessages and DataNodeQueryMessages, with the matching zh translations. * LOAD TsFile: precalculate the staged chunk offsets and route the load as pieces - TsFilePrecalculatedChunkWriter tracks a high-water mark instead of zeroing the range it pads, so filling the hole left by an out-of-order piece can no longer overwrite a chunk that is already on disk; the absolute offset of a chunk is produced by ChunkOffsetCalculator over a counting stream. - ChunkData hands the payload of a chunk out as a ChunkPayloadRef (path, offset, size), so the WAL and a consensus piece reference the staged bytes instead of carrying a second copy of them. - LoadTsFileProgress, LoadStagingDirs, LoadTaskRetention, TsFileWriterManager, LoadTsFileCleaner and LoadTsFileSnapshot describe the staged files, their offsets and their terminal states through .progress ledgers, repair a torn tail, and drop the staged directory of a task only once the whole batch has ended. - The coordinator splits a source TsFile and streams the result: TsFileSplitConsumer buffers chunks inside a memory budget, DataPartitionRouter maps every chunk to the region that owns it, PieceDispatcher groups them per region and flushes the largest piece first under memory pressure, and LoadConsensusSubmitter drives the PREPARE / COMMIT / ABORT round on the write node of each route. LocalLoadStrategy keeps the local path and LoadFallbackHandler converts a failed TsFile into tablets and retries it. - IWALNode declares getSafelyDeletedSearchIndex(), which WALNode already overrides. - Every new log, exception and message string lives in StorageEngineMessages and DataNodeQueryMessages, with the matching zh translations. - The class javadoc of LoadTsFileScheduler documents the end-to-end picture: the coordinator, phase 1 streaming piece dispatch and phase 2 two-phase consensus decision. * LOAD TsFile: fail a piece of a staged file with the file it cannot resume The writer of a partition is opened inside computeIfAbsent, and its mapping function turned every IOException into an IllegalStateException naming only the partition. A piece of a staged file that no writer could resume must fail with the file and the load task, so that wrapper hid the information the failure is about and replaced the checked failure the caller handles with an unchecked one. Open the writer through openWriter, which carries the failure out of the mapping function, which cannot throw a checked exception, and throws it as it is, and create the file, the resource, the progress log and the writer in createStagedWriter. A staged file that is already there fails with the staged file and the task; only a failure to open the writer itself reports the partition that could not be initialized. Verified with the load tests of iotdb-core/datanode: 95 cases over 12 classes, including LoadTsFileManagerTest#testPieceOfAStagedFileWithoutResumableWriterFails, which failed before this change. * LOAD TsFile: cover the staged deletions of a LOAD piece A piece may carry deletions next to its chunks, and they are staged as the exclusive modification file of the staged TsFile of the task. Nothing covered that path: the manager tests built pieces out of chunks only, so writeDeletion, the sidecar it creates, its lifecycle and its recovery were untested. LoadTsFileManagerDeletionTest pins the six properties of the path: - a deletion piece creates the .mods2 sidecar next to the staged TsFile and the deletion reads back with its path and time range; - such a piece announces no staged TsFile bytes, because the modification file belongs to the staged TsFile and a replica must not be asked for it; - a deletion of a task that staged no TsFile is a no-op that creates no modification file, which is what the dispatch order (chunks first) relies on; - a restart keeps the deletions staged before it and the recovered task appends its next deletion to the file it finds on disk instead of replacing it; - PREPARE seals the staged file with its deletions, and the sealed file stays readable; - ABORT takes the deletions away with the staged files and the task directory. The restart cases are verified to be discriminating: with the modification file no longer re-attached by recoverFromDisk, the resumed task loses its deletions again, and with the sidecar announced as staged bytes the piece refs are wrong. * LOAD TsFile: do not invent a failure for a load stopped without one The coordinator stops the execution of a query as soon as its state is done, and it passes no failure when the load reached FINISHED: every successful LOAD calls stop(null). The scheduler recorded an explicit stop in that case, and QueryExecution answers a done query with stateMachine.getFailureStatus(), which synthesizes a status from the failure that was recorded. A LOAD whose files were all committed therefore answered the client with "1100: Load task explicitly stopped", which is what IoTDBLoadTsFileIT#testLoadWithTableMod and #testLoadWithTimeColumn hit. stop now reports only the failure it is given. That matches ClusterScheduler, whose stop never touches the state machine, and keeps the failure path: a stop that carries a cause still fails the task. Verified with LoadTsFileSchedulerTest, which reproduces the bogus failure on a FINISHED load before the change, and with the two integration tests above, which pass against a DataNode built from this commit. * LOAD TsFile: address the review comments of the consensus LOAD path Consensus commands and their serialization: - keep PREPARE as the name of the sealing phase; with BEGIN gone it no longer collides with anything - drop BEGIN (the first PIECE opens the staging directory of its region) and PULL, which no node ever sent or handled - write and read back only the fields of a command's own phase, instead of all fourteen for every command, and drop deleteAfterLoad, which no receiver read - measure serializedSize() with a byte counter and size the buffer of a PIECE from its payload instead of growing it - answer the command with a switch expression, so a new phase cannot be added without being handled - report a clear error for a malformed body instead of a bare IOException Staged data and its release: - state the real release rule: the staged bytes go away only once the replicated watermark has passed COMMIT or ABORT, and a directory whose terminal command never arrives is kept so a retried LOAD can resume it - restore the OBJECT_FILE_NODE branch of WALNode that the LOAD branch had replaced, which had dropped the read-back of an object file - clear the LOAD manager of a region inside the monitor that publishes it Routing and retries: - look the route of a region up again only for a failure that may mean it is stale (dispatch error, region group gone), and follow the value putIfAbsent returned - keep the per-region state in plain maps: one coordinator thread owns it - read back the chunk payloads only when a piece is actually sent, and say which path feeds a lagging IoT consensus V1 follower - lower the per-piece dispatch log to debug and drop a log that told nothing Splitter and progress file: - hoist the per-page type-service lookup out of the per-point loop - build the staged chunk header and statistics as views of one buffer instead of copying them twice, cache the header size, and fix the imports - replace the dead writePointToChunkWriter helper - explain in place why the terminal record is read forward and why a finished task is recorded both in its progress files and in terminal.marker Tests: - start one cluster for the whole class and bring back a DataNode a case stops - add a concurrent LOAD case and a case that asserts a meaningful failure when two of three DataNodes are down - cover the per-phase serialization of a command * LOAD TsFile: sync the staged files once per piece instead of per chunk Every chunk of a piece forced its bytes onto disk, and the piece then forced its staged files again, so one piece paid one fsync per chunk plus one per staged file. The piece is the unit the WAL entry and the consensus command work in, so each staged file is now synced once per piece. The progress records of the piece follow that sync: they are queued while the piece is staged and appended once its bytes are on disk. A record tells the recovery path which ranges are staged, so a record that reached the file before its bytes would make a restart trust a hole; a record that is lost only makes the chunk be written again. Seeding the in-memory state from the file is now done once, and it keeps the records that are still queued. Reading the file again on every query used to rebuild that state from the file alone, which dropped the queued records and made a re-applied piece rewrite a chunk it had already staged. * LOAD TsFile: reclaim the staged data of a load that never finishes A LOAD whose coordinator dies before COMMIT or ABORT left its staged directory behind forever: the cleaner only reclaimed directories that carry a terminal record. An unfinished task is now reclaimed once its region received no command of it for load_clean_up_task_idle_timeout_seconds (15 min by default, hot reloadable) and once the watermark has passed the chunk payloads it recorded, so no replica loses payloads it can still be fed from. The idle time lives on the task itself, in memory, and is never read from the timestamps of the file system: every command stamps its task, and a task recovered after a restart counts as heard from at startup. A LOAD command also sweeps the tasks of its region, so an idle one is reclaimed without waiting for the periodic sweep, while a directory the sweep does not find among them is dated from the first sweep that saw it, which is what reclaims the leftovers of an earlier run. A task whose coordinator is only stalled fails its PREPARE or COMMIT and the client retries the load. The region drops its writers of that task before the files are deleted, because they are open on them, and the sweep runs at the smaller of the two intervals so the idle timeout is not delayed. * LOAD TsFile: size the buffer of a piece from the size its request knows Expanding a piece allocated its buffer with a capacity guessed from the payload size, so the bytes were copied into it again and again while the buffer grew. The replication request already holds both numbers that add up to the piece: the exact length of the reference form it retained, and the payload size the node inlines when the piece is sent. It passes that total down, so the buffer of a piece is allocated once, and the reference-only fallback passes the exact length of that form. A caller with no better number keeps the small default start, and the size is clamped so that a piece close to 2 GiB of payloads cannot overflow the int cast. * LOAD TsFile: calculate the size of a command instead of serializing it serializedSize() writes the whole body into a byte counter only to read the number back, which allocates a stream and, for every string it passes, the bytes of that string. The WAL asks for that size for every entry it buffers, so the size is now added up field by field: the phase code, the strings, the fixed-width fields of the phase, the progress indexes and, for a PIECE, the references and the records it carries. A record calculates its own size beside its serializer, and both forms of it (references and inlined payloads) are covered. Tests keep the calculation and the serializer in step by comparing the number with the bytes that are actually written: for every record kind, for every phase, for a real staged piece, and for the counted length of a string against the encoder. * LOAD TsFile: cover the sizes a piece is logged in, without a shared helper The counted size of a string is now compared with the bytes its encoder writes, for ASCII, for every length of an encoded character and for the characters an encoder has to replace, and a string test caught an ASCII character behind a wider one being counted as two bytes. A record is sized in the state the WAL asks about, with its pages already ended, which the calculated size assumes and now documents; the oracle asks for the size before it writes the bytes, as the WAL does, and the aligned records are covered with their chunks, their references and their layout. * LOAD TsFile: build the chunk layout once, and size the group header from its device - allocate() returns the ChunkData.ChunkLayout the caller persists, so assign() no longer projects six of its fields into a second record; the scheduler-side ChunkLayout was used only here and is gone - size the staged chunk group header from the device as it is: the staged writer serializes the device of the ChunkData it is handed, so normalizing it here could reserve one byte more than the writer emits * LOAD TsFile: drop only the cached route of the region being re-resolved Re-resolving a route needs a cache miss: getRegionReplicaSet answers with the cached route when it holds one, so the entry has to go first. invalidAllCache() also cleared the database, data partition and schema partition caches of the whole DataNode, on every retry attempt. - PartitionCache.invalidReplicaSetCache(regionId) removes that one region - ClusterPartitionFetcher exposes it, so the load path stops clearing the rest * LOAD TsFile: reuse the partitions the judgement resolved when routing the load The judgement of a TsFile looks the data partitions of its devices up to tell whether the file has to be decoded, and the routing of its chunks then asked the fetcher for the same (device, time partition) pairs again. Nothing of what the judgement resolved was kept, so the second lookup was cheap only because the partition fetcher caches, which is a property of the fetcher and not of the load. - DataPartitionResolver resolves the partitions of a file and keeps what it resolved: the scheduler resolves through it to judge the file, and the routing of its chunks reuses those results - only the slots the judgement did not resolve are fetched, in one batch - a judgement that answers from its own cache, for instance the same file loaded again, resolves nothing and the routing fetches as it did before - the strategy and the split consumer take the resolver instead of the fetcher and the user name, which it holds DataPartitionRouterTest: the fetcher is asked once when the routing asks for the slots the judgement resolved, only the slots the judgement missed are fetched otherwise, and without a judgement the routing asks for them as it always did. * LOAD memory: account the per-file budget in a sub-block of the shared 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 * LOAD TsFile: tear the LOAD manager down with the region it belongs to 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. * LOAD TsFile: keep the one manager of a region instead of clearing it 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 * LOAD TsFile: drop the always-false needDecodeChunk byte from the non-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. * LOAD TsFile: keep the terminal record in the progress files instead of 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. * LOAD TsFile: bound every progress entry and force the records of a piece 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. * LOAD TsFile: open the staged directories less often when sweeping 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. * LOAD TsFile: size the list of active task directories up front The list holds exactly the task directories of the writers the manager tracks, so it can be created with that size instead of growing. * LOAD TsFile: filter the progress files while listing a task directory 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. * LOAD TsFile: link a retained staged file instead of copying 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. * LOAD TsFile: say why the torn-tail scan marks the stream 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. * LOAD TsFile: keep the chunk bodies as views instead of copies 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. * LOAD TsFile: size the serialized bodies up front and stop fsyncing a 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. * LOAD TsFile: stop flushing a piece, and size the bodies to what they 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. * LOAD TsFile: force the staged files once a piece is complete, and not 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. * LOAD TsFile: mark a task ended only on staged files the command can describe 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. * LOAD TsFile: require an aborted staged file to reach the end of its records 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. * LOAD TsFile: tell every replica how far the coordinator staged each file 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. * LOAD TsFile: adapt the staged files to IoTConsensusV2 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. * LOAD TsFile: tighten the comments and the expected end lookup 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. * LOAD TsFile: take the chunk header size from the header itself 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. * LOAD TsFile: import the collector the load utility uses The stream of a transfer was collected through its fully qualified name, the last of its kind in the load package. * LOAD TsFile: compare the consensus protocol by its canonical name 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. * LOAD TsFile: link the files a snapshot publishes when possible 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. * LOAD TsFile: refuse a chunk a non-seekable output cannot place 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. * LOAD TsFile: verify the series an out-of-order test imports, and reformat 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. * LOAD TsFile: key the staged files by their time partition and keep their 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. * LOAD TsFile: verify every point an out-of-order test imports, and reformat 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.
…18866) * [Subscription] Replay uncommitted delivery payloads from WAL * [Subscription] Avoid WAL validation on the delivery path * [Subscription] Drop eager WAL replayability checks
Refresh candidates before fallback selection and resume after the current endpoint. Preserve candidates and cursor on empty refreshes. Add routing and connection tests and re-enable node status failover IT.
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
20 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
Content1 ...
Content2 ...
Content3 ...
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