fix(querier): drop the block comment guard now the parser handles it

The parser no longer loops on an unterminated block comment, so the hand-rolled
scan that worked around it is redundant, and it was the riskier of the two: it
duplicated the lexer's handling of string literals and could have rejected a
legitimate query. Leave the parser as the single source of truth.

The panic case likewise no longer panics, so its test can no longer cover the
recover and is removed. The recover stays as insurance, since this reaches
user-authored SQL and the parser has regressed this way before.

Cover INTERSECT and EXCEPT, which the parser only started accepting in this
version and which reach a second query through the same rules.

go.mod

@@ -4,7 +4,7 @@ go 1.25.7
 
 require (
 	dario.cat/mergo v1.0.2
-	github.com/AfterShip/clickhouse-sql-parser v0.4.16
+	github.com/AfterShip/clickhouse-sql-parser v0.5.2
 	github.com/ClickHouse/clickhouse-go/v2 v2.44.0
 	github.com/DATA-DOG/go-sqlmock v1.5.2
 	github.com/SigNoz/clickhouse-go-mock v0.14.0

go.sum

@@ -66,8 +66,8 @@ dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
 filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
-github.com/AfterShip/clickhouse-sql-parser v0.4.16 h1:gpl+wXclYUKT0p4+gBq22XeRYWwEoZ9f35vogqMvkLQ=
-github.com/AfterShip/clickhouse-sql-parser v0.4.16/go.mod h1:W0Z82wJWkJxz2RVun/RMwxue3g7ut47Xxl+SFqdJGus=
+github.com/AfterShip/clickhouse-sql-parser v0.5.2 h1:KCxdmvuVawVF4yl0ZS33q7olgmLZwvZG5BjWHp6o414=
+github.com/AfterShip/clickhouse-sql-parser v0.5.2/go.mod h1:Qi3qvPTfZb/aFwI5V4WFOahgjsLJa4MzVijIAfwOhDw=
 github.com/Azure/azure-sdk-for-go v68.0.0+incompatible h1:fcYLmCpyNYRnvJbPerq7U0hS+6+I79yEDJBqVNcqUzU=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0 h1:fou+2+WFTib47nS+nz/ozhEBnvU96bKHy6LjRsY4E28=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0/go.mod h1:t76Ruy8AHvUAC8GfMWJMa0ElSbuIcO03NLpynfbgsPA=

pkg/querier/clickhouse_query.go

@@ -100,9 +100,22 @@ func (q *chSQLQuery) renderVars(query string, vars map[string]qbtypes.VariableIt
 	return newQuery.String(), nil
 }
 
+func (q *chSQLQuery) render() (string, error) {
+	rendered, err := q.renderVars(q.query.Query, q.vars, q.fromMS, q.toMS)
+	if err != nil {
+		return "", err
+	}
+
+	if err := querybuilder.ValidateReadOnlySelect(rendered); err != nil {
+		return "", err
+	}
+
+	return rendered, nil
+}
+
 // Statement renders the SQL without executing it, for the preview path.
 func (q *chSQLQuery) Statement(_ context.Context) (*qbtypes.Statement, error) {
-	rendered, err := q.renderVars(q.query.Query, q.vars, q.fromMS, q.toMS)
+	rendered, err := q.render()
 	if err != nil {
 		return nil, err
 	}
@@ -124,11 +137,13 @@ func (q *chSQLQuery) Execute(ctx context.Context) (*qbtypes.Result, error) {
 		elapsed += p.Elapsed
 	}))
 
-	query, err := q.renderVars(q.query.Query, q.vars, q.fromMS, q.toMS)
+	query, err := q.render()
 	if err != nil {
 		return nil, err
 	}
 
+	ctx = ctxtypes.SetClickhouseReadOnly(ctx)
+
 	rows, err := q.telemetryStore.ClickhouseDB().Query(ctx, query, q.args...)
 	if err != nil {
 		return nil, err

pkg/query-service/app/http_handler.go

@@ -11,6 +11,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/flagger"
 	"github.com/SigNoz/signoz/pkg/modules/thirdpartyapi"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/queryparser"
 
 	"log/slog"
@@ -1021,21 +1022,6 @@ func prepareQuery(r *http.Request) (string, error) {
 		return "", fmt.Errorf("query is required")
 	}
 
-	notAllowedOps := []string{
-		"alter table",
-		"drop table",
-		"truncate table",
-		"drop database",
-		"drop view",
-		"drop function",
-	}
-
-	for _, op := range notAllowedOps {
-		if strings.Contains(strings.ToLower(query), op) {
-			return "", fmt.Errorf("operation %s is not allowed", op)
-		}
-	}
-
 	vars := make(map[string]string)
 	for k, v := range postData.Variables {
 		vars[k] = metrics.FormattedValue(v)
@@ -1051,30 +1037,34 @@ func prepareQuery(r *http.Request) (string, error) {
 		return "", tmplErr
 	}
 
-	if !constants.IsDotMetricsEnabled {
-		return queryBuf.String(), nil
-	}
-
 	query = queryBuf.String()
 
-	// Now handle $var replacements (simple string replace)
-	keys := make([]string, 0, len(vars))
-	for k := range vars {
-		keys = append(keys, k)
+	if constants.IsDotMetricsEnabled {
+		// Now handle $var replacements (simple string replace)
+		keys := make([]string, 0, len(vars))
+		for k := range vars {
+			keys = append(keys, k)
+		}
+
+		sort.Slice(keys, func(i, j int) bool {
+			return len(keys[i]) > len(keys[j])
+		})
+
+		newQuery := query
+		for _, k := range keys {
+			placeholder := "$" + k
+			v := vars[k]
+			newQuery = strings.ReplaceAll(newQuery, placeholder, v)
+		}
+
+		query = newQuery
 	}
 
-	sort.Slice(keys, func(i, j int) bool {
-		return len(keys[i]) > len(keys[j])
-	})
-
-	newQuery := query
-	for _, k := range keys {
-		placeholder := "$" + k
-		v := vars[k]
-		newQuery = strings.ReplaceAll(newQuery, placeholder, v)
+	if err := querybuilder.ValidateReadOnlySelect(query); err != nil {
+		return "", err
 	}
 
-	return newQuery, nil
+	return query, nil
 }
 
 func (aH *APIHandler) Get(rw http.ResponseWriter, r *http.Request) {
@@ -1092,7 +1082,7 @@ func (aH *APIHandler) queryDashboardVarsV2(w http.ResponseWriter, r *http.Reques
 		return
 	}
 
-	dashboardVars, err := aH.reader.QueryDashboardVars(r.Context(), query)
+	dashboardVars, err := aH.reader.QueryDashboardVars(ctxtypes.SetClickhouseReadOnly(r.Context()), query)
 	if err != nil {
 		RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
 		return

pkg/query-service/app/http_handler_test.go

@@ -67,7 +67,26 @@ func TestPrepareQuery(t *testing.T) {
 				Query: "ALTER TABLE signoz_table DELETE where true",
 			},
 			expectedErr: true,
-			errMsg:      "operation alter table is not allowed",
+			errMsg:      "only SELECT statements are allowed in ClickHouse SQL queries",
+		},
+		{
+			name: "query smuggles a statement through a variable",
+			postData: &model.DashboardVars{
+				Query: "select * from table where id = {{.id}}",
+				Variables: map[string]interface{}{
+					"id": "1'; DROP TABLE signoz_table; --",
+				},
+			},
+			expectedErr: true,
+			errMsg:      "ClickHouse SQL must contain exactly one statement",
+		},
+		{
+			name: "query reads a system table",
+			postData: &model.DashboardVars{
+				Query: "SELECT name FROM system.users",
+			},
+			expectedErr: true,
+			errMsg:      "the ClickHouse system database is not allowed in SQL queries",
 		},
 		{
 			name: "query text produces template exec error",

pkg/querybuilder/agg_rewrite.go

@@ -96,9 +96,9 @@ func (r *aggExprRewriter) Rewrite(
 	}
 
 	if visitor.isRate {
-		return fmt.Sprintf("%s/%d", sel.SelectItems[0].String(), rateInterval), visitor.chArgs, nil
+		return fmt.Sprintf("%s/%d", chparser.Format(sel.SelectItems[0]), rateInterval), visitor.chArgs, nil
 	}
-	return sel.SelectItems[0].String(), visitor.chArgs, nil
+	return chparser.Format(sel.SelectItems[0]), visitor.chArgs, nil
 }
 
 // RewriteMulti rewrites a slice of expressions.
@@ -207,7 +207,7 @@ func (v *exprVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
 	// Handle *If functions with predicate + values
 	if aggFunc.FuncCombinator {
 		// Map the predicate (last argument)
-		origPred := args[len(args)-1].String()
+		origPred := chparser.Format(args[len(args)-1])
 		whereClause, err := PrepareWhereClause(
 			origPred,
 			FilterExprVisitorOpts{
@@ -242,7 +242,7 @@ func (v *exprVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
 
 		// Map each value column argument
 		for i := 0; i < len(args)-1; i++ {
-			origVal := args[i].String()
+			origVal := chparser.Format(args[i])
 			fieldKey := telemetrytypes.GetFieldKeyFromKeyText(origVal)
 			expr, err := v.fieldMapper.ColumnExpressionFor(v.ctx, v.orgID, v.startNs, v.endNs, &fieldKey, dataType, v.fieldKeys)
 			if err != nil {
@@ -259,7 +259,7 @@ func (v *exprVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
 	} else {
 		// Non-If functions: map every argument as a column/value
 		for i, arg := range args {
-			orig := arg.String()
+			orig := chparser.Format(arg)
 			fieldKey := telemetrytypes.GetFieldKeyFromKeyText(orig)
 			expr, err := v.fieldMapper.ColumnExpressionFor(v.ctx, v.orgID, v.startNs, v.endNs, &fieldKey, dataType, v.fieldKeys)
 			if err != nil {

pkg/querybuilder/clickhouse_sql.go

@@ -0,0 +1,68 @@
+package querybuilder
+
+import (
+	"strings"
+
+	chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
+	"github.com/SigNoz/signoz/pkg/errors"
+)
+
+// internalDatabases hold server metadata, credentials and grants rather than telemetry.
+var internalDatabases = map[string]struct{}{
+	"system":             {},
+	"information_schema": {},
+}
+
+// ValidateReadOnlySelect rejects a user-authored ClickHouse statement unless it is a
+// single SELECT that reads telemetry: no table function, no internal database and no
+// lowering of the readonly setting. It must run on the rendered statement, since the
+// substituted variable values are user input too.
+func ValidateReadOnlySelect(query string) (err error) {
+	// The parser has a history of panicking on malformed input rather than returning
+	// an error. No input is known to still do so, but this reaches user-authored SQL,
+	// so a regression must not take the process down.
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			err = errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid ClickHouse SQL: %v", recovered)
+		}
+	}()
+
+	stmts, parseErr := chparser.NewParser(query).ParseStmts()
+	if parseErr != nil {
+		// The cause is carried in the message because the renderers drop the wrapped error.
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid ClickHouse SQL: %s", parseErr.Error())
+	}
+
+	if len(stmts) != 1 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "ClickHouse SQL must contain exactly one statement")
+	}
+
+	selectQuery, ok := stmts[0].(*chparser.SelectQuery)
+	if !ok {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "only SELECT statements are allowed in ClickHouse SQL queries")
+	}
+
+	visitor := &chparser.DefaultASTVisitor{Visit: func(node chparser.Expr) error {
+		switch expr := node.(type) {
+		case *chparser.TableFunctionExpr:
+			// Source table functions remain usable in ClickHouse read-only mode.
+			return errors.NewInvalidInputf(errors.CodeInvalidInput, "ClickHouse table functions are not allowed in SQL queries: %s", chparser.Format(expr.Name))
+		case *chparser.TableIdentifier:
+			// Reading these is unaffected by ClickHouse read-only mode.
+			if expr.Database == nil {
+				return nil
+			}
+			if _, ok := internalDatabases[strings.ToLower(expr.Database.Name)]; ok {
+				return errors.NewInvalidInputf(errors.CodeInvalidInput, "the ClickHouse %s database is not allowed in SQL queries", expr.Database.Name)
+			}
+		case *chparser.SettingExpr:
+			// A query-level setting takes precedence over the context setting.
+			if strings.EqualFold(expr.Name.Name, "readonly") {
+				return errors.NewInvalidInputf(errors.CodeInvalidInput, "the ClickHouse readonly setting cannot be overridden")
+			}
+		}
+		return nil
+	}}
+
+	return selectQuery.Accept(visitor)
+}

pkg/querybuilder/clickhouse_sql_test.go

@@ -0,0 +1,326 @@
+package querybuilder
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidateReadOnlySelect(t *testing.T) {
+	testCases := []struct {
+		name  string
+		query string
+		pass  bool
+	}{
+		{
+			name:  "Select_Valid",
+			query: "SELECT region AS r, zone FROM metrics WHERE metric_name = 'cpu' GROUP BY region, zone",
+			pass:  true,
+		},
+		{
+			name:  "SelectWithTrailingSemicolon_Valid",
+			query: "SELECT count() FROM signoz_logs.distributed_logs_v2;",
+			pass:  true,
+		},
+		{
+			name:  "CommonTableExpression_Valid",
+			query: "WITH t AS (SELECT fingerprint FROM signoz_metrics.time_series_v4 GROUP BY fingerprint) SELECT * FROM t",
+			pass:  true,
+		},
+		{
+			name:  "Join_Valid",
+			query: "SELECT * FROM t1 LEFT JOIN t2 ON t1.a = t2.b",
+			pass:  true,
+		},
+		{
+			name:  "GlobalIn_Valid",
+			query: "SELECT a FROM t WHERE a GLOBAL IN (SELECT b FROM t2)",
+			pass:  true,
+		},
+		{
+			name:  "Union_Valid",
+			query: "SELECT * FROM t UNION ALL SELECT * FROM t2",
+			pass:  true,
+		},
+		{
+			name:  "WindowFunction_Valid",
+			query: "SELECT sum(v) OVER (PARTITION BY a ORDER BY t) FROM t",
+			pass:  true,
+		},
+		{
+			name:  "UnrelatedSetting_Valid",
+			query: "SELECT * FROM t SETTINGS max_threads = 4",
+			pass:  true,
+		},
+		{
+			name:  "Empty_Invalid",
+			query: "",
+			pass:  false,
+		},
+		{
+			name:  "Unparseable_Invalid",
+			query: "SELECT FROM WHERE",
+			pass:  false,
+		},
+		{
+			name:  "TerminatedBlockComment_Valid",
+			query: "SELECT /* keep me */ count() FROM t",
+			pass:  true,
+		},
+		{
+			name:  "BlockCommentMarkerInsideStringLiteral_Valid",
+			query: "SELECT count() FROM t WHERE body = '/* not a comment'",
+			pass:  true,
+		},
+		{
+			// The parser used to loop forever on this. It now reads the comment to the
+			// end of the input, so this doubles as a canary for that regression.
+			name:  "TrailingUnterminatedBlockComment_Valid",
+			query: "SELECT count() FROM t /* unterminated",
+			pass:  true,
+		},
+		{
+			name:  "UnterminatedBlockCommentOnly_Invalid",
+			query: "/* x",
+			pass:  false,
+		},
+		{
+			name:  "Intersect_Valid",
+			query: "SELECT * FROM t INTERSECT SELECT * FROM t2",
+			pass:  true,
+		},
+		{
+			name:  "IntersectReadingInternalDatabase_Invalid",
+			query: "SELECT * FROM t INTERSECT SELECT * FROM system.users",
+			pass:  false,
+		},
+		{
+			name:  "ExceptReadingTableFunction_Invalid",
+			query: "SELECT * FROM t EXCEPT SELECT * FROM url('http://x', CSV, 'a String')",
+			pass:  false,
+		},
+		{
+			name:  "MultipleStatements_Invalid",
+			query: "SELECT 1; DROP TABLE signoz_logs.logs_v2",
+			pass:  false,
+		},
+		{
+			name:  "Drop_Invalid",
+			query: "DROP TABLE signoz_logs.logs_v2",
+			pass:  false,
+		},
+		{
+			name:  "Insert_Invalid",
+			query: "INSERT INTO signoz_logs.logs_v2 SELECT * FROM signoz_logs.logs_v2",
+			pass:  false,
+		},
+		{
+			name:  "AlterDelete_Invalid",
+			query: "ALTER TABLE signoz_logs.logs_v2 DELETE WHERE 1 = 1",
+			pass:  false,
+		},
+		{
+			name:  "Truncate_Invalid",
+			query: "TRUNCATE TABLE signoz_logs.logs_v2",
+			pass:  false,
+		},
+		{
+			name:  "CreateTable_Invalid",
+			query: "CREATE TABLE evil (a Int) ENGINE = Memory",
+			pass:  false,
+		},
+		{
+			name:  "AttachTable_Invalid",
+			query: "ATTACH TABLE x",
+			pass:  false,
+		},
+		{
+			name:  "Optimize_Invalid",
+			query: "OPTIMIZE TABLE signoz_logs.logs_v2 FINAL",
+			pass:  false,
+		},
+		{
+			name:  "Grant_Invalid",
+			query: "GRANT ALL ON *.* TO admin",
+			pass:  false,
+		},
+		{
+			name:  "Describe_Invalid",
+			query: "DESCRIBE TABLE signoz_logs.logs_v2",
+			pass:  false,
+		},
+		{
+			name:  "Set_Invalid",
+			query: "SET readonly = 0",
+			pass:  false,
+		},
+		{
+			name:  "ShowGrants_Invalid",
+			query: "SHOW GRANTS",
+			pass:  false,
+		},
+		{
+			name:  "System_Invalid",
+			query: "SYSTEM SHUTDOWN",
+			pass:  false,
+		},
+		{
+			name:  "Kill_Invalid",
+			query: "KILL QUERY WHERE 1",
+			pass:  false,
+		},
+		{
+			name:  "IntoOutfile_Invalid",
+			query: "SELECT * FROM t INTO OUTFILE '/tmp/x.csv'",
+			pass:  false,
+		},
+		{
+			name:  "UrlTableFunction_Invalid",
+			query: "SELECT * FROM url('http://attacker.example/x', CSV, 'a String')",
+			pass:  false,
+		},
+		{
+			name:  "FileTableFunction_Invalid",
+			query: "SELECT * FROM file('/etc/passwd', CSV, 'a String')",
+			pass:  false,
+		},
+		{
+			name:  "RemoteTableFunction_Invalid",
+			query: "SELECT * FROM remote('attacker:9000', system.users)",
+			pass:  false,
+		},
+		{
+			name:  "S3TableFunction_Invalid",
+			query: "SELECT * FROM s3('https://x/y.csv', 'CSV')",
+			pass:  false,
+		},
+		{
+			name:  "MysqlTableFunction_Invalid",
+			query: "SELECT * FROM mysql('host:3306', 'db', 'tbl', 'u', 'p')",
+			pass:  false,
+		},
+		{
+			name:  "ExecutableTableFunction_Invalid",
+			query: "SELECT * FROM executable('script.sh', CSV, 'a String')",
+			pass:  false,
+		},
+		{
+			name:  "ClusterTableFunction_Invalid",
+			query: "SELECT * FROM cluster('c', system, users)",
+			pass:  false,
+		},
+		{
+			name:  "TableFunctionInJoin_Invalid",
+			query: "SELECT * FROM t1 JOIN url('http://x', CSV, 'a String') u ON 1 = 1",
+			pass:  false,
+		},
+		{
+			name:  "TableFunctionInScalarSubquery_Invalid",
+			query: "SELECT (SELECT * FROM url('http://x', CSV, 'a String'))",
+			pass:  false,
+		},
+		{
+			name:  "TableFunctionInCommonTableExpression_Invalid",
+			query: "WITH c AS (SELECT * FROM url('http://x', CSV, 'a String')) SELECT * FROM c",
+			pass:  false,
+		},
+		{
+			name:  "TableFunctionInNestedSubquery_Invalid",
+			query: "SELECT * FROM (SELECT * FROM (SELECT * FROM file('/etc/passwd', CSV, 'a String')))",
+			pass:  false,
+		},
+		{
+			name:  "TableFunctionInWhereSubquery_Invalid",
+			query: "SELECT * FROM t WHERE a IN (SELECT * FROM url('http://x', CSV, 'a String'))",
+			pass:  false,
+		},
+		{
+			name:  "TableFunctionInUnion_Invalid",
+			query: "SELECT * FROM t UNION ALL SELECT * FROM url('http://x', CSV, 'a String')",
+			pass:  false,
+		},
+		{
+			name:  "SystemUsers_Invalid",
+			query: "SELECT * FROM system.users",
+			pass:  false,
+		},
+		{
+			name:  "SystemGrants_Invalid",
+			query: "SELECT * FROM system.grants",
+			pass:  false,
+		},
+		{
+			name:  "SystemQuoted_Invalid",
+			query: "SELECT count() FROM `system`.`tables`",
+			pass:  false,
+		},
+		{
+			name:  "SystemUppercase_Invalid",
+			query: "SELECT * FROM SYSTEM.USERS",
+			pass:  false,
+		},
+		{
+			name:  "SystemInSubquery_Invalid",
+			query: "SELECT * FROM (SELECT name FROM system.parts)",
+			pass:  false,
+		},
+		{
+			name:  "SystemInCommonTableExpression_Invalid",
+			query: "WITH c AS (SELECT * FROM system.query_log) SELECT * FROM c",
+			pass:  false,
+		},
+		{
+			name:  "SystemInJoin_Invalid",
+			query: "SELECT * FROM signoz_logs.distributed_logs_v2 AS l JOIN system.users AS u ON 1 = 1",
+			pass:  false,
+		},
+		{
+			name:  "SystemInUnion_Invalid",
+			query: "SELECT name FROM t UNION ALL SELECT name FROM system.tables",
+			pass:  false,
+		},
+		{
+			name:  "InformationSchema_Invalid",
+			query: "SELECT * FROM information_schema.tables",
+			pass:  false,
+		},
+		{
+			name:  "InformationSchemaUppercase_Invalid",
+			query: "SELECT * FROM INFORMATION_SCHEMA.COLUMNS",
+			pass:  false,
+		},
+		{
+			name:  "TableNamedSystemInTelemetryDatabase_Valid",
+			query: "SELECT * FROM signoz_logs.system",
+			pass:  true,
+		},
+		{
+			name:  "ReadonlySettingOverride_Invalid",
+			query: "SELECT * FROM t SETTINGS readonly = 0",
+			pass:  false,
+		},
+		{
+			name:  "ReadonlySettingOverrideUppercase_Invalid",
+			query: "SELECT * FROM t SETTINGS READONLY = 0",
+			pass:  false,
+		},
+		{
+			name:  "ReadonlySettingOverrideAlongsideOthers_Invalid",
+			query: "SELECT * FROM t SETTINGS max_threads = 4, readonly = 0",
+			pass:  false,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			err := ValidateReadOnlySelect(testCase.query)
+
+			if testCase.pass {
+				assert.NoError(t, err)
+				return
+			}
+
+			assert.Error(t, err)
+		})
+	}
+}

pkg/queryparser/queryfilterextractor/clickhouse.go

@@ -316,17 +316,17 @@ func (e *ClickHouseFilterExtractor) extractColumnStrByExpr(expr clickhouse.Expr)
 	// FunctionExpr is a function call like "toDate(timestamp)"
 	case *clickhouse.FunctionExpr:
 		// For function expressions, return the complete function call string
-		return ex.String()
+		return clickhouse.Format(ex)
 	// ColumnExpr is a column expression like "m.region", "toDate(timestamp)"
 	case *clickhouse.ColumnExpr:
 		// ColumnExpr wraps another expression - extract the underlying expression
 		if ex.Expr != nil {
 			return e.extractColumnStrByExpr(ex.Expr)
 		}
-		return ex.String()
+		return clickhouse.Format(ex)
 	default:
 		// For other expression types, return the string representation
-		return expr.String()
+		return clickhouse.Format(expr)
 	}
 }
 
@@ -524,7 +524,7 @@ func (e *ClickHouseFilterExtractor) extractFullExpression(expr clickhouse.Expr)
 	if expr == nil {
 		return ""
 	}
-	return expr.String()
+	return clickhouse.Format(expr)
 }
 
 // isSimpleColumnReference checks if an expression is just a simple column reference
@@ -657,7 +657,7 @@ func (e *ClickHouseFilterExtractor) extractCTEName(cte *clickhouse.CTEStmt) stri
 	case *clickhouse.Ident:
 		return name.Name
 	default:
-		return cte.Expr.String()
+		return clickhouse.Format(cte.Expr)
 	}
 }
 

pkg/telemetrymetadata/stmt_parse.go

@@ -27,7 +27,7 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
 	}
 
 	// Parse column name to extract context and data type
-	columnName := expr.Name.String()
+	columnName := parser.Format(expr.Name)
 
 	// Remove backticks if present
 	columnName = strings.TrimPrefix(columnName, "`")
@@ -75,7 +75,7 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
 	// Extract field name from the DEFAULT expression
 	// The DEFAULT expression should be something like: resources_string['k8s.cluster.name']
 	// We need to extract the key inside the square brackets
-	defaultExprStr := expr.DefaultExpr.String()
+	defaultExprStr := parser.Format(expr.DefaultExpr)
 
 	// Look for the pattern: map['key']
 	startIdx := strings.Index(defaultExprStr, "['")

pkg/telemetrystore/telemetrystorehook/settings.go

@@ -72,6 +72,12 @@ func (h *provider) BeforeQuery(ctx context.Context, _ *telemetrystore.QueryEvent
 		settings["result_overflow_mode"] = ctx.Value("result_overflow_mode")
 	}
 
+	// readonly = 2 allows reads and per-query setting changes, but no writes, DDL or
+	// SYSTEM statements, and ClickHouse refuses to lower it from within the statement.
+	if readOnly, ok := ctx.Value(ctxtypes.ClickhouseContextReadOnlyKey).(bool); ok && readOnly {
+		settings["readonly"] = 2
+	}
+
 	// TODO(srikanthccv): enable it when the "Cannot read all data" issue is fixed
 	// https://github.com/ClickHouse/ClickHouse/issues/82283
 	settings["secondary_indices_enable_bulk_filtering"] = false

pkg/types/ctxtypes/clickhouse.go

@@ -6,9 +6,17 @@ type ctxKey string
 
 const (
 	ClickhouseContextMaxThreadsKey ctxKey = "clickhouse_max_threads"
+	ClickhouseContextReadOnlyKey   ctxKey = "clickhouse_readonly"
 )
 
 // SetClickhouseMaxThreads stores the max threads value in context.
 func SetClickhouseMaxThreads(ctx context.Context, maxThreads int) context.Context {
 	return context.WithValue(ctx, ClickhouseContextMaxThreadsKey, maxThreads)
 }
+
+// SetClickhouseReadOnly marks the context so the statement runs under a read-only
+// ClickHouse session. The telemetry store connection is shared with write paths, so
+// those must never set this.
+func SetClickhouseReadOnly(ctx context.Context) context.Context {
+	return context.WithValue(ctx, ClickhouseContextReadOnlyKey, true)
+}

pkg/variables/clickhouse/processor.go

@@ -69,7 +69,7 @@ func (qp *QueryProcessor) ProcessQuery(query string, transformer FilterTransform
 	// Reconstruct the query
 	var resultBuilder strings.Builder
 	for _, stmt := range stmts {
-		resultBuilder.WriteString(stmt.String())
+		resultBuilder.WriteString(parser.Format(stmt))
 		resultBuilder.WriteString(";")
 	}
 

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

@@ -20,12 +20,15 @@ from fixtures.querier import (
 # Numeric aggregation-function coverage for logs — the logs counterpart of
 # queriertraces/02_aggregation.py::test_traces_aggregate_functions. Logs querier
 # tests elsewhere only ever use count()/count_distinct(); here a grouped scalar
-# query computes sum / avg / min / max / p50 / p90 / p95 / p99 / countIf /
-# count_distinct over a numeric log attribute and asserts every value.
+# query computes sum / avg / min / max / p50 / p90 / p95 / p99 / countIf / rate /
+# rate_sum / count_distinct over a numeric log attribute and asserts every value.
 #
 # Log number attributes are stored as Float64, so aggregates come back as floats;
 # percentiles are matched with pytest.approx (ClickHouse quantile() and
 # numpy.percentile both linear-interpolate, but avoid ULP-level exact equality).
+#
+# The rate family divides by a window the caller does not otherwise see, so the
+# lookback is passed explicitly instead of taken from the request helper default.
 
 
 def test_logs_aggregate_functions(
@@ -41,11 +44,14 @@ def test_logs_aggregate_functions(
 
     Tests:
     A grouped scalar query computes count / sum / avg / min / max / p50 / p90 /
-    p95 / p99 over latency_ms, countIf over a numeric threshold, and
-    count_distinct over a string attribute — all matching values derived from the
-    inserted logs, ordered by count() desc.
+    p95 / p99 over latency_ms, countIf over a numeric threshold, rate and
+    rate_sum over the query window, and count_distinct over a string attribute —
+    all matching values derived from the inserted logs, ordered by count() desc.
     """
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # A scalar rate divides by the whole query window, so both sides agree on it.
+    lookback_minutes = 5
+    rate_interval_seconds = lookback_minutes * 60
     # (service, latency_ms, endpoint)
     specs = [
         ("svc-a", 10, "/x"),
@@ -80,12 +86,14 @@ def test_logs_aggregate_functions(
             build_aggregation("p95(latency_ms)", "p95_l"),
             build_aggregation("p99(latency_ms)", "p99_l"),
             build_aggregation("countIf(latency_ms >= 25)", "slow"),
+            build_aggregation("rate()", "rate_all"),
+            build_aggregation("rate_sum(latency_ms)", "rate_sum_l"),
             build_aggregation("count_distinct(endpoint)", "endpoints"),
         ],
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", "desc")],
     )
-    response = make_scalar_query_request(signoz, token, now, [query])
+    response = make_scalar_query_request(signoz, token, now, [query], lookback_minutes=lookback_minutes)
 
     assert response.status_code == HTTPStatus.OK, response.text
     data = get_scalar_table_data(response.json())
@@ -109,6 +117,12 @@ def test_logs_aggregate_functions(
             float(np.percentile(latencies, 95)),  # p95(latency_ms)
             float(np.percentile(latencies, 99)),  # p99(latency_ms)
             sum(1 for latency in latencies if latency >= 25),  # countIf(latency_ms >= 25)
+            # Response floats are rounded to 3 decimals at or above 1 and to 3
+            # significant figures below it (roundToNonZeroDecimals). Every other
+            # value here is exact; the rates are the only ones that repeat, and
+            # they stay below 1 for this fixture, so mirror the latter form.
+            float(f"{len(latencies) / rate_interval_seconds:.3g}"),  # rate()
+            float(f"{sum(latencies) / rate_interval_seconds:.3g}"),  # rate_sum(latency_ms)
             len(set(endpoints)),  # count_distinct(endpoint)
         ]
         assert by_service[service] == pytest.approx(expected), f"{service}: {by_service[service]} != {expected}"

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

@@ -150,12 +150,17 @@ def test_traces_aggregate_functions(
     Tests:
     A grouped scalar query computes count / sum / avg / min / max / p50 / p90 over
     duration_nano, countIf over the intrinsic status_code and the calculated
-    response_status_code, and avg over a numeric attribute — all matching values
-    derived from the inserted spans. Under the corrupt variant the same-named
-    colliding attributes must not change any of these.
+    response_status_code, rate and rate_sum over the query window, and avg over a
+    numeric attribute — all matching values derived from the inserted spans. Under
+    the corrupt variant the same-named colliding attributes must not change any of
+    these.
     """
     extra_attrs, extra_resources = trace_noise(noise)
     now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # A scalar rate divides by the whole query window, so both sides agree on it.
+    # Traces derives this separately from logs (telemetrytraces/statement_builder.go).
+    lookback_minutes = 5
+    rate_interval_seconds = lookback_minutes * 60
 
     def mk(service: str, dur_s: float, status: TracesStatusCode, rsc: str, latency: float) -> Traces:
         return Traces(
@@ -189,6 +194,8 @@ def test_traces_aggregate_functions(
         build_aggregation("p50(duration_nano)", "p50_d"),
         build_aggregation("p90(duration_nano)", "p90_d"),
         build_aggregation("countIf(status_code = 2)", "errs"),
+        build_aggregation("rate()", "rate_all"),
+        build_aggregation("rate_sum(duration_nano)", "rate_sum_d"),
         build_aggregation("avg(latency_ms)", "avg_lat"),
     ]
     query = build_traces_scalar_query(
@@ -196,7 +203,7 @@ def test_traces_aggregate_functions(
         group_by=[build_group_by_field("service.name", "string", "resource")],
         order=[build_order_by("count()", "desc")],
     )
-    response = make_scalar_query_request(signoz, token, now, [query])
+    response = make_scalar_query_request(signoz, token, now, [query], lookback_minutes=lookback_minutes)
 
     assert response.status_code == HTTPStatus.OK, response.text
     data = get_scalar_table_data(response.json())
@@ -214,6 +221,11 @@ def test_traces_aggregate_functions(
             float(np.percentile(durations, 50)),  # p50(duration_nano)
             float(np.percentile(durations, 90)),  # p90(duration_nano)
             sum(1 for s in group if int(s.status_code) == 2),  # countIf(status_code = 2)
+            # Response floats are rounded to 3 significant figures below 1 and to 3
+            # decimals at or above it (roundToNonZeroDecimals). The two rates land on
+            # either side of that boundary, so each mirrors its own form.
+            float(f"{len(group) / rate_interval_seconds:.3g}"),  # rate()
+            round(sum(durations) / rate_interval_seconds, 3),  # rate_sum(duration_nano)
             sum(latencies) / len(latencies),  # avg(latency_ms)
         )