parquet: support sharded parquet file conversion

Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>

Author: SungJin1212

docs/configuration/config-file-reference.md

@@ -186,6 +186,13 @@ parquet_converter:
   # CLI flag: -parquet-converter.max-rows-per-row-group
   [max_rows_per_row_group: <int> | default = 1000000]
 
+  # Maximum number of row groups per parquet shard. Each shard holds at most
+  # num-row-groups * max-rows-per-row-group series, so lowering this value
+  # splits a block into more parquet shards for better read parallelization.
+  # Default is unlimited (single shard).
+  # CLI flag: -parquet-converter.num-row-groups
+  [num_row_groups: <int> | default = 2147483647]
+
   # Enable disk-based write buffering to reduce memory consumption during
   # parquet file generation.
   # CLI flag: -parquet-converter.file-buffer-enabled

pkg/parquetconverter/converter.go

@@ -5,6 +5,7 @@ import (
 	"flag"
 	"fmt"
 	"hash/fnv"
+	"math"
 	"math/rand"
 	"net/http"
 	"os"
@@ -57,6 +58,7 @@ type Config struct {
 	MetaSyncConcurrency int           `yaml:"meta_sync_concurrency"`
 	ConversionInterval  time.Duration `yaml:"conversion_interval"`
 	MaxRowsPerRowGroup  int           `yaml:"max_rows_per_row_group"`
+	NumRowGroups        int           `yaml:"num_row_groups"`
 	FileBufferEnabled   bool          `yaml:"file_buffer_enabled"`
 
 	DataDir string `yaml:"data_dir"`
@@ -107,6 +109,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.StringVar(&cfg.DataDir, "parquet-converter.data-dir", "./data", "Local directory path for caching TSDB blocks during parquet conversion.")
 	f.IntVar(&cfg.MetaSyncConcurrency, "parquet-converter.meta-sync-concurrency", 20, "Maximum concurrent goroutines for downloading block metadata from object storage.")
 	f.IntVar(&cfg.MaxRowsPerRowGroup, "parquet-converter.max-rows-per-row-group", 1e6, "Maximum number of time series per parquet row group. Larger values improve compression but may reduce performance during reads.")
+	f.IntVar(&cfg.NumRowGroups, "parquet-converter.num-row-groups", math.MaxInt32, "Maximum number of row groups per parquet shard. Each shard holds at most num-row-groups * max-rows-per-row-group series, so lowering this value splits a block into more parquet shards for better read parallelization. Default is unlimited (single shard).")
 	f.DurationVar(&cfg.ConversionInterval, "parquet-converter.conversion-interval", time.Minute, "How often to check for new TSDB blocks to convert to parquet format.")
 	f.BoolVar(&cfg.FileBufferEnabled, "parquet-converter.file-buffer-enabled", true, "Enable disk-based write buffering to reduce memory consumption during parquet file generation.")
 }
@@ -126,6 +129,13 @@ func NewConverter(cfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, blockR
 }
 
 func newConverter(cfg Config, bkt objstore.InstrumentedBucket, storageCfg cortex_tsdb.BlocksStorageConfig, blockRanges []int64, logger log.Logger, registerer prometheus.Registerer, limits *validation.Overrides, usersScanner users.Scanner) *Converter {
+	// A non-positive number of row groups is invalid and would lead to a division by zero
+	// while sharding the block.
+	numRowGroups := cfg.NumRowGroups
+	if numRowGroups <= 0 {
+		numRowGroups = math.MaxInt32
+	}
+
 	c := &Converter{
 		cfg:            cfg,
 		reg:            registerer,
@@ -141,6 +151,7 @@ func newConverter(cfg Config, bkt objstore.InstrumentedBucket, storageCfg cortex
 		baseConverterOptions: []convert.ConvertOption{
 			convert.WithColDuration(time.Hour * 8),
 			convert.WithRowGroupSize(cfg.MaxRowsPerRowGroup),
+			convert.WithNumRowGroups(numRowGroups),
 		},
 	}
 

pkg/parquetconverter/converter_test.go

@@ -173,6 +173,85 @@ func prepareConfig() Config {
 	return cfg
 }
 
+func TestConverter_SplitsBlockIntoMultipleShards(t *testing.T) {
+	cfg := prepareConfig()
+	// Configure the converter so that each parquet shard holds at most
+	// numRowGroups * maxRowsPerRowGroup = 1 * 2 = 2 series.
+	cfg.NumRowGroups = 1
+	cfg.MaxRowsPerRowGroup = 2
+
+	user := "user-1"
+	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()
+
+	// Create 5 unique series so that the block is split into
+	// ceil(5 / 2) = 3 parquet shards.
+	const numSeries = 5
+	const expectedShards = 3
+	series := make([]labels.Labels, 0, numSeries)
+	for i := range numSeries {
+		series = append(series, labels.FromStrings("__name__", "test", "series", fmt.Sprintf("%d", i)))
+	}
+
+	// Create and upload a 24h block. It must be larger than the first configured
+	// block range (2h) so that the converter does not skip it as a raw TSDB block.
+	rnd := rand.New(rand.NewSource(time.Now().Unix()))
+	blockID, err := e2e.CreateBlock(ctx, rnd, dir, series, 2, 0, 24*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)
+
+	// Start the converter.
+	err = services.StartAndAwaitRunning(context.Background(), c)
+	require.NoError(t, err)
+	defer services.StopAndAwaitTerminated(ctx, c) // nolint:errcheck
+
+	// Wait until the block is converted and assert it was split into multiple shards.
+	test.Poll(t, 3*time.Minute, expectedShards, func() any {
+		m, err := parquet.ReadConverterMark(ctx, blockID, userBucket, logger)
+		require.NoError(t, err)
+		if m.Version != parquet.CurrentVersion {
+			return -1
+		}
+		return m.Shards
+	})
+
+	// Verify that one labels/chunks parquet file exists per shard.
+	for shard := range expectedShards {
+		for _, file := range []string{
+			fmt.Sprintf("%s/%d.chunks.parquet", blockID.String(), shard),
+			fmt.Sprintf("%s/%d.labels.parquet", blockID.String(), shard),
+		} {
+			ok, err := userBucket.Exists(ctx, file)
+			require.NoError(t, err)
+			require.True(t, ok, "expected shard file %s to exist", file)
+		}
+	}
+
+	// Verify there is no extra shard beyond the expected count.
+	ok, err := userBucket.Exists(ctx, fmt.Sprintf("%s/%d.chunks.parquet", blockID.String(), expectedShards))
+	require.NoError(t, err)
+	require.False(t, ok, "expected no shard file at index %d", expectedShards)
+}
+
 func prepare(t *testing.T, cfg Config, bucketClient objstore.InstrumentedBucket, limits *validation.Limits, tenantLimits validation.TenantLimits) (*Converter, log.Logger, prometheus.Gatherer) {
 	storageCfg := cortex_tsdb.BlocksStorageConfig{}
 	blockRanges := cortex_tsdb.DurationList{2 * time.Hour, 12 * time.Hour, 24 * time.Hour}

schemas/cortex-config-schema.json

@@ -9112,6 +9112,12 @@
           "type": "number",
           "x-cli-flag": "parquet-converter.meta-sync-concurrency"
         },
+        "num_row_groups": {
+          "default": 2147483647,
+          "description": "Maximum number of row groups per parquet shard. Each shard holds at most num-row-groups * max-rows-per-row-group series, so lowering this value splits a block into more parquet shards for better read parallelization. Default is unlimited (single shard).",
+          "type": "number",
+          "x-cli-flag": "parquet-converter.num-row-groups"
+        },
         "ring": {
           "properties": {
             "auto_forget_delay": {