snleee commented on a change in pull request #5828:
URL: https://github.com/apache/incubator-pinot/pull/5828#discussion_r468200530



##########
File path: 
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
##########
@@ -177,4 +193,63 @@ private boolean shouldDeleteInProgressLLCSegment(String 
segmentName, IdealState
       return states.size() == 1 && 
states.contains(CommonConstants.Helix.StateModel.SegmentStateModel.OFFLINE);
     }
   }
+
+  private void manageSegmentLineageCleanupForTable(String tableNameWithType) {
+    try {
+      DEFAULT_RETRY_POLICY.attempt(() -> {
+        // Fetch segment lineage
+        ZNRecord segmentLineageZNRecord = SegmentLineageAccessHelper
+            
.getSegmentLineageZNRecord(_pinotHelixResourceManager.getPropertyStore(), 
tableNameWithType);
+        if (segmentLineageZNRecord == null) {
+          return true;
+        }
+        SegmentLineage segmentLineage = 
SegmentLineage.fromZNRecord(segmentLineageZNRecord);
+        int expectedVersion = segmentLineageZNRecord.getVersion();
+
+        // 1. The original segments can be deleted once the merged segments 
are successfully uploaded
+        // 2. The zombie lineage entry & merged segments should be deleted if 
the segment replacement failed in
+        //    the middle
+        Set<String> segmentsForTable = new 
HashSet<>(_pinotHelixResourceManager.getSegmentsFor(tableNameWithType));
+        List<String> segmentsToDelete = new ArrayList<>();
+        for (String lineageEntryId : segmentLineage.getLineageEntryIds()) {
+          LineageEntry lineageEntry = 
segmentLineage.getLineageEntry(lineageEntryId);
+          if (lineageEntry.getState() == LineageEntryState.COMPLETED) {
+            // If the lineage state is 'COMPLETED', it is safe to delete all 
segments from 'segmentsFrom'
+            segmentsToDelete.addAll(lineageEntry.getSegmentsFrom());
+
+            // The lineage entry with 'COMPLETED' state can only be safely 
removed when both segmentFrom & segmentTo
+            // are all removed from the table.
+            if (Collections.disjoint(segmentsForTable, 
lineageEntry.getSegmentsFrom()) && Collections
+                .disjoint(segmentsForTable, lineageEntry.getSegmentsTo())) {
+              segmentLineage.deleteLineageEntry(lineageEntryId);
+            }

Review comment:
       Changed. Please double check the logic for `IN_PROGRESS` as well.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org

Reply via email to