Compactor: honor the per-tenant float chunk encoding

CHANGELOG.md

@@ -24,6 +24,7 @@
 * [ENHANCEMENT] Block-builder-scheduler: Add `cortex_blockbuilder_scheduler_end_offset_probe_failed_total`, counting failures to list a cluster's end offsets, and `cortex_blockbuilder_scheduler_startup_jobs_skipped_total`, counting observed jobs that startup recovery could not import. #16134
 * [FEATURE] Querier: Add experimental per-tenant limit `-querier.max-blocks-per-store-request` to cap the number of blocks a single store-gateway request may reference. Disabled by default. #16292
 * [FEATURE] Validation: Add optional `id`, `note`, `created_by`, `created_at`, and `expires_at` fields to `blocked_queries` and `limited_queries` rules, for tooling to attach ownership/context metadata to a rule. For rules with `expires_at` set, the earliest `expires_at` per tenant and `id` (rules without an `id` are grouped together) is exported as the `cortex_blocked_query_rule_expires_at`/`cortex_limited_query_rule_expires_at` metrics, so an alert can fire on stale rules; this is informational only and never affects enforcement. The query-frontend's `"query blocked"` log line now also includes the matched rule's `id` and whether it is expired, and rate-limited queries are now logged with a new `"query limited"` line carrying the same fields. #16395
+* [BUGFIX] Compactor: Honor the per-tenant `float_chunk_encoding` limit (`-ingester.float-chunk-encoding`) when re-encoding float chunks during compaction. Previously the compactor was built without a float chunk encoding, so every float chunk it re-encoded was written back as `xor`, undoing `xor2` for tenants that had it enabled. Only chunks that overlap in time are re-encoded, so compacted blocks can stay mixed-encoding, and blocks already compacted are not repaired. #16488
 * [BUGFIX] Query-frontend: Wait for the querier ring to be populated during startup, up to 30 seconds, before reporting the query-frontend as ready. Previously a query-frontend could become ready before it had seen any querier in the ring and fail every query it received until the ring was populated. Only applies when remote execution is enabled, and can be disabled with the experimental `-query-frontend.wait-for-querier-ring-on-startup=false`. #16333
 * [BUGFIX] Query-frontend: Fail queries with a clear error, rather than planning them against an invalid maximum supported query plan version, when the querier ring contains only unhealthy queriers. #16333
 * [BUGFIX] Query-frontend: Fix `cortex_query_frontend_queries_in_progress` drifting permanently below zero. The response body returned to the middleware chain is closed more than once, and every close decremented the gauge against a single increment. #16429

pkg/compactor/blocks_cleaner_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	prom_tsdb "github.com/prometheus/prometheus/tsdb"
+	"github.com/prometheus/prometheus/tsdb/chunkenc"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
@@ -39,6 +40,7 @@ import (
 	mimir_testutil "github.com/grafana/mimir/pkg/storage/tsdb/testutil"
 	"github.com/grafana/mimir/pkg/util"
 	"github.com/grafana/mimir/pkg/util/test"
+	"github.com/grafana/mimir/pkg/util/validation"
 )
 
 type testBlocksCleanerOptions struct {
@@ -1857,6 +1859,7 @@ func (m *mockBucketFailure) Delete(ctx context.Context, name string) error {
 }
 
 type mockConfigProvider struct {
+	floatChunkEncodings          map[string]chunkenc.Encoding
 	userRetentionPeriods         map[string]time.Duration
 	splitAndMergeShards          map[string]int
 	oooSplitAndMergeShards       map[string]int
@@ -1874,6 +1877,7 @@ type mockConfigProvider struct {
 
 func newMockConfigProvider() *mockConfigProvider {
 	return &mockConfigProvider{
+		floatChunkEncodings:          make(map[string]chunkenc.Encoding),
 		userRetentionPeriods:         make(map[string]time.Duration),
 		splitAndMergeShards:          make(map[string]int),
 		oooSplitAndMergeShards:       make(map[string]int),
@@ -1944,6 +1948,13 @@ func (m *mockConfigProvider) CompactorBlockUploadMaxBlockSizeBytes(user string)
 	return m.blockUploadMaxBlockSizeBytes[user]
 }
 
+func (m *mockConfigProvider) FloatChunkEncoding(userID string) chunkenc.Encoding {
+	if result, ok := m.floatChunkEncodings[userID]; ok {
+		return result
+	}
+	return validation.ParseFloatChunkEncoding(validation.DefaultFloatChunkEncodingValue)
+}
+
 func (m *mockConfigProvider) S3SSEType(string) string {
 	return ""
 }

pkg/compactor/compactor.go

@@ -28,6 +28,7 @@ import (
 	"github.com/grafana/dskit/services"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/prometheus/prometheus/tsdb/chunkenc"
 	"github.com/thanos-io/objstore"
 	"go.uber.org/atomic"
 
@@ -86,13 +87,18 @@ type BlocksGrouperFactory func(
 	reg prometheus.Registerer,
 ) Grouper
 
-// BlocksCompactorFactory builds and returns the compactor and planner for compacting a tenant's blocks.
+// BlocksCompactorFactory builds and returns the compactor provider and planner for compacting tenants' blocks.
 type BlocksCompactorFactory func(
 	ctx context.Context,
 	cfg Config,
+	cfgProvider ConfigProvider,
 	logger log.Logger,
 	reg prometheus.Registerer,
-) (Compactor, Planner, error)
+) (BlocksCompactorProvider, Planner, error)
+
+// BlocksCompactorProvider returns the Compactor to use for a given tenant's blocks. It must be safe
+// for concurrent use, because jobs belonging to different tenants can be compacted at the same time.
+type BlocksCompactorProvider func(userID string) Compactor
 
 // Config holds the MultitenantCompactor config.
 type Config struct {
@@ -286,6 +292,10 @@ type ConfigProvider interface {
 
 	// CompactorMaxPerBlockUploadConcurrency returns the maximum number of TSDB files that can be uploaded concurrently for each block.
 	CompactorMaxPerBlockUploadConcurrency(userID string) int
+
+	// FloatChunkEncoding returns the encoding to use for float chunks written for a given user.
+	// An encoding no -ingester.float-chunk-encoding value selects is treated as the default.
+	FloatChunkEncoding(userID string) chunkenc.Encoding
 }
 
 // MultitenantCompactor is a multi-tenant TSDB block compactor based on Thanos.
@@ -308,9 +318,9 @@ type MultitenantCompactor struct {
 	// Blocks cleaner is responsible for hard deletion of blocks marked for deletion.
 	blocksCleaner *BlocksCleaner
 
-	// Underlying compactor and planner for compacting TSDB blocks.
-	blocksCompactor Compactor
-	blocksPlanner   Planner
+	// Underlying compactor provider and planner for compacting TSDB blocks.
+	blocksCompactorProvider BlocksCompactorProvider
+	blocksPlanner           Planner
 
 	// Client used to run operations on the bucket storing blocks.
 	bucketClient objstore.Bucket
@@ -571,7 +581,7 @@ func (c *MultitenantCompactor) starting(ctx context.Context) error {
 	}
 
 	// Create blocks compactor dependencies.
-	c.blocksCompactor, c.blocksPlanner, err = c.blocksCompactorFactory(ctx, c.compactorCfg, c.logger, c.registerer)
+	c.blocksCompactorProvider, c.blocksPlanner, err = c.blocksCompactorFactory(ctx, c.compactorCfg, c.cfgProvider, c.logger, c.registerer)
 	if err != nil {
 		return fmt.Errorf("failed to initialize compactor dependencies: %w", err)
 	}
@@ -944,7 +954,7 @@ func (c *MultitenantCompactor) newBucketCompactor(ctx context.Context, userID st
 		userLogger,
 		c.blocksGrouperFactory(ctx, c.compactorCfg, c.cfgProvider, userID, userLogger, reg),
 		c.blocksPlanner,
-		c.blocksCompactor,
+		c.blocksCompactorProvider(userID),
 		compactDir,
 		userBucket,
 		c.compactorCfg.CompactionConcurrency,

pkg/compactor/compactor_test.go

@@ -1884,8 +1884,8 @@ func prepareWithConfigProvider(t *testing.T, compactorCfg Config, bucketClient o
 		return bucketClient, nil
 	}
 
-	blocksCompactorFactory := func(context.Context, Config, log.Logger, prometheus.Registerer) (Compactor, Planner, error) {
-		return tsdbCompactor, tsdbPlanner, nil
+	blocksCompactorFactory := func(context.Context, Config, ConfigProvider, log.Logger, prometheus.Registerer) (BlocksCompactorProvider, Planner, error) {
+		return func(string) Compactor { return tsdbCompactor }, tsdbPlanner, nil
 	}
 
 	c, err := newMultitenantCompactor(compactorCfg, storageCfg, limits, logger, registry, bucketClientFactory, splitAndMergeGrouperFactory, blocksCompactorFactory)

pkg/compactor/executor_test.go

@@ -162,8 +162,11 @@ func newTestSchedulerExecutor(t *testing.T, cfg Config, client compactorschedule
 
 func prepareCompactorForExecutorTest(t *testing.T, cfg Config, bkt objstore.Bucket, cfgProvider ConfigProvider) *MultitenantCompactor {
 	t.Helper()
-	c, _, _, _, _ := prepareWithConfigProvider(t, cfg, bkt, cfgProvider)
+	c, tsdbCompactor, _, _, _ := prepareWithConfigProvider(t, cfg, bkt, cfgProvider)
 	c.bucketClient = bkt
+	// These tests don't start the service, so the dependencies normally built by starting() have to
+	// be installed by hand.
+	c.blocksCompactorProvider = func(string) Compactor { return tsdbCompactor }
 	c.shardingStrategy = newSplitAndMergeShardingStrategy(nil, nil, nil, c.cfgProvider)
 	return c
 }
@@ -911,9 +914,9 @@ func TestSchedulerExecutor_ExecuteCompactionJob_Compaction(t *testing.T) {
 			schedulerExec := newTestSchedulerExecutor(t, cfg, nil)
 			c := prepareCompactorForExecutorTest(t, cfg, bkt, mockCfg)
 
-			compactor, planner, err := splitAndMergeCompactorFactory(context.Background(), cfg, log.NewNopLogger(), prometheus.NewRegistry())
+			compactor, planner, err := splitAndMergeCompactorFactory(t.Context(), cfg, mockCfg, log.NewNopLogger(), prometheus.NewRegistry())
 			require.NoError(t, err)
-			c.blocksCompactor = compactor
+			c.blocksCompactorProvider = compactor
 			c.blocksPlanner = planner
 
 			blockIDBytes := make([][]byte, len(setup.blockIDsToCompact))

pkg/compactor/split_merge_compactor.go

@@ -4,12 +4,15 @@ package compactor
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/go-kit/log"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/tsdb"
+	"github.com/prometheus/prometheus/tsdb/chunkenc"
 
 	util_log "github.com/grafana/mimir/pkg/util/log"
+	"github.com/grafana/mimir/pkg/util/validation"
 )
 
 func splitAndMergeGrouperFactory(_ context.Context, cfg Config, cfgProvider ConfigProvider, userID string, logger log.Logger, _ prometheus.Registerer) Grouper {
@@ -20,22 +23,48 @@ func splitAndMergeGrouperFactory(_ context.Context, cfg Config, cfgProvider Conf
 		logger)
 }
 
-func splitAndMergeCompactorFactory(ctx context.Context, cfg Config, logger log.Logger, reg prometheus.Registerer) (Compactor, Planner, error) {
-	// We don't need to customise the TSDB compactor so we're just using the Prometheus one.
-	compactor, err := tsdb.NewLeveledCompactor(ctx, reg, util_log.SlogFromGoKit(logger), cfg.BlockRanges.ToMilliseconds(), nil, nil)
-	if err != nil {
-		return nil, nil, err
-	}
+func splitAndMergeCompactorFactory(ctx context.Context, cfg Config, cfgProvider ConfigProvider, logger log.Logger, reg prometheus.Registerer) (BlocksCompactorProvider, Planner, error) {
+	blockRanges := cfg.BlockRanges.ToMilliseconds()
+
+	concurrencyOpts := tsdb.DefaultLeveledCompactorConcurrencyOptions()
+	concurrencyOpts.MaxOpeningBlocks = cfg.MaxOpeningBlocksConcurrency
+	concurrencyOpts.MaxClosingBlocks = cfg.MaxClosingBlocksConcurrency
+	concurrencyOpts.SymbolsFlushersCount = cfg.SymbolsFlushersConcurrency
+
+	// The metrics are built once and shared by every compactor: registering them twice would panic,
+	// and they are aggregated across tenants anyway.
+	metrics := tsdb.NewCompactorMetrics(reg)
 
-	opts := tsdb.DefaultLeveledCompactorConcurrencyOptions()
-	opts.MaxOpeningBlocks = cfg.MaxOpeningBlocksConcurrency
-	opts.MaxClosingBlocks = cfg.MaxClosingBlocksConcurrency
-	opts.SymbolsFlushersCount = cfg.SymbolsFlushersConcurrency
+	// The encoding reaches the merge function through a callback taking no tenant, and jobs for
+	// different tenants compact concurrently, so we build one compactor per encoding.
+	compactors := make(map[chunkenc.Encoding]Compactor, len(validation.FloatChunkEncodingValues))
+	for _, value := range validation.FloatChunkEncodingValues {
+		enc := validation.ParseFloatChunkEncoding(value)
+		compactor, err := tsdb.NewLeveledCompactorWithOptions(ctx, nil, util_log.SlogFromGoKit(logger), blockRanges, nil, tsdb.LeveledCompactorOptions{
+			Metrics:            metrics,
+			FloatChunkEncoding: func() chunkenc.Encoding { return enc },
+			// Inert here, since Mimir plans compaction itself, but NewLeveledCompactor() set it
+			// and this keeps the switch to NewLeveledCompactorWithOptions() behaviour-preserving.
+			EnableOverlappingCompaction: true,
+		})
+		if err != nil {
+			return nil, nil, fmt.Errorf("creating compactor for float chunk encoding %s: %w", value, err)
+		}
 
-	compactor.SetConcurrencyOptions(opts)
+		compactor.SetConcurrencyOptions(concurrencyOpts)
+		compactors[enc] = compactor
+	}
+
+	// A downstream ConfigProvider may return an encoding the limit cannot select.
+	defaultCompactor := compactors[validation.ParseFloatChunkEncoding(validation.DefaultFloatChunkEncodingValue)]
+	provider := func(userID string) Compactor {
+		if compactor, ok := compactors[cfgProvider.FloatChunkEncoding(userID)]; ok {
+			return compactor
+		}
+		return defaultCompactor
+	}
 
-	planner := NewSplitAndMergePlanner(cfg.BlockRanges.ToMilliseconds())
-	return compactor, planner, nil
+	return provider, NewSplitAndMergePlanner(blockRanges), nil
 }
 
 // configureSplitAndMergeCompactor updates the provided configuration injecting the split-and-merge compactor.