Compactor cleaner deletes all visit markers when partition group file is deleted

CHANGELOG.md

@@ -30,7 +30,7 @@
 * [ENHANCEMENT] Alertmanager/Ruler: Introduce a user scanner to reduce the number of list calls to object storage. #6999
 * [ENHANCEMENT] Ruler: Add DecodingConcurrency config flag for Thanos Engine. #7118
 * [ENHANCEMENT] Query Frontend: Add query priority based on operation. #7128
-* [ENHANCEMENT] Compactor: Avoid double compaction by cleaning partition files in 2 cycles. #7130 #7209
+* [ENHANCEMENT] Compactor: Avoid double compaction by cleaning partition files in 2 cycles. #7130 #7209 #7257
 * [ENHANCEMENT] Distributor: Optimize memory usage by recycling v2 requests. #7131
 * [ENHANCEMENT] Compactor: Avoid double compaction by not filtering delete blocks on real time when using bucketIndex lister. #7156
 * [ENHANCEMENT] Upgrade to go 1.25. #7164

pkg/compactor/blocks_cleaner.go

@@ -798,6 +798,7 @@ func (c *BlocksCleaner) cleanPartitionedGroupInfo(ctx context.Context, userBucke
 		partitionedGroupInfoFile := extraInfo.path
 		deletedBlocksCount := 0
 		partitionedGroupLogger := log.With(userLogger, "partitioned_group_id", partitionedGroupInfo.PartitionedGroupID, "partitioned_group_creation_time", partitionedGroupInfo.CreationTimeString())
+		isPartitionGroupInfoDeleted := false
 
 		if extraInfo.status.CanDelete {
 			if extraInfo.status.IsCompleted {
@@ -820,11 +821,13 @@ func (c *BlocksCleaner) cleanPartitionedGroupInfo(ctx context.Context, userBucke
 					level.Warn(partitionedGroupLogger).Log("msg", "failed to delete partitioned group info", "partitioned_group_file", partitionedGroupInfoFile, "err", err)
 				} else {
 					level.Info(partitionedGroupLogger).Log("msg", "deleted partitioned group info", "partitioned_group_file", partitionedGroupInfoFile)
+					isPartitionGroupInfoDeleted = true
 				}
 			}
 		}
 
-		if extraInfo.status.CanDelete {
+		if extraInfo.status.CanDelete && (isPartitionGroupInfoDeleted || c.cannotFindPartitionGroupInfoFile(ctx, userBucket, userLogger, partitionedGroupInfoFile)) {
+			level.Info(partitionedGroupLogger).Log("msg", "partition group info file has been deleted or does not exist, removing all related partition visit markers")
 			// Remove all partition visit markers for completed partitions
 			if _, err := bucket.DeletePrefix(ctx, userBucket, GetPartitionVisitMarkerDirectoryPath(partitionedGroupInfo.PartitionedGroupID), userLogger, defaultDeleteBlocksConcurrency); err != nil {
 				level.Warn(partitionedGroupLogger).Log("msg", "failed to delete all partition visit markers for partitioned group", "err", err)
@@ -845,6 +848,11 @@ func (c *BlocksCleaner) cleanPartitionedGroupInfo(ctx context.Context, userBucke
 	})
 }
 
+func (c *BlocksCleaner) cannotFindPartitionGroupInfoFile(ctx context.Context, userBucket objstore.InstrumentedBucket, userLogger log.Logger, partitionedGroupInfoFile string) bool {
+	_, err := ReadPartitionedGroupInfoFile(ctx, userBucket, userLogger, partitionedGroupInfoFile)
+	return err != nil && errors.Is(err, ErrorPartitionedGroupInfoNotFound)
+}
+
 func (c *BlocksCleaner) emitUserPartitionMetrics(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket, userID string) {
 	existentPartitionedGroupInfo, err := c.iterPartitionGroups(ctx, userBucket, userLogger)
 	if err != nil {

pkg/compactor/blocks_cleaner_test.go

@@ -984,6 +984,10 @@ func TestBlocksCleaner_CleanPartitionedGroupInfo(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, partitionedGroupFileExists)
 
+	visitMarkerExists, err := userBucket.Exists(ctx, visitMarker.GetVisitMarkerFilePath())
+	require.NoError(t, err)
+	require.True(t, visitMarkerExists)
+
 	block1DeletionMarkerExists, err := userBucket.Exists(ctx, path.Join(block1.String(), metadata.DeletionMarkFilename))
 	require.NoError(t, err)
 	require.True(t, block1DeletionMarkerExists)
@@ -1000,7 +1004,7 @@ func TestBlocksCleaner_CleanPartitionedGroupInfo(t *testing.T) {
 	require.NoError(t, err)
 	require.False(t, partitionedGroupFileExists)
 
-	visitMarkerExists, err := userBucket.Exists(ctx, visitMarker.GetVisitMarkerFilePath())
+	visitMarkerExists, err = userBucket.Exists(ctx, visitMarker.GetVisitMarkerFilePath())
 	require.NoError(t, err)
 	require.False(t, visitMarkerExists)
 }