Do not use ignoreDeletionMarker filter for bucketindex lister

CHANGELOG.md

@@ -14,6 +14,7 @@
 * [ENHANCEMENT] Ruler: Add DecodingConcurrency config flag for Thanos Engine. #7118
 * [ENHANCEMENT] Compactor: Avoid double compaction by cleaning partition files in 2 cycles. #7129
 * [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
 * [BUGFIX] Ring: Change DynamoDB KV to retry indefinitely for WatchKey. #7088
 * [BUGFIX] Ruler: Add XFunctions validation support. #7111
 * [BUGFIX] Querier: propagate Prometheus info annotations in protobuf responses. #7132

pkg/compactor/compactor.go

@@ -1024,7 +1024,8 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 	noCompactMarkerFilter := compact.NewGatherNoCompactionMarkFilter(ulogger, bucket, c.compactorCfg.MetaSyncConcurrency)
 
 	var blockLister block.Lister
-	switch cortex_tsdb.BlockDiscoveryStrategy(c.storageCfg.BucketStore.BlockDiscoveryStrategy) {
+	blockDiscoveryStrategy := cortex_tsdb.BlockDiscoveryStrategy(c.storageCfg.BucketStore.BlockDiscoveryStrategy)
+	switch blockDiscoveryStrategy {
 	case cortex_tsdb.ConcurrentDiscovery:
 		blockLister = block.NewConcurrentLister(ulogger, bucket)
 	case cortex_tsdb.RecursiveDiscovery:
@@ -1038,6 +1039,24 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 		return cortex_tsdb.ErrBlockDiscoveryStrategy
 	}
 
+	// List of filters to apply (order matters).
+	filterList := []block.MetadataFilter{
+		// Remove the ingester ID because we don't shard blocks anymore, while still
+		// honoring the shard ID if sharding was done in the past.
+		NewLabelRemoverFilter([]string{cortex_tsdb.IngesterIDExternalLabel}),
+		block.NewConsistencyDelayMetaFilter(ulogger, c.compactorCfg.ConsistencyDelay, reg),
+	}
+
+	// Add ignoreDeletionMarkFilter only when not using bucket index discovery.
+	if blockDiscoveryStrategy != cortex_tsdb.BucketIndexDiscovery {
+		filterList = append(filterList, ignoreDeletionMarkFilter)
+	}
+
+	filterList = append(filterList,
+		deduplicateBlocksFilter,
+		noCompactMarkerFilter,
+	)
+
 	fetcher, err := block.NewMetaFetcherWithMetrics(
 		ulogger,
 		c.compactorCfg.MetaSyncConcurrency,
@@ -1046,16 +1065,7 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 		c.metaSyncDirForUser(userID),
 		c.compactorMetrics.getBaseFetcherMetrics(),
 		c.compactorMetrics.getMetaFetcherMetrics(),
-		// List of filters to apply (order matters).
-		[]block.MetadataFilter{
-			// Remove the ingester ID because we don't shard blocks anymore, while still
-			// honoring the shard ID if sharding was done in the past.
-			NewLabelRemoverFilter([]string{cortex_tsdb.IngesterIDExternalLabel}),
-			block.NewConsistencyDelayMetaFilter(ulogger, c.compactorCfg.ConsistencyDelay, reg),
-			ignoreDeletionMarkFilter,
-			deduplicateBlocksFilter,
-			noCompactMarkerFilter,
-		},
+		filterList,
 	)
 	if err != nil {
 		return err