SungJin1212 · Nov 6, 2026
@@ -10,6 +10,7 @@
 * [ENHANCEMENT] gRPC: When `-auth.enabled=true`, a gRPC request without the `X-Scope-OrgID` metadata is now rejected with the `UNAUTHENTICATED` status code instead of `UNKNOWN`. #7873
 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889
 * [ENHANCEMENT] Ruler: Retry a notification the Alertmanager failed to accept with a transport error or a 5xx response, instead of dropping it and leaving its alert undelivered until the next evaluation of the rule group. Enable the retries with `-ruler.notification.backoff-retries` (`0`, no retry, by default), and tune the delay between them with `-ruler.notification.backoff-min-period` and `-ruler.notification.backoff-max-period`. The retries are bounded by `-ruler.notification-timeout` and counted by the new `cortex_ruler_notification_retries_total` metric. #7909
+* [ENHANCEMENT] Parquet Converter: Allow 2h blocks conversion if the block has a no-compact-mark.json file. #6865
 * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877
 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869
 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880

@@ -781,11 +781,16 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userLogger log.Logger, us
 		}
 		level.Info(userLogger).Log("msg", "finish writing new index", "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds())
 	}
-	c.updateBucketMetrics(userID, parquetEnabled, idx, float64(len(partials)), float64(totalBlocksBlocksMarkedForNoCompaction))
+
+	noCompactMarkCheckFunc := func(blockID ulid.ULID) bool {
+		return cortex_parquet.ExistBlockNoCompact(ctx, userBucket, userLogger, blockID)
+	}
+
+	c.updateBucketMetrics(userID, parquetEnabled, idx, float64(len(partials)), float64(totalBlocksBlocksMarkedForNoCompaction), noCompactMarkCheckFunc)
 	return nil
 }
 
-func (c *BlocksCleaner) updateBucketMetrics(userID string, parquetEnabled bool, idx *bucketindex.Index, partials, totalBlocksBlocksMarkedForNoCompaction float64) {
+func (c *BlocksCleaner) updateBucketMetrics(userID string, parquetEnabled bool, idx *bucketindex.Index, partials, totalBlocksBlocksMarkedForNoCompaction float64, noCompactMarkCheckFunc func(ulid.ULID) bool) {
 	c.tenantBlocks.WithLabelValues(userID).Set(float64(len(idx.Blocks)))
 	c.tenantBlocksMarkedForDelete.WithLabelValues(userID).Set(float64(len(idx.BlockDeletionMarks)))
 	c.tenantBlocksMarkedForNoCompaction.WithLabelValues(userID).Set(totalBlocksBlocksMarkedForNoCompaction)
@@ -794,8 +799,10 @@ func (c *BlocksCleaner) updateBucketMetrics(userID string, parquetEnabled bool,
 	if parquetEnabled {
 		c.tenantParquetBlocks.WithLabelValues(userID).Set(float64(len(idx.ParquetBlocks())))
 		remainingBlocksToConvert := 0
+		now := time.Now().UnixMilli()
 		for _, b := range idx.NonParquetBlocks() {
-			if cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.cfg.BlockRanges) {
+			hasNoCompactMark := func() bool { return noCompactMarkCheckFunc(b.ID) }
+			if cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, now, c.cfg.BlockRanges, hasNoCompactMark) {
 				remainingBlocksToConvert++
 			}
 		}

@@ -366,7 +366,7 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions
 		cortex_bucket_parquet_blocks_count{user="user-6"} 1
 		# HELP cortex_bucket_parquet_unconverted_blocks_count Total number of unconverted parquet blocks in the bucket. Blocks marked for deletion are included.
 		# TYPE cortex_bucket_parquet_unconverted_blocks_count gauge
-		cortex_bucket_parquet_unconverted_blocks_count{user="user-5"} 0
+		cortex_bucket_parquet_unconverted_blocks_count{user="user-5"} 1
 		cortex_bucket_parquet_unconverted_blocks_count{user="user-6"} 0
 	`),
 		"cortex_bucket_blocks_count",
@@ -1225,8 +1225,12 @@ func TestBlocksCleaner_ParquetMetrics(t *testing.T) {
 		},
 	}
 
+	mockNoCompactMarkCheckFunc := func(blockID ulid.ULID) bool {
+		return false
+	}
+
 	// Update metrics
-	cleaner.updateBucketMetrics("user1", true, idx, 0, 0)
+	cleaner.updateBucketMetrics("user1", true, idx, 0, 0, mockNoCompactMarkCheckFunc)
 
 	// Verify metrics
 	require.NoError(t, prom_testutil.CollectAndCompare(cleaner.tenantParquetBlocks, strings.NewReader(`

@@ -423,7 +423,11 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 			continue
 		}
 
-		if !cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.blockRanges) {
+		hasNoCompactMark := func() bool {
+			return cortex_parquet.ExistBlockNoCompact(ctx, uBucket, logger, b.ULID)
+		}
+
+		if !cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, time.Now().UnixMilli(), c.blockRanges, hasNoCompactMark) {
 			continue
 		}
 

@@ -19,10 +19,12 @@ import (
 	parquetgo "github.com/parquet-go/parquet-go"
 	"github.com/prometheus-community/parquet-common/convert"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/tsdb"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/objstore/providers/filesystem"
@@ -327,6 +329,87 @@ func TestConverter_CleanupMetricsForNotOwnedUser(t *testing.T) {
 	assert.Equal(t, 0.0, testutil.ToFloat64(converter.metrics.convertBlockFailures.WithLabelValues(userID)))
 }
 
+func TestConverter_ShouldConvertNoCompactMarkBlocks(t *testing.T) {
+	cfg := prepareConfig()
+	userID := "user"
+	ringStore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil)
+	t.Cleanup(func() { assert.NoError(t, closer.Close()) })
+	dir := t.TempDir()
+
+	cfg.Ring.InstanceID = "parquet-converter-1"
+	cfg.Ring.InstanceAddr = "1.2.3.4"
+	cfg.Ring.KVStore.Mock = ringStore
+	bucketClient, err := filesystem.NewBucket(t.TempDir())
+	require.NoError(t, err)
+	userBucket := bucket.NewPrefixedBucketClient(bucketClient, userID)
+	limits := &validation.Limits{}
+	flagext.DefaultValues(limits)
+	limits.ParquetConverterEnabled = true
+
+	c, logger, _ := prepare(t, cfg, objstore.WithNoopInstr(bucketClient), limits, nil)
+	c.ringLifecycler = &ring.Lifecycler{
+		Addr: "1.2.3.4",
+	}
+
+	ctx := context.Background()
+
+	lbls := labels.Labels{labels.Label{
+		Name:  "__name__",
+		Value: "test",
+	}}
+
+	blocks := []ulid.ULID{}
+	numBlocks := 2
+	mint := time.Hour * 13
+	maxt := time.Hour * 15
+	// Create 2h blocks
+	for range numBlocks {
+		rnd := rand.New(rand.NewSource(time.Now().Unix()))
+		id, err := e2e.CreateBlock(ctx, rnd, dir, []labels.Labels{lbls}, 2, mint.Milliseconds(), maxt.Milliseconds(), time.Minute.Milliseconds(), 10)
+		require.NoError(t, err)
+		blocks = append(blocks, id)
+	}
+
+	// Create a 2h block that is too young to have its no-compact mark checked.
+	recentMaxt := time.Now().Truncate(2 * time.Hour)
+	recentMint := recentMaxt.Add(-2 * time.Hour)
+	recentBlock, err := e2e.CreateBlock(ctx, rand.New(rand.NewSource(time.Now().Unix())), dir, []labels.Labels{lbls}, 2, recentMint.UnixMilli(), recentMaxt.UnixMilli(), time.Minute.Milliseconds(), 10)
+	require.NoError(t, err)
+
+	for _, bId := range append(blocks, recentBlock) {
+		blockDir := fmt.Sprintf("%s/%s", dir, bId.String())
+		b, err := tsdb.OpenBlock(nil, blockDir, nil, nil)
+		require.NoError(t, err)
+		// upload block
+		err = block.Upload(ctx, logger, userBucket, b.Dir(), metadata.NoneFunc)
+		require.NoError(t, err)
+
+		// upload no-compact-mark.json
+		err = block.MarkForNoCompact(ctx, logger, userBucket, bId, metadata.ManualNoCompactReason, "", promauto.With(nil).NewCounter(prometheus.CounterOpts{}))
+		require.NoError(t, err)
+	}
+
+	ringMock := &ring.RingMock{}
+	ringMock.On("Get", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ring.ReplicationSet{
+		Instances: []ring.InstanceDesc{
+			{
+				Addr: "1.2.3.4",
+			},
+		},
+	}, nil)
+
+	err = c.convertUser(ctx, log.NewNopLogger(), ringMock, userID)
+	require.NoError(t, err)
+
+	// Verify the converted metric was incremented only for the old blocks
+	assert.Equal(t, float64(numBlocks), testutil.ToFloat64(c.metrics.convertedBlocks.WithLabelValues(userID)))
+
+	// Verify the recent block was not converted, even though it has a no-compact mark
+	mark, err := parquet.ReadConverterMark(ctx, recentBlock, objstore.WithNoopInstr(userBucket), logger)
+	require.NoError(t, err)
+	assert.False(t, parquet.ValidConverterMarkVersion(mark.Version))
+}
+
 func TestConverter_BlockConversionFailure(t *testing.T) {
 	// Create a new registry for testing
 	reg := prometheus.NewRegistry()
@@ -376,7 +459,16 @@ func TestConverter_BlockConversionFailure(t *testing.T) {
 		Addr: "1.2.3.4",
 	}
 
-	err = converter.convertUser(context.Background(), logger, &RingMock{ReadRing: &ring.Ring{}}, userID)
+	ringMock := &ring.RingMock{}
+	ringMock.On("Get", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ring.ReplicationSet{
+		Instances: []ring.InstanceDesc{
+			{
+				Addr: "1.2.3.4",
+			},
+		},
+	}, nil)
+
+	err = converter.convertUser(context.Background(), logger, ringMock, userID)
 	require.NoError(t, err)
 
 	// Verify the failure metric was incremented
@@ -443,7 +535,16 @@ func TestConverter_ShouldNotFailOnAccessDenyError(t *testing.T) {
 		Addr: "1.2.3.4",
 	}
 
-	err = converter.convertUser(context.Background(), logger, &RingMock{ReadRing: &ring.Ring{}}, userID)
+	ringMock := &ring.RingMock{}
+	ringMock.On("Get", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ring.ReplicationSet{
+		Instances: []ring.InstanceDesc{
+			{
+				Addr: "1.2.3.4",
+			},
+		},
+	}, nil)
+
+	err = converter.convertUser(context.Background(), logger, ringMock, userID)
 	require.Error(t, err)
 
 	// Verify the failure metric was not incremented
@@ -471,20 +572,6 @@ func (m *mockBucket) Get(ctx context.Context, name string) (io.ReadCloser, error
 	return m.Bucket.Get(ctx, name)
 }
 
-type RingMock struct {
-	ring.ReadRing
-}
-
-func (r *RingMock) Get(key uint32, op ring.Operation, bufDescs []ring.InstanceDesc, bufHosts []string, bufZones map[string]int) (ring.ReplicationSet, error) {
-	return ring.ReplicationSet{
-		Instances: []ring.InstanceDesc{
-			{
-				Addr: "1.2.3.4",
-			},
-		},
-	}, nil
-}
-
 // mockTenantLimits implements the validation.TenantLimits interface for testing
 type mockTenantLimits struct {
 	limits map[string]*validation.Limits

@@ -9,8 +9,10 @@ import (
 
 	"github.com/efficientgo/core/errors"
 	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
 	"github.com/oklog/ulid/v2"
 	"github.com/thanos-io/objstore"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/runutil"
 
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
@@ -34,6 +36,15 @@ type ConverterMark struct {
 	Shards int `json:"shards,omitempty"`
 }
 
+func ExistBlockNoCompact(ctx context.Context, userBkt objstore.InstrumentedBucket, logger log.Logger, blockID ulid.ULID) bool {
+	noCompactMarkerExists, err := userBkt.Exists(ctx, path.Join(blockID.String(), metadata.NoCompactMarkFilename))
+	if err != nil {
+		level.Warn(logger).Log("msg", "unable to get stats of no-compact-mark.json for block", "block", blockID.String(), "err", err)
+		return false
+	}
+	return noCompactMarkerExists
+}
+
 func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.InstrumentedBucket, logger log.Logger) (*ConverterMark, error) {
 	markerPath := path.Join(id.String(), ConverterMarkerFileName)
 	reader, err := userBkt.WithExpectedErrs(bucket.IsOneOfTheExpectedErrors(userBkt.IsAccessDeniedErr, userBkt.IsObjNotFoundErr)).Get(ctx, markerPath)

@@ -1,8 +1,26 @@
 package parquet
 
-func ShouldConvertBlockToParquet(mint, maxt int64, timeRanges []int64) bool {
-	// We assume timeRanges[0] is the TSDB block duration (2h), and we don't convert them.
-	return getBlockTimeRange(mint, maxt, timeRanges) > timeRanges[0]
+// ShouldConvertBlockToParquet returns whether the block should be converted to parquet.
+// mint, maxt and now are timestamps in milliseconds.
+func ShouldConvertBlockToParquet(mint, maxt, now int64, timeRanges []int64, hasNoCompactMark func() bool) bool {
+	// We assume timeRanges[0] is the TSDB block duration (2h). Bigger blocks are always converted.
+	if getBlockTimeRange(mint, maxt, timeRanges) > timeRanges[0] {
+		return true
+	}
+
+	// We should check if 2h blocks have a `no-compact-mark.json` file
+	// since these will never be compacted to 12h block.
+	// We check if the `no-compact-mark.json` file exists only for blocks older
+	// than the next block range to reduce object storage calls, since younger
+	// blocks are still expected to be compacted.
+	return now-maxt >= noCompactMarkCheckAge(timeRanges) && hasNoCompactMark()
+}
+
+func noCompactMarkCheckAge(timeRanges []int64) int64 {
+	if len(timeRanges) > 1 {
+		return timeRanges[1]
+	}
+	return timeRanges[0]
 }
 
 func getBlockTimeRange(mint, maxt int64, timeRanges []int64) int64 {