Repository navigation
Conversation
d207c17 to
29a3153
Compare
|
Self-review passed. Fixed the compaction sink setting so sink.enable-compaction=false (and the compatibility default for restored Pipes) now prevents queue replacement even when real-time-first scheduling is enabled. The queue still prioritizes real-time events independently, while only explicitly enabled user Pipes perform compaction replacement; the no-arg queue keeps internal Pipes disabled by default. Added a regression test for the disabled-by-default case. Verification: mvn spotless:apply -pl iotdb-core/datanode passed; focused DataNode reactor tests passed (14 tests, 0 failures/errors): PipeRealtimePriorityBlockingQueueTest and PipeDataNodeTaskBuilderTest. |
|
|
||
| @RunWith(IoTDBTestRunner.class) | ||
| @Category({MultiClusterIT1.class}) | ||
| public class IoTDBPipeCompactionSinkParameterIT extends AbstractPipeSingleIT { |
There was a problem hiding this comment.
Is it possible to put this into an existing class?
There was a problem hiding this comment.
Moved the compaction parameter assertions into the existing IoTDBPipeSingleEnvDemoIT.testSingleEnv test and removed the dedicated test class. This reuses the existing cluster startup/teardown. Verified with the MultiClusterIT1 profile: 1 test, 0 failures/errors/skips. Included in eae8427.
| // Replace if all source files are present in the queue | ||
| .filter(entry -> entry.getValue().size() == sourceFiles.size()) | ||
| // Replace only when every source file is present exactly once for this pipe. | ||
| .filter(entry -> entry.getValue().size() == sourceFilePaths.size()) | ||
| .filter( | ||
| entry -> | ||
| entry.getValue().stream() | ||
| .map(PipeTsFileInsertionEvent::getSourceTsFile) | ||
| .map(PipeRealtimePriorityBlockingQueue::normalizePath) | ||
| .collect(Collectors.toSet()) | ||
| .equals(sourceFilePaths)) |
There was a problem hiding this comment.
Is it possible to ease the condition to allow more files to be replaced when possible?
One suggestion is to allow a suffix with a total size over twice the sum of the targets to be replaced.
For example, the compaction involves f0, f1, f2, f3, f4 and results in f5 and f6.
However, f0 is already sent by Pipe and only f1-f4 remain in the queue.
If sum_size(f1, f2, f3, f4) > 2 * sum_size(f5, f6), then we still admit the replacement since it is expected to reduce the IO amount significantly.
There was a problem hiding this comment.
I am keeping the complete-source requirement for this fix. A partial replacement also replays data from the files that have already been sent, so the size ratio alone does not establish safe overwrite/replay and progress behavior. The proposed IO optimization needs dedicated tests for those cases before relaxing the condition. This update fixes source matching and preserves the replacement's queue position.
| // The deque is concurrent, so a source event may be polled while replacement is in progress. | ||
| // Only release events that are still owned by this queue. If a poll wins the race, both the | ||
| // source event and its compacted replacement may be delivered; this deliberately favors the | ||
| // hot-path throughput over strict de-duplication. | ||
| final Set<PipeTsFileInsertionEvent> removedEvents = | ||
| eventsToRemove.stream().filter(tsfileInsertEventDeque::remove).collect(Collectors.toSet()); | ||
| for (final PipeTsFileInsertionEvent event : successfullyReferenceIncreasedEvents) { | ||
| tsfileInsertEventDeque.add(event); | ||
| eventCounter.increaseEventCount(event); | ||
| } |
There was a problem hiding this comment.
Will this break the order of the files?
For example, we have f0, f1, f2, f3, f4, f5, and f0, f1, f2 are replaced with f6, then the queue becomes f4, f5, f6.
If some data points of f0 are overwritten in f4, the new transfer order may cause some trouble.
There was a problem hiding this comment.
Yes, appending the replacement changes its position relative to the remaining files. Fixed in eae8427: replacement now requires the complete oldest queued source prefix for the same committer, and inserts the target files at the front in their original order. For example, [f0,f1,f2,f3] with f0,f1 -> g becomes [g,f2,f3]. If an unrelated older file is interleaved for that committer, the original events stay queued. Added regressions for multiple target files, a non-prefix source group, and independent committers; all 8 queue tests pass. The existing two-ended real-time scheduling and the documented poll/replacement race tradeoff remain.
|
Comment-handling review of 9bc711d completed. Addressed the replacement-position issue by requiring an oldest source prefix per committer and inserting ordered target files at the front. Added regression coverage and merged the compaction parameter assertions into the existing single-environment IT. The partial-source size optimization was not applied pending overwrite/replay and progress validation. Verification: Spotless passed; clean reactor passed 8 queue unit tests (0 failures/errors/skips); a separate clean MultiClusterIT1 verify run passed the existing IT with the merged parameter assertions (1 test, 0 failures/errors/skips). Build telemetry extensions were disabled only in the verification copy after an extension-resolution failure. Remote CI was not awaited. Final local and confirmed source-branch/PR head: eae8427. All three review comments have replies and discussions remain unresolved. Independent re-review is pending for the behavioral update. |
Description
Queued TsFile events were selected for compaction replacement by event count, which could remove unrelated files. This change matches normalized original source paths, including after an event acquires a Pipe hard link.
Replacement requires every compaction source exactly once for the same committer and an oldest queued source prefix. Target files keep their order and replace that prefix at the front. For example,
[f0,f1,f2,f3]withf0,f1 -> gbecomes[g,f2,f3]; the previous implementation appended it as[f2,f3,g]. Groups with an unrelated older event for the same committer stay queued. Partial-source replacement based on file size is deferred until overwrite/replay and progress behavior can be verified.New user Pipes persist
sink.enable-compaction=trueunless an explicit value or theconnector.enable-compactionalias is supplied. Restored metadata without the setting retains the compatibility defaultfalse. Internal subscription and consensus Pipes are not enabled automatically.Only replacement operations are serialized. Normal offer, poll, peek and discard operations retain the concurrent deque. All replacement references are acquired before queue mutation, and only removed source events release queue ownership. A poll/replacement race may deliver both the source and its replacement; the existing duplicate-delivery tradeoff remains.
Validation
PipeRealtimePriorityBlockingQueueTesttests (0 failures/errors/skips); integration-test sources compiled.verifyrun withMultiClusterIT1passedIoTDBPipeSingleEnvDemoIT: 1 test, 0 failures/errors/skips. The compaction parameter assertions now reuse that existing test's cluster.