Fix v2 function names in ClickHouse query results

.github/workflows/build-test.yml

@@ -96,3 +96,31 @@ jobs:
         if-no-files-found: error
         path: |
           bin
+
+  clickhouse-test:
+    name: Test with ClickHouse storage
+    runs-on: ubuntu-latest
+    timeout-minutes: 90
+    services:
+      clickhouse:
+        image: clickhouse/clickhouse-server:25.8
+        env:
+          CLICKHOUSE_SKIP_USER_SETUP: "1"
+        ports:
+          - 9000:9000
+        options: >-
+          --health-cmd "clickhouse-client --query 'SELECT 1'"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 12
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+      - name: Set up Go
+        uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5.6.0
+        with:
+          go-version-file: .go-version
+      - name: Test
+        env:
+          PARCA_TEST_STORAGE: clickhouse
+          PARCA_TEST_CLICKHOUSE_ADDRESS: localhost:9000
+        run: make go/test ENABLE_RACE=yes

internal/testutil/storage.go

@@ -0,0 +1,152 @@
+// Copyright 2026 The Parca Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package testutil
+
+import (
+	"context"
+	"fmt"
+	"hash/fnv"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/go-kit/log"
+	"github.com/polarsignals/frostdb"
+	frostquery "github.com/polarsignals/frostdb/query"
+	"go.opentelemetry.io/otel/trace/noop"
+
+	pb "github.com/parca-dev/parca/gen/proto/go/parca/query/v1alpha1"
+	"github.com/parca-dev/parca/pkg/clickhouse"
+	"github.com/parca-dev/parca/pkg/demangle"
+	"github.com/parca-dev/parca/pkg/ingester"
+	"github.com/parca-dev/parca/pkg/parcacol"
+	"github.com/parca-dev/parca/pkg/profile"
+	"github.com/parca-dev/parca/pkg/symbolizer"
+)
+
+type Querier interface {
+	Labels(context.Context, []string, time.Time, time.Time, string) ([]string, error)
+	Values(context.Context, string, []string, time.Time, time.Time, string) ([]string, error)
+	QueryRange(context.Context, string, time.Time, time.Time, time.Duration, uint32, []string) ([]*pb.MetricsSeries, error)
+	ProfileTypes(context.Context, time.Time, time.Time) ([]*pb.ProfileType, error)
+	QuerySingle(context.Context, string, time.Time, bool) (profile.Profile, error)
+	QueryMerge(context.Context, string, time.Time, time.Time, []string, bool, string) (profile.Profile, error)
+	GetProfileMetadataMappings(context.Context, string, time.Time, time.Time) ([]string, error)
+	GetProfileMetadataLabels(context.Context, string, time.Time, time.Time) ([]string, error)
+	HasProfileData(context.Context) (bool, error)
+}
+
+type Storage struct {
+	Ingester    ingester.Ingester
+	Querier     Querier
+	ensureReady func() error
+}
+
+func (s *Storage) EnsureReady(t *testing.T) {
+	t.Helper()
+	if err := s.ensureReady(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func NewStorage(t *testing.T, allocator memory.Allocator) *Storage {
+	t.Helper()
+	switch backend := os.Getenv("PARCA_TEST_STORAGE"); backend {
+	case "", "frostdb":
+		return newFrostDBStorage(t, allocator)
+	case "clickhouse":
+		return newClickHouseStorage(t, allocator)
+	default:
+		t.Fatalf("unsupported PARCA_TEST_STORAGE %q", backend)
+		return nil
+	}
+}
+
+func newFrostDBStorage(t *testing.T, allocator memory.Allocator) *Storage {
+	col, err := frostdb.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	db, err := col.DB(context.Background(), "parca")
+	if err != nil {
+		t.Fatal(err)
+	}
+	table, err := db.Table("stacktraces", frostdb.NewTableConfig(profile.SchemaDefinition()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	logger := log.NewNopLogger()
+	tracer := noop.NewTracerProvider().Tracer("")
+	return &Storage{
+		Ingester: ingester.NewIngester(logger, table),
+		Querier: parcacol.NewQuerier(
+			logger,
+			tracer,
+			frostquery.NewEngine(allocator, db.TableProvider()),
+			"stacktraces",
+			nil,
+			nil,
+			allocator,
+		),
+		ensureReady: table.EnsureCompaction,
+	}
+}
+
+type nopSymbolizer struct{}
+
+func (nopSymbolizer) Symbolize(context.Context, symbolizer.SymbolizationRequest) error { return nil }
+
+func newClickHouseStorage(t *testing.T, allocator memory.Allocator) *Storage {
+	address := os.Getenv("PARCA_TEST_CLICKHOUSE_ADDRESS")
+	if address == "" {
+		t.Fatal("PARCA_TEST_CLICKHOUSE_ADDRESS must be set when PARCA_TEST_STORAGE=clickhouse")
+	}
+
+	h := fnv.New64a()
+	_, _ = fmt.Fprintf(h, "%s-%d", t.Name(), os.Getpid())
+	table := fmt.Sprintf("stacktraces_%x", h.Sum64())
+	ctx := context.Background()
+	client, err := clickhouse.NewClient(ctx, clickhouse.Config{
+		Address:     address,
+		Database:    "parca_test",
+		Table:       table,
+		Compression: "none",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := client.EnsureSchema(ctx); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		_ = client.Exec(context.Background(), "DROP TABLE IF EXISTS "+client.FullTableName())
+		_ = client.Close()
+	})
+
+	logger := log.NewNopLogger()
+	tracer := noop.NewTracerProvider().Tracer("")
+	return &Storage{
+		Ingester: clickhouse.NewIngester(logger, client),
+		Querier: clickhouse.NewQuerier(
+			client,
+			logger,
+			tracer,
+			allocator,
+			nopSymbolizer{},
+			demangle.MustNewDefaultDemangler(),
+		),
+		ensureReady: func() error { return nil },
+	}
+}

pkg/clickhouse/querier.go

@@ -15,7 +15,9 @@ package clickhouse
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"sort"
 	"strings"
 	"time"
 
@@ -43,6 +45,7 @@ type Querier struct {
 	tracer     trace.Tracer
 	mem        memory.Allocator
 	symbolizer symbolizer.SymbolizationClient
+	demangler  profile.Demangler
 }
 
 // NewQuerier creates a new ClickHouse querier.
@@ -52,13 +55,15 @@ func NewQuerier(
 	tracer trace.Tracer,
 	mem memory.Allocator,
 	sym symbolizer.SymbolizationClient,
+	demangler profile.Demangler,
 ) *Querier {
 	return &Querier{
 		client:     client,
 		logger:     logger,
 		tracer:     tracer,
 		mem:        mem,
 		symbolizer: sym,
+		demangler:  demangler,
 	}
 }
 
@@ -320,10 +325,10 @@ func (q *Querier) QueryRange(
 		return nil, err
 	}
 
-	// Build sumBy label selections - cast to String because by default it is forbidden to group by
-	// dynamic types in ClickHouse
-	sumBySelects := ""
-	outerLabelSelects := ""
+	// Build label selections - cast to String because by default it is forbidden to group by
+	// dynamic types in ClickHouse. Without sumBy, preserve the complete label set like FrostDB.
+	sumBySelects := ", toString(labels) AS labels_json"
+	outerLabelSelects := "labels_json, "
 	if len(sumBy) > 0 {
 		selects := make([]string, len(sumBy))
 		outerSelects := make([]string, len(sumBy))
@@ -385,30 +390,35 @@ func (q *Querier) QueryRange(
 
 	var resSeries []*pb.MetricsSeries
 
-	// Each row represents one unique labelset with all its samples
+	// Each row represents one unique labelset with all its samples.
 	for rows.Next() {
-		// Scan label columns
-		labelValues := make([]string, len(sumBy))
-		scanArgs := make([]interface{}, 0, len(sumBy)+1)
-		for i := range sumBy {
-			scanArgs = append(scanArgs, &labelValues[i])
+		// Without sumBy the single label column is the JSON labelset; with sumBy
+		// there is one String column per requested label.
+		labelColumns := make([]string, max(len(sumBy), 1))
+		scanArgs := make([]interface{}, 0, len(labelColumns)+1)
+		for i := range labelColumns {
+			scanArgs = append(scanArgs, &labelColumns[i])
 		}
 
-		// Scan samples array - ClickHouse returns array of tuples as slice of slices
-		// Each tuple element becomes []interface{}{timestamp_bucket, total_sum, duration_min}
+		// ClickHouse returns the array of tuples as a slice of
+		// []interface{}{timestamp_bucket, total_sum, duration_min}.
 		var samples [][]interface{}
 		scanArgs = append(scanArgs, &samples)
 
 		if err := rows.Scan(scanArgs...); err != nil {
 			return nil, fmt.Errorf("failed to scan row: %w", err)
 		}
 
-		// Build labelset
-		pbLabelSet := make([]*profilestorepb.Label, len(sumBy))
-		for i, s := range sumBy {
-			pbLabelSet[i] = &profilestorepb.Label{
-				Name:  s,
-				Value: labelValues[i],
+		var pbLabelSet []*profilestorepb.Label
+		if len(sumBy) == 0 {
+			pbLabelSet, err = labelsFromJSON(labelColumns[0])
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			pbLabelSet = make([]*profilestorepb.Label, 0, len(sumBy))
+			for i, name := range sumBy {
+				pbLabelSet = append(pbLabelSet, &profilestorepb.Label{Name: name, Value: labelColumns[i]})
 			}
 		}
 
@@ -836,12 +846,73 @@ type sampleData struct {
 }
 
 func (s sampleData) hasStoredFunction(i int) bool {
-	return (i < len(s.functionNames) && s.functionNames[i] != "") ||
-		(i < len(s.functionSystemNames) && s.functionSystemNames[i] != "") ||
-		(i < len(s.functionFilenames) && s.functionFilenames[i] != "") ||
+	return stringAt(s.functionNames, i) != "" ||
+		stringAt(s.functionSystemNames, i) != "" ||
+		stringAt(s.functionFilenames, i) != "" ||
 		(i < len(s.functionStartLines) && s.functionStartLines[i] != 0)
 }
 
+// displayFunctionName returns the stored display name, falling back to the
+// system name for parca-agent v2 rows, then applies the configured demangler.
+func (q *Querier) displayFunctionName(s sampleData, idx int) string {
+	name := stringAt(s.functionNames, idx)
+	if name == "" {
+		name = stringAt(s.functionSystemNames, idx)
+	}
+	if name == "" || q.demangler == nil {
+		return name
+	}
+	return q.demangler.Demangle([]byte(name))
+}
+
+func stringAt(values []string, idx int) string {
+	if idx < len(values) {
+		return values[idx]
+	}
+	return ""
+}
+
+// labelsFromJSON decodes the toString() rendering of the ClickHouse JSON labels
+// column into a sorted labelset. ClickHouse stores dotted label names such as
+// k8s.pod as nested objects, so nested keys are joined back with dots.
+func labelsFromJSON(data string) ([]*profilestorepb.Label, error) {
+	var decoded map[string]any
+	if err := json.Unmarshal([]byte(data), &decoded); err != nil {
+		return nil, fmt.Errorf("failed to decode labels: %w", err)
+	}
+	labels := map[string]string{}
+	flattenLabels("", decoded, labels)
+
+	names := make([]string, 0, len(labels))
+	for name := range labels {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	res := make([]*profilestorepb.Label, 0, len(names))
+	for _, name := range names {
+		res = append(res, &profilestorepb.Label{Name: name, Value: labels[name]})
+	}
+	return res, nil
+}
+
+func flattenLabels(prefix string, in map[string]any, out map[string]string) {
+	for key, value := range in {
+		name := prefix + key
+		switch v := value.(type) {
+		case nil:
+		case map[string]any:
+			flattenLabels(name+".", v, out)
+		case string:
+			out[name] = v
+		default:
+			// ClickHouse infers scalar JSON types per path, so a label value
+			// may come back unquoted; keep its textual form.
+			out[name] = fmt.Sprint(v)
+		}
+	}
+}
+
 // rowsToArrowRecords converts ClickHouse query results to Arrow records.
 func (q *Querier) rowsToArrowRecords(
 	ctx context.Context,
@@ -1055,7 +1126,7 @@ func (q *Querier) rowsToArrowRecords(
 					w.LineNumber.AppendNull()
 				}
 				w.ColumnNumber.AppendNull()
-				if err := w.FunctionName.Append([]byte(s.functionNames[idx])); err != nil {
+				if err := w.FunctionName.Append([]byte(q.displayFunctionName(s, idx))); err != nil {
 					level.Error(q.logger).Log("msg", "failed to append function name", "err", err)
 					w.FunctionName.AppendNull()
 				}