Add support for orphan segment cleanup #15142

9aman · 2025-02-27T07:56:34Z

Context

Followup of this PR: #15048
The first PR ensured that we don't miss any segments that are created from now on.

This PR aims to fix the orphan segments that are present in deepstore and have passed the retention time but are neither present in ZK or IdealState.

Scope of the PR.

The PR aims to handle two scenarios:

Old segments that did not get deleted from the deepstore but their SegmentZKMetadata was deleted by the retention manager. This happened due to the issue mentioned and resolved in: Ensure SegmentDeletionManager deletes deepstore files created by BaseMultipleSegmentsConversionExecutor #15048
The deletion by retention manager is performed as follows:
- Find all the segments present in the ZK ( i.e. have ZK Metadata entry)
- Delete Segment from IS
- Delete Segment from ZK
- Delete Segment from deepstore
  The RententionManager will not be able to delete a segment in the subsequent runs if the controller restarted etc. between the last two steps i.e. any failure between the last two steps of the process.

Testing

Unit Tests were added to ensure that segments present in deepstore that are missing in ZK are picked for deletion.

Update

After analyzing the test results:

https://docs.google.com/document/d/1ZvNefSsRL716NspQc1f5VrRYjd9CSqocac5TIfLJWPA/edit?usp=sharing

I concluded that limiting the number of segments deleted in a single run would be beneficial. To address this, I have introduced a configurable batch size for segment deletion. (untrackedSegmentsDeletionBatchSize part of SegmentsValidationAndRetentionConfig)

Reasons for This Change

Deletion

Prevent starvation of deepstore cleanup: Ensures that the RetentionManager for other tables can also delete files without delays.
Avoid blocking API-based segment deletions:
- Some APIs rely on the same code for deleting segments from deepstore.
- The API will respond but the deletion will stay blocked
Segment Lineage update function at the end does deepstore cleanup. It will not stay blocked as we just submit the request in a single threaded executor service but deepstore cleanup will stay blocked
UpsertCompactionTaskGenerator and UpsertCompactMergeTaskGenerator also use the same code run using a single threaded executor.

  protected void deleteSegmentsWithDelay(String tableName, Collection<String> segmentIds,
      Long deletedSegmentsRetentionMs, long deletionDelaySeconds) {
    _executorService.schedule(new Runnable() {
      @Override
      public void run() {
        deleteSegmentFromPropertyStoreAndLocal(tableName, segmentIds, deletedSegmentsRetentionMs, deletionDelaySeconds);
      }
    }, deletionDelaySeconds, TimeUnit.SECONDS);
  }


protected synchronized void deleteSegmentFromPropertyStoreAndLocal

codecov-commenter · 2025-02-27T08:39:22Z

Codecov Report

Attention: Patch coverage is 87.50000% with 9 lines in your changes missing coverage. Please review.

Project coverage is 63.68%. Comparing base (59551e4) to head (6894d6b).
Report is 1837 commits behind head on master.

Files with missing lines	Patch %	Lines
...troller/helix/core/retention/RetentionManager.java	88.33%	4 Missing and 3 partials ⚠️
.../controller/helix/core/SegmentDeletionManager.java	0.00%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #15142      +/-   ##
============================================
+ Coverage     61.75%   63.68%   +1.92%     
- Complexity      207     1461    +1254     
============================================
  Files          2436     2772     +336     
  Lines        133233   156306   +23073     
  Branches      20636    23988    +3352     
============================================
+ Hits          82274    99536   +17262     
- Misses        44911    49283    +4372     
- Partials       6048     7487    +1439

Flag	Coverage Δ
custom-integration1	`100.00% <ø> (+99.99%)`	⬆️
integration	`100.00% <ø> (+99.99%)`	⬆️
integration1	`100.00% <ø> (+99.99%)`	⬆️
integration2	`0.00% <ø> (ø)`
java-11	`63.65% <87.50%> (+1.94%)`	⬆️
java-21	`63.57% <87.50%> (+1.94%)`	⬆️
skip-bytebuffers-false	`63.67% <87.50%> (+1.92%)`	⬆️
skip-bytebuffers-true	`63.50% <87.50%> (+35.77%)`	⬆️
temurin	`63.68% <87.50%> (+1.92%)`	⬆️
unittests	`63.67% <87.50%> (+1.92%)`	⬆️
unittests1	`56.16% <100.00%> (+9.27%)`	⬆️
unittests2	`34.25% <87.50%> (+6.52%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

9aman · 2025-02-27T09:32:09Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

@@ -124,8 +138,19 @@ private void manageRetentionForTable(TableConfig tableConfig) {
  }

  private void manageRetentionForOfflineTable(String offlineTableName, RetentionStrategy retentionStrategy) {
+    List<SegmentZKMetadata> segmentZKMetadataList = _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName);


Segment metadata is added to ZK before it's added to IdealState
Segment metadata is removed from ZK after the segment has been removed from the ideal state.

The segment list from ZK will always be a super set of segments in Ideal State. Thus, I have not checked IS for filtering segments that are still in IS.

Can we not just rely on the ideal state as source of truth because that would avoid reading all of segment Zk ? We are going to check retention window as well prior to deletion (if the concern is deleting a new segment that just got added, which can happen even when the segment is not in Zk metadata but just in deep store).

The deletion occurs based on the endTime present in the segmentZKMetadata. That's why the isPurgeable relies on segmentZKMetadata to decide whether we need to remove a segment or not.
Are you suggesting some other approach where we move away from endTime of the segment?

Would it happen that segment's ZKMetadata was still there in ZK but segment was not in IS any more?

I initially did not realize that we already loop through all of segmentZk metadata to check if segment is purge-able (retentionStrategy.isPurgeable).
Yes its possible for a segment to be Zk metadata but not in IS @klsince, but we want to handle that case as well right?

If the deletion fails in the middle then the segment will be in ZK but not in IS.
Moreover, as per Xiang we have seen cases where the SegmentRefreshTask etc push segment in Deepstore and ZK but fail to update the IS.
In that case as well we have such segments.

9aman · 2025-02-27T09:33:00Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+
+      long lastModifiedTime = fileMetadata.getLastModifiedTime();
+
+      if (retentionStrategy.isPurgeable(segmentName, offlineTableName, lastModifiedTime)) {


This ensures that a segment that has not entry in the ZK but has been uploaded to deepstore does not get deleted.

9aman · 2025-02-27T09:34:16Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+      }
+
+      String segmentName = extractSegmentName(fileMetadata.getFilePath());
+      if (segmentName == null || segmentsToExclude.contains(segmentName)) {


Rely on endTime in ZK Metadata for retention in case the segment ZK metadata is present.
This function relies on file modification time to decide whether a segment should be purged or not.

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

swaminathanmanish · 2025-02-27T16:27:59Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

@@ -124,8 +138,19 @@ private void manageRetentionForTable(TableConfig tableConfig) {
  }

  private void manageRetentionForOfflineTable(String offlineTableName, RetentionStrategy retentionStrategy) {
+    List<SegmentZKMetadata> segmentZKMetadataList = _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName);


Can we not just rely on the ideal state as source of truth because that would avoid reading all of segment Zk ? We are going to check retention window as well prior to deletion (if the concern is deleting a new segment that just got added, which can happen even when the segment is not in Zk metadata but just in deep store).

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

swaminathanmanish · 2025-02-27T16:33:28Z

...c/main/java/org/apache/pinot/controller/helix/core/retention/strategy/RetentionStrategy.java

+   * @param endTimeMs The end time of the segment in milliseconds
+   * @return Whether the segment should be purged
+   */
+  boolean isPurgeable(String segmentName, String tableNameWithType, long endTimeMs);


Suggest having this method local to SegmentDeletionManager since endTime is supplied from outside. It'll be confusing as to what needs to be passed in (file modification time/segment time etc..). The caller can get the retention time and do its own checks ?

Yes, but I would still keep it as part of the RetentionManager. I feel that the SegmentDeletionManager should just have the job of deleting (moving to Deleted_Segments dir) and not know what segments needs to be deleted.

For our case, the RetentionManager can fetch the segments from deepstore and decide based on the file modification time whether they need to be deleted or not.

On a second pass I think it's fine to have this.
One can pass the endTimeMs that needs to be compared to System.currentTimeMillis() - retentionMS i.e.
System.currentTimeMillis() - retentionMS > endTimeMs.

It's a more generic implementation of the

public boolean isPurgeable(String tableNameWithType, SegmentZKMetadata segmentZKMetadata)

Where one only needs to rely on endTime present in segmentZKMetadata

Have changed as per @klsince suggestions.

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

klsince · 2025-03-05T22:47:28Z

...-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java

@@ -221,6 +221,7 @@ private void deleteSegmentMetadataFromStore(PinotFS pinotFS, URI segmentFileUri,
      URI segmentMetadataUri = SegmentPushUtils.generateSegmentMetadataURI(segmentFileUri.toString(), segmentId);
      if (pinotFS.exists(segmentMetadataUri)) {
        LOGGER.info("Deleting segment metadata {} from {}", segmentId, segmentMetadataUri);
+        // TODO: check if the deletion was successful and add a warning here.


is this TODO supposed to check the return of the delete() method and log warning if false? if so, maybe we can just add it in this PR.

klsince · 2025-03-05T22:47:46Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

@@ -77,6 +88,7 @@ protected void processTable(String tableNameWithType) {
      return;
    }

+    // Did not understand


Done. Thanks for pointing it out.

klsince · 2025-03-05T22:54:52Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

@@ -124,8 +138,19 @@ private void manageRetentionForTable(TableConfig tableConfig) {
  }

  private void manageRetentionForOfflineTable(String offlineTableName, RetentionStrategy retentionStrategy) {
+    List<SegmentZKMetadata> segmentZKMetadataList = _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName);


Would it happen that segment's ZKMetadata was still there in ZK but segment was not in IS any more?

klsince · 2025-03-05T22:56:42Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+        segmentsToDelete.add(segmentName);
+      }
+    }
+


add an INFO log about how many segments to delete

klsince · 2025-03-05T22:58:24Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+    PinotFS pinotFS = PinotFSFactory.create(tableDataUri.getScheme());
+
+    List<FileMetadata> deepstoreFiles = pinotFS.listFilesWithMetadata(tableDataUri, false);
+


maybe print a INFO log here about how many segments found, and the timestamp in the log, with another INFO log after the for-loop, can be used to calculate how long it'd take to figure out the files to delete.

klsince · 2025-03-05T23:10:26Z

...in/java/org/apache/pinot/controller/helix/core/retention/strategy/TimeRetentionStrategy.java

@@ -50,4 +50,17 @@ public boolean isPurgeable(String tableNameWithType, SegmentZKMetadata segmentZK

    return System.currentTimeMillis() - endTimeMs > _retentionMs;
  }
+
+  @Override
+  public boolean isPurgeable(String segmentName, String tableNameWithType, long endTimeMs) {


better move tableNameWithType as the first param for consistency

can reuse this method to implement the method above

perhaps call endTimeMs segmentTimeMs to be generic? and leave a comment that segmentTimeMs can be endTime or mtime etc. to be decided by the caller

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

swaminathanmanish · 2025-03-06T16:18:27Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+    URI tableDataUri = URIUtils.getUri(_pinotHelixResourceManager.getDataDir(), rawTableName);
+    PinotFS pinotFS = PinotFSFactory.create(tableDataUri.getScheme());
+
+    List<FileMetadata> deepstoreFiles = pinotFS.listFilesWithMetadata(tableDataUri, false);


Can we add a metric to get count of files that are in deep store but not in segmentZkMetadata ? This is to see how many dangling files are in the deep store

swaminathanmanish · 2025-03-06T16:20:54Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

@@ -124,8 +138,19 @@ private void manageRetentionForTable(TableConfig tableConfig) {
  }

  private void manageRetentionForOfflineTable(String offlineTableName, RetentionStrategy retentionStrategy) {
+    List<SegmentZKMetadata> segmentZKMetadataList = _pinotHelixResourceManager.getSegmentsZKMetadata(offlineTableName);


I initially did not realize that we already loop through all of segmentZk metadata to check if segment is purge-able (retentionStrategy.isPurgeable).
Yes its possible for a segment to be Zk metadata but not in IS @klsince, but we want to handle that case as well right?

klsince · 2025-03-07T18:21:10Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+
+    List<FileMetadata> deepstoreFiles = pinotFS.listFilesWithMetadata(tableDataUri, false);
+    long listEndTimeMs = System.currentTimeMillis();
+    LOGGER.info("Found: {} segments in deepstore for table: {}. Time taken to list segments: {} ms",


nit: Found ... in {} ms to be short

klsince · 2025-03-07T18:23:16Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+      }
+    }
+    long endTimeMs = System.currentTimeMillis();
+    LOGGER.info(


perhaps combine the two INFO together, as "Found {} segments ... have no corresponding ZK metadata, in {} msg"

klsince · 2025-03-07T18:25:39Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+    } catch (IOException e) {
+      LOGGER.warn("Unable to fetch segments from deep store that are beyond retention period for table: {}",
+          realtimeTableName);
+    }


make a helper method from L241-L258 and use it for both manageRetentionForOfflineTable and manageRetentionForREaltimeTable?

… to be deleted in an single run of RetentionManager

9aman · 2025-03-12T11:53:27Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+      _controllerMetrics.setValueOfTableGauge(tableNameWithType, ControllerGauge.UNTRACKED_SEGMENTS_COUNT,
+          segmentsToDelete.size());
+
+      if (segmentsToDelete.size() > untrackedSegmentsDeletionBatchSize) {


Added the logic to pick as per the configured batch size.

cc: @swaminathanmanish @klsince

The ideal way would be to do pagination on the FS and bring only untrackedSegmentsDeletionBatchSize number of files/ segments from deepstore.
We can add this improvement later.

9aman · 2025-03-12T11:53:52Z

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java

+      List<SegmentZKMetadata> segmentZKMetadataList, int untrackedSegmentsDeletionBatchSize) {
+    List<String> segmentsToDelete = new ArrayList<>();
+
+    if (untrackedSegmentsDeletionBatchSize <= 0) {


Logging and returning early in case it's set to zero to avoid listing all the files in the deepstore.

9aman force-pushed the deeptore_cleanup_for_historical_segments branch from a14c5bc to d2d1bcf Compare February 27, 2025 09:28

9aman commented Feb 27, 2025

View reviewed changes

9aman force-pushed the deeptore_cleanup_for_historical_segments branch from 2fd0b0e to 4060631 Compare February 27, 2025 11:17

swaminathanmanish reviewed Feb 27, 2025

View reviewed changes

...troller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java Show resolved Hide resolved

swaminathanmanish reviewed Feb 27, 2025

View reviewed changes

9aman marked this pull request as ready for review February 28, 2025 07:42

klsince reviewed Mar 5, 2025

View reviewed changes

swaminathanmanish reviewed Mar 6, 2025

View reviewed changes

9aman force-pushed the deeptore_cleanup_for_historical_segments branch from 5cdcfef to abac218 Compare March 7, 2025 09:53

klsince reviewed Mar 7, 2025

View reviewed changes

Jackie-Jiang added ingestion bugfix labels Mar 7, 2025

9aman force-pushed the deeptore_cleanup_for_historical_segments branch 2 times, most recently from a77116a to 168a771 Compare March 10, 2025 07:39

9aman added 8 commits March 12, 2025 14:10

Add support for orphan segment cleanup

8e73a34

Add unit tests to the code

840fdaf

Minor refactoring

1047802

Fixing checkstyle violations

22ff47e

Add support for realtime tables

aaacb8b

Fix test failures

7724d26

Adding metrics, improving logs for better debuggability

6c8eb3f

Reducing code repetition

e3b7d7e

9aman force-pushed the deeptore_cleanup_for_historical_segments branch from 168a771 to e3b7d7e Compare March 12, 2025 08:40

9aman added 2 commits March 12, 2025 16:44

Add support for providing match size for number of untracked segments…

833bc0d

… to be deleted in an single run of RetentionManager

Add test cases to test batch sizes

4db5030

9aman commented Mar 12, 2025

View reviewed changes

9aman added 2 commits March 12, 2025 17:31

Provide additional comments for test class

c44524f

Fixing linting issues

6894d6b


		long lastModifiedTime = fileMetadata.getLastModifiedTime();

		if (retentionStrategy.isPurgeable(segmentName, offlineTableName, lastModifiedTime)) {

		PinotFS pinotFS = PinotFSFactory.create(tableDataUri.getScheme());

		List<FileMetadata> deepstoreFiles = pinotFS.listFilesWithMetadata(tableDataUri, false);

Add support for orphan segment cleanup #15142

Are you sure you want to change the base?

Add support for orphan segment cleanup #15142

Conversation

9aman commented Feb 27, 2025 • edited Loading

Context

Scope of the PR.

Testing

Update

Reasons for This Change

codecov-commenter commented Feb 27, 2025 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

9aman Mar 2, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

9aman commented Feb 27, 2025 •

edited

Loading

codecov-commenter commented Feb 27, 2025 •

edited

Loading

9aman Mar 2, 2025 •

edited

Loading