From 58d2e8bc09db0d132e168287f3bfdf9e1a271147 Mon Sep 17 00:00:00 2001 From: Siddarth Gundu Date: Tue, 16 Jun 2026 15:36:45 +0530 Subject: [PATCH 1/6] parquetconverter: skip blocks with too many label names Changes: - Add parquet no-convert marker and read/write logic - Add max-block-label-names limit, blocks exceeding it get a no-convert marker instead of being converted. - Add parquet_converter_max_block_label_names to exporter test - Add integration test for parquet no-convert marker Signed-off-by: Siddarth Gundu --- integration/parquet_converter_test.go | 142 +++++++++++++++++++++++ pkg/parquetconverter/converter.go | 76 +++++++++++- pkg/parquetconverter/converter_test.go | 131 +++++++++++++++++++++ pkg/parquetconverter/metrics.go | 6 + pkg/storage/parquet/no_convert_marker.go | 72 ++++++++++++ pkg/util/validation/exporter_test.go | 1 + pkg/util/validation/limits.go | 13 ++- 7 files changed, 433 insertions(+), 8 deletions(-) create mode 100644 integration/parquet_converter_test.go create mode 100644 pkg/storage/parquet/no_convert_marker.go diff --git a/integration/parquet_converter_test.go b/integration/parquet_converter_test.go new file mode 100644 index 00000000000..4a10d3f1e0f --- /dev/null +++ b/integration/parquet_converter_test.go @@ -0,0 +1,142 @@ +//go:build integration + +package integration + +import ( + "context" + "fmt" + "math/rand" + "path/filepath" + "testing" + "time" + + "github.com/prometheus/prometheus/model/labels" + "github.com/stretchr/testify/require" + "github.com/thanos-io/objstore" + "github.com/thanos-io/thanos/pkg/block" + "github.com/thanos-io/thanos/pkg/block/metadata" + + "github.com/cortexproject/cortex/integration/e2e" + e2ecache "github.com/cortexproject/cortex/integration/e2e/cache" + e2edb "github.com/cortexproject/cortex/integration/e2e/db" + "github.com/cortexproject/cortex/integration/e2ecortex" + "github.com/cortexproject/cortex/pkg/storage/bucket" + "github.com/cortexproject/cortex/pkg/storage/tsdb" + "github.com/cortexproject/cortex/pkg/util/log" + cortex_testutil "github.com/cortexproject/cortex/pkg/util/test" +) + +func TestParquetConverter_NoConvertMarkWithTooManyLabels(t *testing.T) { + s, err := e2e.NewScenario(networkName) + require.NoError(t, err) + defer s.Close() + + consul := e2edb.NewConsulWithName("consul") + memcached := e2ecache.NewMemcached() + require.NoError(t, s.StartAndWaitReady(consul, memcached)) + + baseFlags := mergeFlags(AlertmanagerLocalFlags(), BlocksStorageFlags()) + flags := mergeFlags( + baseFlags, + map[string]string{ + "-target": "all,parquet-converter", + "-blocks-storage.tsdb.block-ranges-period": "1m,24h", + "-blocks-storage.tsdb.ship-interval": "1s", + "-blocks-storage.bucket-store.sync-interval": "1s", + "-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl": "1s", + "-blocks-storage.bucket-store.bucket-index.idle-timeout": "1s", + "-blocks-storage.bucket-store.bucket-index.enabled": "true", + "-blocks-storage.bucket-store.index-cache.backend": tsdb.IndexCacheBackendInMemory, + // compactor + "-compactor.cleanup-interval": "1s", + // Ingester. + "-ring.store": "consul", + "-consul.hostname": consul.NetworkHTTPEndpoint(), + // Distributor. + "-distributor.replication-factor": "1", + // Store-gateway. + "-store-gateway.sharding-enabled": "false", + "--querier.store-gateway-addresses": "nonExistent", // Make sure we do not call Store gateways + // alert manager + "-alertmanager.web.external-url": "http://localhost/alertmanager", + // Enable vertical sharding. + "-frontend.query-vertical-shard-size": "3", + "-frontend.max-cache-freshness": "1m", + // enable experimental promQL funcs + "-querier.enable-promql-experimental-functions": "true", + // parquet-converter + "-parquet-converter.ring.consul.hostname": consul.NetworkHTTPEndpoint(), + "-parquet-converter.conversion-interval": "1s", + "-parquet-converter.enabled": "true", + "-parquet-converter.max-block-label-names": "1", + // Querier + "-querier.enable-parquet-queryable": "true", + // Enable cache for parquet labels and chunks + "-blocks-storage.bucket-store.parquet-labels-cache.backend": "inmemory,memcached", + "-blocks-storage.bucket-store.parquet-labels-cache.memcached.addresses": "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort), + "-blocks-storage.bucket-store.chunks-cache.backend": "inmemory,memcached", + "-blocks-storage.bucket-store.chunks-cache.memcached.addresses": "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort), + }, + ) + + // make alert manager config dir + require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{})) + + ctx := context.Background() + rnd := rand.New(rand.NewSource(time.Now().Unix())) + dir := filepath.Join(s.SharedDir(), "data") + lbls := []labels.Labels{ + labels.FromStrings("__name__", "test_series_a", "job", "test"), + } + + numSamples := 60 + scrapeInterval := time.Minute + now := time.Now() + start := now.Add(-time.Hour * 24) + end := now.Add(-time.Hour) + + minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"]) + require.NoError(t, s.StartAndWaitReady(minio)) + + cortex := e2ecortex.NewSingleBinary("cortex", flags, "") + require.NoError(t, s.StartAndWaitReady(cortex)) + storage, err := e2ecortex.NewS3ClientForMinio(minio, flags["-blocks-storage.s3.bucket-name"]) + require.NoError(t, err) + bkt := bucket.NewUserBucketClient("user-1", storage.GetBucket(), nil) + + id, err := e2e.CreateBlock(ctx, rnd, dir, lbls, numSamples, + start.UnixMilli(), + end.UnixMilli(), + scrapeInterval.Milliseconds(), 10, + ) + require.NoError(t, err) + + err = block.Upload(ctx, log.Logger, bkt, filepath.Join(dir, id.String()), metadata.NoneFunc) + require.NoError(t, err) + + // Wait for the converter to write the no-convert marker + cortex_testutil.Poll(t, 30*time.Second, true, func() interface{} { + noConvertMarkerPath := fmt.Sprintf("%s/parquet-no-convert-mark.json", id.String()) + found := false + err := bkt.Iter(ctx, "", func(name string) error { + if name == noConvertMarkerPath { + found = true + } + return nil + }, objstore.WithRecursiveIter()) + require.NoError(t, err) + return found + }) + + // confirm the conversion did not happen (check both paths) + blockID := id.String() + markerPaths := []string{ + fmt.Sprintf("%s/parquet-converter-mark.json", blockID), + fmt.Sprintf("parquet-markers/%s-parquet-converter-mark.json", blockID), + } + for _, markerPath := range markerPaths { + exists, err := bkt.Exists(ctx, markerPath) + require.NoError(t, err) + require.False(t, exists, "converter mark should not exist at %s", markerPath) + } +} diff --git a/pkg/parquetconverter/converter.go b/pkg/parquetconverter/converter.go index 5184fe9f68e..d6baad211bb 100644 --- a/pkg/parquetconverter/converter.go +++ b/pkg/parquetconverter/converter.go @@ -49,6 +49,9 @@ const ( ringKey = "parquet-converter" converterMetaPrefix = "converter-meta-" + + parquetConverterDataColumnDuration = time.Hour * 8 + parquetConverterSystemColumnCount = 2 // s_col_indexes and s_series_hash. ) var RingOp = ring.NewOp([]ring.InstanceState{ring.ACTIVE}, nil) @@ -424,12 +427,29 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin continue } - if err := os.RemoveAll(c.compactRootDir()); err != nil { - level.Error(logger).Log("msg", "failed to remove work directory", "path", c.compactRootDir(), "err", err) - if c.checkConvertError(userID, err) { - return err + configuredMaxBlockLabelNames := c.limits.ParquetConverterMaxBlockLabelNames(userID) + maxBlockLabelNames := effectiveMaxBlockLabelNames(configuredMaxBlockLabelNames, b.MinTime, b.MaxTime) + + // If the threshold is enabled, check for no-convert mark + if configuredMaxBlockLabelNames > 0 { + noConvertMark, err := cortex_parquet.ReadNoConvertMark(ctx, b.ULID, uBucket, logger) + if err != nil { + level.Error(logger).Log("msg", "failed to read parquet no-convert marker", "block", b.ULID.String(), "err", err) + continue + } + if cortex_parquet.ValidNoConvertMarkVersion(noConvertMark.Version) { + level.Debug(logger).Log("msg", "skipping block, no-convert marker already exists", "block", b.ULID.String()) + c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonMarkerExists).Inc() + continue + } + + if err := os.RemoveAll(c.compactRootDir()); err != nil { + level.Error(logger).Log("msg", "failed to remove work directory", "path", c.compactRootDir(), "err", err) + if c.checkConvertError(userID, err) { + return err + } + continue } - continue } bdir := filepath.Join(c.compactDirForUser(userID), b.ULID.String()) @@ -453,6 +473,33 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin continue } + if configuredMaxBlockLabelNames > 0 { + labelNames, err := tsdbBlock.LabelNames(ctx) + if err != nil { + _ = tsdbBlock.Close() + level.Error(logger).Log("msg", "failed to get label names", "block", b.ULID.String(), "err", err) + if c.checkConvertError(userID, err) { + return err + } + continue + } + labelNamesCount := len(labelNames) + if labelNamesCount > maxBlockLabelNames { + if err := cortex_parquet.WriteNoConvertMark(ctx, b.ULID, uBucket, labelNamesCount, maxBlockLabelNames); err != nil { + _ = tsdbBlock.Close() + level.Error(logger).Log("msg", "failed to write parquet no-convert marker", "block", b.ULID.String(), "err", err) + if c.checkConvertError(userID, err) { + return err + } + continue + } + level.Debug(logger).Log("msg", "skipping parquet conversion for block with too many label names", "block", b.ULID.String(), "label_names", labelNamesCount, "limit", maxBlockLabelNames) + c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonTooManyLabels).Inc() + _ = tsdbBlock.Close() + continue + } + } + level.Info(logger).Log("msg", "converting block", "block", b.ULID.String(), "dir", bdir) start := time.Now() @@ -514,6 +561,25 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin return nil } +func effectiveMaxBlockLabelNames(configuredMaxBlockLabelNames int, mint, maxt int64) int { + if configuredMaxBlockLabelNames <= 0 { + return configuredMaxBlockLabelNames + } + + dataColumnCount := 0 + if maxt >= mint { + dataColumnCount = int((maxt-mint)/parquetConverterDataColumnDuration.Milliseconds()) + 1 + } + + // Reserve for s_col_indexes, s_series_hash, and generated s_data_* columns. + maxBlockLabelNames := max(parquet.MaxColumnIndex-parquetConverterSystemColumnCount-dataColumnCount, 0) + + if configuredMaxBlockLabelNames > maxBlockLabelNames { + return maxBlockLabelNames + } + return configuredMaxBlockLabelNames +} + func (c *Converter) checkConvertError(userID string, err error) (terminate bool) { if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || c.isCausedByPermissionDenied(err) { terminate = true diff --git a/pkg/parquetconverter/converter_test.go b/pkg/parquetconverter/converter_test.go index 9cdcf31f6ea..e0c03906b8c 100644 --- a/pkg/parquetconverter/converter_test.go +++ b/pkg/parquetconverter/converter_test.go @@ -16,6 +16,7 @@ import ( "github.com/go-kit/log" "github.com/oklog/ulid/v2" + 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/testutil" @@ -734,3 +735,133 @@ func TestConverter_RingLifecyclerShouldAutoForgetUnhealthyInstances(t *testing.T err = services.StopAndAwaitTerminated(context.Background(), converters[0]) require.True(t, err == nil || errors.Is(err, context.Canceled), "unexpected error stopping converter: %v", err) } + +func TestConverter_WriteNoConvertMarkForBlockWithTooManyLabels(t *testing.T) { + cfg := prepareConfig() + user := "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, user) + limits := &validation.Limits{} + flagext.DefaultValues(limits) + limits.ParquetConverterEnabled = true + limits.ParquetConverterMaxBlockLabelNames = 1 + + c, logger, _ := prepare(t, cfg, objstore.WithNoopInstr(bucketClient), limits, nil) + + ctx := context.Background() + + lbls := labels.FromStrings("__name__", "test", "job", "foo") + + // Create a block + rnd := rand.New(rand.NewSource(time.Now().Unix())) + + // 2h blocks are skipped by ShouldConvertBlockToParquet + blockID, err := e2e.CreateBlock(ctx, rnd, dir, []labels.Labels{lbls}, 2, 0, 4*time.Hour.Milliseconds(), time.Minute.Milliseconds(), 10) + require.NoError(t, err) + + // Upload the block to the bucket + blockDir := fmt.Sprintf("%s/%s", dir, blockID.String()) + b, err := tsdb.OpenBlock(nil, blockDir, nil, nil) + require.NoError(t, err) + err = block.Upload(ctx, logger, userBucket, b.Dir(), metadata.NoneFunc) + require.NoError(t, err) + + err = services.StartAndAwaitRunning(context.Background(), c) + require.NoError(t, err) + defer services.StopAndAwaitTerminated(ctx, c) // nolint:errcheck + + // Start the converter + err = c.convertUser(ctx, logger, c.ring, user) + require.NoError(t, err) + + // Verify the marker was written correctly + readNoConvertMark, err := parquet.ReadNoConvertMark(ctx, blockID, userBucket, logger) + require.NoError(t, err) + require.True(t, parquet.ValidNoConvertMarkVersion(readNoConvertMark.Version)) + require.Equal(t, fmt.Sprintf("%s: label_names_count=%d threshold=%d", parquet.NoConvertReasonTooManyLabels, 2, 1), readNoConvertMark.Reason) + + // Confirm conversion did not happen + assert.Equal(t, 0.0, testutil.ToFloat64(c.metrics.convertedBlocks.WithLabelValues(user))) + assert.Equal(t, 1.0, testutil.ToFloat64(c.metrics.skippedBlocks.WithLabelValues(user, parquet.NoConvertReasonTooManyLabels))) +} + +func TestConverter_SkipBlockWhenNoConvertMarkAlreadyExists(t *testing.T) { + cfg := prepareConfig() + user := "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, user) + limits := &validation.Limits{} + flagext.DefaultValues(limits) + limits.ParquetConverterEnabled = true + + c, logger, _ := prepare(t, cfg, objstore.WithNoopInstr(bucketClient), limits, nil) + + ctx := context.Background() + + lbls := labels.FromStrings("__name__", "test", "job", "foo") + rnd := rand.New(rand.NewSource(time.Now().Unix())) + + // 2h blocks are skipped by ShouldConvertBlockToParquet + blockID, err := e2e.CreateBlock(ctx, rnd, dir, []labels.Labels{lbls}, 2, 0, + 4*time.Hour.Milliseconds(), time.Minute.Milliseconds(), 10) + require.NoError(t, err) + + blockDir := fmt.Sprintf("%s/%s", dir, blockID.String()) + b, err := tsdb.OpenBlock(nil, blockDir, nil, nil) + require.NoError(t, err) + err = block.Upload(ctx, logger, userBucket, b.Dir(), metadata.NoneFunc) + require.NoError(t, err) + + markerV1 := parquet.NoConvertMark{ + Version: parquet.CurrentNoConvertMarkVersion, + Reason: "manually uploaded", + } + markerBytes, err := json.Marshal(markerV1) + require.NoError(t, err) + markerPath := path.Join(blockID.String(), parquet.NoConvertMarkerFileName) + err = userBucket.Upload(ctx, markerPath, bytes.NewReader(markerBytes)) + require.NoError(t, err) + + err = services.StartAndAwaitRunning(context.Background(), c) + require.NoError(t, err) + defer services.StopAndAwaitTerminated(ctx, c) // nolint:errcheck + + // start converter + err = c.convertUser(ctx, logger, c.ring, user) + require.NoError(t, err) + + // confirm conversion was skipped + assert.Equal(t, 0.0, testutil.ToFloat64(c.metrics.convertedBlocks.WithLabelValues(user))) + assert.Equal(t, 1.0, testutil.ToFloat64(c.metrics.skippedBlocks.WithLabelValues(user, parquet.NoConvertReasonMarkerExists))) + + markerAfter, err := parquet.ReadNoConvertMark(ctx, blockID, userBucket, logger) + require.NoError(t, err) + require.True(t, parquet.ValidNoConvertMarkVersion(markerAfter.Version)) + require.Equal(t, "manually uploaded", markerAfter.Reason) +} + +func TestEffectiveMaxBlockLabelNamesLeavesRoomForGeneratedColumns(t *testing.T) { + mint := int64(0) + maxt := 2 * parquetConverterDataColumnDuration.Milliseconds() + expectedReservedColumns := parquetConverterSystemColumnCount + 3 + + require.Equal(t, 10, effectiveMaxBlockLabelNames(10, mint, maxt)) + require.Equal(t, parquetgo.MaxColumnIndex-expectedReservedColumns, effectiveMaxBlockLabelNames(parquetgo.MaxColumnIndex, mint, maxt)) + require.Equal(t, 0, effectiveMaxBlockLabelNames(0, mint, maxt)) +} diff --git a/pkg/parquetconverter/metrics.go b/pkg/parquetconverter/metrics.go index d926ed6d465..4de08271d01 100644 --- a/pkg/parquetconverter/metrics.go +++ b/pkg/parquetconverter/metrics.go @@ -10,6 +10,7 @@ import ( type metrics struct { convertedBlocks *prometheus.CounterVec convertBlockFailures *prometheus.CounterVec + skippedBlocks *prometheus.CounterVec convertBlockDuration *prometheus.GaugeVec convertParquetBlockDelay prometheus.Histogram ownedUsers prometheus.Gauge @@ -25,6 +26,10 @@ func newMetrics(reg prometheus.Registerer) *metrics { Name: "cortex_parquet_converter_block_convert_failures_total", Help: "Total number of failed block conversions per user.", }, []string{"user"}), + skippedBlocks: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "cortex_parquet_converter_blocks_skipped_total", + Help: "Total number of blocks skipped during parquet conversion per user and reason.", + }, []string{"user", "reason"}), convertBlockDuration: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{ Name: "cortex_parquet_converter_convert_block_duration_seconds", Help: "Time taken to for the latest block conversion for the user.", @@ -47,5 +52,6 @@ func newMetrics(reg prometheus.Registerer) *metrics { func (m *metrics) deleteMetricsForTenant(userID string) { m.convertedBlocks.DeleteLabelValues(userID) m.convertBlockFailures.DeleteLabelValues(userID) + m.skippedBlocks.DeletePartialMatch(prometheus.Labels{"user": userID}) m.convertBlockDuration.DeleteLabelValues(userID) } diff --git a/pkg/storage/parquet/no_convert_marker.go b/pkg/storage/parquet/no_convert_marker.go new file mode 100644 index 00000000000..9f48ff97511 --- /dev/null +++ b/pkg/storage/parquet/no_convert_marker.go @@ -0,0 +1,72 @@ +package parquet + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "path" + + "github.com/efficientgo/core/errors" + "github.com/go-kit/log" + "github.com/oklog/ulid/v2" + "github.com/thanos-io/objstore" + "github.com/thanos-io/thanos/pkg/runutil" + + "github.com/cortexproject/cortex/pkg/storage/bucket" +) + +const ( + NoConvertMarkerFileName = "parquet-no-convert-mark.json" + + CurrentNoConvertMarkVersion = NoConvertMarkVersion1 + NoConvertMarkVersion1 = 1 + + NoConvertReasonTooManyLabels = "too_many_labels" + NoConvertReasonMarkerExists = "marker_exists" +) + +type NoConvertMark struct { + Version int `json:"version"` + Reason string `json:"reason"` +} + +func ReadNoConvertMark(ctx context.Context, id ulid.ULID, userBkt objstore.InstrumentedBucket, logger log.Logger) (*NoConvertMark, error) { + markerPath := path.Join(id.String(), NoConvertMarkerFileName) + reader, err := userBkt.WithExpectedErrs(bucket.IsOneOfTheExpectedErrors(userBkt.IsAccessDeniedErr, userBkt.IsObjNotFoundErr)).Get(ctx, markerPath) + if err != nil { + if userBkt.IsObjNotFoundErr(err) || userBkt.IsAccessDeniedErr(err) { + return &NoConvertMark{}, nil + } + + return &NoConvertMark{}, err + } + defer runutil.CloseWithLogOnErr(logger, reader, "close parquet no-convert marker file reader") + + markerContent, err := io.ReadAll(reader) + if err != nil { + return &NoConvertMark{}, errors.Wrapf(err, "read file: %s", NoConvertMarkerFileName) + } + + marker := NoConvertMark{} + err = json.Unmarshal(markerContent, &marker) + return &marker, err +} + +func WriteNoConvertMark(ctx context.Context, id ulid.ULID, userBkt objstore.Bucket, labelNamesCount int, maxBlockLabelNames int) error { + noConvertMarker := NoConvertMark{ + Version: CurrentNoConvertMarkVersion, + Reason: fmt.Sprintf("%s: label_names_count=%d threshold=%d", NoConvertReasonTooManyLabels, labelNamesCount, maxBlockLabelNames), + } + noConvertMarkerPath := path.Join(id.String(), NoConvertMarkerFileName) + b, err := json.Marshal(noConvertMarker) + if err != nil { + return err + } + return userBkt.Upload(ctx, noConvertMarkerPath, bytes.NewReader(b)) +} + +func ValidNoConvertMarkVersion(version int) bool { + return version == NoConvertMarkVersion1 +} diff --git a/pkg/util/validation/exporter_test.go b/pkg/util/validation/exporter_test.go index 6067ed96067..1414d3f2bb5 100644 --- a/pkg/util/validation/exporter_test.go +++ b/pkg/util/validation/exporter_test.go @@ -101,6 +101,7 @@ func TestOverridesExporter_withConfig(t *testing.T) { cortex_overrides{limit_name="out_of_order_results_cache_ttl",user="tenant-a"} 0 cortex_overrides{limit_name="out_of_order_time_window",user="tenant-a"} 0 cortex_overrides{limit_name="parquet_converter_enabled",user="tenant-a"} 0 + cortex_overrides{limit_name="parquet_converter_max_block_label_names",user="tenant-a"} 0 cortex_overrides{limit_name="parquet_converter_tenant_shard_size",user="tenant-a"} 0 cortex_overrides{limit_name="parquet_max_fetched_chunk_bytes",user="tenant-a"} 0 cortex_overrides{limit_name="parquet_max_fetched_data_bytes",user="tenant-a"} 0 diff --git a/pkg/util/validation/limits.go b/pkg/util/validation/limits.go index 019a5adc3ed..fc37e4a4054 100644 --- a/pkg/util/validation/limits.go +++ b/pkg/util/validation/limits.go @@ -241,9 +241,10 @@ type Limits struct { CompactorPartitionSeriesCount int64 `yaml:"compactor_partition_series_count" json:"compactor_partition_series_count"` // Parquet converter - ParquetConverterEnabled bool `yaml:"parquet_converter_enabled" json:"parquet_converter_enabled"` - ParquetConverterTenantShardSize float64 `yaml:"parquet_converter_tenant_shard_size" json:"parquet_converter_tenant_shard_size"` - ParquetConverterSortColumns []string `yaml:"parquet_converter_sort_columns" json:"parquet_converter_sort_columns"` + ParquetConverterEnabled bool `yaml:"parquet_converter_enabled" json:"parquet_converter_enabled"` + ParquetConverterTenantShardSize float64 `yaml:"parquet_converter_tenant_shard_size" json:"parquet_converter_tenant_shard_size"` + ParquetConverterSortColumns []string `yaml:"parquet_converter_sort_columns" json:"parquet_converter_sort_columns"` + ParquetConverterMaxBlockLabelNames int `yaml:"parquet_converter_max_block_label_names" json:"parquet_converter_max_block_label_names"` // This config doesn't have a CLI flag registered here because they're registered in // their own original config struct. S3SSEType string `yaml:"s3_sse_type" json:"s3_sse_type" doc:"nocli|description=S3 server-side encryption type. Required to enable server-side encryption overrides for a specific tenant. If not set, the default S3 client settings are used."` @@ -369,6 +370,7 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) { f.Float64Var(&l.ParquetConverterTenantShardSize, "parquet-converter.tenant-shard-size", 0, "The default tenant's shard size when the shuffle-sharding strategy is used by the parquet converter. When this setting is specified in the per-tenant overrides, a value of 0 disables shuffle sharding for the tenant. If the value is < 1 and > 0 the shard size will be a percentage of the total parquet converters.") f.BoolVar(&l.ParquetConverterEnabled, "parquet-converter.enabled", false, "If set, enables the Parquet converter to create the parquet files.") f.Var((*flagext.StringSlice)(&l.ParquetConverterSortColumns), "parquet-converter.sort-columns", "Additional label names for specific tenants to sort by after metric name, in order of precedence. These are applied during Parquet file generation.") + f.IntVar(&l.ParquetConverterMaxBlockLabelNames, "parquet-converter.max-block-label-names", 0, "[Experimental] Maximum number of distinct label names allowed in a TSDB block for parquet conversion. If exceeded, the converter writes a no-convert marker. 0 to disable.") // Parquet Queryable enforced limits. f.IntVar(&l.ParquetMaxFetchedRowCount, "querier.parquet-queryable.max-fetched-row-count", 0, "The maximum number of rows that can be fetched when querying parquet storage. Each row maps to a series in a parquet file. This limit applies before materializing chunks. 0 to disable.") @@ -1029,6 +1031,11 @@ func (o *Overrides) ParquetConverterSortColumns(userID string) []string { return o.GetOverridesForUser(userID).ParquetConverterSortColumns } +// ParquetConverterMaxBlockLabelNames returns the maximum number of distinct label names allowed in a TSDB block for parquet conversion. +func (o *Overrides) ParquetConverterMaxBlockLabelNames(userID string) int { + return o.GetOverridesForUser(userID).ParquetConverterMaxBlockLabelNames +} + // ParquetMaxFetchedRowCount returns the maximum number of rows that can be fetched when querying parquet storage. func (o *Overrides) ParquetMaxFetchedRowCount(userID string) int { return o.GetOverridesForUser(userID).ParquetMaxFetchedRowCount From fe2a759a43a2ddc626eb6fae0be43acb799e8103 Mon Sep 17 00:00:00 2001 From: Siddarth Gundu Date: Tue, 16 Jun 2026 15:41:16 +0530 Subject: [PATCH 2/6] Update config docs for max-block-label-names Signed-off-by: Siddarth Gundu --- docs/configuration/config-file-reference.md | 6 ++++++ docs/configuration/v1-guarantees.md | 1 + schemas/cortex-config-schema.json | 6 ++++++ 3 files changed, 13 insertions(+) diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index b7922a94376..6e701d7304b 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -4923,6 +4923,12 @@ query_rejection: # CLI flag: -parquet-converter.sort-columns [parquet_converter_sort_columns: | default = []] +# [Experimental] Maximum number of distinct label names allowed in a TSDB block +# for parquet conversion. If exceeded, the converter writes a no-convert marker. +# 0 to disable. +# CLI flag: -parquet-converter.max-block-label-names +[parquet_converter_max_block_label_names: | default = 0] + # S3 server-side encryption type. Required to enable server-side encryption # overrides for a specific tenant. If not set, the default S3 client settings # are used. diff --git a/docs/configuration/v1-guarantees.md b/docs/configuration/v1-guarantees.md index a957e03c8b1..14ad6976b1f 100644 --- a/docs/configuration/v1-guarantees.md +++ b/docs/configuration/v1-guarantees.md @@ -157,3 +157,4 @@ Currently experimental features are: - Parquet Converter: Maximum number of columns per file - `-parquet-converter.max-num-columns` (int) CLI flag - Automatically shards parquet files when the number of columns exceeds the configured limit +- Parquet Converter: `-parquet-converter.max-block-label-names` (int) - If enabled, adds a no-convert mark and skips blocks with too many labels. diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index aed2998e063..7536029444d 100644 --- a/schemas/cortex-config-schema.json +++ b/schemas/cortex-config-schema.json @@ -5993,6 +5993,12 @@ "type": "boolean", "x-cli-flag": "parquet-converter.enabled" }, + "parquet_converter_max_block_label_names": { + "default": 0, + "description": "[Experimental] Maximum number of distinct label names allowed in a TSDB block for parquet conversion. If exceeded, the converter writes a no-convert marker. 0 to disable.", + "type": "number", + "x-cli-flag": "parquet-converter.max-block-label-names" + }, "parquet_converter_sort_columns": { "default": [], "description": "Additional label names for specific tenants to sort by after metric name, in order of precedence. These are applied during Parquet file generation.", From 63846e7105c1642dd59beebd7f2eaf407e779c30 Mon Sep 17 00:00:00 2001 From: Siddarth Gundu Date: Tue, 16 Jun 2026 17:30:02 +0530 Subject: [PATCH 3/6] parquetconverter: skip manually marked no-convert blocks The converter only read no-convert markers when the label-name limit was enabled, so manually marked blocks were still converted when the limit was 0. Read the marker unconditionally before conversion so these blocks stay skipped. Signed-off-by: Siddarth Gundu --- CHANGELOG.md | 1 + pkg/parquetconverter/converter.go | 33 ++++++++++++++----------------- 2 files changed, 16 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 16148eeae63..a3698171765 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ * [FEATURE] StoreGateway: Add experimental optional limit `blocks-storage.bucket-store.max-concurrent-data-bytes` on the data bytes (postings, series and chunks) fetched via the Series() API call and processed concurrently across all queries per store gateway to protect from oomkill. This returns an error that is retryable at querier level. #7271 * [ENHANCEMENT] Upgrade prometheus alertmanager version to v0.32.1. #7462 * [ENHANCEMENT] Tenant Federation: Avoid purging the regex resolver LRU cache on user-sync ticks when the set of known users has not changed. #7489 +* [ENHANCEMENT] Parquet Converter: Add `parquet-converter.max-block-label-names` limit to skip conversion of TSDB blocks with too many label names. #7625 * [ENHANCEMENT] Parquet Converter: Add a ring status page to expose the ring status. #7455 * [ENHANCEMENT] Parquet: Add `-blocks-storage.bucket-store.parquet-query-concurrency` flag to configure the maximum number of concurrent goroutines applied at each level of parquet query processing in store-gateway: shard querying, row group processing, and column materialization. #7613 * [ENHANCEMENT] Parquet: Add a row ranges cache for parquet query filtering in querier and store-gateway. #7478 diff --git a/pkg/parquetconverter/converter.go b/pkg/parquetconverter/converter.go index d6baad211bb..4b883b54a6a 100644 --- a/pkg/parquetconverter/converter.go +++ b/pkg/parquetconverter/converter.go @@ -430,26 +430,23 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin configuredMaxBlockLabelNames := c.limits.ParquetConverterMaxBlockLabelNames(userID) maxBlockLabelNames := effectiveMaxBlockLabelNames(configuredMaxBlockLabelNames, b.MinTime, b.MaxTime) - // If the threshold is enabled, check for no-convert mark - if configuredMaxBlockLabelNames > 0 { - noConvertMark, err := cortex_parquet.ReadNoConvertMark(ctx, b.ULID, uBucket, logger) - if err != nil { - level.Error(logger).Log("msg", "failed to read parquet no-convert marker", "block", b.ULID.String(), "err", err) - continue - } - if cortex_parquet.ValidNoConvertMarkVersion(noConvertMark.Version) { - level.Debug(logger).Log("msg", "skipping block, no-convert marker already exists", "block", b.ULID.String()) - c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonMarkerExists).Inc() - continue - } + noConvertMark, err := cortex_parquet.ReadNoConvertMark(ctx, b.ULID, uBucket, logger) + if err != nil { + level.Error(logger).Log("msg", "failed to read parquet no-convert marker", "block", b.ULID.String(), "err", err) + continue + } + if cortex_parquet.ValidNoConvertMarkVersion(noConvertMark.Version) { + level.Debug(logger).Log("msg", "skipping block, no-convert marker already exists", "block", b.ULID.String()) + c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonMarkerExists).Inc() + continue + } - if err := os.RemoveAll(c.compactRootDir()); err != nil { - level.Error(logger).Log("msg", "failed to remove work directory", "path", c.compactRootDir(), "err", err) - if c.checkConvertError(userID, err) { - return err - } - continue + if err := os.RemoveAll(c.compactRootDir()); err != nil { + level.Error(logger).Log("msg", "failed to remove work directory", "path", c.compactRootDir(), "err", err) + if c.checkConvertError(userID, err) { + return err } + continue } bdir := filepath.Join(c.compactDirForUser(userID), b.ULID.String()) From 93979192b9740f276f4fd6fe334a4cf05ee984bd Mon Sep 17 00:00:00 2001 From: Siddarth Gundu Date: Sun, 28 Jun 2026 00:42:31 +0530 Subject: [PATCH 4/6] parquetconverter: track limits in no-convert Retry conversion if the current limit has increased beyond the label count stored in the old no-convert mark. Update converter tests for the new marker fields and retry behavior Signed-off-by: Siddarth Gundu --- pkg/parquetconverter/converter.go | 4 +- pkg/parquetconverter/converter_test.go | 73 ++++++++++++++++-------- pkg/storage/parquet/no_convert_marker.go | 28 +++++++-- 3 files changed, 76 insertions(+), 29 deletions(-) diff --git a/pkg/parquetconverter/converter.go b/pkg/parquetconverter/converter.go index 4b883b54a6a..8f713163fc3 100644 --- a/pkg/parquetconverter/converter.go +++ b/pkg/parquetconverter/converter.go @@ -435,7 +435,9 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin level.Error(logger).Log("msg", "failed to read parquet no-convert marker", "block", b.ULID.String(), "err", err) continue } - if cortex_parquet.ValidNoConvertMarkVersion(noConvertMark.Version) { + + // Retry conversion if the current limit has increased beyond the label count stored in the old no-convert mark + if cortex_parquet.ValidNoConvertMarkVersion(noConvertMark.Version) && noConvertMark.ShouldSkipBlock(maxBlockLabelNames) { level.Debug(logger).Log("msg", "skipping block, no-convert marker already exists", "block", b.ULID.String()) c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonMarkerExists).Inc() continue diff --git a/pkg/parquetconverter/converter_test.go b/pkg/parquetconverter/converter_test.go index e0c03906b8c..766788e43a6 100644 --- a/pkg/parquetconverter/converter_test.go +++ b/pkg/parquetconverter/converter_test.go @@ -786,14 +786,16 @@ func TestConverter_WriteNoConvertMarkForBlockWithTooManyLabels(t *testing.T) { readNoConvertMark, err := parquet.ReadNoConvertMark(ctx, blockID, userBucket, logger) require.NoError(t, err) require.True(t, parquet.ValidNoConvertMarkVersion(readNoConvertMark.Version)) - require.Equal(t, fmt.Sprintf("%s: label_names_count=%d threshold=%d", parquet.NoConvertReasonTooManyLabels, 2, 1), readNoConvertMark.Reason) + require.Equal(t, parquet.NoConvertReasonTooManyLabels, readNoConvertMark.Reason) + require.Equal(t, 2, readNoConvertMark.LabelNamesCount) + require.Equal(t, 1, readNoConvertMark.MaxBlockLabelNames) // Confirm conversion did not happen assert.Equal(t, 0.0, testutil.ToFloat64(c.metrics.convertedBlocks.WithLabelValues(user))) assert.Equal(t, 1.0, testutil.ToFloat64(c.metrics.skippedBlocks.WithLabelValues(user, parquet.NoConvertReasonTooManyLabels))) } -func TestConverter_SkipBlockWhenNoConvertMarkAlreadyExists(t *testing.T) { +func TestConverter_NoConvertMarkHandling(t *testing.T) { cfg := prepareConfig() user := "user" ringStore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil) @@ -809,51 +811,76 @@ func TestConverter_SkipBlockWhenNoConvertMarkAlreadyExists(t *testing.T) { limits := &validation.Limits{} flagext.DefaultValues(limits) limits.ParquetConverterEnabled = true + limits.ParquetConverterMaxBlockLabelNames = 3 c, logger, _ := prepare(t, cfg, objstore.WithNoopInstr(bucketClient), limits, nil) ctx := context.Background() - lbls := labels.FromStrings("__name__", "test", "job", "foo") rnd := rand.New(rand.NewSource(time.Now().Unix())) - // 2h blocks are skipped by ShouldConvertBlockToParquet - blockID, err := e2e.CreateBlock(ctx, rnd, dir, []labels.Labels{lbls}, 2, 0, - 4*time.Hour.Milliseconds(), time.Minute.Milliseconds(), 10) - require.NoError(t, err) + createAndUploadBlock := func(mint, maxt int64) ulid.ULID { + t.Helper() - blockDir := fmt.Sprintf("%s/%s", dir, blockID.String()) - b, err := tsdb.OpenBlock(nil, blockDir, nil, nil) - require.NoError(t, err) - err = block.Upload(ctx, logger, userBucket, b.Dir(), metadata.NoneFunc) - require.NoError(t, err) + blockID, err := e2e.CreateBlock(ctx, rnd, dir, []labels.Labels{lbls}, 2, mint, maxt, time.Minute.Milliseconds(), 10) + require.NoError(t, err) + + blockDir := fmt.Sprintf("%s/%s", dir, blockID.String()) + b, err := tsdb.OpenBlock(nil, blockDir, nil, nil) + require.NoError(t, err) + err = block.Upload(ctx, logger, userBucket, b.Dir(), metadata.NoneFunc) + require.NoError(t, err) + + return blockID + } + + manuallyMarkedBlockID := createAndUploadBlock(0, 4*time.Hour.Milliseconds()) + limitIncreasedBlockID := createAndUploadBlock(4*time.Hour.Milliseconds(), 8*time.Hour.Milliseconds()) - markerV1 := parquet.NoConvertMark{ + writeNoConvertMark := func(blockID ulid.ULID, noConvertMark parquet.NoConvertMark) { + t.Helper() + + markerBytes, err := json.Marshal(noConvertMark) + require.NoError(t, err) + markerPath := path.Join(blockID.String(), parquet.NoConvertMarkerFileName) + err = userBucket.Upload(ctx, markerPath, bytes.NewReader(markerBytes)) + require.NoError(t, err) + } + + writeNoConvertMark(manuallyMarkedBlockID, parquet.NoConvertMark{ Version: parquet.CurrentNoConvertMarkVersion, Reason: "manually uploaded", - } - markerBytes, err := json.Marshal(markerV1) - require.NoError(t, err) - markerPath := path.Join(blockID.String(), parquet.NoConvertMarkerFileName) - err = userBucket.Upload(ctx, markerPath, bytes.NewReader(markerBytes)) - require.NoError(t, err) + }) + writeNoConvertMark(limitIncreasedBlockID, parquet.NoConvertMark{ + Version: parquet.CurrentNoConvertMarkVersion, + Reason: parquet.NoConvertReasonTooManyLabels, + LabelNamesCount: 2, + MaxBlockLabelNames: 1, + }) err = services.StartAndAwaitRunning(context.Background(), c) require.NoError(t, err) defer services.StopAndAwaitTerminated(ctx, c) // nolint:errcheck - // start converter err = c.convertUser(ctx, logger, c.ring, user) require.NoError(t, err) - // confirm conversion was skipped - assert.Equal(t, 0.0, testutil.ToFloat64(c.metrics.convertedBlocks.WithLabelValues(user))) + assert.Equal(t, 1.0, testutil.ToFloat64(c.metrics.convertedBlocks.WithLabelValues(user))) assert.Equal(t, 1.0, testutil.ToFloat64(c.metrics.skippedBlocks.WithLabelValues(user, parquet.NoConvertReasonMarkerExists))) + assert.Equal(t, 0.0, testutil.ToFloat64(c.metrics.skippedBlocks.WithLabelValues(user, parquet.NoConvertReasonTooManyLabels))) - markerAfter, err := parquet.ReadNoConvertMark(ctx, blockID, userBucket, logger) + markerAfter, err := parquet.ReadNoConvertMark(ctx, manuallyMarkedBlockID, userBucket, logger) require.NoError(t, err) require.True(t, parquet.ValidNoConvertMarkVersion(markerAfter.Version)) require.Equal(t, "manually uploaded", markerAfter.Reason) + + converterMark, err := parquet.ReadConverterMark(ctx, manuallyMarkedBlockID, userBucket, logger) + require.NoError(t, err) + require.False(t, parquet.ValidConverterMarkVersion(converterMark.Version)) + + converterMark, err = parquet.ReadConverterMark(ctx, limitIncreasedBlockID, userBucket, logger) + require.NoError(t, err) + require.True(t, parquet.ValidConverterMarkVersion(converterMark.Version)) } func TestEffectiveMaxBlockLabelNamesLeavesRoomForGeneratedColumns(t *testing.T) { diff --git a/pkg/storage/parquet/no_convert_marker.go b/pkg/storage/parquet/no_convert_marker.go index 9f48ff97511..0e879c7375c 100644 --- a/pkg/storage/parquet/no_convert_marker.go +++ b/pkg/storage/parquet/no_convert_marker.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "encoding/json" - "fmt" "io" "path" @@ -28,8 +27,10 @@ const ( ) type NoConvertMark struct { - Version int `json:"version"` - Reason string `json:"reason"` + Version int `json:"version"` + Reason string `json:"reason"` + LabelNamesCount int `json:"label_names_count,omitempty"` + MaxBlockLabelNames int `json:"max_block_label_names,omitempty"` } func ReadNoConvertMark(ctx context.Context, id ulid.ULID, userBkt objstore.InstrumentedBucket, logger log.Logger) (*NoConvertMark, error) { @@ -56,8 +57,10 @@ func ReadNoConvertMark(ctx context.Context, id ulid.ULID, userBkt objstore.Instr func WriteNoConvertMark(ctx context.Context, id ulid.ULID, userBkt objstore.Bucket, labelNamesCount int, maxBlockLabelNames int) error { noConvertMarker := NoConvertMark{ - Version: CurrentNoConvertMarkVersion, - Reason: fmt.Sprintf("%s: label_names_count=%d threshold=%d", NoConvertReasonTooManyLabels, labelNamesCount, maxBlockLabelNames), + Version: CurrentNoConvertMarkVersion, + Reason: NoConvertReasonTooManyLabels, + LabelNamesCount: labelNamesCount, + MaxBlockLabelNames: maxBlockLabelNames, } noConvertMarkerPath := path.Join(id.String(), NoConvertMarkerFileName) b, err := json.Marshal(noConvertMarker) @@ -70,3 +73,18 @@ func WriteNoConvertMark(ctx context.Context, id ulid.ULID, userBkt objstore.Buck func ValidNoConvertMarkVersion(version int) bool { return version == NoConvertMarkVersion1 } + +func (m NoConvertMark) ShouldSkipBlock(currentMaxBlockLabelNamesLimit int) bool { + // Manual no-convert marks are not tied to the label-name limit + if m.Reason != NoConvertReasonTooManyLabels { + return true + } + + // limit=0 means the label-name guard is disabled, + if currentMaxBlockLabelNamesLimit <= 0 { + return false + } + + // m.LabelNamesCount is recorded when the old no-convert marker was written + return currentMaxBlockLabelNamesLimit < m.LabelNamesCount +} From b40cdb386be5ccbb69b9b40730d44132dfbe8536 Mon Sep 17 00:00:00 2001 From: Siddarth Gundu Date: Mon, 29 Jun 2026 14:46:15 +0530 Subject: [PATCH 5/6] parquetconverter: split no-convert skip reasons - Update tests to check skip with lower current limit Signed-off-by: Siddarth Gundu --- pkg/parquetconverter/converter.go | 17 ++++++++++++----- pkg/parquetconverter/converter_test.go | 13 ++++++++++++- pkg/storage/parquet/no_convert_marker.go | 4 ---- 3 files changed, 24 insertions(+), 10 deletions(-) diff --git a/pkg/parquetconverter/converter.go b/pkg/parquetconverter/converter.go index 8f713163fc3..3f72a78af29 100644 --- a/pkg/parquetconverter/converter.go +++ b/pkg/parquetconverter/converter.go @@ -436,11 +436,18 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin continue } - // Retry conversion if the current limit has increased beyond the label count stored in the old no-convert mark - if cortex_parquet.ValidNoConvertMarkVersion(noConvertMark.Version) && noConvertMark.ShouldSkipBlock(maxBlockLabelNames) { - level.Debug(logger).Log("msg", "skipping block, no-convert marker already exists", "block", b.ULID.String()) - c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonMarkerExists).Inc() - continue + if cortex_parquet.ValidNoConvertMarkVersion(noConvertMark.Version) { + if noConvertMark.Reason != cortex_parquet.NoConvertReasonTooManyLabels { + level.Debug(logger).Log("msg", "skipping block, no-convert marker already exists", "block", b.ULID.String()) + c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonMarkerExists).Inc() + continue + } + + if noConvertMark.ShouldSkipBlock(maxBlockLabelNames) { + level.Debug(logger).Log("msg", "skipping block because label count still exceeds current limit", "block", b.ULID.String(), "label_names_count", noConvertMark.LabelNamesCount, "current_limit", maxBlockLabelNames) + c.metrics.skippedBlocks.WithLabelValues(userID, cortex_parquet.NoConvertReasonTooManyLabels).Inc() + continue + } } if err := os.RemoveAll(c.compactRootDir()); err != nil { diff --git a/pkg/parquetconverter/converter_test.go b/pkg/parquetconverter/converter_test.go index 766788e43a6..77ad7cfcc60 100644 --- a/pkg/parquetconverter/converter_test.go +++ b/pkg/parquetconverter/converter_test.go @@ -836,6 +836,7 @@ func TestConverter_NoConvertMarkHandling(t *testing.T) { manuallyMarkedBlockID := createAndUploadBlock(0, 4*time.Hour.Milliseconds()) limitIncreasedBlockID := createAndUploadBlock(4*time.Hour.Milliseconds(), 8*time.Hour.Milliseconds()) + stillTooManyLabelsBlockID := createAndUploadBlock(8*time.Hour.Milliseconds(), 12*time.Hour.Milliseconds()) writeNoConvertMark := func(blockID ulid.ULID, noConvertMark parquet.NoConvertMark) { t.Helper() @@ -857,6 +858,12 @@ func TestConverter_NoConvertMarkHandling(t *testing.T) { LabelNamesCount: 2, MaxBlockLabelNames: 1, }) + writeNoConvertMark(stillTooManyLabelsBlockID, parquet.NoConvertMark{ + Version: parquet.CurrentNoConvertMarkVersion, + Reason: parquet.NoConvertReasonTooManyLabels, + LabelNamesCount: 4, + MaxBlockLabelNames: 1, + }) err = services.StartAndAwaitRunning(context.Background(), c) require.NoError(t, err) @@ -867,7 +874,7 @@ func TestConverter_NoConvertMarkHandling(t *testing.T) { assert.Equal(t, 1.0, testutil.ToFloat64(c.metrics.convertedBlocks.WithLabelValues(user))) assert.Equal(t, 1.0, testutil.ToFloat64(c.metrics.skippedBlocks.WithLabelValues(user, parquet.NoConvertReasonMarkerExists))) - assert.Equal(t, 0.0, testutil.ToFloat64(c.metrics.skippedBlocks.WithLabelValues(user, parquet.NoConvertReasonTooManyLabels))) + assert.Equal(t, 1.0, testutil.ToFloat64(c.metrics.skippedBlocks.WithLabelValues(user, parquet.NoConvertReasonTooManyLabels))) markerAfter, err := parquet.ReadNoConvertMark(ctx, manuallyMarkedBlockID, userBucket, logger) require.NoError(t, err) @@ -881,6 +888,10 @@ func TestConverter_NoConvertMarkHandling(t *testing.T) { converterMark, err = parquet.ReadConverterMark(ctx, limitIncreasedBlockID, userBucket, logger) require.NoError(t, err) require.True(t, parquet.ValidConverterMarkVersion(converterMark.Version)) + + converterMark, err = parquet.ReadConverterMark(ctx, stillTooManyLabelsBlockID, userBucket, logger) + require.NoError(t, err) + require.False(t, parquet.ValidConverterMarkVersion(converterMark.Version)) } func TestEffectiveMaxBlockLabelNamesLeavesRoomForGeneratedColumns(t *testing.T) { diff --git a/pkg/storage/parquet/no_convert_marker.go b/pkg/storage/parquet/no_convert_marker.go index 0e879c7375c..4f0d4185828 100644 --- a/pkg/storage/parquet/no_convert_marker.go +++ b/pkg/storage/parquet/no_convert_marker.go @@ -75,10 +75,6 @@ func ValidNoConvertMarkVersion(version int) bool { } func (m NoConvertMark) ShouldSkipBlock(currentMaxBlockLabelNamesLimit int) bool { - // Manual no-convert marks are not tied to the label-name limit - if m.Reason != NoConvertReasonTooManyLabels { - return true - } // limit=0 means the label-name guard is disabled, if currentMaxBlockLabelNamesLimit <= 0 { From 6674a96824969e1dd8b5c95311e4b1e8e3da41ad Mon Sep 17 00:00:00 2001 From: Siddarth Gundu Date: Sun, 6 Sep 2026 14:55:47 +0530 Subject: [PATCH 6/6] parquetconverter: use requires_docker build tag so integration test runs in CI Signed-off-by: Siddarth Gundu --- integration/parquet_converter_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/integration/parquet_converter_test.go b/integration/parquet_converter_test.go index 4a10d3f1e0f..b8ef4b4fd76 100644 --- a/integration/parquet_converter_test.go +++ b/integration/parquet_converter_test.go @@ -1,4 +1,4 @@ -//go:build integration +//go:build requires_docker package integration