Skip to content

Optimize pipe TsFile cleanup on drop#18163

Open
luoluoyuyu wants to merge 8 commits into
apache:masterfrom
luoluoyuyu:optimize-pipe-tsfile-cleanup
Open

Optimize pipe TsFile cleanup on drop#18163
luoluoyuyu wants to merge 8 commits into
apache:masterfrom
luoluoyuyu:optimize-pipe-tsfile-cleanup

Conversation

@luoluoyuyu

Copy link
Copy Markdown
Member

Description

As the title said


This PR has:

  • been self-reviewed.
    • concurrent read
    • concurrent write
    • concurrent read and write
  • added documentation for new or modified features or behaviors.
  • added Javadocs for most classes and all non-trivial methods.
  • added or updated version, license, or notice information
  • added comments explaining the "why" and the intent of the code wherever would not be obvious
    for an unfamiliar reader.
  • added unit tests or modified existing tests to cover new code paths, ensuring the threshold
    for code coverage.
  • added integration tests.
  • been tested in a test IoTDB cluster.

Key changed/added classes (or packages if there are too many classes) in this PR

@luoluoyuyu luoluoyuyu force-pushed the optimize-pipe-tsfile-cleanup branch from 2c0d324 to c03d06c Compare July 9, 2026 04:23

@jt2594838 jt2594838 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the system restarts before removing all files and after dropping the pipe, will the files still be removed?

@luoluoyuyu

Copy link
Copy Markdown
Member Author

If the system restarts before removing all files and after dropping the pipe, will the files still be removed?

After startup finds stale pipe hardlink directories, it does not delete them synchronously in the startup path. Instead, it registers a periodic cleanup job:

periodicalJobRegistrar.register(
    PERIODICAL_CLEANUP_JOB_ID,
    new PeriodicalStalePipeDirCleaner(stalePipeDirs)::cleanOneRound,
    PipeConfig.getInstance().getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds());

The cleanup job deletes stale directories in throttled rounds. Each round stops when either condition is met:

private static final long DELETE_MAX_PATH_COUNT_PER_ROUND = 100_000L;
private static final long DELETE_MAX_TIME_PER_ROUND_MS = 1_000L;

So one cleanup round deletes at most 100,000 paths or runs for at most 1,000 ms.

This prevents DataNode startup from being blocked by a large number of leftover pipe TsFile hardlinks. If cleanup is not finished in one round, the next periodic round continues from the remaining files.

@jt2594838

Copy link
Copy Markdown
Contributor

If the system restarts before removing all files and after dropping the pipe, will the files still be removed?

After startup finds stale pipe hardlink directories, it does not delete them synchronously in the startup path. Instead, it registers a periodic cleanup job:

periodicalJobRegistrar.register(
    PERIODICAL_CLEANUP_JOB_ID,
    new PeriodicalStalePipeDirCleaner(stalePipeDirs)::cleanOneRound,
    PipeConfig.getInstance().getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds());

The cleanup job deletes stale directories in throttled rounds. Each round stops when either condition is met:

private static final long DELETE_MAX_PATH_COUNT_PER_ROUND = 100_000L;
private static final long DELETE_MAX_TIME_PER_ROUND_MS = 1_000L;

So one cleanup round deletes at most 100,000 paths or runs for at most 1,000 ms.

This prevents DataNode startup from being blocked by a large number of leftover pipe TsFile hardlinks. If cleanup is not finished in one round, the next periodic round continues from the remaining files.

Is it possible to reuse the logic for dropping pipe? And hence avoid introducing a new thread.

@luoluoyuyu

Copy link
Copy Markdown
Member Author

If the system restarts before removing all files and after dropping the pipe, will the files still be removed?

After startup finds stale pipe hardlink directories, it does not delete them synchronously in the startup path. Instead, it registers a periodic cleanup job:

periodicalJobRegistrar.register(
    PERIODICAL_CLEANUP_JOB_ID,
    new PeriodicalStalePipeDirCleaner(stalePipeDirs)::cleanOneRound,
    PipeConfig.getInstance().getPipeSubtaskExecutorCronHeartbeatEventIntervalSeconds());

The cleanup job deletes stale directories in throttled rounds. Each round stops when either condition is met:

private static final long DELETE_MAX_PATH_COUNT_PER_ROUND = 100_000L;
private static final long DELETE_MAX_TIME_PER_ROUND_MS = 1_000L;

So one cleanup round deletes at most 100,000 paths or runs for at most 1,000 ms.
This prevents DataNode startup from being blocked by a large number of leftover pipe TsFile hardlinks. If cleanup is not finished in one round, the next periodic round continues from the remaining files.

Is it possible to reuse the logic for dropping pipe? And hence avoid introducing a new thread.

Yes, I agree. Reusing the existing stale pipe directory cleanup logic is better than introducing a new dedicated thread.
I updated the drop-pipe cleanup path to follow the same mechanism: when a pipe is dropped, its pipe-specific TsFile directory is first renamed to a stale directory like <pipeName_createTime>.startup-cleaning--, then submitted to the existing periodical stale-dir cleaner. The cleaner deletes it in throttled rounds with the existing limits: at most 100,000 paths or 1,000 ms per round.
This avoids blocking the drop path with synchronous per-file deletion, avoids adding another cleanup thread, and keeps restart recovery consistent. If the DataNode restarts before deletion finishes, startup will find the stale directory and the same periodical cleanup job will continue removing it.

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.

2 participants