Merge remote-tracking branch 'origin/main' into tvats-query-result-budget

They return every row like raw; only aggregation request types are exempt.

Assisted-by: Claude Opus 5.5

.github/workflows/integrationci.yaml

@@ -72,6 +72,7 @@ jobs:
           - tracedetail
           - querier_json_body
           - querier_skip_resource_fingerprint
+          - querierclickhousequery
           - ttl
           - clickhousecluster
           - metricreduction

conf/example.yaml

@@ -178,6 +178,17 @@ querier:
   # range with padding to include slightly delayed log exports. Logs only; set
   # to 0 to disable.
   log_trace_id_window_padding: 5m
+  # Limits of one user-written ClickHouse SQL query; 0 disables a limit. Applies to the
+  # query_range API, and to modules that do not pass their own limits.
+  clickhouse_query:
+    # Covers ClickHouse execution and reading the result.
+    max_execution_time: 55s
+    # ClickHouse memory limit of the query, in bytes.
+    max_memory_usage: 10000000000
+    # Result row limit of list requests (raw, raw_stream, trace).
+    max_result_rows: 100000
+    # Result size limit, in bytes.
+    max_result_bytes: 524288000
   schemametadatastore:
     # How long found schema metadata stays cached; 0 caches it for the life of the process.
     cache_ttl: 0s

pkg/clickhousesql/statement.go

@@ -20,8 +20,25 @@ var (
 	CodeReadingFunction    = errors.MustNewCode("clickhouse_sql_reading_function")
 	CodeInternalDatabase   = errors.MustNewCode("clickhouse_sql_internal_database")
 	CodeReadonlyOverride   = errors.MustNewCode("clickhouse_sql_readonly_override")
+	CodeLimitOverride      = errors.MustNewCode("clickhouse_sql_limit_override")
 )
 
+// limitSettings bound a query's time, memory, reads and result. The caller applies them, and a
+// query-level setting takes precedence over the context setting.
+//
+// Keyed by the lowercased name, since ClickHouse resolves setting names case-insensitively.
+var limitSettings = map[string]struct{}{
+	"max_execution_time":      {},
+	"max_execution_time_leaf": {},
+	"timeout_overflow_mode":   {},
+	"max_memory_usage":        {},
+	"max_bytes_to_read":       {},
+	"read_overflow_mode":      {},
+	"max_result_rows":         {},
+	"max_result_bytes":        {},
+	"result_overflow_mode":    {},
+}
+
 // internalDatabases hold server metadata, credentials and grants rather than telemetry.
 var internalDatabases = map[string]struct{}{
 	"system":             {},
@@ -144,10 +161,7 @@ func ErrIfStatementIsNotValid(query string) (err error) {
 			}
 
 		case *chparser.SettingExpr:
-			// A query-level setting takes precedence over the context setting.
-			if strings.EqualFold(expr.Name.Name, "readonly") {
-				return errors.NewInvalidInputf(CodeReadonlyOverride, "the ClickHouse readonly setting cannot be overridden")
-			}
+			return errIfSettingOverridden(expr.Name.Name)
 		}
 
 		return nil
@@ -163,6 +177,18 @@ func LogIfStatementIsNotValid(ctx context.Context, logger *slog.Logger, query st
 	}
 }
 
+func errIfSettingOverridden(name string) error {
+	if strings.EqualFold(name, "readonly") {
+		return errors.NewInvalidInputf(CodeReadonlyOverride, "the ClickHouse readonly setting cannot be overridden")
+	}
+
+	if _, ok := limitSettings[strings.ToLower(name)]; ok {
+		return errors.NewInvalidInputf(CodeLimitOverride, "the ClickHouse %s setting cannot be overridden", name)
+	}
+
+	return nil
+}
+
 func errIfFunctionReads(name string) error {
 	lowered := strings.ToLower(name)
 	if _, ok := readingFunctions[lowered]; !ok && !strings.HasPrefix(lowered, dictionaryFunctionPrefix) {

pkg/clickhousesql/statement_test.go

@@ -31,6 +31,7 @@ func TestErrIfStatementIsNotValid_Pass(t *testing.T) {
 		{"ParenthesisedUnionLeftOperandAtStatementLevel", "(SELECT 1 AS a) UNION ALL (SELECT 2 AS a)"},
 		{"WindowFunction", "SELECT sum(v) OVER (PARTITION BY a ORDER BY t) FROM t"},
 		{"UnrelatedSetting", "SELECT * FROM t SETTINGS max_threads = 4"},
+		{"LimitSettingNameInStringLiteral", "SELECT * FROM t WHERE body = 'max_result_rows'"},
 		{"TerminatedBlockComment", "SELECT /* keep me */ count() FROM t"},
 		{"BlockCommentMarkerInsideStringLiteral", "SELECT count() FROM t WHERE body = '/* not a comment'"},
 		// Looped forever before v0.5.2.
@@ -168,6 +169,9 @@ func TestErrIfStatementIsNotValid_Fail(t *testing.T) {
 		// Takes precedence over the setting the caller applies.
 		{"ReadonlySettingOverride", "SELECT * FROM t SETTINGS readonly = 0", CodeReadonlyOverride},
 		{"ReadonlySettingOverrideAmongOthers", "SELECT * FROM t SETTINGS max_threads = 4, readonly = 0", CodeReadonlyOverride},
+		{"LimitSettingOverride", "SELECT * FROM t SETTINGS max_result_rows = 0", CodeLimitOverride},
+		{"LimitSettingOverrideUppercase", "SELECT * FROM t SETTINGS MAX_MEMORY_USAGE = 0", CodeLimitOverride},
+		{"LimitSettingOverrideInSubquery", "SELECT * FROM (SELECT * FROM t SETTINGS max_execution_time = 0)", CodeLimitOverride},
 	}
 
 	for _, testCase := range testCases {

pkg/querier/clickhouse_query.go

@@ -30,6 +30,7 @@ type chSQLQuery struct {
 	toMS   uint64
 	kind   qbtypes.RequestType
 	vars   map[string]qbtypes.VariableItem
+	limits ClickHouseQueryConfig
 }
 
 var _ qbtypes.Query = (*chSQLQuery)(nil)
@@ -43,6 +44,7 @@ func newchSQLQuery(
 	tr qbtypes.TimeRange,
 	kind qbtypes.RequestType,
 	variables map[string]qbtypes.VariableItem,
+	limits ClickHouseQueryConfig,
 ) *chSQLQuery {
 	return &chSQLQuery{
 		logger:         logger,
@@ -53,6 +55,7 @@ func newchSQLQuery(
 		toMS:           tr.To,
 		kind:           kind,
 		vars:           variables,
+		limits:         limits,
 	}
 }
 
@@ -141,6 +144,15 @@ func (q *chSQLQuery) Execute(ctx context.Context) (*qbtypes.Result, error) {
 		return nil, err
 	}
 
+	if q.limits.MaxExecutionTime > 0 {
+		var cancel context.CancelFunc
+		// TODO: clickhouse-go replaces a lower max_execution_time setting with the ctx deadline, so the limit is a
+		// deadline until https://github.com/ClickHouse/clickhouse-go/issues/2100 is fixed.
+		ctx, cancel = context.WithTimeout(ctx, q.limits.MaxExecutionTime)
+		defer cancel()
+	}
+	ctx = telemetrystore.NewContextWithQuerySettings(ctx, q.limits.settings(q.kind))
+
 	rows, err := q.telemetryStore.ClickhouseDB().Query(ctx, query, q.args...)
 	if err != nil {
 		return nil, err
@@ -162,3 +174,22 @@ func (q *chSQLQuery) Execute(ctx context.Context) (*qbtypes.Result, error) {
 		},
 	}, nil
 }
+
+// settings returns the limits as ClickHouse query settings. A SETTINGS clause in the user's SQL
+// overrides them. MaxExecutionTime is a context deadline instead.
+func (config ClickHouseQueryConfig) settings(requestType qbtypes.RequestType) clickhouse.Settings {
+	settings := clickhouse.Settings{}
+	if config.MaxMemoryUsage > 0 {
+		settings["max_memory_usage"] = config.MaxMemoryUsage
+	}
+	// Aggregation request types return one row per series and point; MaxResultBytes bounds them.
+	if config.MaxResultRows > 0 && !requestType.IsAggregation() {
+		settings["max_result_rows"] = config.MaxResultRows
+		settings["result_overflow_mode"] = "throw"
+	}
+	if config.MaxResultBytes > 0 {
+		settings["max_result_bytes"] = config.MaxResultBytes
+		settings["result_overflow_mode"] = "throw"
+	}
+	return settings
+}

pkg/querier/clickhouse_query_test.go

@@ -0,0 +1,74 @@
+package querier
+
+import (
+	"testing"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/stretchr/testify/assert"
+
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+)
+
+func TestClickHouseQueryConfigSettings(t *testing.T) {
+	allSet := ClickHouseQueryConfig{MaxExecutionTime: time.Minute, MaxMemoryUsage: 1 << 30, MaxResultRows: 1000, MaxResultBytes: 1 << 20}
+
+	testCases := []struct {
+		name        string
+		config      ClickHouseQueryConfig
+		requestType qbtypes.RequestType
+		want        clickhouse.Settings
+	}{
+		{
+			name:        "AllZero_NoSettings",
+			config:      ClickHouseQueryConfig{},
+			requestType: qbtypes.RequestTypeRaw,
+			want:        clickhouse.Settings{},
+		},
+		{
+			name:        "AllSet_Raw_NoExecutionTimeSetting",
+			config:      allSet,
+			requestType: qbtypes.RequestTypeRaw,
+			want: clickhouse.Settings{
+				"max_memory_usage":     int64(1 << 30),
+				"max_result_rows":      int64(1000),
+				"max_result_bytes":     int64(1 << 20),
+				"result_overflow_mode": "throw",
+			},
+		},
+		{
+			name:        "RowsOnly_Trace_RowLimit",
+			config:      ClickHouseQueryConfig{MaxResultRows: 1000},
+			requestType: qbtypes.RequestTypeTrace,
+			want:        clickhouse.Settings{"max_result_rows": int64(1000), "result_overflow_mode": "throw"},
+		},
+		{
+			name:        "RowsOnly_RawStream_RowLimit",
+			config:      ClickHouseQueryConfig{MaxResultRows: 1000},
+			requestType: qbtypes.RequestTypeRawStream,
+			want:        clickhouse.Settings{"max_result_rows": int64(1000), "result_overflow_mode": "throw"},
+		},
+		{
+			name:        "AllSet_TimeSeries_NoRowLimit",
+			config:      allSet,
+			requestType: qbtypes.RequestTypeTimeSeries,
+			want: clickhouse.Settings{
+				"max_memory_usage":     int64(1 << 30),
+				"max_result_bytes":     int64(1 << 20),
+				"result_overflow_mode": "throw",
+			},
+		},
+		{
+			name:        "RowsOnly_Scalar_NoSettings",
+			config:      ClickHouseQueryConfig{MaxResultRows: 1000},
+			requestType: qbtypes.RequestTypeScalar,
+			want:        clickhouse.Settings{},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, testCase.config.settings(testCase.requestType))
+		})
+	}
+}

pkg/querier/config.go

@@ -20,11 +20,26 @@ type Config struct {
 	MaxConcurrentQueries int `yaml:"max_concurrent_queries" mapstructure:"max_concurrent_queries"`
 	// LogTraceIDWindowPadding is the padding added to narrowed down timerange from trace summary to logs with trace_id filter.
 	LogTraceIDWindowPadding time.Duration `yaml:"log_trace_id_window_padding" mapstructure:"log_trace_id_window_padding"`
+	// ClickHouseQuery bounds user-written ClickHouse SQL queries. A QueryRange caller can replace it with
+	// WithClickHouseQueryLimits.
+	ClickHouseQuery ClickHouseQueryConfig `yaml:"clickhouse_query" mapstructure:"clickhouse_query"`
 
 	// Keys sit under querier.skip_resource_fingerprint.
 	statementbuilder.Config `mapstructure:",squash" yaml:",squash"`
 }
 
+// ClickHouseQueryConfig is the budget of one user-written ClickHouse SQL query. A zero value disables that limit.
+type ClickHouseQueryConfig struct {
+	// MaxExecutionTime is enforced as a context deadline.
+	MaxExecutionTime time.Duration `yaml:"max_execution_time" mapstructure:"max_execution_time"`
+	// MaxMemoryUsage is the ClickHouse memory limit of the query, in bytes.
+	MaxMemoryUsage int64 `yaml:"max_memory_usage" mapstructure:"max_memory_usage"`
+	// MaxResultRows applies to list requests (raw, raw_stream, trace) only.
+	MaxResultRows int64 `yaml:"max_result_rows" mapstructure:"max_result_rows"`
+	// MaxResultBytes caps the uncompressed result size, in bytes.
+	MaxResultBytes int64 `yaml:"max_result_bytes" mapstructure:"max_result_bytes"`
+}
+
 // NewConfigFactory creates a new config factory for querier.
 func NewConfigFactory() factory.ConfigFactory {
 	return factory.NewConfigFactory(factory.MustNewName("querier"), newConfig)
@@ -37,7 +52,13 @@ func newConfig() factory.Config {
 		FluxInterval:            5 * time.Minute,
 		MaxConcurrentQueries:    DefaultMaxConcurrentQueries,
 		LogTraceIDWindowPadding: 5 * time.Minute,
-		Config:                  statementbuilder.NewConfig(),
+		ClickHouseQuery: ClickHouseQueryConfig{
+			MaxExecutionTime: 55 * time.Second,
+			MaxMemoryUsage:   10_000_000_000,
+			MaxResultRows:    100_000,
+			MaxResultBytes:   500 << 20,
+		},
+		Config: statementbuilder.NewConfig(),
 	}
 }
 
@@ -55,6 +76,18 @@ func (c Config) Validate() error {
 	if c.LogTraceIDWindowPadding < 0 {
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "log_trace_id_window_padding must not be negative, got %v", c.LogTraceIDWindowPadding)
 	}
+	if c.ClickHouseQuery.MaxExecutionTime < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "clickhouse_query.max_execution_time must not be negative, got %v", c.ClickHouseQuery.MaxExecutionTime)
+	}
+	if c.ClickHouseQuery.MaxMemoryUsage < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "clickhouse_query.max_memory_usage must not be negative, got %v", c.ClickHouseQuery.MaxMemoryUsage)
+	}
+	if c.ClickHouseQuery.MaxResultRows < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "clickhouse_query.max_result_rows must not be negative, got %v", c.ClickHouseQuery.MaxResultRows)
+	}
+	if c.ClickHouseQuery.MaxResultBytes < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "clickhouse_query.max_result_bytes must not be negative, got %v", c.ClickHouseQuery.MaxResultBytes)
+	}
 	// Embedded Validate is shadowed by this one; call it explicitly.
 	if err := c.Config.Validate(); err != nil {
 		return err

pkg/querier/preview.go

@@ -231,7 +231,7 @@ func (q *querier) buildPreviewProviders(
 			sub.CompositeQuery = qbtypes.CompositeQuery{Queries: []qbtypes.QueryEnvelope{query}}
 		}
 
-		built, _, bErr := q.buildQueries(orgID, &sub, deps, missingMetricQuerySet, event)
+		built, _, bErr := q.buildQueries(orgID, &sub, deps, missingMetricQuerySet, event, q.clickHouseQueryBudget)
 		if bErr != nil {
 			errs[name] = bErr
 			continue

pkg/querier/querier.go

@@ -38,13 +38,28 @@ var (
 
 // Querier interface defines the contract for querying data.
 type Querier interface {
-	QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error)
+	QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, opts ...QueryRangeOption) (*qbtypes.QueryRangeResponse, error)
 	QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, client *qbtypes.RawStream)
 	statsreporter.StatsCollector
 	// QueryRangePreview validates and renders the queries without executing them.
 	QueryRangePreview(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, opts qbtypes.QueryRangePreviewOptions) (*qbtypes.QueryRangePreviewResponse, error)
 }
 
+type QueryRangeOption func(*queryRangeOptions)
+
+// WithClickHouseQueryLimits replaces the configured querier.clickhouse_query limits for the ClickHouse SQL queries
+// of the call. A zero ClickHouseQueryConfig runs them without limits. The query_range API does not pass it, so API
+// callers always get the configured limits.
+func WithClickHouseQueryLimits(limits ClickHouseQueryConfig) QueryRangeOption {
+	return func(options *queryRangeOptions) {
+		options.clickHouseQueryLimits = limits
+	}
+}
+
+type queryRangeOptions struct {
+	clickHouseQueryLimits ClickHouseQueryConfig
+}
+
 type querier struct {
 	logger                   *slog.Logger
 	fl                       flagger.Flagger
@@ -62,6 +77,7 @@ type querier struct {
 	liveDataRefresh          time.Duration
 	builderConfig            builderConfig
 	maxConcurrentQueries     int
+	clickHouseQueryBudget    ClickHouseQueryConfig
 }
 
 var _ Querier = (*querier)(nil)
@@ -82,6 +98,7 @@ func New(
 	flagger flagger.Flagger,
 	logTraceIDWindowPadding time.Duration,
 	maxConcurrentQueries int,
+	clickHouseQueryBudget ClickHouseQueryConfig,
 ) *querier {
 	querierSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/querier")
 	if maxConcurrentQueries <= 0 {
@@ -105,11 +122,13 @@ func New(
 		builderConfig: builderConfig{
 			logTraceIDWindowPaddingMS: uint64(logTraceIDWindowPadding.Milliseconds()),
 		},
-		maxConcurrentQueries: maxConcurrentQueries,
+		maxConcurrentQueries:  maxConcurrentQueries,
+		clickHouseQueryBudget: clickHouseQueryBudget,
 	}
 }
 
-func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error) {
+func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, opts ...QueryRangeOption) (*qbtypes.QueryRangeResponse, error) {
+	options := q.newQueryRangeOptions(opts)
 
 	// Callers that build the request in-process never pass through
 	// UnmarshalJSON, so normalise here too.
@@ -144,7 +163,7 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 		missingMetricQuerySet[name] = true
 	}
 
-	queries, steps, err := q.buildQueries(orgID, req, dependencyQueries, missingMetricQuerySet, event)
+	queries, steps, err := q.buildQueries(orgID, req, dependencyQueries, missingMetricQuerySet, event, options.clickHouseQueryLimits)
 	if err != nil {
 		return nil, err
 	}
@@ -187,12 +206,21 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 	return qbResp, qbErr
 }
 
+func (q *querier) newQueryRangeOptions(opts []QueryRangeOption) queryRangeOptions {
+	options := queryRangeOptions{clickHouseQueryLimits: q.clickHouseQueryBudget}
+	for _, opt := range opts {
+		opt(&options)
+	}
+	return options
+}
+
 func (q *querier) buildQueries(
 	orgID valuer.UUID,
 	req *qbtypes.QueryRangeRequest,
 	dependencyQueries map[string]bool,
 	missingMetricQuerySet map[string]bool,
 	event *qbtypes.QBEvent,
+	clickHouseQueryLimits ClickHouseQueryConfig,
 ) (map[string]qbtypes.Query, map[string]qbtypes.Step, error) {
 
 	tmplVars := req.Variables
@@ -225,7 +253,7 @@ func (q *querier) buildQueries(
 			if !ok {
 				return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid clickhouse query spec %T", query.Spec)
 			}
-			chSQLQuery := newchSQLQuery(q.logger, q.telemetryStore, chQuery, nil, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType, tmplVars)
+			chSQLQuery := newchSQLQuery(q.logger, q.telemetryStore, chQuery, nil, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType, tmplVars, clickHouseQueryLimits)
 			queries[chQuery.Name] = chSQLQuery
 		case qbtypes.QueryTypeTraceOperator:
 			traceOpQuery, ok := query.Spec.(qbtypes.QueryBuilderTraceOperator)
@@ -920,7 +948,7 @@ func (q *querier) createRangedQuery(_ valuer.UUID, originalQuery qbtypes.Query,
 		queryCopy := qt.query.Copy()
 		argsCopy := make([]any, len(qt.args))
 		copy(argsCopy, qt.args)
-		return newchSQLQuery(q.logger, q.telemetryStore, queryCopy, argsCopy, timeRange, qt.kind, qt.vars)
+		return newchSQLQuery(q.logger, q.telemetryStore, queryCopy, argsCopy, timeRange, qt.kind, qt.vars, qt.limits)
 
 	case *builderQuery[qbtypes.TraceAggregation]:
 		specCopy := qt.spec.Copy()

pkg/querier/querier_test.go

@@ -59,6 +59,7 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
 		flaggertest.New(t), // flagger
 		0,                  // logTraceIDWindowPadding
 		0,                  // maxConcurrentQueries
+		ClickHouseQueryConfig{},
 	)
 
 	req := &qbtypes.QueryRangeRequest{
@@ -132,6 +133,7 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
 		flaggertest.New(t), // flagger
 		0,                  // logTraceIDWindowPadding
 		0,                  // maxConcurrentQueries
+		ClickHouseQueryConfig{},
 	)
 
 	req := &qbtypes.QueryRangeRequest{
@@ -310,3 +312,36 @@ func TestRunQueryErrorCancelsSiblings(t *testing.T) {
 	require.ErrorContains(t, err, "query A failed")
 	assert.True(t, bCanceled.Load(), "query B should be canceled once query A fails")
 }
+
+func TestQueryRangeClickHouseQueryLimits(t *testing.T) {
+	configured := ClickHouseQueryConfig{MaxExecutionTime: time.Second, MaxResultRows: 10}
+	moduleLimits := ClickHouseQueryConfig{MaxResultRows: 5}
+	q := New(instrumentationtest.New().ToProviderSettings(), nil, telemetrytypestest.NewMockMetadataStore(), nil, nil, nil, nil, nil, nil, nil, nil, nil, flaggertest.New(t), 0, 0, configured)
+
+	testCases := []struct {
+		name string
+		opts []QueryRangeOption
+		want ClickHouseQueryConfig
+	}{
+		{name: "NoOption_ConfiguredLimits", want: configured},
+		{name: "ModuleLimits_Replace", opts: []QueryRangeOption{WithClickHouseQueryLimits(moduleLimits)}, want: moduleLimits},
+		{name: "ZeroLimits_NoLimits", opts: []QueryRangeOption{WithClickHouseQueryLimits(ClickHouseQueryConfig{})}, want: ClickHouseQueryConfig{}},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			req := &qbtypes.QueryRangeRequest{
+				RequestType: qbtypes.RequestTypeRaw,
+				CompositeQuery: qbtypes.CompositeQuery{Queries: []qbtypes.QueryEnvelope{{
+					Type: qbtypes.QueryTypeClickHouseSQL,
+					Spec: qbtypes.ClickHouseQuery{Name: "A", Query: "SELECT 1"},
+				}}},
+			}
+
+			queries, _, err := q.buildQueries(valuer.GenerateUUID(), req, nil, nil, &qbtypes.QBEvent{}, q.newQueryRangeOptions(testCase.opts).clickHouseQueryLimits)
+			require.NoError(t, err)
+			require.IsType(t, &chSQLQuery{}, queries["A"])
+			assert.Equal(t, testCase.want, queries["A"].(*chSQLQuery).limits)
+		})
+	}
+}

pkg/querier/queriertest/querier.go

@@ -24,7 +24,7 @@ func NewMockQuerier() *MockQuerier {
 	return &MockQuerier{}
 }
 
-func (m *MockQuerier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error) {
+func (m *MockQuerier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, _ ...querier.QueryRangeOption) (*qbtypes.QueryRangeResponse, error) {
 	if m.QueryRangeFunc != nil {
 		return m.QueryRangeFunc(ctx, orgID, req)
 	}

pkg/querier/signozquerier/provider.go

@@ -52,6 +52,7 @@ func NewFactory(
 				flagger,
 				cfg.LogTraceIDWindowPadding,
 				cfg.MaxConcurrentQueries,
+				cfg.ClickHouseQuery,
 			), nil
 		},
 	)

pkg/query-service/rules/setups_test.go

@@ -52,6 +52,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		fl,
 		0,
 		0, // maxConcurrentQueries (0 means default)
+		querier.ClickHouseQueryConfig{},
 	), metadataStore
 }
 
@@ -87,6 +88,7 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
 		fl,
 		5*time.Minute, // logTraceIDWindowPadding
 		0,             // maxConcurrentQueries (0 means default)
+		querier.ClickHouseQueryConfig{},
 	)
 }
 
@@ -123,6 +125,7 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
 		fl,
 		0,
 		0, // maxConcurrentQueries (0 means default)
+		querier.ClickHouseQueryConfig{},
 	)
 }
 
@@ -159,5 +162,6 @@ func prepareQuerierForAITraces(t *testing.T, telemetryStore telemetrystore.Telem
 		fl,
 		0,
 		0, // maxConcurrentQueries (0 means default)
+		querier.ClickHouseQueryConfig{},
 	)
 }

pkg/query-service/rules/threshold_rule.go

@@ -138,7 +138,8 @@ func (r *ThresholdRule) buildAndRunQuery(ctx context.Context, orgID valuer.UUID,
 		instrumentationtypes.CodeFunctionName: "buildAndRunQuery",
 	})
 
-	v5Result, err := r.querier.QueryRange(ctx, orgID, params)
+	// Rules run their ClickHouse SQL without the query_range API limits.
+	v5Result, err := r.querier.QueryRange(ctx, orgID, params, querier.WithClickHouseQueryLimits(querier.ClickHouseQueryConfig{}))
 	if err != nil {
 		return nil, err
 	}

pkg/telemetrystore/clickhousetelemetrystore/provider.go

@@ -2,6 +2,7 @@ package clickhousetelemetrystore
 
 import (
 	"context"
+	"os"
 
 	chproto "github.com/ClickHouse/ch-go/proto"
 	"github.com/ClickHouse/clickhouse-go/v2"
@@ -22,12 +23,16 @@ var (
 
 	ErrCodeDistributedSubqueryDenied = errors.MustNewCode("distributed_subquery_denied")
 	ErrCodeDecimalOverflow           = errors.MustNewCode("decimal_overflow")
+	ErrCodeMemoryLimitExceeded       = errors.MustNewCode("memory_limit_exceeded")
+	ErrCodeResultTooLarge            = errors.MustNewCode("result_too_large")
 
 	ErrCodeQueryCanceled   = errors.MustNewCode("query_canceled")
 	ErrCodeQueryTimeout    = errors.MustNewCode("query_timeout")
 	ErrCodeExecutionFailed = errors.MustNewCode("execution_failed")
 )
 
+const queryTimeoutSuggestion = "optimize your query: narrow the time range or add filters to read less data"
+
 // Codes absent from this map fall through to the raw driver error in castError.
 var clickHouseExceptionWrappers = map[chproto.Error]func(cause error, ex *clickhouse.Exception) error{
 	chproto.ErrSyntaxError: func(cause error, ex *clickhouse.Exception) error {
@@ -87,6 +92,23 @@ var clickHouseExceptionWrappers = map[chproto.Error]func(cause error, ex *clickh
 		return errors.WrapInvalidInputf(cause, ErrCodeDecimalOverflow, "decimal overflow: %s", ex.Message).
 			WithSuggestions("cast the Decimal values to Float64, or to a wider Decimal such as Decimal128, before the arithmetic")
 	},
+	chproto.ErrTimeoutExceeded: func(cause error, ex *clickhouse.Exception) error {
+		return errors.WrapTimeoutf(cause, ErrCodeQueryTimeout, "query timed out: %s", ex.Message).
+			WithSuggestions(queryTimeoutSuggestion)
+	},
+	// ClickHouse estimates the query will run past max_execution_time and stops it early.
+	chproto.ErrTooSlow: func(cause error, ex *clickhouse.Exception) error {
+		return errors.WrapTimeoutf(cause, ErrCodeQueryTimeout, "query is estimated to time out: %s", ex.Message).
+			WithSuggestions(queryTimeoutSuggestion)
+	},
+	chproto.ErrMemoryLimitExceeded: func(cause error, ex *clickhouse.Exception) error {
+		return errors.WrapInvalidInputf(cause, ErrCodeMemoryLimitExceeded, "query memory limit exceeded: %s", ex.Message).
+			WithSuggestions("narrow the time range, add filters or reduce the number of groups")
+	},
+	chproto.ErrTooManyRowsOrBytes: func(cause error, ex *clickhouse.Exception) error {
+		return errors.WrapInvalidInputf(cause, ErrCodeResultTooLarge, "query result too large: %s", ex.Message).
+			WithSuggestions("add a LIMIT, narrow the time range or select fewer columns")
+	},
 }
 
 type provider struct {
@@ -292,8 +314,9 @@ func castError(err error) error {
 	if errors.Is(err, context.Canceled) {
 		return errors.WrapCanceledf(err, ErrCodeQueryCanceled, "query canceled")
 	}
-	if errors.Is(err, context.DeadlineExceeded) {
-		return errors.WrapTimeoutf(err, ErrCodeQueryTimeout, "query timed out")
+	// A socket read can hit the deadline the driver copies from ctx before the driver sees ctx is done.
+	if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, os.ErrDeadlineExceeded) {
+		return errors.WrapTimeoutf(err, ErrCodeQueryTimeout, "query timed out").WithSuggestions(queryTimeoutSuggestion)
 	}
 
 	var ex *clickhouse.Exception

pkg/telemetrystore/clickhousetelemetrystore/provider_test.go

@@ -0,0 +1,72 @@
+package clickhousetelemetrystore
+
+import (
+	"context"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	chproto "github.com/ClickHouse/ch-go/proto"
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+)
+
+func TestCastError_LimitExceptions(t *testing.T) {
+	testCases := []struct {
+		name        string
+		code        chproto.Error
+		wantTimeout bool
+		wantCode    errors.Code
+	}{
+		{name: "TooManyRowsOrBytes_ResultTooLarge", code: chproto.ErrTooManyRowsOrBytes, wantCode: ErrCodeResultTooLarge},
+		{name: "MemoryLimitExceeded_InvalidInput", code: chproto.ErrMemoryLimitExceeded, wantCode: ErrCodeMemoryLimitExceeded},
+		{name: "TimeoutExceeded_Timeout", code: chproto.ErrTimeoutExceeded, wantTimeout: true, wantCode: ErrCodeQueryTimeout},
+		{name: "TooSlow_Timeout", code: chproto.ErrTooSlow, wantTimeout: true, wantCode: ErrCodeQueryTimeout},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			err := castError(&clickhouse.Exception{Code: int32(testCase.code), Message: "limit exceeded"})
+
+			// Required rather than asserted: errors.Asc dereferences the error it is given.
+			require.Error(t, err)
+			assert.True(t, errors.Asc(err, testCase.wantCode), "got %v", err)
+			assert.NotEmpty(t, errors.AsJSON(err).Suggestions)
+			if testCase.wantTimeout {
+				assert.True(t, errors.Ast(err, errors.TypeTimeout))
+			} else {
+				assert.True(t, errors.Ast(err, errors.TypeInvalidInput))
+			}
+		})
+	}
+}
+
+func TestCastError_Timeout(t *testing.T) {
+	deadlineCtx, cancel := context.WithTimeout(context.Background(), time.Nanosecond)
+	defer cancel()
+	<-deadlineCtx.Done()
+
+	testCases := []struct {
+		name string
+		err  error
+	}{
+		{name: "ContextDeadline", err: deadlineCtx.Err()},
+		// The driver sets the socket deadline from ctx, so a read can fail before it sees ctx is done.
+		{name: "SocketTimeout", err: &net.OpError{Op: "read", Net: "tcp", Err: os.ErrDeadlineExceeded}},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			err := castError(testCase.err)
+
+			require.Error(t, err)
+			assert.True(t, errors.Asc(err, ErrCodeQueryTimeout), "got %v", err)
+			assert.True(t, errors.Ast(err, errors.TypeTimeout))
+			assert.Equal(t, []string{queryTimeoutSuggestion}, errors.AsJSON(err).Suggestions)
+		})
+	}
+}

pkg/telemetrystore/query_settings.go

@@ -0,0 +1,25 @@
+package telemetrystore
+
+import (
+	"context"
+	"maps"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+)
+
+type querySettingsKey struct{}
+
+// NewContextWithQuerySettings attaches ClickHouse settings for the queries run with ctx.
+// The settings hook applies them over its defaults, so they win on a clash.
+func NewContextWithQuerySettings(ctx context.Context, settings clickhouse.Settings) context.Context {
+	merged := clickhouse.Settings{}
+	maps.Copy(merged, QuerySettingsFromContext(ctx))
+	maps.Copy(merged, settings)
+	return context.WithValue(ctx, querySettingsKey{}, merged)
+}
+
+// QuerySettingsFromContext returns the settings attached with NewContextWithQuerySettings, or nil.
+func QuerySettingsFromContext(ctx context.Context) clickhouse.Settings {
+	settings, _ := ctx.Value(querySettingsKey{}).(clickhouse.Settings)
+	return settings
+}

pkg/telemetrystore/query_settings_test.go

@@ -0,0 +1,37 @@
+package telemetrystore
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestQuerySettingsFromContext(t *testing.T) {
+	testCases := []struct {
+		name string
+		ctx  func() context.Context
+		want clickhouse.Settings
+	}{
+		{
+			name: "Unset_Nil",
+			ctx:  context.Background,
+			want: nil,
+		},
+		{
+			name: "NestedCalls_MergeLaterWins",
+			ctx: func() context.Context {
+				ctx := NewContextWithQuerySettings(context.Background(), clickhouse.Settings{"max_execution_time": 30, "max_memory_usage": 100})
+				return NewContextWithQuerySettings(ctx, clickhouse.Settings{"max_execution_time": 10})
+			},
+			want: clickhouse.Settings{"max_execution_time": 10, "max_memory_usage": 100},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, QuerySettingsFromContext(testCase.ctx()))
+		})
+	}
+}

pkg/telemetrystore/telemetrystorehook/settings.go

@@ -2,6 +2,7 @@ package telemetrystorehook
 
 import (
 	"context"
+	"maps"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/SigNoz/signoz/pkg/factory"
@@ -72,6 +73,9 @@ func (h *provider) BeforeQuery(ctx context.Context, _ *telemetrystore.QueryEvent
 		settings["result_overflow_mode"] = ctx.Value("result_overflow_mode")
 	}
 
+	// Per-query settings override the ones above.
+	maps.Copy(settings, telemetrystore.QuerySettingsFromContext(ctx))
+
 	ctx = clickhouse.Context(ctx, clickhouse.WithSettings(settings))
 	return ctx
 }

tests/fixtures/querier.py

@@ -30,6 +30,7 @@ class TelemetryFieldKey:
 
 class RequestType:
     RAW = "raw"
+    TRACE = "trace"
     TIME_SERIES = "time_series"
     SCALAR = "scalar"
     TABLE = "table"

tests/integration/tests/querierclickhousequery/01_budget.py

@@ -0,0 +1,146 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+
+import pytest
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.querier import RequestType, make_query_request
+
+
+def test_clickhouse_query_within_budget(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    now = datetime.now(tz=UTC)
+    response = make_query_request(
+        signoz,
+        get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD),
+        int((now - timedelta(hours=1)).timestamp() * 1000),
+        int(now.timestamp() * 1000),
+        [{"type": "clickhouse_sql", "spec": {"name": "A", "query": "SELECT number FROM numbers(1000)"}}],
+        request_type=RequestType.RAW,
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert len(response.json()["data"]["data"]["results"][0]["rows"]) == 1000
+
+
+def test_clickhouse_query_row_limit_only_for_raw(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    now = datetime.now(tz=UTC)
+    response = make_query_request(
+        signoz,
+        get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD),
+        int((now - timedelta(hours=1)).timestamp() * 1000),
+        int(now.timestamp() * 1000),
+        [{"type": "clickhouse_sql", "spec": {"name": "A", "query": "SELECT toString(number) AS k, toFloat64(number) AS __result_0 FROM numbers(2000)"}}],
+        request_type=RequestType.SCALAR,
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+
+
+@pytest.mark.parametrize(
+    "query, request_type, expected_status, expected_code, expected_suggestion",
+    [
+        pytest.param(
+            "SELECT number FROM numbers(2000)",
+            RequestType.RAW,
+            HTTPStatus.BAD_REQUEST,
+            "result_too_large",
+            "add a LIMIT, narrow the time range or select fewer columns",
+            id="max_result_rows",
+        ),
+        pytest.param(
+            "SELECT number FROM numbers(2000)",
+            RequestType.TRACE,
+            HTTPStatus.BAD_REQUEST,
+            "result_too_large",
+            "add a LIMIT, narrow the time range or select fewer columns",
+            id="max_result_rows_trace",
+        ),
+        pytest.param(
+            "SELECT repeat(toString(number), 4000) AS s FROM numbers(500)",
+            RequestType.RAW,
+            HTTPStatus.BAD_REQUEST,
+            "result_too_large",
+            "add a LIMIT, narrow the time range or select fewer columns",
+            id="max_result_bytes",
+        ),
+        pytest.param(
+            "SELECT groupArray(number) FROM numbers(50000000)",
+            RequestType.RAW,
+            HTTPStatus.BAD_REQUEST,
+            "memory_limit_exceeded",
+            "narrow the time range, add filters or reduce the number of groups",
+            id="max_memory_usage",
+        ),
+        pytest.param(
+            "SELECT sleep(3)",
+            RequestType.RAW,
+            HTTPStatus.GATEWAY_TIMEOUT,
+            "query_timeout",
+            "optimize your query: narrow the time range or add filters to read less data",
+            id="max_execution_time",
+        ),
+    ],
+)
+def test_clickhouse_query_over_budget(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    query: str,
+    request_type: str,
+    expected_status: HTTPStatus,
+    expected_code: str,
+    expected_suggestion: str,
+) -> None:
+    now = datetime.now(tz=UTC)
+    response = make_query_request(
+        signoz,
+        get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD),
+        int((now - timedelta(hours=1)).timestamp() * 1000),
+        int(now.timestamp() * 1000),
+        [{"type": "clickhouse_sql", "spec": {"name": "A", "query": query}}],
+        request_type=request_type,
+    )
+
+    assert response.status_code == expected_status, response.text
+    error = response.json()["error"]
+    assert error["code"] == expected_code
+    assert error["suggestions"] == [expected_suggestion]
+
+
+def test_clickhouse_query_budget_reaches_clickhouse(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    now = datetime.now(tz=UTC)
+    response = make_query_request(
+        signoz,
+        get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD),
+        int((now - timedelta(hours=1)).timestamp() * 1000),
+        int(now.timestamp() * 1000),
+        [{"type": "clickhouse_sql", "spec": {"name": "A", "query": "SELECT 'budget_precedence_probe' AS probe"}}],
+        request_type=RequestType.RAW,
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+
+    conn = signoz.telemetrystore.conn
+    conn.command("SYSTEM FLUSH LOGS")
+    result = conn.query("SELECT Settings FROM system.query_log WHERE type = 'QueryFinish' AND position(query, 'budget_precedence_probe') > 0 AND position(query, 'system.query_log') = 0 ORDER BY event_time_microseconds DESC LIMIT 1")
+
+    assert result.result_rows, "the probe query is missing from system.query_log"
+    settings = result.result_rows[0][0]
+    # clickhouse-go sets max_execution_time from the 2s budget deadline plus 5s, not from the 60s request timeout.
+    assert int(settings["max_execution_time"]) <= 7
+    assert settings["max_memory_usage"] == "100000000"
+    assert settings["max_result_rows"] == "1000"
+    assert settings["max_result_bytes"] == "1000000"

tests/integration/tests/querierclickhousequery/conftest.py

@@ -0,0 +1,34 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.signoz import create_signoz
+
+
+@pytest.fixture(name="signoz", scope="package")
+def signoz_clickhouse_query(
+    network: Network,
+    migrator: types.Operation,  # pylint: disable=unused-argument
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.SigNoz:
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-clickhouse-query",
+        env_overrides={
+            "SIGNOZ_QUERIER_CLICKHOUSE__QUERY_MAX__EXECUTION__TIME": "2s",
+            "SIGNOZ_QUERIER_CLICKHOUSE__QUERY_MAX__MEMORY__USAGE": 100_000_000,
+            "SIGNOZ_QUERIER_CLICKHOUSE__QUERY_MAX__RESULT__ROWS": 1_000,
+            "SIGNOZ_QUERIER_CLICKHOUSE__QUERY_MAX__RESULT__BYTES": 1_000_000,
+        },
+    )