diff --git a/.github/workflows/build-test.yml b/.github/workflows/build-test.yml index 21c7039b3cd..5423c1606bd 100644 --- a/.github/workflows/build-test.yml +++ b/.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 diff --git a/internal/testutil/storage.go b/internal/testutil/storage.go new file mode 100644 index 00000000000..b285978c44e --- /dev/null +++ b/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 }, + } +} diff --git a/pkg/clickhouse/querier.go b/pkg/clickhouse/querier.go index cadefaa8cff..5ac880c5529 100644 --- a/pkg/clickhouse/querier.go +++ b/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,6 +55,7 @@ func NewQuerier( tracer trace.Tracer, mem memory.Allocator, sym symbolizer.SymbolizationClient, + demangler profile.Demangler, ) *Querier { return &Querier{ client: client, @@ -59,6 +63,7 @@ func NewQuerier( 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,17 +390,18 @@ 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) @@ -403,12 +409,16 @@ func (q *Querier) QueryRange( 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() } diff --git a/pkg/clickhouse/querier_test.go b/pkg/clickhouse/querier_test.go index 56aa3781cc1..ccab10bc3c3 100644 --- a/pkg/clickhouse/querier_test.go +++ b/pkg/clickhouse/querier_test.go @@ -14,9 +14,18 @@ package clickhouse import ( + "context" + "os" "testing" + "time" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/go-kit/log" "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/trace/noop" + + "github.com/parca-dev/parca/pkg/demangle" + "github.com/parca-dev/parca/pkg/profile" ) func TestSampleDataHasStoredFunction(t *testing.T) { @@ -24,3 +33,115 @@ func TestSampleDataHasStoredFunction(t *testing.T) { require.True(t, (sampleData{functionNames: []string{"function"}}).hasStoredFunction(0)) require.False(t, (sampleData{functionNames: []string{""}, functionSystemNames: []string{""}}).hasStoredFunction(0)) } + +func TestDisplayFunctionName(t *testing.T) { + q := &Querier{demangler: demangle.MustNewDefaultDemangler()} + + // parca-agent v2 rows: function_name empty, system_name populated. + require.Equal(t, "run_until_complete", q.displayFunctionName(sampleData{ + functionNames: []string{""}, + functionSystemNames: []string{"run_until_complete"}, + }, 0)) + + // Mangled system names are demangled, like the FrostDB path. + require.Equal(t, "TB::RowBinaryEncoder::convert", q.displayFunctionName(sampleData{ + functionNames: []string{""}, + functionSystemNames: []string{"_ZN2TB16RowBinaryEncoder7convertEv"}, + }, 0)) + + // v1 rows keep working when no system name is stored. + require.Equal(t, "plain", q.displayFunctionName(sampleData{ + functionNames: []string{"plain"}, + }, 0)) + + // An already-demangled stored name wins over the mangled system name and + // is returned untouched, even when it is more detailed than the default + // demangle mode would produce. + require.Equal(t, "TB::RowBinaryEncoder::convert(int const&)", q.displayFunctionName(sampleData{ + functionNames: []string{"TB::RowBinaryEncoder::convert(int const&)"}, + functionSystemNames: []string{"_ZN2TB16RowBinaryEncoder7convertERKi"}, + }, 0)) + + // A mangled stored name is demangled, exactly as profile.DecodeInto does. + require.Equal(t, "TB::RowBinaryEncoder::convert", q.displayFunctionName(sampleData{ + functionNames: []string{"_ZN2TB16RowBinaryEncoder7convertEv"}, + functionSystemNames: []string{"_ZN2TB16RowBinaryEncoder7convertEv"}, + }, 0)) + + // Without a demangler the raw system name is still better than nothing. + noDemangler := &Querier{} + require.Equal(t, "sys", noDemangler.displayFunctionName(sampleData{ + functionNames: []string{""}, + functionSystemNames: []string{"sys"}, + }, 0)) +} + +func TestLabelsFromJSON(t *testing.T) { + labels, err := labelsFromJSON(`{"job":"api","k8s":{"pod":"web-1","node":{"name":"n1"}},"replica":3,"gone":null}`) + require.NoError(t, err) + + got := map[string]string{} + for _, l := range labels { + got[l.Name] = l.Value + } + require.Equal(t, map[string]string{ + "job": "api", + "k8s.node.name": "n1", + "k8s.pod": "web-1", + "replica": "3", + }, got) + require.Equal(t, "job", labels[0].Name, "labels are sorted by name") + + _, err = labelsFromJSON("not json") + require.Error(t, err) +} + +func TestClickHouseV2FunctionNameIntegration(t *testing.T) { + address := os.Getenv("PARCA_TEST_CLICKHOUSE_ADDRESS") + if address == "" { + t.Skip("PARCA_TEST_CLICKHOUSE_ADDRESS is not set") + } + + ctx := context.Background() + client, err := NewClient(ctx, Config{ + Address: address, + Database: "parca_test", + Table: "stacktraces_v2", + Compression: "none", + }) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + require.NoError(t, client.EnsureSchema(ctx)) + t.Cleanup(func() { require.NoError(t, client.Exec(ctx, "DROP TABLE IF EXISTS parca_test.stacktraces_v2")) }) + + ts := time.Unix(1700000000, 0) + batch, err := client.PrepareBatch(ctx, InsertSQL(client.Database(), client.Table())) + require.NoError(t, err) + require.NoError(t, batch.Append( + "process_cpu", "samples", "count", "cpu", "nanoseconds", + int64(1), int64(0), ts.UnixMilli(), ts.UnixNano(), int64(1), map[string]string{}, + []uint64{0}, []uint64{0}, []uint64{0}, []uint64{0}, []string{""}, []string{""}, + []int64{1}, []string{""}, []string{"_ZN2TB16RowBinaryEncoder7convertEv"}, []string{"row_binary.cpp"}, []int64{1}, + )) + require.NoError(t, batch.Send()) + + q := NewQuerier( + client, + log.NewNopLogger(), + noop.NewTracerProvider().Tracer("test"), + memory.NewGoAllocator(), + nil, + demangle.MustNewDefaultDemangler(), + ) + result, err := q.QuerySingle(ctx, "process_cpu:samples:count:cpu:nanoseconds{}", ts, false) + require.NoError(t, err) + for _, record := range result.Samples { + defer record.Release() + } + + reader, err := profile.NewReader(result) + require.NoError(t, err) + require.Len(t, reader.RecordReaders, 1) + r := reader.RecordReaders[0] + require.Equal(t, "TB::RowBinaryEncoder::convert", string(r.LineFunctionNameDict.Value(int(r.LineFunctionNameIndices.Value(0))))) +} diff --git a/pkg/demangle/demangle.go b/pkg/demangle/demangle.go index ac62f56033f..58a730e271a 100644 --- a/pkg/demangle/demangle.go +++ b/pkg/demangle/demangle.go @@ -64,7 +64,7 @@ func stringsToDemanglerOptions(stringOptions []string) ([]demangle.Option, error } func MustNewDefaultDemangler() Demangler { - d, err := NewDemangler() + d, err := NewDefaultDemangler() if err != nil { panic(err) } @@ -75,6 +75,24 @@ func NewDefaultDemangler() (Demangler, error) { return NewDemangler("no_params", "no_template_params") } +// NewDemanglerForMode creates a Demangler for a --symbolizer-demangle-mode +// value. The "none" mode has no demangler and returns ok=false. +func NewDemanglerForMode(mode string) (d Demangler, ok bool, err error) { + switch mode { + case "", "simple": + d, err = NewDefaultDemangler() + case "templates": + d, err = NewDemangler("no_params") + case "full": + d, err = NewDemangler("no_clones") + case "none": + return Demangler{}, false, nil + default: + return Demangler{}, false, fmt.Errorf("unknown demangle mode %q", mode) + } + return d, err == nil, err +} + // NewDemangler creates a new Demangler with a given demangler options. func NewDemangler(options ...string) (Demangler, error) { demanglerOptions, err := stringsToDemanglerOptions(options) diff --git a/pkg/demangle/demangle_test.go b/pkg/demangle/demangle_test.go new file mode 100644 index 00000000000..02b63a884fe --- /dev/null +++ b/pkg/demangle/demangle_test.go @@ -0,0 +1,43 @@ +// 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 demangle + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNewDemanglerForMode(t *testing.T) { + mangled := []byte("_ZNSt6vectorIiE9push_backERKi") + + for mode, want := range map[string]string{ + "": "std::vector::push_back", + "simple": "std::vector::push_back", + "templates": "std::vector::push_back", + "full": "std::vector::push_back(int const&)", + } { + d, ok, err := NewDemanglerForMode(mode) + require.NoError(t, err, mode) + require.True(t, ok, mode) + require.Equal(t, want, d.Demangle(mangled), mode) + } + + _, ok, err := NewDemanglerForMode("none") + require.NoError(t, err) + require.False(t, ok) + + _, _, err = NewDemanglerForMode("bogus") + require.Error(t, err) +} diff --git a/pkg/parca/parca.go b/pkg/parca/parca.go index d9e92c0d42d..a18e6b4017d 100644 --- a/pkg/parca/parca.go +++ b/pkg/parca/parca.go @@ -330,6 +330,17 @@ func Run(ctx context.Context, logger log.Logger, reg *prometheus.Registry, flags return err } + // The query-time demangler derives function names from system names; both + // storage backends need it since parca-agent v2 only reports system names. + // It follows the same mode as the symbolizer so frames render consistently. + var queryDemangler profile.Demangler + if d, ok, err := demangle.NewDemanglerForMode(flags.Symbolizer.DemangleMode); err != nil { + level.Error(logger).Log("msg", "failed to initialize demangler", "err", err) + return err + } else if ok { + queryDemangler = d + } + // Initialize storage backend - either ClickHouse or FrostDB var ( profileIngester ingester.Ingester @@ -377,6 +388,7 @@ func Run(ctx context.Context, logger log.Logger, reg *prometheus.Registry, flags flags.Symbolizer.ExternalAddr2linePath, symbolizer.WithDemangleMode(flags.Symbolizer.DemangleMode), ), + queryDemangler, ) // We still need the schema for ProfileColumnStore @@ -452,7 +464,8 @@ func Run(ctx context.Context, logger log.Logger, reg *prometheus.Registry, flags {Level: index.L0, MaxSize: 1024 * 1024 * 15, Type: index.CompactionTypeParquetDisk}, {Level: index.L1, MaxSize: 1024 * 1024 * 128, Type: index.CompactionTypeParquetDisk}, {Level: index.L2, MaxSize: 1024 * 1024 * 512}, - })) + }, + )) } } @@ -486,11 +499,6 @@ func Run(ctx context.Context, logger log.Logger, reg *prometheus.Registry, flags } profileIngester = ingester.NewIngester(logger, table) - queryDemangler, err := demangle.NewDefaultDemangler() - if err != nil { - level.Error(logger).Log("msg", "failed to initialize demangler", "err", err) - return err - } querier = parcacol.NewQuerier( logger, tracerProvider.Tracer("querier"), diff --git a/pkg/parca/parca_test.go b/pkg/parca/parca_test.go index 0bad6c70c11..08c2b9f26b6 100644 --- a/pkg/parca/parca_test.go +++ b/pkg/parca/parca_test.go @@ -45,6 +45,7 @@ import ( profilestorepb "github.com/parca-dev/parca/gen/proto/go/parca/profilestore/v1alpha1" querypb "github.com/parca-dev/parca/gen/proto/go/parca/query/v1alpha1" sharepb "github.com/parca-dev/parca/gen/proto/go/parca/share/v1alpha1" + "github.com/parca-dev/parca/internal/testutil" "github.com/parca-dev/parca/pkg/ingester" "github.com/parca-dev/parca/pkg/kv" "github.com/parca-dev/parca/pkg/parcacol" @@ -170,6 +171,33 @@ func MustReadAllGzip(t require.TestingT, filename string) []byte { return content } +func newTestStoreAndAPI(t *testing.T, mem memory.Allocator) (*profilestore.ProfileColumnStore, *queryservice.ColumnQueryAPI, *testutil.Storage) { + t.Helper() + logger := log.NewNopLogger() + tracer := noop.NewTracerProvider().Tracer("") + storage := testutil.NewStorage(t, mem) + schema, err := parcaprofile.Schema() + require.NoError(t, err) + store := profilestore.NewProfileColumnStore( + prometheus.NewRegistry(), + logger, + tracer, + storage.Ingester, + schema, + memory.DefaultAllocator, + ) + api := queryservice.NewColumnQueryAPI( + logger, + tracer, + getShareServerConn(t), + storage.Querier, + mem, + parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), + nil, + ) + return store, api, storage +} + func TestConsistency(t *testing.T) { t.Skipf("skipped, need to think how we want to bring back consistency exports without pprof_labels") @@ -294,39 +322,13 @@ func TestPGOE2e(t *testing.T) { runCmd(t, "./testdata/pgotest") ctx := context.Background() - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - tracer := noop.NewTracerProvider().Tracer("") - col, err := frostdb.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - schema, err := parcaprofile.Schema() - require.NoError(t, err) - - table, err := colDB.Table( - "stacktraces", - frostdb.NewTableConfig(parcaprofile.SchemaDefinition()), - ) - require.NoError(t, err) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api, storage := newTestStoreAndAPI(t, mem) fileContent, err := os.ReadFile("./testdata/pgotest.prof") require.NoError(t, err) - ingester := ingester.NewIngester( - logger, - table, - ) - store := profilestore.NewProfileColumnStore( - reg, - logger, - tracer, - ingester, - schema, - memory.DefaultAllocator, - ) - _, err = store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ Series: []*profilestorepb.RawProfileSeries{{ Labels: &profilestorepb.LabelSet{ @@ -348,30 +350,7 @@ func TestPGOE2e(t *testing.T) { }) require.NoError(t, err) - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - require.NoError(t, table.EnsureCompaction()) - api := queryservice.NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) - + storage.EnsureReady(t) res, err := api.Query(ctx, &querypb.QueryRequest{ Mode: querypb.QueryRequest_MODE_MERGE, ReportType: querypb.QueryRequest_REPORT_TYPE_PPROF, @@ -395,38 +374,12 @@ func TestLabels(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - tracer := noop.NewTracerProvider().Tracer("") - col, err := frostdb.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - schema, err := parcaprofile.Schema() - require.NoError(t, err) - - table, err := colDB.Table( - "labels", - frostdb.NewTableConfig(parcaprofile.SchemaDefinition()), - ) - require.NoError(t, err) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api, storage := newTestStoreAndAPI(t, mem) fileContent, err := os.ReadFile("testdata/labels.pb.gz") require.NoError(t, err) - - ingester := ingester.NewIngester( - logger, - table, - ) - store := profilestore.NewProfileColumnStore( - reg, - logger, - tracer, - ingester, - schema, - memory.DefaultAllocator, - ) _, err = store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ Series: []*profilestorepb.RawProfileSeries{{ Labels: &profilestorepb.LabelSet{ @@ -448,30 +401,7 @@ func TestLabels(t *testing.T) { }) require.NoError(t, err) - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - require.NoError(t, table.EnsureCompaction()) - api := queryservice.NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "labels", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) - + storage.EnsureReady(t) ts := timestamppb.New(timestamp.Time(1677488315039)) // time_nanos of the profile divided by 1e6 res, err := api.Query(ctx, &querypb.QueryRequest{ GroupBy: &querypb.GroupBy{ diff --git a/pkg/query/columnquery_test.go b/pkg/query/columnquery_test.go index 799f840d11b..28a02578e79 100644 --- a/pkg/query/columnquery_test.go +++ b/pkg/query/columnquery_test.go @@ -29,8 +29,6 @@ import ( "github.com/apache/arrow-go/v18/arrow/memory" "github.com/go-kit/log" pprofprofile "github.com/google/pprof/profile" - columnstore "github.com/polarsignals/frostdb" - "github.com/polarsignals/frostdb/query" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/prometheus/model/timestamp" "github.com/stretchr/testify/require" @@ -46,9 +44,8 @@ import ( profilestorepb "github.com/parca-dev/parca/gen/proto/go/parca/profilestore/v1alpha1" pb "github.com/parca-dev/parca/gen/proto/go/parca/query/v1alpha1" sharepb "github.com/parca-dev/parca/gen/proto/go/parca/share/v1alpha1" - "github.com/parca-dev/parca/pkg/ingester" + "github.com/parca-dev/parca/internal/testutil" "github.com/parca-dev/parca/pkg/kv" - "github.com/parca-dev/parca/pkg/normalizer" "github.com/parca-dev/parca/pkg/parcacol" "github.com/parca-dev/parca/pkg/profile" "github.com/parca-dev/parca/pkg/profilestore" @@ -64,39 +61,10 @@ func TestColumnQueryAPIQueryRangeEmpty(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - _, err = colDB.Table("stacktraces", columnstore.NewTableConfig(profile.SchemaDefinition())) - require.NoError(t, err) - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) defer mem.AssertSize(t, 0) - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) - _, err = api.QueryRange(ctx, &pb.QueryRangeRequest{ + _, api := newTestColumnQueryAPI(t, mem) + _, err := api.QueryRange(ctx, &pb.QueryRangeRequest{ Query: `memory:alloc_objects:count:space:bytes{job="default"}`, Start: timestamppb.New(timestamp.Time(0)), End: timestamppb.New(timestamp.Time(9223372036854775807)), @@ -136,43 +104,44 @@ func MustDecompressGzip(t Testing, b []byte) []byte { return content } -func TestColumnQueryAPIQueryRange(t *testing.T) { - t.Parallel() - - ctx := context.Background() +func newTestColumnQueryAPI(t *testing.T, mem memory.Allocator) (*profilestore.ProfileColumnStore, *ColumnQueryAPI) { + t.Helper() logger := log.NewNopLogger() - reg := prometheus.NewRegistry() tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - + storage := testutil.NewStorage(t, mem) schema, err := profile.Schema() require.NoError(t, err) - - table, err := colDB.Table( - "stacktraces", - columnstore.NewTableConfig(profile.SchemaDefinition()), - ) - require.NoError(t, err) - - dir := "./testdata/many/" - files, err := os.ReadDir(dir) - require.NoError(t, err) - - ingester := ingester.NewIngester( - logger, - table, - ) store := profilestore.NewProfileColumnStore( - reg, + prometheus.NewRegistry(), logger, tracer, - ingester, + storage.Ingester, schema, memory.DefaultAllocator, ) + api := NewColumnQueryAPI( + logger, + tracer, + getShareServerConn(t), + storage.Querier, + mem, + parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), + nil, + ) + return store, api +} + +func TestColumnQueryAPIQueryRange(t *testing.T) { + t.Parallel() + + ctx := context.Background() + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) + + dir := "./testdata/many/" + files, err := os.ReadDir(dir) + require.NoError(t, err) for _, f := range files { fileContent, err := os.ReadFile(dir + f.Name()) @@ -200,28 +169,6 @@ func TestColumnQueryAPIQueryRange(t *testing.T) { require.NoError(t, err) } - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) res, err := api.QueryRange(ctx, &pb.QueryRangeRequest{ Query: `memory:alloc_objects:count:space:bytes{job="default"}`, Start: timestamppb.New(time.Unix(0, 0)), @@ -233,38 +180,64 @@ func TestColumnQueryAPIQueryRange(t *testing.T) { require.Equal(t, 10, len(res.Series[0].Samples)) } -func TestColumnQueryAPIQuerySingle(t *testing.T) { +func TestColumnQueryAPIQueryRangePerLabelset(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) - schema, err := profile.Schema() + fileContent, err := os.ReadFile("testdata/alloc_objects.pb.gz") require.NoError(t, err) - table, err := colDB.Table( - "stacktraces", - columnstore.NewTableConfig(profile.SchemaDefinition()), - ) + for _, job := range []string{"api", "worker"} { + _, err = store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ + Series: []*profilestorepb.RawProfileSeries{{ + Labels: &profilestorepb.LabelSet{ + Labels: []*profilestorepb.Label{ + {Name: "__name__", Value: "memory"}, + {Name: "job", Value: job}, + {Name: "k8s.pod", Value: "pod-" + job}, + }, + }, + Samples: []*profilestorepb.RawSample{{RawProfile: fileContent}}, + }}, + }) + require.NoError(t, err) + } + + jobsOf := func(series []*pb.MetricsSeries) []string { + jobs := make([]string, 0, len(series)) + for _, s := range series { + labels := make(map[string]string, len(s.Labelset.Labels)) + for _, label := range s.Labelset.Labels { + labels[label.Name] = label.Value + } + require.Len(t, s.Samples, 1) + jobs = append(jobs, labels["job"]) + require.Equal(t, "pod-"+labels["job"], labels["k8s.pod"]) + } + return jobs + } + + // Without sumBy every distinct labelset is its own series. + res, err := api.QueryRange(ctx, &pb.QueryRangeRequest{ + Query: `memory:alloc_objects:count:space:bytes`, + Start: timestamppb.New(time.Unix(0, 0)), + End: timestamppb.New(time.Unix(0, 9223372036854775807)), + }) require.NoError(t, err) - ingester := ingester.NewIngester( - logger, - table, - ) - store := profilestore.NewProfileColumnStore( - reg, - logger, - tracer, - ingester, - schema, - memory.DefaultAllocator, - ) + require.ElementsMatch(t, []string{"api", "worker"}, jobsOf(res.Series)) +} + +func TestColumnQueryAPIQuerySingle(t *testing.T) { + t.Parallel() + + ctx := context.Background() + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) fileContent, err := os.ReadFile("testdata/alloc_objects.pb.gz") require.NoError(t, err) @@ -293,28 +266,6 @@ func TestColumnQueryAPIQuerySingle(t *testing.T) { }) require.NoError(t, err) - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) ts := timestamppb.New(timestamp.Time(p.TimeNanos / time.Millisecond.Nanoseconds())) res, err := api.Query(ctx, &pb.QueryRequest{ Options: &pb.QueryRequest_Single{ @@ -387,40 +338,13 @@ func TestColumnQueryAPIQueryFgprof(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - schema, err := profile.Schema() - require.NoError(t, err) - - table, err := colDB.Table( - "stacktraces", - columnstore.NewTableConfig(profile.SchemaDefinition()), - ) - require.NoError(t, err) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) fileContent, err := os.ReadFile("testdata/fgprof.pb.gz") require.NoError(t, err) - ingester := ingester.NewIngester( - logger, - table, - ) - - store := profilestore.NewProfileColumnStore( - reg, - logger, - tracer, - ingester, - schema, - memory.DefaultAllocator, - ) - _, err = store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ Series: []*profilestorepb.RawProfileSeries{{ Labels: &profilestorepb.LabelSet{ @@ -442,29 +366,6 @@ func TestColumnQueryAPIQueryFgprof(t *testing.T) { }) require.NoError(t, err) - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) - res, err := api.QueryRange(ctx, &pb.QueryRangeRequest{ Query: `fgprof:samples:count:wallclock:nanoseconds:delta`, Start: timestamppb.New(time.Unix(0, 0)), @@ -481,36 +382,9 @@ func TestColumnQueryAPIQueryCumulative(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - schema, err := profile.Schema() - require.NoError(t, err) - - table, err := colDB.Table( - "stacktraces", - columnstore.NewTableConfig(profile.SchemaDefinition()), - ) - require.NoError(t, err) - - ingester := ingester.NewIngester( - logger, - table, - ) - - store := profilestore.NewProfileColumnStore( - reg, - logger, - tracer, - ingester, - schema, - memory.DefaultAllocator, - ) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) // Load CPU and memory profiles fileNames := []string{ @@ -549,29 +423,6 @@ func TestColumnQueryAPIQueryCumulative(t *testing.T) { require.NoError(t, err) } - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) - // These have been extracted from the profiles above. queries := []struct { name string @@ -640,21 +491,9 @@ func TestColumnQueryAPIQueryDiff(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - schema, err := profile.Schema() - require.NoError(t, err) - - table, err := colDB.Table( - "stacktraces", - columnstore.NewTableConfig(profile.SchemaDefinition()), - ) - require.NoError(t, err) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) p := &pprofpb.Profile{ StringTable: []string{ @@ -699,14 +538,7 @@ func TestColumnQueryAPIQueryDiff(t *testing.T) { }}, } - ingester := ingester.NewIngester( - logger, - table, - ) - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - - r1, err := normalizer.WriteRawRequestToArrowRecord(ctx, mem, &profilestorepb.WriteRawRequest{ + _, err := store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ Series: []*profilestorepb.RawProfileSeries{{ Labels: &profilestorepb.LabelSet{ Labels: []*profilestorepb.Label{ @@ -724,17 +556,15 @@ func TestColumnQueryAPIQueryDiff(t *testing.T) { RawProfile: MustCompressGzip(t, p), }}, }}, - }, schema) + }) require.NoError(t, err) - require.NoError(t, ingester.Ingest(ctx, r1)) - r1.Release() p.Sample = []*pprofpb.Sample{{ Value: []int64{2}, LocationId: []uint64{2}, }} p.TimeNanos = 2000000 - r2, err := normalizer.WriteRawRequestToArrowRecord(ctx, mem, &profilestorepb.WriteRawRequest{ + _, err = store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ Series: []*profilestorepb.RawProfileSeries{{ Labels: &profilestorepb.LabelSet{ Labels: []*profilestorepb.Label{ @@ -752,31 +582,8 @@ func TestColumnQueryAPIQueryDiff(t *testing.T) { RawProfile: MustCompressGzip(t, p), }}, }}, - }, schema) + }) require.NoError(t, err) - require.NoError(t, ingester.Ingest(ctx, r2)) - r2.Release() - - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) res, err := api.Query(ctx, &pb.QueryRequest{ Mode: pb.QueryRequest_MODE_DIFF, @@ -915,49 +722,19 @@ func TestColumnQueryAPIQueryDiff(t *testing.T) { require.Equal(t, 2, len(testProf.Sample)) require.Equal(t, []int64{2}, testProf.Sample[0].Value) require.Equal(t, []int64{-1}, testProf.Sample[1].Value) - - // Need to double release them because the storage will keep a reference to them. - r1.Release() - r2.Release() } func TestColumnQueryAPITypes(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - schema, err := profile.Schema() - require.NoError(t, err) - - table, err := colDB.Table( - "stacktraces", - columnstore.NewTableConfig(profile.SchemaDefinition()), - ) - require.NoError(t, err) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) fileContent, err := os.ReadFile("testdata/alloc_space_delta.pb.gz") require.NoError(t, err) - ingester := ingester.NewIngester( - logger, - table, - ) - store := profilestore.NewProfileColumnStore( - reg, - logger, - tracer, - ingester, - schema, - memory.DefaultAllocator, - ) - _, err = store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ Series: []*profilestorepb.RawProfileSeries{{ Labels: &profilestorepb.LabelSet{ @@ -979,30 +756,6 @@ func TestColumnQueryAPITypes(t *testing.T) { }) require.NoError(t, err) - require.NoError(t, table.EnsureCompaction()) - - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) res, err := api.ProfileTypes(ctx, &pb.ProfileTypesRequest{}) require.NoError(t, err) @@ -1023,39 +776,13 @@ func TestColumnQueryAPILabelNames(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - schema, err := profile.Schema() - require.NoError(t, err) - - table, err := colDB.Table( - "stacktraces", - columnstore.NewTableConfig(profile.SchemaDefinition()), - ) - require.NoError(t, err) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) fileContent, err := os.ReadFile("testdata/alloc_objects.pb.gz") require.NoError(t, err) - ingester := ingester.NewIngester( - logger, - table, - ) - store := profilestore.NewProfileColumnStore( - reg, - logger, - tracer, - ingester, - schema, - memory.DefaultAllocator, - ) - _, err = store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ Series: []*profilestorepb.RawProfileSeries{{ Labels: &profilestorepb.LabelSet{ @@ -1077,28 +804,6 @@ func TestColumnQueryAPILabelNames(t *testing.T) { }) require.NoError(t, err) - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) res, err := api.Labels(ctx, &pb.LabelsRequest{}) require.NoError(t, err) @@ -1111,39 +816,13 @@ func TestColumnQueryAPILabelValues(t *testing.T) { t.Parallel() ctx := context.Background() - logger := log.NewNopLogger() - reg := prometheus.NewRegistry() - tracer := noop.NewTracerProvider().Tracer("") - col, err := columnstore.New() - require.NoError(t, err) - colDB, err := col.DB(context.Background(), "parca") - require.NoError(t, err) - - schema, err := profile.Schema() - require.NoError(t, err) - - table, err := colDB.Table( - "stacktraces", - columnstore.NewTableConfig(profile.SchemaDefinition()), - ) - require.NoError(t, err) + mem := memory.NewCheckedAllocator(memory.DefaultAllocator) + defer mem.AssertSize(t, 0) + store, api := newTestColumnQueryAPI(t, mem) fileContent, err := os.ReadFile("testdata/alloc_objects.pb.gz") require.NoError(t, err) - ingester := ingester.NewIngester( - logger, - table, - ) - store := profilestore.NewProfileColumnStore( - reg, - logger, - tracer, - ingester, - schema, - memory.DefaultAllocator, - ) - _, err = store.WriteRaw(ctx, &profilestorepb.WriteRawRequest{ Series: []*profilestorepb.RawProfileSeries{{ Labels: &profilestorepb.LabelSet{ @@ -1165,28 +844,6 @@ func TestColumnQueryAPILabelValues(t *testing.T) { }) require.NoError(t, err) - mem := memory.NewCheckedAllocator(memory.DefaultAllocator) - defer mem.AssertSize(t, 0) - api := NewColumnQueryAPI( - logger, - tracer, - getShareServerConn(t), - parcacol.NewQuerier( - logger, - tracer, - query.NewEngine( - mem, - colDB.TableProvider(), - ), - "stacktraces", - nil, - nil, - mem, - ), - mem, - parcacol.NewArrowToProfileConverter(tracer, kv.NewKeyMaker()), - nil, - ) res, err := api.Values(ctx, &pb.ValuesRequest{ LabelName: "job", })