feat(audit): adding kafka audit exporter (#13357)

<!--A few plain bullets saying what changed and why, for a reviewer
skimming it - not a wall of text, not a restatement of the diff, not
generated boilerplate.-->
#### Description
- add `kafka` auditor provider that produces audit logs to a
kafka/redpanda topic as `otlp_proto`
- uses franz-go v1.21.7.
- no delivery timeout or retry cap: on franz-go < v1.22.2 these can ack
a record that was never written (twmb/franz-go#1484).
- added unit tests

<!--Reference issues using `Closes #issue-number` to enable automatic
closure on merge. -->
#### Issues closed by this PR
closes https://github.com/SigNoz/keystone-pod/issues/2

<!--Please delete paragraphs that you did not use before submitting.-->

cmd/enterprise/server.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/SigNoz/signoz/cmd"
 	"github.com/SigNoz/signoz/ee/auditor/fileauditor"
+	"github.com/SigNoz/signoz/ee/auditor/kafkaauditor"
 	"github.com/SigNoz/signoz/ee/auditor/otlphttpauditor"
 	"github.com/SigNoz/signoz/ee/authn/callbackauthn/oidccallbackauthn"
 	"github.com/SigNoz/signoz/ee/authn/callbackauthn/samlcallbackauthn"
@@ -156,6 +157,9 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
 			if err := factories.Add(fileauditor.NewFactory(licensing, version.Info)); err != nil {
 				panic(err)
 			}
+			if err := factories.Add(kafkaauditor.NewFactory(licensing, version.Info)); err != nil {
+				panic(err)
+			}
 			return factories
 		},
 		func(ctx context.Context, providerSettings factory.ProviderSettings, flagger pkgflagger.Flagger, licensing licensing.Licensing, telemetryStore telemetrystore.TelemetryStore, retentionGetter retention.Getter, orgGetter organization.Getter, zeus zeus.Zeus) (factory.NamedMap[factory.ProviderFactory[meterreporter.Reporter, meterreporter.Config]], string) {

conf/example.yaml

@@ -436,6 +436,7 @@ auditor:
   # Specifies the auditor provider to use.
   # noop: discards all audit events (community default).
   # otlphttp: exports audit events via OTLP HTTP (enterprise).
+  # kafka: exports audit events to a Kafka topic as OTLP protobuf (enterprise).
   provider: noop
   # The async channel capacity for audit events. Events are dropped when full (fail-open).
   buffer_size: 1000
@@ -461,6 +462,18 @@ auditor:
       max_interval: 30s
       # The total maximum time spent retrying.
       max_elapsed_time: 60s
+  kafka:
+    # The seed broker addresses (host:port) used to discover the cluster.
+    brokers:
+      - localhost:9092
+    # The topic audit events are written to. It must already exist.
+    topic: ""
+    # The client id the producer reports to the brokers.
+    client_id: signoz
+    # The compression codec for produced batches: none, gzip, snappy, lz4 or zstd.
+    compression: lz4
+    # The maximum time to drain buffered events on shutdown. Undelivered events are dropped after it.
+    shutdown_timeout: 30s
 
 ##################### Cloud Integration #####################
 cloudintegration:

ee/auditor/kafkaauditor/export.go

@@ -0,0 +1,31 @@
+package kafkaauditor
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+
+	"github.com/SigNoz/signoz/pkg/auditor"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/types/audittypes"
+)
+
+// Produces the whole batch as a single record so the collector's otlp_proto unmarshaler reads it as one ExportLogsServiceRequest.
+func (provider *provider) export(ctx context.Context, events []audittypes.AuditEvent) error {
+	logs := audittypes.NewPLogsFromAuditEvents(events, "signoz", provider.build.Version(), "signoz.audit")
+
+	value, err := provider.marshaler.MarshalLogs(logs)
+	if err != nil {
+		return errors.Wrapf(err, errors.TypeInternal, auditor.ErrCodeAuditExportFailed, "failed to marshal audit logs")
+	}
+
+	record := &kgo.Record{Topic: provider.config.Kafka.Topic, Value: value}
+	if err := provider.client.ProduceSync(ctx, record).FirstErr(); err != nil {
+		provider.settings.Logger().ErrorContext(ctx, "audit export failed", errors.Attr(err), slog.Int("dropped_log_records", len(events)))
+
+		return errors.Wrapf(err, errors.TypeInternal, auditor.ErrCodeAuditExportFailed, "failed to produce audit logs to topic %q", provider.config.Kafka.Topic)
+	}
+
+	return nil
+}

ee/auditor/kafkaauditor/provider.go

@@ -0,0 +1,132 @@
+package kafkaauditor
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+	"github.com/twmb/franz-go/plugin/kslog"
+	"go.opentelemetry.io/collector/pdata/plog"
+
+	"github.com/SigNoz/signoz/pkg/auditor"
+	"github.com/SigNoz/signoz/pkg/auditor/auditorserver"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/licensing"
+	"github.com/SigNoz/signoz/pkg/types/audittypes"
+	"github.com/SigNoz/signoz/pkg/version"
+)
+
+var _ auditor.Auditor = (*provider)(nil)
+
+type provider struct {
+	settings  factory.ScopedProviderSettings
+	config    auditor.Config
+	licensing licensing.Licensing
+	build     version.Build
+	server    *auditorserver.Server
+	marshaler plog.ProtoMarshaler
+	client    *kgo.Client
+}
+
+func NewFactory(licensing licensing.Licensing, build version.Build) factory.ProviderFactory[auditor.Auditor, auditor.Config] {
+	return factory.NewProviderFactory(factory.MustNewName("kafka"), func(ctx context.Context, providerSettings factory.ProviderSettings, config auditor.Config) (auditor.Auditor, error) {
+		return newProvider(ctx, providerSettings, config, licensing, build)
+	})
+}
+
+func newProvider(_ context.Context, providerSettings factory.ProviderSettings, config auditor.Config, licensing licensing.Licensing, build version.Build) (auditor.Auditor, error) {
+	settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/ee/auditor/kafkaauditor")
+
+	// Acks from all in-sync replicas, idempotent writes and unbounded record retries are franz-go defaults and are kept as is.
+	// A delivery timeout or retry limit lets franz-go < v1.22.2 ack a record that was never written (twmb/franz-go#1484).
+	client, err := kgo.NewClient(
+		kgo.SeedBrokers(config.Kafka.Brokers...),
+		kgo.ClientID(config.Kafka.ClientID),
+		kgo.ProducerBatchCompression(compressionCodec(config.Kafka.Compression)),
+		kgo.ProducerLinger(0),
+		kgo.WithLogger(kslog.New(settings.Logger())),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	provider := &provider{
+		settings:  settings,
+		config:    config,
+		licensing: licensing,
+		build:     build,
+		marshaler: plog.ProtoMarshaler{},
+		client:    client,
+	}
+
+	server, err := auditorserver.New(settings,
+		auditorserver.Config{
+			BufferSize:    config.BufferSize,
+			BatchSize:     config.BatchSize,
+			FlushInterval: config.FlushInterval,
+		},
+		provider.export,
+	)
+	if err != nil {
+		client.Close()
+		return nil, err
+	}
+
+	provider.server = server
+	return provider, nil
+}
+
+func (provider *provider) Start(ctx context.Context) error {
+	return provider.server.Start(ctx)
+}
+
+func (provider *provider) Audit(ctx context.Context, event audittypes.AuditEvent) {
+	if event.PrincipalAttributes.PrincipalOrgID.IsZero() {
+		provider.settings.Logger().WarnContext(ctx, "audit event dropped as org_id is zero")
+		return
+	}
+
+	if _, err := provider.licensing.GetActive(ctx, event.PrincipalAttributes.PrincipalOrgID); err != nil {
+		return
+	}
+
+	provider.server.Add(ctx, event)
+}
+
+func (provider *provider) Healthy() <-chan struct{} {
+	return provider.server.Healthy()
+}
+
+func (provider *provider) Stop(ctx context.Context) error {
+	ctx, cancel := context.WithTimeout(ctx, provider.config.Kafka.ShutdownTimeout)
+	defer cancel()
+
+	stoppedC := make(chan error, 1)
+	go func() { stoppedC <- provider.server.Stop(ctx) }()
+
+	select {
+	case err := <-stoppedC:
+		provider.client.Close()
+		return err
+	case <-ctx.Done():
+		provider.settings.Logger().WarnContext(ctx, "audit drain timed out, closing kafka client", slog.Duration("shutdown_timeout", provider.config.Kafka.ShutdownTimeout))
+		// Closing fails buffered and in-flight records, so the blocked export and the rest of the drain return immediately.
+		provider.client.Close()
+		return <-stoppedC
+	}
+}
+
+func compressionCodec(name string) kgo.CompressionCodec {
+	switch name {
+	case "gzip":
+		return kgo.GzipCompression()
+	case "snappy":
+		return kgo.SnappyCompression()
+	case "lz4":
+		return kgo.Lz4Compression()
+	case "zstd":
+		return kgo.ZstdCompression()
+	default:
+		return kgo.NoCompression()
+	}
+}

ee/auditor/kafkaauditor/provider_test.go

@@ -0,0 +1,227 @@
+package kafkaauditor
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/twmb/franz-go/pkg/kerr"
+	"github.com/twmb/franz-go/pkg/kfake"
+	"github.com/twmb/franz-go/pkg/kgo"
+	"go.opentelemetry.io/collector/pdata/plog"
+
+	"github.com/SigNoz/signoz/pkg/auditor"
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/types/audittypes"
+	"github.com/SigNoz/signoz/pkg/types/coretypes"
+	"github.com/SigNoz/signoz/pkg/version"
+)
+
+func newTestCluster(t *testing.T, topics ...string) *kfake.Cluster {
+	t.Helper()
+
+	cluster, err := kfake.NewCluster(kfake.NumBrokers(1), kfake.SeedTopics(1, topics...))
+	require.NoError(t, err)
+	t.Cleanup(cluster.Close)
+
+	return cluster
+}
+
+func newTestProvider(t *testing.T, brokers []string, topic string, compression string) *provider {
+	t.Helper()
+
+	config := auditor.Config{
+		Provider:      "kafka",
+		BufferSize:    10,
+		BatchSize:     5,
+		FlushInterval: 50 * time.Millisecond,
+		Kafka: auditor.KafkaConfig{
+			Brokers:         brokers,
+			Topic:           topic,
+			ClientID:        "signoz-test",
+			Compression:     compression,
+			ShutdownTimeout: 30 * time.Second,
+		},
+	}
+
+	auditor, err := newProvider(context.Background(), instrumentationtest.New().ToProviderSettings(), config, nil, version.Info)
+	require.NoError(t, err)
+
+	return auditor.(*provider)
+}
+
+func newTestEvent(resource coretypes.Resource, action coretypes.Verb, body string) audittypes.AuditEvent {
+	return audittypes.AuditEvent{
+		Timestamp: time.Now(),
+		Body:      body,
+		EventName: audittypes.NewEventName(resource.Kind(), action),
+		AuditAttributes: audittypes.AuditAttributes{
+			Action:  action,
+			Outcome: audittypes.OutcomeSuccess,
+		},
+		ResourceAttributes: audittypes.ResourceAttributes{
+			Resource: resource,
+		},
+	}
+}
+
+// Polls the topic until the consumed records hold wantLogRecords audit log records in total.
+func consumeRecords(t *testing.T, brokers []string, topic string, wantLogRecords int) []*kgo.Record {
+	t.Helper()
+
+	client, err := kgo.NewClient(
+		kgo.SeedBrokers(brokers...),
+		kgo.ConsumeTopics(topic),
+		kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
+	)
+	require.NoError(t, err)
+	defer client.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	var records []*kgo.Record
+	gotLogRecords := 0
+	for gotLogRecords < wantLogRecords {
+		fetches := client.PollFetches(ctx)
+		require.NoError(t, ctx.Err(), "timed out waiting for %d log records, got %d", wantLogRecords, gotLogRecords)
+		require.Empty(t, fetches.Errors())
+
+		for _, record := range fetches.Records() {
+			logs, err := (&plog.ProtoUnmarshaler{}).UnmarshalLogs(record.Value)
+			require.NoError(t, err)
+			gotLogRecords += logs.LogRecordCount()
+			records = append(records, record)
+		}
+	}
+
+	return records
+}
+
+func TestExport(t *testing.T) {
+	testCases := []struct {
+		name            string
+		compression     string
+		compressionType uint8
+	}{
+		{name: "CompressionNone", compression: "none", compressionType: 0},
+		{name: "CompressionGzip", compression: "gzip", compressionType: 1},
+		{name: "CompressionSnappy", compression: "snappy", compressionType: 2},
+		{name: "CompressionLz4", compression: "lz4", compressionType: 3},
+		{name: "CompressionZstd", compression: "zstd", compressionType: 4},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			cluster := newTestCluster(t, "tenant_audit")
+			provider := newTestProvider(t, cluster.ListenAddrs(), "tenant_audit", testCase.compression)
+			defer provider.client.Close()
+
+			events := []audittypes.AuditEvent{
+				newTestEvent(coretypes.ResourceUser, coretypes.VerbUpdate, "user updated"),
+				newTestEvent(coretypes.ResourceMetaResourceDashboard, coretypes.VerbCreate, "dashboard created"),
+				newTestEvent(coretypes.ResourceMetaResourceDashboard, coretypes.VerbDelete, "dashboard deleted"),
+			}
+			require.NoError(t, provider.export(context.Background(), events))
+
+			records := consumeRecords(t, cluster.ListenAddrs(), "tenant_audit", len(events))
+			require.Len(t, records, 1)
+			assert.Equal(t, "tenant_audit", records[0].Topic)
+			assert.Equal(t, testCase.compressionType, records[0].Attrs.CompressionType())
+
+			logs, err := (&plog.ProtoUnmarshaler{}).UnmarshalLogs(records[0].Value)
+			require.NoError(t, err)
+			assert.Equal(t, len(events), logs.LogRecordCount())
+
+			serviceName, ok := logs.ResourceLogs().At(0).Resource().Attributes().Get("service.name")
+			require.True(t, ok)
+			assert.Equal(t, "signoz", serviceName.Str())
+		})
+	}
+}
+
+func TestExport_RecordTooLarge(t *testing.T) {
+	cluster := newTestCluster(t, "large_audit")
+	provider := newTestProvider(t, cluster.ListenAddrs(), "large_audit", "none")
+	defer provider.client.Close()
+
+	event := newTestEvent(coretypes.ResourceUser, coretypes.VerbCreate, strings.Repeat("a", 2<<20))
+
+	err := provider.export(context.Background(), []audittypes.AuditEvent{event})
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, kerr.MessageTooLarge))
+}
+
+func TestExport_UnknownTopic(t *testing.T) {
+	cluster := newTestCluster(t)
+	provider := newTestProvider(t, cluster.ListenAddrs(), "missing_audit", "lz4")
+	defer provider.client.Close()
+
+	err := provider.export(context.Background(), []audittypes.AuditEvent{newTestEvent(coretypes.ResourceUser, coretypes.VerbCreate, "user created")})
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, kerr.UnknownTopicOrPartition))
+}
+
+func TestExport_BrokerDown_RetriesUntilContextDone(t *testing.T) {
+	cluster := newTestCluster(t, "down_audit")
+	brokers := cluster.ListenAddrs()
+	cluster.Close()
+
+	provider := newTestProvider(t, brokers, "down_audit", "lz4")
+	defer provider.client.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	start := time.Now()
+	err := provider.export(ctx, []audittypes.AuditEvent{newTestEvent(coretypes.ResourceUser, coretypes.VerbCreate, "user created")})
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, context.DeadlineExceeded))
+	assert.GreaterOrEqual(t, time.Since(start), 500*time.Millisecond)
+}
+
+func TestStop_DrainsQueuedEvents(t *testing.T) {
+	cluster := newTestCluster(t, "drain_audit")
+	provider := newTestProvider(t, cluster.ListenAddrs(), "drain_audit", "lz4")
+
+	done := make(chan error, 1)
+	go func() { done <- provider.Start(context.Background()) }()
+	<-provider.Healthy()
+
+	for range 7 {
+		provider.server.Add(context.Background(), newTestEvent(coretypes.ResourceUser, coretypes.VerbUpdate, "user updated"))
+	}
+
+	require.NoError(t, provider.Stop(context.Background()))
+	require.NoError(t, <-done)
+
+	records := consumeRecords(t, cluster.ListenAddrs(), "drain_audit", 7)
+	assert.NotEmpty(t, records)
+}
+
+func TestStop_BrokerDown_ClosesAfterShutdownTimeout(t *testing.T) {
+	cluster := newTestCluster(t, "stuck_audit")
+	brokers := cluster.ListenAddrs()
+	cluster.Close()
+
+	provider := newTestProvider(t, brokers, "stuck_audit", "lz4")
+	provider.config.Kafka.ShutdownTimeout = 500 * time.Millisecond
+
+	done := make(chan error, 1)
+	go func() { done <- provider.Start(context.Background()) }()
+	<-provider.Healthy()
+
+	for range 7 {
+		provider.server.Add(context.Background(), newTestEvent(coretypes.ResourceUser, coretypes.VerbUpdate, "user updated"))
+	}
+
+	start := time.Now()
+	require.NoError(t, provider.Stop(context.Background()))
+	require.NoError(t, <-done)
+	assert.GreaterOrEqual(t, time.Since(start), 500*time.Millisecond)
+	assert.Less(t, time.Since(start), 5*time.Second)
+}

go.mod

@@ -61,6 +61,9 @@ require (
 	github.com/swaggest/jsonschema-go v0.3.78
 	github.com/swaggest/rest v0.2.75
 	github.com/tidwall/gjson v1.18.0
+	github.com/twmb/franz-go v1.21.7
+	github.com/twmb/franz-go/pkg/kfake v0.0.0-20260421215025-4e7a1e1569ac
+	github.com/twmb/franz-go/plugin/kslog v1.0.0
 	github.com/uptrace/bun v1.2.9
 	github.com/uptrace/bun/dialect/pgdialect v1.2.9
 	github.com/uptrace/bun/dialect/sqlitedialect v1.2.9
@@ -144,6 +147,7 @@ require (
 	github.com/swaggest/refl v1.4.0 // indirect
 	github.com/swaggest/usecase v1.3.1 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.13.1 // indirect
 	github.com/ugorji/go/codec v1.3.0 // indirect
 	github.com/uptrace/opentelemetry-go-extra/otelsql v0.3.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
@@ -243,7 +247,7 @@ require (
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/jpillora/backoff v1.0.0 // indirect
 	github.com/jtolds/gls v4.20.0+incompatible // indirect
-	github.com/klauspost/compress v1.18.5 // indirect
+	github.com/klauspost/compress v1.19.2 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
 	github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
 	github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
@@ -272,7 +276,7 @@ require (
 	github.com/openfga/openfga v1.14.1
 	github.com/paulmach/orb v0.12.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
-	github.com/pierrec/lz4/v4 v4.1.25 // indirect
+	github.com/pierrec/lz4/v4 v4.1.26 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
 	github.com/pressly/goose/v3 v3.27.0 // indirect

go.sum

@@ -675,8 +675,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
-github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
-github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8=
+github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
 github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
 github.com/knadh/koanf v1.5.0 h1:q2TSd/3Pyc/5yP9ldIrSdIz26MCcyNQzW0pEAugLPNs=
@@ -853,8 +853,8 @@ github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8
 github.com/perses/spec v0.1.2 h1:yGoygcR3ZusuGDCmRMwsVXCMvMwi1qZndKV6NYNreEw=
 github.com/perses/spec v0.1.2/go.mod h1:NoGI5jmGwRdkdPgyYSZJTBL4/Py+dqIPKS2QV8NOvGE=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
-github.com/pierrec/lz4/v4 v4.1.25 h1:kocOqRffaIbU5djlIBr7Wh+cx82C0vtFb0fOurZHqD0=
-github.com/pierrec/lz4/v4 v4.1.25/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4=
+github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY=
+github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4=
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ=
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
@@ -1055,6 +1055,16 @@ github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc/go.mod h1:bciPuU6GH
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
 github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
+github.com/twmb/franz-go v1.21.7 h1:/DkA/o8wQN55gZWtpj2QNb9SIdxwFR7M+NecQWMdmc0=
+github.com/twmb/franz-go v1.21.7/go.mod h1:89kLt1uhE1GkyossLHGdpAMFNK9mV8GYk1lfWu9FiNs=
+github.com/twmb/franz-go/pkg/kadm v1.18.0 h1:WRf/LZmDdcDXwX7WMbtDU++v+b3NzYh2bCGoPMmzirw=
+github.com/twmb/franz-go/pkg/kadm v1.18.0/go.mod h1:XeLhGoLXLFzK8/ryv5FfpxPxGwj4oFEGpPJMB/x6KDE=
+github.com/twmb/franz-go/pkg/kfake v0.0.0-20260421215025-4e7a1e1569ac h1:6Dn8at7J4+5A8H7NgVLcPj4dfVm4pRtUZMHxXmBKxJ4=
+github.com/twmb/franz-go/pkg/kfake v0.0.0-20260421215025-4e7a1e1569ac/go.mod h1:6ofUZL6jAhajJ5WAaJTX9rK3B1ozGDOwKyZczdGK8uE=
+github.com/twmb/franz-go/pkg/kmsg v1.13.1 h1:fG5kItwysTk5UXqVwb64EpQEy3TydF3vYYK21nUQ+bI=
+github.com/twmb/franz-go/pkg/kmsg v1.13.1/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY=
+github.com/twmb/franz-go/plugin/kslog v1.0.0 h1:I64oEmF+0PDvmyLgwrlOtg4mfpSE9GwlcLxM4af2t60=
+github.com/twmb/franz-go/plugin/kslog v1.0.0/go.mod h1:8pMjK3OJJJNNYddBSbnXZkIK5dCKFIk9GcVVCDgvnQc=
 github.com/twmb/murmur3 v1.1.8 h1:8Yt9taO/WN3l08xErzjeschgZU2QSrwm1kclYq+0aRg=
 github.com/twmb/murmur3 v1.1.8/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
 github.com/ua-parser/uap-go v0.0.0-20240611065828-3a4781585db6 h1:SIKIoA4e/5Y9ZOl0DCe3eVMLPOQzJxgZpfdHHeauNTM=

pkg/auditor/config.go

@@ -27,6 +27,8 @@ type Config struct {
 	OTLPHTTP OTLPHTTPConfig `mapstructure:"otlphttp"`
 
 	File FileConfig `mapstructure:"file"`
+
+	Kafka KafkaConfig `mapstructure:"kafka"`
 }
 
 // OTLPHTTPConfig holds configuration for the OTLP HTTP exporter provider.
@@ -54,6 +56,23 @@ type FileConfig struct {
 	Path string `mapstructure:"path"`
 }
 
+type KafkaConfig struct {
+	// Brokers are the seed broker addresses (host:port) used to discover the cluster.
+	Brokers []string `mapstructure:"brokers"`
+
+	// Topic must already exist; the producer does not create it.
+	Topic string `mapstructure:"topic"`
+
+	// ClientID identifies this producer in broker logs and metrics.
+	ClientID string `mapstructure:"client_id"`
+
+	// Compression is the codec for produced batches: none, gzip, snappy, lz4 or zstd.
+	Compression string `mapstructure:"compression"`
+
+	// ShutdownTimeout bounds draining buffered events on stop; events still undelivered when it expires are dropped.
+	ShutdownTimeout time.Duration `mapstructure:"shutdown_timeout"`
+}
+
 // RetryConfig configures exponential backoff for the OTLP HTTP exporter.
 type RetryConfig struct {
 	// Enabled controls whether retries are attempted on transient failures.
@@ -89,6 +108,12 @@ func newConfig() factory.Config {
 				MaxElapsedTime:  time.Minute,
 			},
 		},
+		Kafka: KafkaConfig{
+			Brokers:         []string{"localhost:9092"},
+			ClientID:        "signoz",
+			Compression:     "lz4",
+			ShutdownTimeout: 30 * time.Second,
+		},
 	}
 }
 
@@ -125,5 +150,25 @@ func (c Config) Validate() error {
 		}
 	}
 
+	if c.Provider == "kafka" {
+		if len(c.Kafka.Brokers) == 0 {
+			return errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::kafka::brokers must be set when provider is kafka")
+		}
+
+		if c.Kafka.Topic == "" {
+			return errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::kafka::topic must be set when provider is kafka")
+		}
+
+		switch c.Kafka.Compression {
+		case "none", "gzip", "snappy", "lz4", "zstd":
+		default:
+			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::kafka::compression must be one of none, gzip, snappy, lz4 or zstd, got %q", c.Kafka.Compression)
+		}
+
+		if c.Kafka.ShutdownTimeout <= 0 {
+			return errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "auditor::kafka::shutdown_timeout must be greater than 0")
+		}
+	}
+
 	return nil
 }

tests/fixtures/querier.py

@@ -173,7 +173,7 @@ def make_query_request(
     end_ms: int,
     queries: list[dict],
     *,
-    request_type: str,
+    request_type: str = RequestType.TIME_SERIES,
     format_options: dict | None = None,
     variables: dict | None = None,
     no_cache: bool = True,
@@ -209,7 +209,7 @@ def make_preview_query_request(
     end_ms: int,
     queries: list[dict],
     *,
-    request_type: str,
+    request_type: str = RequestType.TIME_SERIES,
     format_options: dict | None = None,
     variables: dict | None = None,
     verbose: bool = True,
@@ -282,7 +282,6 @@ def query_metric_values(  # pylint: disable=too-many-arguments,too-many-position
         start_ms=start_epoch * 1000,
         end_ms=end_epoch * 1000,
         queries=[build_builder_query("A", metric_name, time_agg, space_agg, step_interval=step_interval)],
-        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
     return sorted(get_series_values(response.json(), "A"), key=lambda v: v["timestamp"])
@@ -680,7 +679,7 @@ def assert_raw_row_subset(
     assert expected.items() <= data.items(), f"expected subset {expected}, got data {data}"
 
 
-def build_aggregation_query(
+def build_scalar_query(
     name: str,
     signal: str,
     aggregations: list[dict],
@@ -769,7 +768,7 @@ def build_heatmap_query(
     return {"type": "builder_query", "spec": spec}
 
 
-def build_traces_aggregation_query(
+def build_traces_scalar_query(
     aggregations: list[dict],
     *,
     name: str = "A",
@@ -782,7 +781,8 @@ def build_traces_aggregation_query(
     disabled: bool = False,
     functions: list[dict] | None = None,
 ) -> dict:
-    return build_aggregation_query(
+    """build_scalar_query pinned to the traces signal, with name defaulting to 'A'."""
+    return build_scalar_query(
         name=name,
         signal="traces",
         aggregations=aggregations,
@@ -1321,7 +1321,7 @@ def run_query_case(signoz: types.SigNoz, token: str, now: datetime, case: dict[s
             aggregations = aggregation
         else:
             aggregations = []
-        query = build_aggregation_query(
+        query = build_scalar_query(
             name=case["name"],
             signal="logs",
             aggregations=aggregations,

tests/integration/tests/auditquerier/01_audit_logs.py

@@ -10,8 +10,8 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.querier import (
     BuilderQuery,
     build_aggregation,
-    build_aggregation_query,
     build_order_by,
+    build_scalar_query,
     make_query_request,
 )
 
@@ -361,7 +361,7 @@ def test_audit_scalar_count_failures(
         start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
         end_ms=int(now.timestamp() * 1000),
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="logs",
                 source="audit",

tests/integration/tests/metricreduction/03_reduced_recent_window.py

@@ -6,7 +6,6 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metricreduction import build_recent_gauge_data
 from fixtures.querier import (
-    RequestType,
     aligned_epoch,
     build_builder_query,
     get_all_series,
@@ -60,7 +59,6 @@ def test_recent_queries_group_by_full_labels(
         start_ms=base_epoch * 1000,
         end_ms=(base_epoch + MINUTES * 60) * 1000,
         queries=[build_builder_query("A", metric_name, "sum", "sum", step_interval=300, group_by=["service"])],
-        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
 

tests/integration/tests/promqlconformance/01_upstream_corpus.py

@@ -8,7 +8,7 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.promqltestcorpus import ingest_promqltest_corpus
-from fixtures.querier import RequestType, get_all_series, make_query_request
+from fixtures.querier import get_all_series, make_query_request
 
 TESTDATA_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "testdata")
 # The corpus (see fixtures/promqltestcorpus.py) is frozen from Prometheus' own
@@ -57,7 +57,7 @@ def test_upstream_promqltest_corpus(
         }
         case_id = f"{case['source']}[{case['variant']}]"
 
-        response = make_query_request(signoz, token, req_start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+        response = make_query_request(signoz, token, req_start_ms, end_ms, [query])
         if response.status_code != HTTPStatus.OK:
             failures.append(f"{case_id}: HTTP {response.status_code} for {case['expr']!r}: {response.text[:200]}")
             continue

tests/integration/tests/promqlconformance/02_fingerprint_probe.py

@@ -5,7 +5,7 @@ from http import HTTPStatus
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import RequestType, get_all_series, make_query_request
+from fixtures.querier import get_all_series, make_query_request
 
 METRIC = "probe_requests"
 EVOLVED_METRIC = "probe_schema_evolution"
@@ -74,7 +74,7 @@ def test_identical_labelsets_merge_and_grouping(
 
     def run(promql: str, start_ms: int, end_ms: int) -> list[tuple[dict, list]]:
         q = {"type": "promql", "spec": {"name": "A", "query": promql, "step": 60, "skipStepAlignment": True}}
-        resp = make_query_request(signoz, token, start_ms, end_ms, [q], request_type=RequestType.TIME_SERIES)
+        resp = make_query_request(signoz, token, start_ms, end_ms, [q])
         assert resp.status_code == HTTPStatus.OK, f"{promql!r}: {resp.text[:300]}"
         out = []
         for series in get_all_series(resp.json(), "A") or []:

tests/integration/tests/promqlconformance/03_nonfinite_cache.py

@@ -6,7 +6,7 @@ from uuid import uuid4
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import RequestType, get_all_series, make_query_request
+from fixtures.querier import get_all_series, make_query_request
 
 HOUR_MS = 3_600_000
 SAMPLE_INTERVAL_MS = 60_000
@@ -40,7 +40,7 @@ def test_promql_ratio_with_zero_denominator_is_dropped_and_cached(
 
     def run() -> tuple[dict[str, dict[int, object]], int]:
         query = {"type": "promql", "spec": {"name": "A", "query": promql}}
-        response = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
+        response = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
         assert response.status_code == HTTPStatus.OK, response.text[:300]
         body = response.json()
         out: dict[str, dict[int, object]] = {}

tests/integration/tests/promqlconformance/04_no_step_subquery.py

@@ -6,7 +6,7 @@ from uuid import uuid4
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import RequestType, get_all_series, make_query_request
+from fixtures.querier import get_all_series, make_query_request
 
 MINUTE_MS = 60_000
 
@@ -41,7 +41,7 @@ def test_promql_subquery_without_step_evaluates(
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     query = {"type": "promql", "spec": {"name": "A", "query": f"max_over_time({metric}[5m:])", "skipStepAlignment": True}}
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK, response.text[:300]
     series = get_all_series(response.json(), "A")
     assert series, "the subquery must return the inserted series"
@@ -55,6 +55,5 @@ def test_promql_subquery_without_step_evaluates(
         start_ms,
         end_ms,
         [{"type": "promql", "spec": {"name": "A", "query": metric, "skipStepAlignment": True}}],
-        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text[:300]

tests/integration/tests/promqlconformance/05_or_subquery_duplicate_labelset.py

@@ -8,7 +8,7 @@ import requests
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import RequestType, get_all_series, make_query_request
+from fixtures.querier import get_all_series, make_query_request
 
 MINUTE_MS = 60_000
 QUERY_TIMEOUT = 30
@@ -73,7 +73,7 @@ def test_or_arms_merge_under_subquery_name_drop(
 
     # query_range over the same grid must agree point for point.
     spec = {"name": "A", "query": query, "step": 300, "skipStepAlignment": True}
-    response = make_query_request(signoz, token, start_ms, end_ms, [{"type": "promql", "spec": spec}], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [{"type": "promql", "spec": spec}])
     assert response.status_code == HTTPStatus.OK, response.text[:300]
     series = get_all_series(response.json(), "A")
     assert len(series) == 1, f"both or arms must merge into one series: {series}"

tests/integration/tests/querier_json_body/02_non_body_fields.py

@@ -7,10 +7,10 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_order_by,
     build_raw_query,
+    build_scalar_query,
     get_rows,
     get_scalar_table_data,
     make_query_request,
@@ -136,7 +136,7 @@ def test_non_body_filter_groupby_aggregation(
                 step_interval=60,
             )
         else:
-            query = build_aggregation_query(
+            query = build_scalar_query(
                 name=case["name"],
                 signal="logs",
                 aggregations=[build_aggregation(case["aggregation"])],
@@ -233,7 +233,7 @@ def test_non_body_orderby(
                 step_interval=60,
             )
         else:
-            query = build_aggregation_query(
+            query = build_scalar_query(
                 name=case["name"],
                 signal="logs",
                 aggregations=[build_aggregation(case["aggregation"])],

tests/integration/tests/querier_json_body/04_unknown_key_collisions.py

@@ -12,9 +12,9 @@ from fixtures.logs import Logs
 from fixtures.querier import (
     RequestType,
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_raw_query,
+    build_scalar_query,
     get_rows,
     get_scalar_table_data,
     make_query_request,
@@ -68,7 +68,7 @@ def test_json_body_unknown_key_collision_groupby(
         token,
         now,
         [
-            build_aggregation_query(
+            build_scalar_query(
                 "A",
                 "logs",
                 [build_aggregation("count()")],
@@ -175,7 +175,7 @@ def test_json_body_unknown_key_attribute_parity(
             token,
             now,
             [
-                build_aggregation_query(
+                build_scalar_query(
                     "A",
                     "logs",
                     [build_aggregation("count()")],
@@ -225,7 +225,7 @@ def test_json_body_unknown_key_aggregation_over_body(
         token,
         now,
         [
-            build_aggregation_query(
+            build_scalar_query(
                 "A",
                 "logs",
                 [build_aggregation("count(field)", "cnt"), build_aggregation("count_distinct(field)", "distinct")],
@@ -291,7 +291,7 @@ def test_json_body_unknown_key_body_path_parity(
             token,
             now,
             [
-                build_aggregation_query(
+                build_scalar_query(
                     "A",
                     "logs",
                     [build_aggregation("count()")],
@@ -308,7 +308,7 @@ def test_json_body_unknown_key_body_path_parity(
             token,
             now,
             [
-                build_aggregation_query(
+                build_scalar_query(
                     "A",
                     "logs",
                     [build_aggregation(f"count_distinct({field})")],

tests/integration/tests/querier_json_body/07_field_name_quoting.py

@@ -8,7 +8,7 @@ import pytest
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
-from fixtures.querier import build_aggregation, build_aggregation_query, build_raw_query, get_rows, get_scalar_table_data, make_query_request
+from fixtures.querier import build_aggregation, build_raw_query, build_scalar_query, get_rows, get_scalar_table_data, make_query_request
 
 SERVICE = "quoting-json-svc"
 
@@ -55,7 +55,7 @@ def test_body_json_group_by_and_select_hostile_key(
         token,
         start_ms,
         end_ms,
-        [build_aggregation_query("A", "logs", [build_aggregation("count()")], group_by=[field], order=[{"key": {"name": key}, "direction": "asc"}], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_scalar_query("A", "logs", [build_aggregation("count()")], group_by=[field], order=[{"key": {"name": key}, "direction": "asc"}], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text

tests/integration/tests/queriercommon/03_clickhouse_queries.py

@@ -118,7 +118,7 @@ INVALID_REGEXP = "write the pattern in RE2 syntax, which has no lookaround or ba
             id="regexp_in_sql",
         ),
         pytest.param(
-            querier.build_aggregation_query("A", "logs", [querier.build_aggregation("count()")], filter_expression="body REGEXP '(?=a)'"),
+            querier.build_scalar_query("A", "logs", [querier.build_aggregation("count()")], filter_expression="body REGEXP '(?=a)'"),
             "invalid_regexp",
             INVALID_REGEXP,
             id="regexp_in_filter",

tests/integration/tests/queriercommon/06_keyless_semantics.py

@@ -195,7 +195,6 @@ def test_metrics_negative_operators_include_keyless_series(
                 filter_expression=expression,
             )
         ],
-        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
 

tests/integration/tests/queriercommon/08_field_name_quoting.py

@@ -9,11 +9,10 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    RequestType,
     build_aggregation,
-    build_aggregation_query,
     build_builder_query,
     build_raw_query,
+    build_scalar_query,
     get_all_series,
     get_rows,
     get_scalar_table_data,
@@ -67,7 +66,7 @@ def test_logs_group_by_order_by_and_select_hostile_attribute(
         token,
         start_ms,
         end_ms,
-        [build_aggregation_query("A", "logs", [build_aggregation("count()")], group_by=[field], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_scalar_query("A", "logs", [build_aggregation("count()")], group_by=[field], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text
@@ -79,7 +78,7 @@ def test_logs_group_by_order_by_and_select_hostile_attribute(
         token,
         start_ms,
         end_ms,
-        [build_aggregation_query("A", "logs", [build_aggregation("count()")], group_by=[field], order=[{"key": {"name": key}, "direction": "asc"}], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_scalar_query("A", "logs", [build_aggregation("count()")], group_by=[field], order=[{"key": {"name": key}, "direction": "asc"}], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text
@@ -171,7 +170,7 @@ def test_logs_aggregation_with_dollar_in_key(
             token,
             start_ms,
             end_ms,
-            [build_aggregation_query("A", "logs", [build_aggregation(expression)], filter_expression=f"service.name = '{SERVICE}'")],
+            [build_scalar_query("A", "logs", [build_aggregation(expression)], filter_expression=f"service.name = '{SERVICE}'")],
             request_type="scalar",
         )
         assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
@@ -202,7 +201,7 @@ def test_logs_group_by_hostile_resource_attribute(
         token,
         int((now - timedelta(minutes=2)).timestamp() * 1000),
         int((now + timedelta(seconds=1)).timestamp() * 1000),
-        [build_aggregation_query("A", "logs", [build_aggregation("count()")], group_by=[{"name": key, "fieldContext": "resource", "fieldDataType": "string"}], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_scalar_query("A", "logs", [build_aggregation("count()")], group_by=[{"name": key, "fieldContext": "resource", "fieldDataType": "string"}], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text
@@ -237,7 +236,7 @@ def test_traces_group_by_and_select_hostile_attribute(
         token,
         start_ms,
         end_ms,
-        [build_aggregation_query("A", "traces", [build_aggregation("count()")], group_by=[field], filter_expression=f"service.name = '{SERVICE}'")],
+        [build_scalar_query("A", "traces", [build_aggregation("count()")], group_by=[field], filter_expression=f"service.name = '{SERVICE}'")],
         request_type="scalar",
     )
     assert response.status_code == HTTPStatus.OK, response.text
@@ -275,7 +274,6 @@ def test_metrics_group_by_hostile_label(
         int((now - timedelta(minutes=5)).timestamp() * 1000),
         int((now + timedelta(minutes=1)).timestamp() * 1000),
         [build_builder_query("A", metric, "avg", "avg", group_by=[label])],
-        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
     labels = {tuple(sorted((entry["key"]["name"], entry["value"]) for entry in item.get("labels", []))) for item in get_all_series(response.json(), "A")}

tests/integration/tests/queriercommon/09_time_range.py

@@ -12,7 +12,6 @@ from fixtures.querier import (
     Aggregation,
     BuilderQuery,
     MetricAggregation,
-    RequestType,
     get_all_warnings,
     get_error_message,
     get_series_values,
@@ -48,7 +47,6 @@ def test_time_range_rejected(
         start_ms=start_ms,
         end_ms=end_ms,
         queries=[BuilderQuery(signal="logs", aggregations=[Aggregation(expression="count()")], step_interval=60).to_dict()],
-        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.BAD_REQUEST
@@ -81,7 +79,6 @@ def test_two_year_logs_timeseries_step_clamped(
             ).to_dict()
         ],
         format_options={"formatTableResultForUI": False, "fillGaps": True},
-        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK
@@ -120,7 +117,6 @@ def test_two_year_traces_timeseries_step_clamped(
             ).to_dict()
         ],
         format_options={"formatTableResultForUI": False, "fillGaps": True},
-        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK
@@ -159,7 +155,6 @@ def test_two_year_metrics_timeseries_step_clamped(
             ).to_dict()
         ],
         format_options={"formatTableResultForUI": False, "fillGaps": True},
-        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK

tests/integration/tests/queriercommon/10_cache_time_shift.py

@@ -7,11 +7,10 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
-    RequestType,
     build_aggregation,
-    build_aggregation_query,
     build_function,
     build_group_by_field,
+    build_scalar_query,
     get_all_series,
     index_series_by_label,
     make_query_request,
@@ -59,7 +58,7 @@ def test_time_shift_partial_hit_matches_no_cache(
         ]
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],
@@ -72,11 +71,11 @@ def test_time_shift_partial_hit_matches_no_cache(
     first_end_ms = int((now - 30 * MINUTE).timestamp() * 1000)
     second_end_ms = int((now - 10 * MINUTE).timestamp() * 1000)
 
-    warm = make_query_request(signoz, token, start_ms, first_end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
+    warm = make_query_request(signoz, token, start_ms, first_end_ms, [query], no_cache=False)
     assert warm.status_code == HTTPStatus.OK, warm.text
-    cached = make_query_request(signoz, token, start_ms, second_end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
+    cached = make_query_request(signoz, token, start_ms, second_end_ms, [query], no_cache=False)
     assert cached.status_code == HTTPStatus.OK, cached.text
-    fresh = make_query_request(signoz, token, start_ms, second_end_ms, [query], no_cache=True, request_type=RequestType.TIME_SERIES)
+    fresh = make_query_request(signoz, token, start_ms, second_end_ms, [query], no_cache=True)
     assert fresh.status_code == HTTPStatus.OK, fresh.text
 
     fresh_points = {service: {point["timestamp"]: point["value"] for point in series["values"]} for service, series in index_series_by_label(get_all_series(fresh.json(), "A"), "service.name").items()}

tests/integration/tests/queriercommon/12_cache_limited_group_by.py

@@ -7,11 +7,10 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
-    RequestType,
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_order_by,
+    build_scalar_query,
     get_all_series,
     index_series_by_label,
     make_query_request,
@@ -49,7 +48,7 @@ def test_limited_group_by_partial_hit_matches_no_cache(
         ]
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],
@@ -63,11 +62,11 @@ def test_limited_group_by_partial_hit_matches_no_cache(
     head_end_ms = int((head_start + 20 * MINUTE).timestamp() * 1000)
     tail_end_ms = int((head_start + 40 * MINUTE).timestamp() * 1000)
 
-    warm = make_query_request(signoz, token, head_start_ms, head_end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
+    warm = make_query_request(signoz, token, head_start_ms, head_end_ms, [query], no_cache=False)
     assert warm.status_code == HTTPStatus.OK, warm.text
-    cached = make_query_request(signoz, token, head_start_ms, tail_end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
+    cached = make_query_request(signoz, token, head_start_ms, tail_end_ms, [query], no_cache=False)
     assert cached.status_code == HTTPStatus.OK, cached.text
-    fresh = make_query_request(signoz, token, head_start_ms, tail_end_ms, [query], no_cache=True, request_type=RequestType.TIME_SERIES)
+    fresh = make_query_request(signoz, token, head_start_ms, tail_end_ms, [query], no_cache=True)
     assert fresh.status_code == HTTPStatus.OK, fresh.text
 
     fresh_points = {service: {point["timestamp"]: point["value"] for point in series["values"]} for service, series in index_series_by_label(get_all_series(fresh.json(), "A"), "service.name").items()}
@@ -107,7 +106,7 @@ def test_limited_group_by_shrunk_window_matches_no_cache(
         ]
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],
@@ -121,11 +120,11 @@ def test_limited_group_by_shrunk_window_matches_no_cache(
     middle_ms = int((start + 5 * MINUTE).timestamp() * 1000)
     end_ms = int((start + 10 * MINUTE).timestamp() * 1000)
 
-    warm = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
+    warm = make_query_request(signoz, token, start_ms, end_ms, [query], no_cache=False)
     assert warm.status_code == HTTPStatus.OK, warm.text
-    cached = make_query_request(signoz, token, middle_ms, end_ms, [query], no_cache=False, request_type=RequestType.TIME_SERIES)
+    cached = make_query_request(signoz, token, middle_ms, end_ms, [query], no_cache=False)
     assert cached.status_code == HTTPStatus.OK, cached.text
-    fresh = make_query_request(signoz, token, middle_ms, end_ms, [query], no_cache=True, request_type=RequestType.TIME_SERIES)
+    fresh = make_query_request(signoz, token, middle_ms, end_ms, [query], no_cache=True)
     assert fresh.status_code == HTTPStatus.OK, fresh.text
 
     fresh_points = {service: {point["timestamp"]: point["value"] for point in series["values"]} for service, series in index_series_by_label(get_all_series(fresh.json(), "A"), "service.name").items()}

tests/integration/tests/querierlogs/04_formula.py

@@ -9,10 +9,10 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
     build_aggregation,
-    build_aggregation_query,
     build_formula_query,
     build_group_by_field,
     build_order_by,
+    build_scalar_query,
     find_named_result,
     get_all_series,
     get_error_message,
@@ -101,7 +101,7 @@ def test_logs_formula_orderby_and_limit(
         end_ms=int(now.timestamp() * 1000),
         request_type="scalar",
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -109,7 +109,7 @@ def test_logs_formula_orderby_and_limit(
                 filter_expression="severity_text = 'ERROR'",
                 disabled=True,
             ),
-            build_aggregation_query(
+            build_scalar_query(
                 name="B",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -240,8 +240,8 @@ def test_logs_formula_having_time_series_matches_query_having(
         end_ms=int(now.timestamp() * 1000),
         request_type="time_series",
         queries=[
-            build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
-            build_aggregation_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
+            build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
+            build_scalar_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
             build_formula_query("F1", "A", having_expression=formula_having),
         ],
     )
@@ -290,8 +290,8 @@ def test_logs_formula_having_scalar_matches_query_having(
         end_ms=int(now.timestamp() * 1000),
         request_type="scalar",
         queries=[
-            build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
-            build_aggregation_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
+            build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
+            build_scalar_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
             build_formula_query("F1", "A", having_expression=formula_having),
         ],
     )
@@ -325,7 +325,7 @@ def test_logs_formula_having_invalid_reference(
         end_ms=int(now.timestamp() * 1000),
         request_type="time_series",
         queries=[
-            build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], disabled=True),
+            build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], disabled=True),
             build_formula_query("F1", "A", having_expression="count() > 0"),
         ],
     )

tests/integration/tests/querierlogs/11_aggregation_functions.py

@@ -10,9 +10,9 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_order_by,
+    build_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -72,7 +72,7 @@ def test_logs_aggregate_functions(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[
@@ -165,7 +165,7 @@ def test_logs_aggregate_if_predicates_use_filter_syntax(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[
@@ -240,7 +240,7 @@ def test_logs_aggregate_arguments_are_field_names(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[
@@ -288,7 +288,7 @@ def test_logs_aggregate_invalid_expression_is_bad_request(
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation(expression)])
+    query = build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation(expression)])
     response = make_scalar_query_request(signoz, token, now, [query])
 
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
@@ -335,7 +335,7 @@ def test_logs_aggregate_having(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation(expr) for expr in aggregations],

tests/integration/tests/querierlogs/12_aggregation_timeseries.py

@@ -10,9 +10,9 @@ from fixtures.logs import Logs
 from fixtures.querier import (
     RequestType,
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_order_by,
+    build_scalar_query,
     get_all_series,
     get_scalar_table_data,
     index_series_by_label,
@@ -64,7 +64,7 @@ def test_logs_aggregate_min_max(
         end_ms=int((now + timedelta(seconds=5)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[
@@ -135,7 +135,7 @@ def test_logs_aggregate_time_series_limit(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -187,7 +187,7 @@ def test_logs_aggregate_multi_group_by(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],

tests/integration/tests/querierlogs/13_unknown_key_collisions.py

@@ -10,9 +10,9 @@ from fixtures.logs import Logs
 from fixtures.querier import (
     RequestType,
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_raw_query,
+    build_scalar_query,
     get_rows,
     get_scalar_table_data,
     make_query_request,
@@ -92,7 +92,7 @@ def test_logs_unknown_key_collision_groupby_parity(
             token,
             now,
             [
-                build_aggregation_query(
+                build_scalar_query(
                     "A",
                     "logs",
                     [build_aggregation("count()")],
@@ -210,7 +210,7 @@ def test_logs_unknown_key_collision_aggregation_parity(
             token,
             now,
             [
-                build_aggregation_query(
+                build_scalar_query(
                     "A",
                     "logs",
                     [build_aggregation(f"count({key})", "cnt"), build_aggregation(f"count_distinct({key})", "distinct")],

tests/integration/tests/querierlogs/17_heatmap.py

@@ -11,12 +11,12 @@ from fixtures.querier import (
     RequestType,
     assert_distribution_sums_heatmap,
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_heatmap_query,
     build_linear_bucket_options,
     build_log_bucket_options,
     build_order_by,
+    build_scalar_query,
     get_all_series,
     get_buckets,
     get_error_message,
@@ -200,7 +200,7 @@ def test_heatmap_rejections(
         token,
         int((now - timedelta(minutes=5)).timestamp() * 1000),
         int(now.timestamp() * 1000),
-        [build_aggregation_query("A", "logs", aggregations)],
+        [build_scalar_query("A", "logs", aggregations)],
         request_type=RequestType.HEATMAP,
     )
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text

tests/integration/tests/queriermetrics/02_warnings.py

@@ -7,7 +7,6 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    RequestType,
     build_builder_query,
     get_all_warnings,
     make_query_request,
@@ -51,12 +50,12 @@ def test_histogram_p90_returns_warning_outside_data_window(
     end_ms = int(now.timestamp() * 1000)
 
     start_4h = int((now - timedelta(hours=4)).timestamp() * 1000)
-    response = make_query_request(signoz, token, start_4h, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_4h, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
     assert response.json()["status"] == "success"
 
     start_15m = int((now - timedelta(minutes=15)).timestamp() * 1000)
-    response = make_query_request(signoz, token, start_15m, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_15m, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
     data = response.json()
     warnings = get_all_warnings(data)
@@ -84,7 +83,7 @@ def test_non_existent_metrics_returns_warning(
     end_ms = int(now.timestamp() * 1000)
 
     start_2h = int((now - timedelta(hours=2)).timestamp() * 1000)
-    response = make_query_request(signoz, token, start_2h, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_2h, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -112,7 +111,7 @@ def test_non_existent_internal_metrics_returns_no_warning(
     end_ms = int(now.timestamp() * 1000)
 
     start_2h = int((now - timedelta(hours=2)).timestamp() * 1000)
-    response = make_query_request(signoz, token, start_2h, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_2h, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
     data = response.json()
     assert get_all_warnings(data) == []
@@ -173,7 +172,6 @@ def test_variable_in_filter_returns_no_warning(
         end_ms,
         [query],
         variables={"tag": {"type": "query", "value": "service-a"}},
-        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK

tests/integration/tests/queriermetrics/04_histogram.py

@@ -9,7 +9,6 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    RequestType,
     build_builder_query,
     get_all_series,
     get_series_values,
@@ -76,7 +75,7 @@ def test_histogram_count_for_one_endpoint(
         filter_expression='endpoint = "/health"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -143,7 +142,7 @@ def test_histogram_count_for_one_service(
         filter_expression='service = "api"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -213,7 +212,7 @@ def test_histogram_count_for_delta_service(
         filter_expression='service = "web"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -272,7 +271,7 @@ def test_histogram_count_for_all_services(
         ## no services filter, this tests for multitemporality handling as well
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -309,7 +308,7 @@ def test_histogram_count_no_param(
         "count",
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -395,7 +394,7 @@ def test_histogram_percentile_for_all_services(
         space_agg,
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -446,7 +445,7 @@ def test_histogram_percentile_for_cumulative_service(
         filter_expression='service = "api"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -501,7 +500,6 @@ def test_histogram_percentile_with_a_gap_in_one_bucket(
         start_ms,
         end_ms,
         [build_builder_query("A", metric_name, "doesnotreallymatter", "p50")],
-        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
 
@@ -550,7 +548,7 @@ def test_histogram_percentile_for_delta_service(
         filter_expression='service = "web"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -592,7 +590,6 @@ def test_histogram_percentile_explicit_le_group_by(
             build_builder_query("A", metric_name, "doesnotreallymatter", "p95", group_by=["le", "service"]),
             build_builder_query("B", metric_name, "doesnotreallymatter", "p95", group_by=["service"]),
         ],
-        request_type=RequestType.TIME_SERIES,
     )
     assert response.status_code == HTTPStatus.OK, response.text
 

tests/integration/tests/queriermetrics/05_multi_temporality.py

@@ -10,7 +10,6 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    RequestType,
     build_builder_query,
     get_all_series,
     get_series_values,
@@ -60,7 +59,7 @@ def test_with_steady_values_and_reset(
         filter_expression='endpoint = "/orders"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -119,7 +118,7 @@ def test_group_by_endpoint(
         group_by=["endpoint"],
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -243,7 +242,7 @@ def test_for_service_with_switch(
         filter_expression='service = "api"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -295,7 +294,7 @@ def test_for_week_long_time_range(
         "sum",
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
     data = response.json()
     result_values = sorted(get_series_values(data, "A"), key=lambda x: x["timestamp"])
@@ -339,7 +338,7 @@ def test_for_month_long_time_range(
         "sum",
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
     data = response.json()
     result_values = sorted(get_series_values(data, "A"), key=lambda x: x["timestamp"])

tests/integration/tests/queriermetrics/06_gauge.py

@@ -9,7 +9,6 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    RequestType,
     build_builder_query,
     get_series_values,
     make_query_request,
@@ -73,7 +72,7 @@ def test_for_one_service(
         filter_expression=f'service = "{service}"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -129,7 +128,7 @@ def test_for_multiple_aggregations(
         space_agg,
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()

tests/integration/tests/queriermetrics/07_rate_cumulative.py

@@ -8,7 +8,6 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    RequestType,
     build_builder_query,
     get_all_series,
     get_series_values,
@@ -47,7 +46,7 @@ def test_rate_with_steady_values_and_reset(
         filter_expression='endpoint = "/orders"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -91,7 +90,7 @@ def test_rate_group_by_endpoint(
         group_by=["endpoint"],
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()

tests/integration/tests/queriermetrics/08_rate_delta.py

@@ -8,7 +8,6 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    RequestType,
     build_builder_query,
     get_all_series,
     get_series_values,
@@ -46,7 +45,7 @@ def test_rate_with_steady_values_and_reset(
         filter_expression='endpoint = "/orders"',
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()
@@ -88,7 +87,7 @@ def test_rate_group_by_endpoint(
         group_by=["endpoint"],
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()

tests/integration/tests/queriermetrics/09_cost_meter.py

@@ -8,7 +8,6 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.meter import MeterSample, make_meter_samples
 from fixtures.querier import (
-    RequestType,
     build_builder_query,
     get_series_values,
     make_query_request,
@@ -50,7 +49,7 @@ def test_query_range_cost_meter(
         temporality="delta",
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK
 
     data = response.json()

tests/integration/tests/queriermetrics/10_key_resolution.py

@@ -45,7 +45,7 @@ def test_metrics_group_by_context_collapse(
             token,
             now,
             [
-                querier.build_aggregation_query(
+                querier.build_scalar_query(
                     name="A",
                     signal="metrics",
                     aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -93,7 +93,7 @@ def test_metrics_filter_label_context(
             token,
             now,
             [
-                querier.build_aggregation_query(
+                querier.build_scalar_query(
                     name="A",
                     signal="metrics",
                     aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -113,7 +113,7 @@ def test_metrics_filter_label_context(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -154,7 +154,7 @@ def test_metrics_group_by_unknown_label(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -199,7 +199,7 @@ def test_metrics_filter_unknown_label_matches_nothing(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -245,7 +245,7 @@ def test_metrics_full_text_filter_does_not_error(
             token,
             now,
             [
-                querier.build_aggregation_query(
+                querier.build_scalar_query(
                     name="A",
                     signal="metrics",
                     aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -290,7 +290,7 @@ def test_metrics_group_by_label_named_value(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],

tests/integration/tests/queriermetrics/12_delta_nonmonotonic_sum.py

@@ -7,7 +7,7 @@ import pytest
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import RequestType, build_builder_query, get_series_values, make_query_request
+from fixtures.querier import build_builder_query, get_series_values, make_query_request
 
 
 # A delta, non-monotonic Sum queried without an explicit type must be treated as
@@ -53,7 +53,7 @@ def test_delta_nonmonotonic_sum_is_treated_as_sum(
     # No type and no temporality: the server resolves both from the seeded series.
     query = build_builder_query("A", metric_name, time_aggregation, "sum")
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK, response.text
 
     values = get_series_values(response.json(), "A")

tests/integration/tests/queriermetrics/13_bool_label_filter.py

@@ -52,7 +52,7 @@ def test_metrics_filter_bool_label(
             token,
             now,
             [
-                querier.build_aggregation_query(
+                querier.build_scalar_query(
                     name="A",
                     signal="metrics",
                     aggregations=[querier.build_metrics_aggregation(METRIC, "latest", "sum", "unspecified", reduce_to="last")],

tests/integration/tests/queriermetrics/14_exp_histogram.py

@@ -8,7 +8,6 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import ExpHistogramMetrics
 from fixtures.querier import (
-    RequestType,
     build_builder_query,
     get_all_series,
     get_all_warnings,
@@ -75,7 +74,7 @@ def test_exp_histogram_percentile_delta_grouped(
         group_by=["service.name"],
     )
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK, response.text
 
     data = response.json()
@@ -124,7 +123,7 @@ def test_exp_histogram_percentile_delta_merges_across_series(
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
     query = build_builder_query("A", metric_name, "", "p95", temporality="delta")
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
     assert response.status_code == HTTPStatus.OK, response.text
 
     data = response.json()
@@ -165,7 +164,7 @@ def test_exp_histogram_percentile_cumulative_is_rejected(
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
     query = build_builder_query("A", metric_name, "", "p95", temporality="cumulative")
 
-    response = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms, end_ms, [query])
 
     # the request is well formed; it is the stored metric that cannot be read, so
     # this is reported as unsupported rather than as bad input

tests/integration/tests/queriermetrics/17_cache.py

@@ -7,7 +7,6 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    RequestType,
     assert_results_equal,
     build_builder_query,
     get_series_values,
@@ -57,16 +56,16 @@ def test_builder_shortening_the_time_range_at_the_end(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
     assert base_query.status_code == HTTPStatus.OK, base_query.text
     points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
     returned_points = [(point["value"], point.get("partial", False)) for point in points]
     assert returned_points == [(16, False), (4096, False)]
 
-    from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    from_cache = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=False)
     assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
-    uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=True, request_type=RequestType.TIME_SERIES)
+    uncached = make_query_request(signoz, token, start_time_ms, end_time_ms_shortened_query, query, no_cache=True)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened end")
@@ -117,16 +116,16 @@ def test_builder_shortening_the_time_range_at_the_start(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    base_query = make_query_request(signoz, token, start_time_ms_base_query, end_time_ms, query, no_cache=False)
     assert base_query.status_code == HTTPStatus.OK, base_query.text
     points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
     returned_points = [(point["value"], point.get("partial", False)) for point in points]
     assert returned_points == [(65536, False), (4096, False)]
 
-    from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    from_cache = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=False)
     assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
-    uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=True, request_type=RequestType.TIME_SERIES)
+    uncached = make_query_request(signoz, token, start_time_ms_shortened_query, end_time_ms, query, no_cache=True)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "shortened start")
@@ -179,16 +178,16 @@ def test_builder_time_range_shorter_than_the_step(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
     assert base_query.status_code == HTTPStatus.OK, base_query.text
     points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
     returned_points = [(point["value"], point.get("partial", False)) for point in points]
     assert returned_points == [(16, False), (4096, False)]
 
-    from_cache = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    from_cache = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=False)
     assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
-    uncached = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=True, request_type=RequestType.TIME_SERIES)
+    uncached = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=True)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "range shorter than the step")
@@ -239,10 +238,10 @@ def test_promql_running_the_same_query_twice(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    first = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
     assert first.status_code == HTTPStatus.OK, first.text
 
-    second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    second = make_query_request(signoz, token, start_time_ms, end_time_ms, query, no_cache=False)
     assert second.status_code == HTTPStatus.OK, second.text
 
     assert_results_equal(first.json(), second.json(), "A", "the same query twice")
@@ -297,7 +296,7 @@ def test_promql_shifting_the_time_range(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    aligned_and_cached = make_query_request(signoz, token, aligned_start_time_ms, aligned_end_time_ms, query, no_cache=False)
     assert aligned_and_cached.status_code == HTTPStatus.OK, aligned_and_cached.text
 
     ## at each timestamp t, promql takes the highest sample in (t-2minutes, t],
@@ -313,7 +312,7 @@ def test_promql_shifting_the_time_range(
 
     # an unaligned range is rounded down to the step, so this asks for the same
     # instants as the aligned query and gets the same answer
-    unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    unaligned_and_cached = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, query, no_cache=False)
     assert unaligned_and_cached.status_code == HTTPStatus.OK, unaligned_and_cached.text
     points = sorted(get_series_values(unaligned_and_cached.json(), "A"), key=lambda point: point["timestamp"])
     assert [(point["timestamp"], point["value"]) for point in points] == aligned_points, "unaligned query, rounded to the step"
@@ -330,7 +329,7 @@ def test_promql_shifting_the_time_range(
         (unaligned_end_time_ms, 1100),  # t = 3m30s
     ]
     for run in ("first", "second"):
-        unaligned_as_asked = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, unaligned_query, no_cache=False, request_type=RequestType.TIME_SERIES)
+        unaligned_as_asked = make_query_request(signoz, token, unaligned_start_time_ms, unaligned_end_time_ms, unaligned_query, no_cache=False)
         assert unaligned_as_asked.status_code == HTTPStatus.OK, unaligned_as_asked.text
         points = sorted(get_series_values(unaligned_as_asked.json(), "A"), key=lambda point: point["timestamp"])
         assert [(point["timestamp"], point["value"]) for point in points] == unaligned_points, f"unaligned query as asked, {run} run"
@@ -372,17 +371,17 @@ def test_promql_time_range_shorter_than_the_step(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    base_query = make_query_request(signoz, token, start_time_ms, end_time_ms_base_query, query, no_cache=False)
     assert base_query.status_code == HTTPStatus.OK, base_query.text
     points = sorted(get_series_values(base_query.json(), "A"), key=lambda point: point["timestamp"])
     assert [(point["timestamp"], point["value"]) for point in points] == [(start_time_ms, 100), (end_time_ms_base_query, 1100)]
 
     # a range shorter than the step rounds down to the single instant at minute
     # 10, the instant the base query closed on
-    from_cache = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+    from_cache = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=False)
     assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
-    uncached = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=True, request_type=RequestType.TIME_SERIES)
+    uncached = make_query_request(signoz, token, start_time_ms_short_query, end_time_ms_short_query, query, no_cache=True)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "range shorter than the step")
@@ -428,7 +427,7 @@ def test_builder_refreshing_a_sliding_time_range(
     # a dashboard left open on a one hour range, re-running a minute later each time
     for refresh in range(20):
         refresh_start_ms = start_time_ms + refresh * MINUTE_MS
-        from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 60 * MINUTE_MS, query, no_cache=False, request_type=RequestType.TIME_SERIES)
+        from_cache = make_query_request(signoz, token, refresh_start_ms, refresh_start_ms + 60 * MINUTE_MS, query, no_cache=False)
         assert from_cache.status_code == HTTPStatus.OK, from_cache.text
 
         # each refresh is stitched out of overlapping cached ranges, so this catches
@@ -439,7 +438,7 @@ def test_builder_refreshing_a_sliding_time_range(
         assert returned_points == expected_points, f"refresh {refresh} did not return the minutes it covers"
 
     last_refresh_start_ms = start_time_ms + 19 * MINUTE_MS
-    uncached = make_query_request(signoz, token, last_refresh_start_ms, last_refresh_start_ms + 60 * MINUTE_MS, query, no_cache=True, request_type=RequestType.TIME_SERIES)
+    uncached = make_query_request(signoz, token, last_refresh_start_ms, last_refresh_start_ms + 60 * MINUTE_MS, query, no_cache=True)
     assert uncached.status_code == HTTPStatus.OK, uncached.text
 
     assert_results_equal(from_cache.json(), uncached.json(), "A", "the twentieth refresh")

tests/integration/tests/queriermetrics/19_offset.py

@@ -7,7 +7,7 @@ import pytest
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
-from fixtures.querier import RequestType, build_builder_query, get_all_series, make_query_request
+from fixtures.querier import build_builder_query, get_all_series, make_query_request
 
 
 @pytest.mark.parametrize(
@@ -62,7 +62,6 @@ def test_metrics_time_series_offset_skips_series_in_order(
                 offset=offset,
             )
         ],
-        request_type=RequestType.TIME_SERIES,
     )
 
     assert response.status_code == HTTPStatus.OK, response.text

tests/integration/tests/querierscalar/01_logs.py

@@ -46,7 +46,7 @@ def build_logs_query(
     gb = [querier.build_group_by_field(f, "string", "resource") for f in group_by] if group_by else None
     order = [querier.build_order_by(name, direction) for name, direction in order_by] if order_by else None
 
-    return querier.build_aggregation_query(
+    return querier.build_scalar_query(
         name=name,
         signal="logs",
         aggregations=aggs,

tests/integration/tests/querierscalar/02_traces.py

@@ -52,7 +52,7 @@ def build_traces_query(
     gb = [querier.build_group_by_field(f, "string", "resource") for f in group_by] if group_by else None
     order = [querier.build_order_by(name, direction) for name, direction in order_by] if order_by else None
 
-    return querier.build_aggregation_query(
+    return querier.build_scalar_query(
         name=name,
         signal="traces",
         aggregations=aggs,

tests/integration/tests/querierscalar/03_metrics.py

@@ -48,7 +48,7 @@ def build_metrics_query(
     gb = [querier.build_group_by_field(f, "string", "attribute") for f in group_by] if group_by else None
     order = [querier.build_order_by(name, direction) for name, direction in order_by] if order_by else None
 
-    return querier.build_aggregation_query(
+    return querier.build_scalar_query(
         name=name,
         signal="metrics",
         aggregations=aggs,

tests/integration/tests/querierscalar/04_having.py

@@ -8,9 +8,9 @@ from fixtures.logs import Logs
 from fixtures.querier import (
     assert_scalar_result_order,
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_order_by,
+    build_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -41,7 +41,7 @@ def test_logs_scalar_group_by_having(
     insert_logs(logs)
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="logs",
         aggregations=[build_aggregation("count()")],

tests/integration/tests/querierscalar/05_reduce_to.py

@@ -9,10 +9,10 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
     assert_scalar_result_order,
-    build_aggregation_query,
     build_group_by_field,
     build_metrics_aggregation,
     build_order_by,
+    build_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -59,7 +59,7 @@ def test_metrics_scalar_reduce_to_modes(
 
     aggregation = build_metrics_aggregation("test.reduce.metric", "latest", "sum", "unspecified")
     aggregation["reduceTo"] = reduce_to
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="metrics",
         aggregations=[aggregation],
@@ -118,7 +118,7 @@ def test_metrics_scalar_limit_ranks_by_reduced_value(
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation(metric_name, "latest", "sum", "unspecified", reduce_to=reduce_to)],
@@ -175,7 +175,7 @@ def test_metrics_scalar_limit_ranks_after_functions(
     )
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation(metric_name, "latest", "sum", "unspecified", reduce_to="max")],
@@ -200,7 +200,7 @@ def test_metrics_scalar_missing_reduce_to_rejected(
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation("test.reduce.metric", "latest", "sum", "unspecified")],
@@ -221,7 +221,7 @@ def test_metrics_scalar_invalid_reduce_to_rejected(
 
     aggregation = build_metrics_aggregation("test.reduce.metric", "latest", "sum", "unspecified")
     aggregation["reduceTo"] = "p99"
-    query = build_aggregation_query(name="A", signal="metrics", aggregations=[aggregation])
+    query = build_scalar_query(name="A", signal="metrics", aggregations=[aggregation])
     response = make_scalar_query_request(signoz, token, now, [query])
 
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text

tests/integration/tests/querierscalar/06_metrics_aggregations.py

@@ -8,9 +8,9 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
-    build_aggregation_query,
     build_group_by_field,
     build_metrics_aggregation,
+    build_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -52,7 +52,7 @@ def test_metrics_scalar_space_aggregations(
     insert_metrics([Metrics(metric_name="test.metric", labels={"service.name": service}, timestamp=now - timedelta(seconds=1), temporality="Unspecified", type_="Gauge", is_monotonic=False, value=value) for service, value in METRIC_VALUES.items()])
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation("test.metric", "latest", space_aggregation, "unspecified", reduce_to="last")],
@@ -93,7 +93,7 @@ def test_metrics_scalar_having(
     insert_metrics([Metrics(metric_name="test.metric", labels={"service.name": service}, timestamp=now - timedelta(seconds=1), temporality="Unspecified", type_="Gauge", is_monotonic=False, value=value) for service, value in METRIC_VALUES.items()])
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_aggregation_query(
+    query = build_scalar_query(
         name="A",
         signal="metrics",
         aggregations=[build_metrics_aggregation("test.metric", "latest", "sum", "unspecified", reduce_to="last")],

tests/integration/tests/querierscalar/07_unknown_keys.py

@@ -32,7 +32,7 @@ def test_logs_scalar_filter_unknown_key_synthesizes(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[querier.build_aggregation("count()")],
@@ -64,7 +64,7 @@ def test_logs_scalar_group_by_over_empty_filter_returns_no_rows(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[querier.build_aggregation("count()")],
@@ -95,7 +95,7 @@ def test_logs_scalar_group_by_unknown_key_null_bucket(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[querier.build_aggregation("count()")],
@@ -142,7 +142,7 @@ def test_traces_scalar_filter_unknown_key_synthesizes(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="traces",
                 aggregations=[querier.build_aggregation("count()")],
@@ -187,7 +187,7 @@ def test_traces_scalar_group_by_unknown_key_null_bucket(
         token,
         now,
         [
-            querier.build_aggregation_query(
+            querier.build_scalar_query(
                 name="A",
                 signal="traces",
                 aggregations=[querier.build_aggregation("count()")],

tests/integration/tests/querierscalar/09_offset.py

@@ -11,11 +11,11 @@ from fixtures.metrics import Metrics
 from fixtures.querier import (
     RequestType,
     build_aggregation,
-    build_aggregation_query,
     build_formula_query,
     build_group_by_field,
     build_metrics_aggregation,
     build_order_by,
+    build_scalar_query,
     find_named_result,
     get_scalar_table_data,
     make_query_request,
@@ -62,7 +62,7 @@ def test_logs_scalar_offset_skips_rows_in_order(
         token,
         now,
         [
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -117,7 +117,7 @@ def test_traces_scalar_offset_skips_rows_in_order(
         token,
         now,
         [
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="traces",
                 aggregations=[build_aggregation("count()")],
@@ -162,7 +162,7 @@ def test_metrics_scalar_offset_skips_rows_in_order(
         token,
         now,
         [
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation("offset.gauge", "latest", "sum", "unspecified", reduce_to="last")],
@@ -205,7 +205,7 @@ def test_formula_scalar_offset_skips_rows_in_order(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count()")],
@@ -240,7 +240,7 @@ def test_negative_offset_is_rejected(
         signoz,
         token,
         datetime.now(tz=UTC).replace(second=0, microsecond=0),
-        [build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=-1)],
+        [build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=-1)],
     )
 
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
@@ -257,7 +257,7 @@ def test_offset_without_limit_is_rejected(
         signoz,
         token,
         datetime.now(tz=UTC).replace(second=0, microsecond=0),
-        [build_aggregation_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=1)],
+        [build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=1)],
     )
 
     assert response.status_code == HTTPStatus.BAD_REQUEST, response.text

tests/integration/tests/queriertraces/02_aggregation.py

@@ -18,7 +18,7 @@ from fixtures.querier import (
     build_aggregation,
     build_group_by_field,
     build_order_by,
-    build_traces_aggregation_query,
+    build_traces_scalar_query,
     get_all_series,
     get_scalar_table_data,
     index_series_by_label,
@@ -208,7 +208,7 @@ def test_traces_aggregate_functions(
         build_aggregation("rate_sum(duration_nano)", "rate_sum_d"),
         build_aggregation("avg(latency_ms)", "avg_lat"),
     ]
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=aggregations,
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", "desc")],
@@ -284,7 +284,7 @@ def test_traces_aggregate_calculated_range_countif(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=[build_aggregation("countIf(response_status_code >= 400 AND response_status_code < 500)")],
     )
     response = make_scalar_query_request(signoz, token, now, [query])
@@ -342,7 +342,7 @@ def test_traces_aggregate_having(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=[build_aggregation("count()")],
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", "desc")],
@@ -418,7 +418,7 @@ def test_traces_aggregate_time_series_limit(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=[build_aggregation("count()")],
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", direction)],
@@ -495,7 +495,7 @@ def test_traces_aggregate_unknown_key(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_aggregation_query(aggregations=[build_aggregation(expression)])
+    query = build_traces_scalar_query(aggregations=[build_aggregation(expression)])
     response = make_scalar_query_request(signoz, token, now, [query])
 
     assert response.status_code == HTTPStatus.OK, response.text

tests/integration/tests/queriertraces/03_fill.py

@@ -12,7 +12,7 @@ from fixtures.querier import (
     build_aggregation,
     build_formula_query,
     build_group_by_field,
-    build_traces_aggregation_query,
+    build_traces_scalar_query,
     find_named_result,
     get_all_series,
     index_series_by_label,
@@ -72,7 +72,7 @@ def test_traces_fill_plain(
         start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
-        queries=[build_traces_aggregation_query(aggregations=[build_aggregation("count()")], functions=functions)],
+        queries=[build_traces_scalar_query(aggregations=[build_aggregation("count()")], functions=functions)],
         format_options=format_options,
     )
 
@@ -138,7 +138,7 @@ def test_traces_fill_group_by(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("service.name", "string", "resource")],
                 functions=functions,
@@ -212,8 +212,8 @@ def test_traces_fill_formula(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(aggregations=[build_aggregation("count()")], filter_expression="resource.service.name = 'test'", disabled=True),
-            build_traces_aggregation_query(name="B", aggregations=[build_aggregation("count()")], filter_expression="resource.service.name = 'another-test'", disabled=True),
+            build_traces_scalar_query(aggregations=[build_aggregation("count()")], filter_expression="resource.service.name = 'test'", disabled=True),
+            build_traces_scalar_query(name="B", aggregations=[build_aggregation("count()")], filter_expression="resource.service.name = 'another-test'", disabled=True),
             build_formula_query("F1", "A + B", functions=functions),
         ],
         format_options=format_options,
@@ -284,8 +284,8 @@ def test_traces_fill_formula_group_by(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
-            build_traces_aggregation_query(name="B", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
+            build_traces_scalar_query(aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
+            build_traces_scalar_query(name="B", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
             build_formula_query("F1", "A + B", functions=functions),
         ],
         format_options=format_options,

tests/integration/tests/queriertraces/05_resource_evolution.py

@@ -9,7 +9,7 @@ from fixtures.querier import (
     assert_grouped_series,
     build_aggregation,
     build_group_by_field,
-    build_traces_aggregation_query,
+    build_traces_scalar_query,
     get_resource_evolution_time,
     index_series_by_label,
     make_query_request,
@@ -99,7 +99,7 @@ def test_traces_resource_evolution(
         end_ms=int((before_1 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("service.name")],
             )
@@ -131,7 +131,7 @@ def test_traces_resource_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("service.name")],
             )
@@ -163,7 +163,7 @@ def test_traces_resource_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("service.name")],
             )
@@ -202,7 +202,7 @@ def test_traces_resource_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count_distinct(service.name)")],
                 group_by=[build_group_by_field("deployment.environment")],
             )

tests/integration/tests/queriertraces/08_aggregation_min_max.py

@@ -10,7 +10,7 @@ from fixtures.querier import (
     RequestType,
     build_aggregation,
     build_group_by_field,
-    build_traces_aggregation_query,
+    build_traces_scalar_query,
     index_series_by_label,
     make_query_request,
 )
@@ -71,7 +71,7 @@ def test_traces_aggregate_min_max(
         end_ms=int((now + timedelta(seconds=5)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[
                     build_aggregation("max(duration_nano)", "maxDuration"),
                     build_aggregation("min(duration_nano)", "minDuration"),

tests/integration/tests/queriertraces/09_unknown_keys.py

@@ -11,7 +11,7 @@ from fixtures.querier import (
     build_aggregation,
     build_group_by_field,
     build_raw_query,
-    build_traces_aggregation_query,
+    build_traces_scalar_query,
     get_rows,
     get_scalar_table_data,
     make_query_request,
@@ -44,7 +44,7 @@ def test_traces_filter_unknown_key_synthesizes(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 [build_aggregation("count()")],
                 filter_expression='totally.unknown.key = "x"',
             )
@@ -152,7 +152,7 @@ def test_traces_group_by_unknown_key_null_bucket(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 [build_aggregation("count()")],
                 group_by=[build_group_by_field("totally.unknown.key", field_data_type="", field_context="")],
                 filter_expression='service.name = "groupby-unknown-svc"',

tests/integration/tests/queriertraces/11_aggregation_depth.py

@@ -11,7 +11,7 @@ from fixtures.querier import (
     build_aggregation,
     build_group_by_field,
     build_order_by,
-    build_traces_aggregation_query,
+    build_traces_scalar_query,
     get_scalar_table_data,
     make_scalar_query_request,
 )
@@ -73,7 +73,7 @@ def test_traces_aggregate_percentiles(
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
     percentiles = [25, 50, 90, 95, 99]
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=[build_aggregation(f"p{p}(duration_nano)", f"p{p}") for p in percentiles],
     )
     response = make_scalar_query_request(signoz, token, now, [query])
@@ -138,7 +138,7 @@ def test_traces_aggregate_group_by_non_resource(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=[build_aggregation("count()")],
         group_by=[build_group_by_field(group_key, field_data_type="", field_context="")],
         filter_expression="resource.service.name = 'gb-svc'",
@@ -193,7 +193,7 @@ def test_traces_aggregate_multi_group_by(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=[build_aggregation("count()")],
         group_by=[
             build_group_by_field("service.name", "string", "resource"),
@@ -262,7 +262,7 @@ def test_traces_aggregate_having_breadth(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=[build_aggregation("count()"), build_aggregation("sum(duration_nano)")],
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", "desc")],
@@ -325,7 +325,7 @@ def test_traces_duration_nano_qol_filter_with_collision(
 
     token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
 
-    query = build_traces_aggregation_query(
+    query = build_traces_scalar_query(
         aggregations=[build_aggregation("count()", "cnt")],
         filter_expression=duration_filter,
     )

tests/integration/tests/queriertraces/13_attributes_json_evolution.py

@@ -12,7 +12,7 @@ from fixtures.querier import (
     assert_grouped_series,
     build_aggregation,
     build_group_by_field,
-    build_traces_aggregation_query,
+    build_traces_scalar_query,
     get_all_warnings,
     get_preview_sql,
     get_rows,
@@ -90,7 +90,7 @@ def test_traces_attributes_json_evolution(
         end_ms=int((before_1 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("http.route", field_context="attribute")],
             )
@@ -114,7 +114,7 @@ def test_traces_attributes_json_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("http.route", field_context="attribute")],
             )
@@ -138,7 +138,7 @@ def test_traces_attributes_json_evolution(
         end_ms=int((after_2 + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("http.route", field_context="attribute")],
             )
@@ -212,7 +212,7 @@ def test_traces_attributes_json_typed_filters(
             end_ms=end_ms,
             request_type=RequestType.TIME_SERIES,
             queries=[
-                build_traces_aggregation_query(
+                build_traces_scalar_query(
                     aggregations=[build_aggregation("count()")],
                     group_by=[build_group_by_field("http.route", field_context="attribute")],
                     filter_expression=filter_expression,
@@ -291,7 +291,7 @@ def test_traces_attributes_json_collision_and_map_parity(
         end_ms=end_ms,
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 aggregations=[build_aggregation("count()")],
                 group_by=[build_group_by_field("app.status", field_context="attribute")],
                 filter_expression="app.status EXISTS",
@@ -325,7 +325,7 @@ def test_traces_attributes_json_collision_and_map_parity(
             end_ms=end_ms,
             request_type=RequestType.TIME_SERIES,
             queries=[
-                build_traces_aggregation_query(
+                build_traces_scalar_query(
                     aggregations=[build_aggregation("count()")],
                     group_by=[build_group_by_field("http.route", field_context="attribute")],
                     filter_expression=filter_expression,
@@ -382,7 +382,7 @@ def test_traces_attributes_json_number_reads(
             end_ms=end_ms,
             request_type=RequestType.TIME_SERIES,
             queries=[
-                build_traces_aggregation_query(
+                build_traces_scalar_query(
                     aggregations=[build_aggregation("count()")],
                     group_by=[build_group_by_field("http.route", field_context="attribute")],
                     filter_expression=filter_expression,
@@ -513,7 +513,7 @@ def test_traces_attributes_json_unreported_key(
     insert_traces(spans)
 
     queries = [
-        build_traces_aggregation_query(
+        build_traces_scalar_query(
             aggregations=[build_aggregation("count()")],
             filter_expression="unreported.key = 'value'",
         )
@@ -523,11 +523,11 @@ def test_traces_attributes_json_unreported_key(
     start_ms = int((before - timedelta(minutes=1)).timestamp() * 1000)
     end_ms = int((before + timedelta(minutes=1)).timestamp() * 1000)
 
-    sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries, request_type=RequestType.TIME_SERIES), "A")
+    sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries), "A")
     assert "attributes_string['unreported.key']" in sql
     assert "attributes.`unreported.key`" not in sql
 
-    response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries, request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries)
     assert response.status_code == HTTPStatus.OK
     series = response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
     assert [value["value"] for value in series[0]["values"]] == [1]
@@ -537,11 +537,11 @@ def test_traces_attributes_json_unreported_key(
     start_ms = int((after - timedelta(minutes=1)).timestamp() * 1000)
     end_ms = int((after + timedelta(minutes=1)).timestamp() * 1000)
 
-    sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries, request_type=RequestType.TIME_SERIES), "A")
+    sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries), "A")
     assert "attributes.`unreported.key`" in sql
     assert "attributes_string" not in sql
 
-    response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries, request_type=RequestType.TIME_SERIES)
+    response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries)
     assert response.status_code == HTTPStatus.OK
     series = response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
     assert [value["value"] for value in series[0]["values"]] == [1]

tests/integration/tests/queriertraces/14_skip_indexes.py

@@ -7,7 +7,7 @@ import requests
 
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
-from fixtures.querier import RequestType, build_aggregation, build_traces_aggregation_query, get_preview_statements, make_preview_query_request
+from fixtures.querier import RequestType, build_aggregation, build_traces_scalar_query, get_preview_statements, make_preview_query_request
 from fixtures.traces import ATTRIBUTE_JSON_ROLLOUT_TIME, Traces
 
 PROMOTED_PATHS = "/api/v1/promoted_paths"
@@ -72,7 +72,7 @@ def test_traces_filters_use_skip_indexes(
                 end_ms=int(now.timestamp() * 1000),
                 request_type=RequestType.SCALAR,
                 # count() over primary-key filters is answered by the exact count projection, which reads no index
-                queries=[build_traces_aggregation_query(aggregations=[build_aggregation("sum(duration_nano)")], filter_expression=filter_expression)],
+                queries=[build_traces_scalar_query(aggregations=[build_aggregation("sum(duration_nano)")], filter_expression=filter_expression)],
             ),
             "A",
         )

tests/integration/tests/semconvfamilies/01_family_matrix.py

@@ -10,11 +10,11 @@ from fixtures.metadata import AttributesMetadata, get_field_keys, get_field_valu
 from fixtures.querier import (
     RequestType,
     build_aggregation,
-    build_aggregation_query,
     build_group_by_field,
     build_order_by,
     build_raw_query,
-    build_traces_aggregation_query,
+    build_scalar_query,
+    build_traces_scalar_query,
     get_all_warnings,
     get_column_data_from_response,
     get_scalar_columns,
@@ -178,7 +178,7 @@ def test_group_by_merges_and_echoes_requested_spelling(
         end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 [build_aggregation("count_distinct(name)")],
                 filter_expression=f"service.name LIKE '{PREFIX}%'",
                 group_by=[build_group_by_field(requested_key, "string", "resource")],
@@ -214,7 +214,7 @@ def test_log_group_by_merges_and_echoes_requested_spelling(
         end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="logs",
                 aggregations=[build_aggregation("count_distinct(body)")],
@@ -245,7 +245,7 @@ def test_time_series_group_by_merges_family(
         end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
         request_type=RequestType.TIME_SERIES,
         queries=[
-            build_traces_aggregation_query(
+            build_traces_scalar_query(
                 [build_aggregation("count_distinct(name)")],
                 filter_expression=f"service.name LIKE '{PREFIX}%'",
                 group_by=[build_group_by_field(CURRENT_KEY, "string", "resource")],

tests/integration/tests/semconvfamilies/02_metric_families.py

@@ -9,9 +9,9 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.metrics import Metrics
 from fixtures.querier import (
     RequestType,
-    build_aggregation_query,
     build_group_by_field,
     build_metrics_aggregation,
+    build_scalar_query,
     get_scalar_columns,
     get_scalar_table_data,
     make_query_request,
@@ -40,7 +40,7 @@ def test_metric_label_filter_merges_stored_spellings(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(LABEL_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -69,7 +69,7 @@ def test_metric_label_group_by_merges_stored_spellings(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(LABEL_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -113,7 +113,7 @@ def test_span_metric_label_filter_reads_resource_layout(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(SPAN_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -142,7 +142,7 @@ def test_metric_name_family_unions_storage_names(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(requested, "latest", "sum", "unspecified", reduce_to="last")],
@@ -177,7 +177,7 @@ def test_metric_name_union_double_counts_dual_emission(
         end_ms=int(now.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(CURRENT_NAME_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -203,7 +203,7 @@ def test_metric_family_stays_literal_with_flag_off(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(LABEL_METRIC, "latest", "sum", "unspecified", reduce_to="last")],
@@ -223,7 +223,7 @@ def test_metric_family_stays_literal_with_flag_off(
         end_ms=int(metric_family_fleet.timestamp() * 1000),
         request_type=RequestType.SCALAR,
         queries=[
-            build_aggregation_query(
+            build_scalar_query(
                 name="A",
                 signal="metrics",
                 aggregations=[build_metrics_aggregation(OLD_NAME_METRIC, "latest", "sum", "unspecified", reduce_to="last")],