[Caideyipi]

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]

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.