Skip to content

[Pipe] Fix compaction queue event replacement - #18726

Open
Caideyipi wants to merge 8 commits into
apache:masterfrom
Caideyipi:fix/pipe-compaction-queue-replacement
Open

Caideyipi wants to merge 8 commits into
apache:masterfrom
Caideyipi:fix/pipe-compaction-queue-replacement

Conversation

@Caideyipi

@Caideyipi Caideyipi commented Sep 28, 2026 •

Copy link
Copy Markdown
Collaborator

Description

Match the actual compaction source files

  • Match queued PipeTsFileInsertionEvents by normalized original TsFile paths instead of only comparing event counts.
  • Keep using the original TsFileResource path after an event creates a hard link in its Pipe directory.
  • Replace a group only when every source file is present exactly once for the same committer key, leaving unrelated queued events untouched.

Roll out through a Pipe sink parameter

  • Use sink.enable-compaction (and its connector.enable-compaction alias) as a Pipe-scoped setting rather than a global configuration.
  • Persist sink.enable-compaction=true when a new user Pipe is created without an explicit value.
  • Preserve explicit values and keep the compatibility default false when restored existing Pipe metadata does not contain the setting.
  • Do not automatically enable the setting for internal subscription or consensus Pipes.

Concurrency and performance

  • Serialize concurrent replacement operations only; normal offer, poll, peek, and discard paths continue to use the concurrent deque without acquiring the replacement monitor.
  • Acquire references for all replacement events before changing the queue.
  • If polling wins a race with replacement, allow the source event and its compacted replacement to both be delivered instead of blocking the sending hot path.
  • Release references and decrement queue statistics only for source events that were actually removed, preventing premature release of an event already taken by a consumer.

This keeps the synchronization cost off the normal Pipe sending path. The accepted tradeoff is a small duplicate-delivery window during a compaction/poll race, while avoiding data loss and reference-count corruption.

Tests

  • Focused unit tests cover source-path matching, unrelated events, hard-link paths, reference-count rollback, concurrent polling, restored-Pipe compatibility, new-Pipe defaults, and explicit-value precedence.
  • The focused ConfigNode/DataNode unit-test reactor completed with BUILD SUCCESS.
  • The dedicated IoTDBPipeCompactionSinkParameterIT ran independently: 1 test, 0 failures, 0 errors, 0 skipped.
  • Spotless and Checkstyle checks passed in the exercised reactor modules.

This PR has:

  • been self-reviewed.
    • concurrent read and write
    • backward compatibility for existing Pipes
    • hot-path synchronization impact
  • added Javadocs for the source-file accessor.
  • added comments explaining the replacement concurrency tradeoff.
  • added unit and integration tests for the corrected code paths and rollout behavior.

Key changed/added classes in this PR
  • CreatePipeProcedureV2
  • PipeDataNodeTaskBuilder
  • PipeRealtimePriorityBlockingQueue
  • PipeTsFileInsertionEvent
  • CreatePipeProcedureV2Test
  • PipeDataNodeTaskBuilderTest
  • PipeRealtimePriorityBlockingQueueTest
  • IoTDBPipeCompactionSinkParameterIT

@Caideyipi
Caideyipi force-pushed the fix/pipe-compaction-queue-replacement branch from d207c17 to 29a3153 Compare October 23, 2026 09:20
@Caideyipi

Caideyipi commented Oct 23, 2026 •

Copy link
Copy Markdown
Collaborator Author

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant