feat: cross-cluster ClickHouse support with configurable distributed table suffix

In multi-cluster ClickHouse deployments (multi-region, read/write
separation), queries need to fan out across all clusters through a
virtual cluster (e.g. "all"). This requires a second set of distributed
tables with skip_unavailable_shards=1, using a different suffix and
cluster name than the write-path tables.

Changes:

1. Configurable distributed table suffix (CLICKHOUSE_DIST_SUFFIX):
   - New shared/distconfig package reads the suffix at startup
   - All hardcoded "_dist" replaced with distconfig.Suffix()
   - InitDistTableNames() re-registers table name mappings after init
   - Default remains "_dist" for full backward compatibility

2. Cross-cluster read-path DDL (CLICKHOUSE_READ_CLUSTER):
   - When set to a different cluster than CLUSTER_NAME, the DDL phase
     creates additional distributed tables for cross-cluster reads
   - New SQL templates: log_read_dist.sql, traces_read_dist.sql,
     profiles_read_dist.sql (version keys 7, 8, 9)
   - All read tables include SETTINGS skip_unavailable_shards = 1

3. Status attribute mapping:
   - TraceQL: { status = ok } now queries otel.status_code
   - Tempo tag values API: /api/search/tag/status/values maps correctly

4. Bug fix: leading space in health check table name " time_series_dist"

Example multi-cluster configuration:
  CLUSTER_NAME=my_cluster              # write cluster
  CLICKHOUSE_READ_CLUSTER=all          # read cluster spanning all clusters
  CLICKHOUSE_DIST_SUFFIX=_all          # suffix for read-path tables

Closes #780

Author: Cluas

cmd/gigapipe/main.go

@@ -11,6 +11,8 @@ import (
 	"github.com/metrico/qryn/v4/ctrl"
 	"github.com/metrico/qryn/v4/reader"
 	"github.com/metrico/qryn/v4/reader/utils/logger"
+	"github.com/metrico/qryn/v4/reader/utils/tables"
+	"github.com/metrico/qryn/v4/shared/distconfig"
 	"github.com/metrico/qryn/v4/reader/utils/middleware"
 	"github.com/metrico/qryn/v4/shared/commonroutes"
 	"github.com/metrico/qryn/v4/view"
@@ -270,6 +272,8 @@ func main() {
 }
 
 func start() {
+	distconfig.Init()
+	tables.InitDistTableNames()
 	var configPaths []string
 	if _, err := os.Stat(*appFlags.ConfigPath); err == nil {
 		configPaths = append(configPaths, *appFlags.ConfigPath)

ctrl/qryn/maintenance/maintain.go

@@ -3,6 +3,7 @@ package maintenance
 import (
 	"context"
 	"fmt"
+	"os"
 	"strings"
 	"time"
 
@@ -27,8 +28,13 @@ func upgradeDB(dbObject *config.ClokiBaseDataBase, logger logger.ILogger) error
 	if dbObject.TTLDays == 0 {
 		return fmt.Errorf("ttl_days should be set for node#%s", dbObject.Node)
 	}
-	return Update(conn, dbObject.Name, dbObject.ClusterName, mode, dbObject.TTLDays,
-		dbObject.StoragePolicy, dbObject.SamplesOrdering, dbObject.SkipUnavailableShards, logger)
+	readCluster := os.Getenv("CLICKHOUSE_READ_CLUSTER")
+	readSuffix := os.Getenv("CLICKHOUSE_DIST_SUFFIX")
+	if readSuffix == "" {
+		readSuffix = "_dist"
+	}
+	return UpdateWithReadCluster(conn, dbObject.Name, dbObject.ClusterName, readCluster, readSuffix, mode,
+		dbObject.TTLDays, dbObject.StoragePolicy, dbObject.SamplesOrdering, dbObject.SkipUnavailableShards, logger)
 }
 
 func InitDB(dbObject *config.ClokiBaseDataBase, logger logger.ILogger) error {

ctrl/qryn/maintenance/rotate.go

@@ -10,6 +10,7 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/metrico/qryn/v4/ctrl/logger"
 	"github.com/metrico/qryn/v4/ctrl/qryn/helputils"
+	"github.com/metrico/qryn/v4/shared/distconfig"
 )
 
 func getSetting(db clickhouse.Conn, dist bool, tp string, name string) (string, error) {
@@ -18,7 +19,7 @@ func getSetting(db clickhouse.Conn, dist bool, tp string, name string) (string,
 	)
 	settings := "settings"
 	if dist {
-		settings += "_dist"
+		settings += distconfig.Suffix()
 	}
 	rows, err := db.Query(context.Background(),
 		fmt.Sprintf(`SELECT argMax(value, inserted_at) as _value FROM %s WHERE fingerprint = $1 

ctrl/qryn/maintenance/update.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/metrico/qryn/v4/ctrl/logger"
 	"github.com/metrico/qryn/v4/ctrl/qryn/sql"
+	"github.com/metrico/qryn/v4/shared/distconfig"
 )
 
 const (
@@ -23,6 +24,14 @@ const (
 )
 
 func Update(db clickhouse.Conn, dbname string, clusterName string, mode int,
+	ttlDays int, storagePolicy string, advancedSamplesOrdering string, skipUnavailableShards bool,
+	logger logger.ILogger) error {
+	return UpdateWithReadCluster(db, dbname, clusterName, "", "", mode,
+		ttlDays, storagePolicy, advancedSamplesOrdering, skipUnavailableShards, logger)
+}
+
+func UpdateWithReadCluster(db clickhouse.Conn, dbname string, clusterName string,
+	readCluster string, readSuffix string, mode int,
 	ttlDays int, storagePolicy string, advancedSamplesOrdering string, skipUnavailableShards bool,
 	logger logger.ILogger) error {
 	checkMode := func(m int) bool { return mode&m == m }
@@ -65,6 +74,26 @@ func Update(db clickhouse.Conn, dbname string, clusterName string, mode int,
 		}
 	}
 
+	// Cross-cluster read-path tables: when a separate read cluster is configured,
+	// create distributed tables that aggregate queries across multiple clusters.
+	if readCluster != "" && readCluster != clusterName && checkMode(CLUST_MODE_DISTRIBUTED) {
+		err = updateReadDistScripts(db, dbname, clusterName, readCluster, readSuffix,
+			7, sql.LogReadDistScript, logger)
+		if err != nil {
+			return err
+		}
+		err = updateReadDistScripts(db, dbname, clusterName, readCluster, readSuffix,
+			8, sql.TracesReadDistScript, logger)
+		if err != nil {
+			return err
+		}
+		err = updateReadDistScripts(db, dbname, clusterName, readCluster, readSuffix,
+			9, sql.ProfilesReadDistScript, logger)
+		if err != nil {
+			return err
+		}
+	}
+
 	err = Cleanup(db, clusterName, checkMode(CLUST_MODE_DISTRIBUTED), dbname, logger)
 
 	return err
@@ -110,6 +139,51 @@ func getDBExec(db clickhouse.Conn, env map[string]string, logger logger.ILogger)
 	}
 }
 
+func updateReadDistScripts(db clickhouse.Conn, dbname string, clusterName string,
+	readCluster string, readSuffix string, k int64, file string, logger logger.ILogger) error {
+	scripts, err := getSQLFile(file)
+	if err != nil {
+		return err
+	}
+	env := map[string]string{
+		"DB":           dbname,
+		"CLUSTER":      clusterName,
+		"OnCluster":    "ON CLUSTER `" + clusterName + "`",
+		"READ_CLUSTER": readCluster,
+		"READ_SUFFIX":  readSuffix,
+	}
+	exec := getDBExec(db, env, logger)
+	verTable := "ver" + distconfig.Suffix()
+	var ver uint64 = 0
+	if k >= 0 {
+		rows, err := db.Query(context.Background(),
+			fmt.Sprintf("SELECT max(ver) as ver FROM %s WHERE k = $1 FORMAT JSON", verTable), k)
+		if err != nil {
+			return err
+		}
+		for rows.Next() {
+			err = rows.Scan(&ver)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	for i := ver; i < uint64(len(scripts)); i++ {
+		logger.Info(fmt.Sprintf("Upgrade read-dist v.%d to v.%d ", i, i+1))
+		err = exec(scripts[i])
+		if err != nil {
+			logger.Error(scripts[i])
+			return err
+		}
+		err = db.Exec(context.Background(), "INSERT INTO ver (k, ver) VALUES ($1, $2)", k, i+1)
+		if err != nil {
+			return err
+		}
+		logger.Info(fmt.Sprintf("Upgrade read-dist v.%d to v.%d ok", i, i+1))
+	}
+	return nil
+}
+
 func updateScripts(db clickhouse.Conn, dbname string, clusterName string, k int64, file string, replicated bool,
 	ttlDays int, storagePolicy string, advancedSamplesOrdering string, skipUnavailableShards bool, logger logger.ILogger) error {
 	scripts, err := getSQLFile(file)
@@ -165,7 +239,7 @@ ENGINE=Distributed('{{.CLUSTER}}','{{.DB}}', 'ver', rand())`)
 		if err != nil {
 			return err
 		}
-		verTable = "ver_dist"
+		verTable = "ver" + distconfig.Suffix()
 	}
 	var ver uint64 = 0
 	if k >= 0 {

ctrl/qryn/sql/log_read_dist.sql

@@ -0,0 +1,75 @@
+## Cross-cluster read-path distributed tables for logs/metrics
+## Used when CLICKHOUSE_READ_CLUSTER is set (multi-cluster query aggregation)
+## APPEND ONLY!!!!!
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.metrics_15s{{.READ_SUFFIX}} {{.OnCluster}} (
+    `fingerprint` UInt64,
+    `timestamp_ns` Int64 CODEC(DoubleDelta),
+    `last` AggregateFunction(argMax, Float64, Int64),
+    `max` SimpleAggregateFunction(max, Float64),
+    `min` SimpleAggregateFunction(min, Float64),
+    `count` AggregateFunction(count),
+    `sum` SimpleAggregateFunction(sum, Float64),
+    `bytes` SimpleAggregateFunction(sum, Float64)
+) ENGINE = Distributed('{{.READ_CLUSTER}}', '{{.DB}}', 'metrics_15s', fingerprint) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.samples_v3{{.READ_SUFFIX}} {{.OnCluster}} (
+    `fingerprint` UInt64,
+    `timestamp_ns` Int64 CODEC(DoubleDelta),
+    `value` Float64 CODEC(Gorilla),
+    `string` String
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'samples_v3', fingerprint) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.time_series{{.READ_SUFFIX}} {{.OnCluster}} (
+    `date` Date,
+    `fingerprint` UInt64,
+    `labels` String,
+    `name` String
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'time_series', fingerprint) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.settings{{.READ_SUFFIX}} {{.OnCluster}} (
+    `fingerprint` UInt64,
+    `type` String,
+    `name` String,
+    `value` String,
+    `inserted_at` DateTime64(9, 'UTC')
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'settings', rand()) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.time_series_gin{{.READ_SUFFIX}} {{.OnCluster}} (
+    date Date,
+    key String,
+    val String,
+    fingerprint UInt64
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'time_series_gin', rand()) SETTINGS skip_unavailable_shards = 1;
+
+ALTER TABLE {{.DB}}.metrics_15s{{.READ_SUFFIX}} {{.OnCluster}} ADD COLUMN IF NOT EXISTS `type` UInt8;
+
+ALTER TABLE {{.DB}}.samples_v3{{.READ_SUFFIX}} {{.OnCluster}} ADD COLUMN IF NOT EXISTS `type` UInt8;
+
+ALTER TABLE {{.DB}}.time_series{{.READ_SUFFIX}} {{.OnCluster}} ADD COLUMN IF NOT EXISTS `type` UInt8;
+
+ALTER TABLE {{.DB}}.time_series_gin{{.READ_SUFFIX}} {{.OnCluster}} ADD COLUMN IF NOT EXISTS `type` UInt8;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.patterns{{.READ_SUFFIX}} {{.OnCluster}}(
+    timestamp_10m UInt32,
+    fingerprint UInt64,
+    timestamp_s UInt32,
+    tokens Array(String),
+    classes Array(UInt32),
+    overall_cost UInt32,
+    generalized_cost UInt32,
+    samples_count UInt32,
+    pattern_id UInt64,
+    iteration_id UInt64
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'patterns', fingerprint) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.ver{{.READ_SUFFIX}} {{.OnCluster}} (
+    k UInt64,
+    ver UInt64
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'ver', rand()) SETTINGS skip_unavailable_shards = 1;
+
+ALTER TABLE {{.DB}}.time_series{{.READ_SUFFIX}} {{.OnCluster}}
+    ADD COLUMN IF NOT EXISTS metadata String DEFAULT '';
+
+ALTER TABLE {{.DB}}.time_series{{.READ_SUFFIX}} {{.OnCluster}}
+    ADD COLUMN IF NOT EXISTS updated_at_ns Int64 DEFAULT toUnixTimestamp64Nano(now64(9));

ctrl/qryn/sql/profiles_read_dist.sql

@@ -0,0 +1,51 @@
+## Cross-cluster read-path distributed tables for profiles
+## Used when CLICKHOUSE_READ_CLUSTER is set (multi-cluster query aggregation)
+## APPEND ONLY!!!!!
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.profiles{{.READ_SUFFIX}} {{.OnCluster}} (
+    timestamp_ns UInt64,
+    fingerprint UInt64,
+    type_id LowCardinality(String),
+    service_name LowCardinality(String),
+    duration_ns UInt64,
+    payload_type LowCardinality(String),
+    payload String,
+    values_agg Array(Tuple(String, Int64, Int32))
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}','profiles', fingerprint) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.profiles_series{{.READ_SUFFIX}} {{.OnCluster}} (
+    date Date,
+    type_id LowCardinality(String),
+    service_name LowCardinality(String),
+    fingerprint UInt64 CODEC(DoubleDelta, ZSTD(1)),
+    tags Array(Tuple(String, String)) CODEC(ZSTD(1))
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}','profiles_series',fingerprint) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.profiles_series_gin{{.READ_SUFFIX}} {{.OnCluster}} (
+    date Date,
+    key String,
+    val String,
+    type_id LowCardinality(String),
+    service_name LowCardinality(String),
+    fingerprint UInt64 CODEC(DoubleDelta, ZSTD(1))
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}','profiles_series_gin',fingerprint) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.profiles_series_keys{{.READ_SUFFIX}} {{.OnCluster}} (
+    date Date,
+    key String,
+    val String,
+    val_id UInt64
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}','profiles_series_keys', rand()) SETTINGS skip_unavailable_shards = 1;
+
+ALTER TABLE {{.DB}}.profiles{{.READ_SUFFIX}} {{.OnCluster}}
+    ADD COLUMN IF NOT EXISTS `tree` Array(Tuple(UInt64, UInt64, UInt64, Array(Tuple(String, Int64, Int64)))),
+    ADD COLUMN IF NOT EXISTS `functions` Array(Tuple(UInt64, String));
+
+ALTER TABLE {{.DB}}.profiles{{.READ_SUFFIX}} {{.OnCluster}}
+    ADD COLUMN IF NOT EXISTS `sample_types_units` Array(Tuple(String, String));
+
+ALTER TABLE {{.DB}}.profiles_series{{.READ_SUFFIX}} {{.OnCluster}}
+    ADD COLUMN IF NOT EXISTS `sample_types_units` Array(Tuple(String, String));
+
+ALTER TABLE {{.DB}}.profiles_series_gin{{.READ_SUFFIX}} {{.OnCluster}}
+    ADD COLUMN IF NOT EXISTS `sample_types_units` Array(Tuple(String, String));

ctrl/qryn/sql/sql.go

@@ -8,14 +8,23 @@ var LogScript string
 //go:embed log_dist.sql
 var LogDistScript string
 
+//go:embed log_read_dist.sql
+var LogReadDistScript string
+
 //go:embed traces.sql
 var TracesScript string
 
 //go:embed traces_dist.sql
 var TracesDistScript string
 
+//go:embed traces_read_dist.sql
+var TracesReadDistScript string
+
 //go:embed profiles.sql
 var ProfilesScript string
 
 //go:embed profiles_dist.sql
 var ProfilesDistScript string
+
+//go:embed profiles_read_dist.sql
+var ProfilesReadDistScript string

ctrl/qryn/sql/traces_read_dist.sql

@@ -0,0 +1,35 @@
+## Cross-cluster read-path distributed tables for traces
+## Used when CLICKHOUSE_READ_CLUSTER is set (multi-cluster query aggregation)
+## APPEND ONLY!!!!!
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.tempo_traces_kv{{.READ_SUFFIX}} {{.OnCluster}} (
+    oid String,
+    date Date,
+    key String,
+    val_id String,
+    val String
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'tempo_traces_kv', sipHash64(oid, key)) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.tempo_traces{{.READ_SUFFIX}} {{.OnCluster}} (
+    oid String,
+    trace_id FixedString(16),
+    span_id FixedString(8),
+    parent_id String,
+    name String,
+    timestamp_ns Int64 CODEC(DoubleDelta),
+    duration_ns Int64,
+    service_name String,
+    payload_type Int8,
+    payload String
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'tempo_traces', sipHash64(oid, trace_id)) SETTINGS skip_unavailable_shards = 1;
+
+CREATE TABLE IF NOT EXISTS {{.DB}}.tempo_traces_attrs_gin{{.READ_SUFFIX}} {{.OnCluster}} (
+    oid String,
+    date Date,
+    key String,
+    val String,
+    trace_id FixedString(16),
+    span_id FixedString(8),
+    timestamp_ns Int64,
+    duration Int64
+) ENGINE = Distributed('{{.READ_CLUSTER}}','{{.DB}}', 'tempo_traces_attrs_gin', sipHash64(oid, trace_id)) SETTINGS skip_unavailable_shards = 1;

reader/controller/tempo.go

@@ -251,6 +251,9 @@ func (t *TempoController) ValuesV2(w http.ResponseWriter, r *http.Request) {
 		timespan[i] = time.Unix(iT, 0)
 	}
 	tag := mux.Vars(r)["tag"]
+	if tag == "status" {
+		tag = "otel.status_code"
+	}
 
 	limit := 2000
 	if r.URL.Query().Get("limit") != "" {
@@ -302,6 +305,9 @@ func (t *TempoController) Values(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	tag := mux.Vars(r)["tag"]
+	if tag == "status" {
+		tag = "otel.status_code"
+	}
 	cRes, err := t.Service.Values(internalCtx, tag)
 	if err != nil {
 		PromError(500, err.Error(), w)

reader/logql/logql_transpiler/clickhouse_planner/planner_labels_detect.go

@@ -3,6 +3,7 @@ package clickhouse_planner
 import (
 	"github.com/metrico/qryn/v4/reader/logql/logql_transpiler/shared"
 	sql "github.com/metrico/qryn/v4/reader/utils/sql_select"
+	"github.com/metrico/qryn/v4/shared/distconfig"
 )
 
 type DetectLabelsPlanner struct {
@@ -13,7 +14,7 @@ type DetectLabelsPlanner struct {
 func (d *DetectLabelsPlanner) Process(ctx *shared.PlannerContext) (sql.ISelect, error) {
 	from := ctx.TimeSeriesGinTableName
 	if ctx.IsCluster {
-		from += "_dist"
+		from += distconfig.Suffix()
 	}
 
 	req := sql.NewSelect().