refactor(promote): move Target into target.go, enum-style SignalFromText, rename handler method

- Target type definition moves from types.go to target.go alongside its
  constructors, with inline comments
- SignalFromText follows the codebase enum pattern (switch over the
  declared values + Enum method) instead of a string-to-signal map
- generic route handler method renamed HandlePromotePaths -> PromotePaths

pkg/apiserver/signozapiserver/promote.go

@@ -42,5 +42,37 @@ func (provider *provider) addPromoteRoutes(router *mux.Router) error {
 		return err
 	}
 
+	if err := router.Handle("/api/v1/promote_paths/{signal}/{context}", handler.New(provider.authzMiddleware.EditAccess(provider.promoteHandler.PromotePaths), handler.OpenAPIDef{
+		ID:                  "PromotePaths",
+		Tags:                []string{"promote"},
+		Summary:             "Promote paths",
+		Description:         "This endpoint promotes paths of a JSON column to its promoted column. The promotion domain is identified by the signal and field context path variables, e.g. traces/attribute.",
+		Request:             new([]*promotetypes.PromotePath),
+		RequestContentType:  "application/json",
+		Response:            nil,
+		ResponseContentType: "",
+		SuccessStatusCode:   http.StatusCreated,
+		ErrorStatusCodes:    []int{http.StatusBadRequest},
+		SecuritySchemes:     newSecuritySchemes(types.RoleEditor),
+	})).Methods(http.MethodPost).GetError(); err != nil {
+		return err
+	}
+
+	if err := router.Handle("/api/v1/promote_paths/{signal}/{context}", handler.New(provider.authzMiddleware.ViewAccess(provider.promoteHandler.ListPromotedPaths), handler.OpenAPIDef{
+		ID:                  "ListPromotedPaths",
+		Tags:                []string{"promote"},
+		Summary:             "List promoted paths",
+		Description:         "This endpoint lists the promoted paths of a JSON column. The promotion domain is identified by the signal and field context path variables, e.g. traces/attribute.",
+		Request:             nil,
+		RequestContentType:  "",
+		Response:            new([]*promotetypes.PromotePath),
+		ResponseContentType: "",
+		SuccessStatusCode:   http.StatusOK,
+		ErrorStatusCodes:    []int{http.StatusBadRequest},
+		SecuritySchemes:     newSecuritySchemes(types.RoleViewer),
+	})).Methods(http.MethodGet).GetError(); err != nil {
+		return err
+	}
+
 	return nil
 }

pkg/modules/promote/implpromote/handler.go

@@ -9,6 +9,8 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/promote"
 	"github.com/SigNoz/signoz/pkg/types/authtypes"
 	"github.com/SigNoz/signoz/pkg/types/promotetypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/gorilla/mux"
 )
 
 type handler struct {
@@ -19,7 +21,56 @@ func NewHandler(module promote.Module) promote.Handler {
 	return &handler{module: module}
 }
 
+// HandlePromoteAndIndexPaths serves the legacy logs body route; the domain is
+// fixed to logs/body.
 func (h *handler) HandlePromoteAndIndexPaths(w http.ResponseWriter, r *http.Request) {
+	h.promote(w, r, promotetypes.NewLogsBodyTarget())
+}
+
+func (h *handler) ListPromotedAndIndexedPaths(w http.ResponseWriter, r *http.Request) {
+	h.list(w, r, promotetypes.NewLogsBodyTarget())
+}
+
+// PromotePaths serves the generic domain route; the domain is resolved from
+// the {signal}/{context} path variables.
+func (h *handler) PromotePaths(w http.ResponseWriter, r *http.Request) {
+	target, err := targetFromPath(r)
+	if err != nil {
+		render.Error(w, err)
+		return
+	}
+	h.promote(w, r, target)
+}
+
+func (h *handler) ListPromotedPaths(w http.ResponseWriter, r *http.Request) {
+	target, err := targetFromPath(r)
+	if err != nil {
+		render.Error(w, err)
+		return
+	}
+	h.list(w, r, target)
+}
+
+// targetFromPath resolves the promotion domain from the {signal} and
+// {context} path variables.
+func targetFromPath(r *http.Request) (promotetypes.Target, error) {
+	vars := mux.Vars(r)
+	signal, ok := telemetrytypes.SignalFromText(vars["signal"])
+	if !ok {
+		return promotetypes.Target{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid signal: %s", vars["signal"])
+	}
+	context, ok := telemetrytypes.FieldContextFromText(vars["context"])
+	if !ok {
+		return promotetypes.Target{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid context: %s", vars["context"])
+	}
+	target, ok := promotetypes.TargetFor(signal, context)
+	if !ok {
+		return promotetypes.Target{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "promotion is not supported for %s %s", signal.StringValue(), context.StringValue())
+	}
+	return target, nil
+}
+
+func (h *handler) promote(w http.ResponseWriter, r *http.Request, target promotetypes.Target) {
 	// TODO(Nitya): Use in multi tenant setup
 	_, err := authtypes.ClaimsFromContext(r.Context())
 	if err != nil {
@@ -33,7 +84,7 @@ func (h *handler) HandlePromoteAndIndexPaths(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
-	err = h.module.PromoteAndIndexPaths(r.Context(), req...)
+	err = h.module.PromotePaths(r.Context(), target, req...)
 	if err != nil {
 		render.Error(w, err)
 		return
@@ -42,7 +93,7 @@ func (h *handler) HandlePromoteAndIndexPaths(w http.ResponseWriter, r *http.Requ
 	render.Success(w, http.StatusCreated, nil)
 }
 
-func (h *handler) ListPromotedAndIndexedPaths(w http.ResponseWriter, r *http.Request) {
+func (h *handler) list(w http.ResponseWriter, r *http.Request, target promotetypes.Target) {
 	// TODO(Nitya): Use in multi tenant setup
 	_, err := authtypes.ClaimsFromContext(r.Context())
 	if err != nil {
@@ -50,7 +101,7 @@ func (h *handler) ListPromotedAndIndexedPaths(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	paths, err := h.module.ListPromotedAndIndexedPaths(r.Context())
+	paths, err := h.module.ListPromotedPaths(r.Context(), target)
 	if err != nil {
 		render.Error(w, err)
 		return

pkg/modules/promote/implpromote/module.go

@@ -2,14 +2,11 @@ package implpromote
 
 import (
 	"context"
-	"maps"
-	"slices"
 	"strings"
 
 	schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/modules/promote"
-	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
 	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
 	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
@@ -31,45 +28,57 @@ func NewModule(metadataStore telemetrytypes.MetadataStore, telemetrystore teleme
 	return &module{metadataStore: metadataStore, telemetryStore: telemetrystore}
 }
 
-func (m *module) ListPromotedAndIndexedPaths(ctx context.Context) ([]promotetypes.PromotePath, error) {
+// ListPromotedPaths lists the promoted paths of the target JSON column,
+// merged with per-path index metadata where the target supports indexes.
+func (m *module) ListPromotedPaths(ctx context.Context, target promotetypes.Target) ([]promotetypes.PromotePath, error) {
+	promotedPaths, err := m.metadataStore.GetPromotedPaths(ctx, target.Entry)
+	if err != nil {
+		return nil, err
+	}
+
+	response := make([]promotetypes.PromotePath, 0, len(promotedPaths))
+	for path := range promotedPaths {
+		response = append(response, promotetypes.PromotePath{
+			Path:    target.RequiredPathPrefix + path,
+			Promote: true,
+		})
+	}
+
+	// Index metadata is optional per target; merge it in only where supported.
+	if !target.IndexesSupported {
+		return response, nil
+	}
+
 	indexes, err := m.metadataStore.ListLogsJSONIndexes(ctx)
 	if err != nil {
 		return nil, err
 	}
 
+	// index.Name is the bare path and index.BaseColumn carries the column
+	// prefix, so the aggregate key is the full sub-column path.
 	aggr := map[string][]promotetypes.WrappedIndex{}
 	for _, index := range indexes {
-		aggr[index.Name] = append(aggr[index.Name], promotetypes.WrappedIndex{
+		fullPath := index.BaseColumn + index.Name
+		aggr[fullPath] = append(aggr[fullPath], promotetypes.WrappedIndex{
 			FieldDataType: index.FieldDataType,
 			Type:          index.IndexType,
 			Granularity:   index.Granularity,
 		})
 	}
-	promotedPaths, err := m.listPromotedPaths(ctx)
-	if err != nil {
-		return nil, err
-	}
 
-	response := []promotetypes.PromotePath{}
-	for _, path := range promotedPaths {
-		fullPath := logstelemetryschema.BodyPromotedColumnPrefix + path
-		path = telemetrytypes.BodyJSONStringSearchPrefix + path
-		item := promotetypes.PromotePath{
-			Path:    path,
-			Promote: true,
-		}
-		indexes, ok := aggr[fullPath]
-		if ok {
-			item.Indexes = indexes
+	for i := range response {
+		fullPath := target.PromotedColumnPrefix() + strings.TrimPrefix(response[i].Path, target.RequiredPathPrefix)
+		if indexes, ok := aggr[fullPath]; ok {
+			response[i].Indexes = indexes
 			delete(aggr, fullPath)
 		}
-		response = append(response, item)
 	}
 
 	// add the paths that are not promoted but have indexes
-	for path, indexes := range aggr {
-		path := strings.TrimPrefix(path, logstelemetryschema.BodyV2ColumnPrefix)
-		path = telemetrytypes.BodyJSONStringSearchPrefix + path
+	for fullPath, indexes := range aggr {
+		path := strings.TrimPrefix(fullPath, target.BaseColumnPrefix())
+		path = strings.TrimPrefix(path, target.PromotedColumnPrefix())
+		path = target.RequiredPathPrefix + path
 		response = append(response, promotetypes.PromotePath{
 			Path:    path,
 			Indexes: indexes,
@@ -78,54 +87,10 @@ func (m *module) ListPromotedAndIndexedPaths(ctx context.Context) ([]promotetype
 	return response, nil
 }
 
-func (m *module) listPromotedPaths(ctx context.Context) ([]string, error) {
-	paths, err := m.metadataStore.GetPromotedPaths(ctx)
-	if err != nil {
-		return nil, err
-	}
-	return slices.Collect(maps.Keys(paths)), nil
-}
-
-// PromotePaths inserts provided JSON paths into the promoted paths table for logs queries.
-func (m *module) PromotePaths(ctx context.Context, paths []string) error {
-	if len(paths) == 0 {
-		return errors.NewInvalidInputf(errors.CodeInvalidInput, "paths cannot be empty")
-	}
-
-	return m.metadataStore.PromotePaths(ctx, paths...)
-}
-
-// createIndexes creates string ngram + token filter indexes on JSON path subcolumns for LIKE queries.
-func (m *module) createIndexes(ctx context.Context, indexes []schemamigrator.Index) error {
-	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
-		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
-		instrumentationtypes.CodeNamespace:    "promote",
-		instrumentationtypes.CodeFunctionName: "createIndexes",
-	})
-	if len(indexes) == 0 {
-		return nil
-	}
-
-	for _, index := range indexes {
-		alterStmt := schemamigrator.AlterTableAddIndex{
-			Database: logstelemetryschema.DBName,
-			Table:    logstelemetryschema.LogsV2LocalTableName,
-			Index:    index,
-		}
-		op := alterStmt.OnCluster(m.telemetryStore.Cluster())
-		if err := m.telemetryStore.ClickhouseDB().Exec(ctx, op.ToSQL()); err != nil {
-			return errors.WrapInternalf(err, CodeFailedToCreateIndex, "failed to create index")
-		}
-	}
-
-	return nil
-}
-
-// PromoteAndIndexPaths handles promoting paths and creating indexes in one call.
-func (m *module) PromoteAndIndexPaths(
-	ctx context.Context,
-	paths ...*promotetypes.PromotePath,
-) error {
+// PromotePaths records new promotions of the target JSON column in the column
+// evolution table and, for targets with index support, creates the requested
+// per-path skip indexes.
+func (m *module) PromotePaths(ctx context.Context, target promotetypes.Target, paths ...*promotetypes.PromotePath) error {
 	if len(paths) == 0 {
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "paths cannot be empty")
 	}
@@ -133,13 +98,13 @@ func (m *module) PromoteAndIndexPaths(
 	pathsStr := []string{}
 	// validate the paths
 	for _, path := range paths {
-		if err := path.ValidateAndSetDefaults(); err != nil {
+		if err := path.ValidateAndSetDefaults(target); err != nil {
 			return err
 		}
 		pathsStr = append(pathsStr, path.Path)
 	}
 
-	existingPromotedPaths, err := m.metadataStore.GetPromotedPaths(ctx, pathsStr...)
+	existingPromotedPaths, err := m.metadataStore.GetPromotedPaths(ctx, target.Entry, pathsStr...)
 	if err != nil {
 		return err
 	}
@@ -153,10 +118,10 @@ func (m *module) PromoteAndIndexPaths(
 			}
 		}
 		if len(it.Indexes) > 0 {
-			parentColumn := logstelemetryschema.LogsV2BodyV2Column
+			parentColumn := target.BaseColumn
 			// if the path is already promoted or is being promoted, add it to the promoted column
 			if _, promoted := existingPromotedPaths[it.Path]; promoted || it.Promote {
-				parentColumn = logstelemetryschema.LogsV2BodyPromotedColumn
+				parentColumn = target.PromotedColumn()
 			}
 
 			for _, index := range it.Indexes {
@@ -182,17 +147,43 @@ func (m *module) PromoteAndIndexPaths(
 	}
 
 	if len(toInsert) > 0 {
-		err := m.PromotePaths(ctx, toInsert)
+		err := m.metadataStore.PromotePaths(ctx, target.Entry, toInsert...)
 		if err != nil {
 			return err
 		}
 	}
 
 	if len(indexes) > 0 {
-		if err := m.createIndexes(ctx, indexes); err != nil {
+		if err := m.createIndexes(ctx, target, indexes); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
+
+// createIndexes creates string ngram + token filter indexes on JSON path subcolumns for LIKE queries.
+func (m *module) createIndexes(ctx context.Context, target promotetypes.Target, indexes []schemamigrator.Index) error {
+	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
+		instrumentationtypes.TelemetrySignal:  target.Entry.Signal.StringValue(),
+		instrumentationtypes.CodeNamespace:    "promote",
+		instrumentationtypes.CodeFunctionName: "createIndexes",
+	})
+	if len(indexes) == 0 {
+		return nil
+	}
+
+	for _, index := range indexes {
+		alterStmt := schemamigrator.AlterTableAddIndex{
+			Database: target.DBName,
+			Table:    target.LocalTableName,
+			Index:    index,
+		}
+		op := alterStmt.OnCluster(m.telemetryStore.Cluster())
+		if err := m.telemetryStore.ClickhouseDB().Exec(ctx, op.ToSQL()); err != nil {
+			return errors.WrapInternalf(err, CodeFailedToCreateIndex, "failed to create index")
+		}
+	}
+
+	return nil
+}

pkg/modules/promote/implpromote/module_test.go

@@ -0,0 +1,145 @@
+package implpromote
+
+import (
+	"context"
+	"testing"
+
+	"github.com/SigNoz/signoz/pkg/types/promotetypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPromoteAttributes(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("promotes new attributes and is idempotent", func(t *testing.T) {
+		store := telemetrytypestest.NewMockMetadataStore()
+		m := NewModule(store, nil)
+
+		paths := []*promotetypes.PromotePath{
+			{Path: "http.method", Promote: true},
+			{Path: "span.operation", Promote: true},
+		}
+		require.NoError(t, m.PromotePaths(ctx, promotetypes.NewTracesAttributesTarget(), paths...))
+		assert.True(t, store.PromotedPathsMap["http.method"])
+		assert.True(t, store.PromotedPathsMap["span.operation"])
+
+		// promoting again must not fail
+		require.NoError(t, m.PromotePaths(ctx, promotetypes.NewTracesAttributesTarget(), paths...))
+		assert.Len(t, store.PromotedPathsMap, 2)
+	})
+
+	t.Run("non-promote entries are not recorded", func(t *testing.T) {
+		store := telemetrytypestest.NewMockMetadataStore()
+		m := NewModule(store, nil)
+
+		require.NoError(t, m.PromotePaths(ctx, promotetypes.NewTracesAttributesTarget(), &promotetypes.PromotePath{Path: "http.method"}))
+		assert.Empty(t, store.PromotedPathsMap)
+	})
+
+	t.Run("indexes are rejected for trace attributes", func(t *testing.T) {
+		store := telemetrytypestest.NewMockMetadataStore()
+		m := NewModule(store, nil)
+
+		err := m.PromotePaths(ctx, promotetypes.NewTracesAttributesTarget(), &promotetypes.PromotePath{
+			Path:    "http.method",
+			Promote: true,
+			Indexes: []promotetypes.WrappedIndex{
+				{FieldDataType: telemetrytypes.FieldDataTypeString, Type: "ngrambf_v1(4, 1024, 2, 0)", Granularity: 1},
+			},
+		})
+		require.Error(t, err)
+		assert.Empty(t, store.PromotedPathsMap)
+	})
+
+	t.Run("column prefixed and empty paths are rejected", func(t *testing.T) {
+		store := telemetrytypestest.NewMockMetadataStore()
+		m := NewModule(store, nil)
+
+		require.Error(t, m.PromotePaths(ctx, promotetypes.NewTracesAttributesTarget(), &promotetypes.PromotePath{Path: "attributes.http.method", Promote: true}))
+		require.Error(t, m.PromotePaths(ctx, promotetypes.NewTracesAttributesTarget(), &promotetypes.PromotePath{Path: "", Promote: true}))
+		assert.Empty(t, store.PromotedPathsMap)
+	})
+
+	t.Run("empty request is rejected", func(t *testing.T) {
+		store := telemetrytypestest.NewMockMetadataStore()
+		m := NewModule(store, nil)
+
+		require.Error(t, m.PromotePaths(ctx, promotetypes.NewTracesAttributesTarget()))
+	})
+}
+
+func TestListPromotedAttributes(t *testing.T) {
+	ctx := context.Background()
+	store := telemetrytypestest.NewMockMetadataStore()
+	store.PromotedPathsMap["http.method"] = true
+	m := NewModule(store, nil)
+
+	paths, err := m.ListPromotedPaths(ctx, promotetypes.NewTracesAttributesTarget())
+	require.NoError(t, err)
+	require.Len(t, paths, 1)
+	assert.Equal(t, "http.method", paths[0].Path)
+	assert.True(t, paths[0].Promote)
+	assert.Empty(t, paths[0].Indexes)
+}
+
+func TestPromoteAndIndexPaths(t *testing.T) {
+	ctx := context.Background()
+
+	t.Run("promotes body path with prefix stripped", func(t *testing.T) {
+		store := telemetrytypestest.NewMockMetadataStore()
+		m := NewModule(store, nil)
+
+		require.NoError(t, m.PromotePaths(ctx, promotetypes.NewLogsBodyTarget(), &promotetypes.PromotePath{Path: "body.user.name", Promote: true}))
+		assert.True(t, store.PromotedPathsMap["user.name"])
+	})
+}
+
+func TestListPromotedAndIndexedPaths(t *testing.T) {
+	ctx := context.Background()
+	store := telemetrytypestest.NewMockMetadataStore()
+	store.PromotedPathsMap["user.name"] = true
+	store.LogsJSONIndexes = []telemetrytypes.TelemetryFieldKeySkipIndex{
+		{
+			Name:          "user.name",
+			FieldContext:  telemetrytypes.FieldContextBody,
+			FieldDataType: telemetrytypes.FieldDataTypeString,
+			BaseColumn:    "body_promoted.",
+			IndexType:     "ngrambf_v1(4, 1024, 2, 0)",
+			Granularity:   1,
+		},
+		{
+			Name:          "request.duration",
+			FieldContext:  telemetrytypes.FieldContextBody,
+			FieldDataType: telemetrytypes.FieldDataTypeFloat64,
+			BaseColumn:    "body_v2.",
+			IndexType:     "minmax",
+			Granularity:   1,
+		},
+	}
+	m := NewModule(store, nil)
+
+	paths, err := m.ListPromotedPaths(ctx, promotetypes.NewLogsBodyTarget())
+	require.NoError(t, err)
+	require.Len(t, paths, 2)
+
+	byPath := map[string]promotetypes.PromotePath{}
+	for _, p := range paths {
+		byPath[p.Path] = p
+	}
+
+	// the promoted path carries its index from the promoted column
+	promoted := byPath["body.user.name"]
+	assert.True(t, promoted.Promote)
+	require.Len(t, promoted.Indexes, 1)
+	assert.Equal(t, telemetrytypes.FieldDataTypeString, promoted.Indexes[0].FieldDataType)
+	assert.Equal(t, "ngrambf_v1(4, 1024, 2, 0)", promoted.Indexes[0].Type)
+
+	// an indexed but not promoted path shows up without the promote flag
+	indexed := byPath["body.request.duration"]
+	assert.False(t, indexed.Promote)
+	require.Len(t, indexed.Indexes, 1)
+	assert.Equal(t, telemetrytypes.FieldDataTypeFloat64, indexed.Indexes[0].FieldDataType)
+}

pkg/modules/promote/promote.go

@@ -8,11 +8,14 @@ import (
 )
 
 type Module interface {
-	ListPromotedAndIndexedPaths(ctx context.Context) ([]promotetypes.PromotePath, error)
-	PromoteAndIndexPaths(ctx context.Context, paths ...*promotetypes.PromotePath) error
+	ListPromotedPaths(ctx context.Context, target promotetypes.Target) ([]promotetypes.PromotePath, error)
+	PromotePaths(ctx context.Context, target promotetypes.Target, paths ...*promotetypes.PromotePath) error
 }
 
 type Handler interface {
 	HandlePromoteAndIndexPaths(w http.ResponseWriter, r *http.Request)
 	ListPromotedAndIndexedPaths(w http.ResponseWriter, r *http.Request)
+
+	PromotePaths(w http.ResponseWriter, r *http.Request)
+	ListPromotedPaths(w http.ResponseWriter, r *http.Request)
 }

pkg/querier/consume.go

@@ -466,16 +466,60 @@ func readAsRaw(rows driver.Rows, queryName string) (*qbtypes.RawData, error) {
 	}, nil
 }
 
+// jsonAttributeMap returns the decoded document of a scanned JSON attributes column (the driver
+// scans it into telemetrystoretypes.JSONValue) and whether one was present.
+func jsonAttributeMap(v any) (map[string]any, bool) {
+	switch m := v.(type) {
+	case telemetrystoretypes.JSONValue:
+		if m == nil {
+			return nil, false
+		}
+		return m, true
+	case map[string]any:
+		if m == nil {
+			return nil, false
+		}
+		return m, true
+	default:
+		return nil, false
+	}
+}
+
+// flattenJSONPaths flattens a decoded JSON document into dotted keys (a nested {"http":{"route":x}}
+// becomes "http.route": x), matching the legacy map representation so the attributes bag has the
+// same flat shape whether it was read from the maps or the JSON column. Existing keys are
+// overwritten, so a JSON path wins over a same-named map entry.
+func flattenJSONPaths(prefix string, m map[string]any, out map[string]any) {
+	for k, v := range m {
+		key := k
+		if prefix != "" {
+			key = prefix + "." + k
+		}
+		switch child := v.(type) {
+		case map[string]any:
+			flattenJSONPaths(key, child, out)
+		case telemetrystoretypes.JSONValue:
+			flattenJSONPaths(key, child, out)
+		default:
+			out[key] = v
+		}
+	}
+}
+
 // mergeSpanAttributeColumns merges (attributes_string, attributes_number, attributes_bool, resources_string) into
 // unified "attributes" and "resource" keys, and parses the stringified `events`
 // and `links` columns into structured slices. Raw DB columns are removed.
+//
+// After the JSON rollout the attributes bag is read from the `attributes` JSON column instead of,
+// or alongside, the legacy maps; those paths are flattened in and win over the maps on collision.
 func mergeSpanAttributeColumns(data map[string]any) {
 	attrStr, hasStr := data["attributes_string"]
 	attrNum, hasNum := data["attributes_number"]
 	attrBool, hasBool := data["attributes_bool"]
+	attrJSON, hasJSON := jsonAttributeMap(data["attributes"])
 	// todo(nitya): move to resource json
 	resStr, hasRes := data["resources_string"]
-	if hasStr || hasNum || hasBool || hasRes {
+	if hasStr || hasNum || hasBool || hasJSON || hasRes {
 		attributes := make(map[string]any)
 		if m, ok := attrStr.(map[string]string); ok {
 			for k, v := range m {
@@ -492,6 +536,9 @@ func mergeSpanAttributeColumns(data map[string]any) {
 				attributes[k] = v
 			}
 		}
+		if hasJSON {
+			flattenJSONPaths("", attrJSON, attributes)
+		}
 		delete(data, "attributes_string")
 		delete(data, "attributes_number")
 		delete(data, "attributes_bool")

pkg/querier/consume_test.go

@@ -195,3 +195,123 @@ func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
 		t.Fatalf("expected empty []spantypes.Link, got %#v", data["links"])
 	}
 }
+
+func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
+	t.Run("json only flattens nested paths and preserves types", func(t *testing.T) {
+		data := map[string]any{
+			"attributes": telemetrystoretypes.JSONValue{
+				"http": map[string]any{
+					"route": "/api/pay",
+					"retry": map[string]any{"count": float64(3)},
+				},
+				"cache.hit": true,
+			},
+			"resources_string": map[string]string{"service.name": "api"},
+		}
+
+		mergeSpanAttributeColumns(data)
+
+		attrs, ok := data["attributes"].(map[string]any)
+		require.True(t, ok, "attributes should be flattened to map[string]any, got %T", data["attributes"])
+		assert.Equal(t, "/api/pay", attrs["http.route"])
+		assert.Equal(t, float64(3), attrs["http.retry.count"])
+		assert.Equal(t, true, attrs["cache.hit"])
+		_, nested := attrs["http"]
+		assert.False(t, nested, "nested objects must be flattened away, not kept")
+	})
+
+	t.Run("straddle: json paths win over legacy map on collision, union otherwise", func(t *testing.T) {
+		data := map[string]any{
+			"attributes_string": map[string]string{"http.route": "/old", "only.map": "m"},
+			"attributes_number": map[string]float64{"http.status": 500},
+			"attributes":        telemetrystoretypes.JSONValue{"http": map[string]any{"route": "/new"}},
+			"resources_string":  map[string]string{"service.name": "api"},
+		}
+
+		mergeSpanAttributeColumns(data)
+
+		attrs := data["attributes"].(map[string]any)
+		assert.Equal(t, "/new", attrs["http.route"], "json home wins on collision")
+		assert.Equal(t, "m", attrs["only.map"], "map-only key survives")
+		assert.Equal(t, float64(500), attrs["http.status"], "number map key survives")
+		for _, removed := range []string{"attributes_string", "attributes_number", "attributes_bool"} {
+			_, present := data[removed]
+			assert.False(t, present, "%s should be removed", removed)
+		}
+	})
+
+	t.Run("map only row with empty json doc keeps every map value", func(t *testing.T) {
+		data := map[string]any{
+			"attributes_string": map[string]string{"http.route": "/map"},
+			"attributes_number": map[string]float64{"http.status": 200},
+			"attributes_bool":   map[string]bool{"cache.hit": true},
+			"attributes":        telemetrystoretypes.JSONValue{},
+			"resources_string":  map[string]string{"service.name": "api"},
+		}
+
+		mergeSpanAttributeColumns(data)
+
+		attrs := data["attributes"].(map[string]any)
+		assert.Equal(t, map[string]any{"http.route": "/map", "http.status": float64(200), "cache.hit": true}, attrs)
+	})
+
+	t.Run("map only row with nil json value behaves as absent", func(t *testing.T) {
+		data := map[string]any{
+			"attributes_string": map[string]string{"http.route": "/map"},
+			"attributes":        telemetrystoretypes.JSONValue(nil),
+			"resources_string":  map[string]string{"service.name": "api"},
+		}
+
+		mergeSpanAttributeColumns(data)
+
+		attrs := data["attributes"].(map[string]any)
+		assert.Equal(t, map[string]any{"http.route": "/map"}, attrs)
+	})
+
+	t.Run("arrays stay leaf values", func(t *testing.T) {
+		data := map[string]any{
+			"attributes":       telemetrystoretypes.JSONValue{"http": map[string]any{"tags": []any{"a", "b"}, "codes": []any{float64(1), float64(2)}}},
+			"resources_string": map[string]string{"service.name": "api"},
+		}
+
+		mergeSpanAttributeColumns(data)
+
+		attrs := data["attributes"].(map[string]any)
+		assert.Equal(t, []any{"a", "b"}, attrs["http.tags"])
+		assert.Equal(t, []any{float64(1), float64(2)}, attrs["http.codes"])
+	})
+
+	t.Run("json null is kept as a nil value (never emitted by ClickHouse JSON; defensive pin)", func(t *testing.T) {
+		data := map[string]any{
+			"attributes":       telemetrystoretypes.JSONValue{"k": nil},
+			"resources_string": map[string]string{"service.name": "api"},
+		}
+
+		mergeSpanAttributeColumns(data)
+
+		attrs := data["attributes"].(map[string]any)
+		v, present := attrs["k"]
+		assert.True(t, present)
+		assert.Nil(t, v)
+	})
+
+	t.Run("same key is a leaf in one row and a parent in another", func(t *testing.T) {
+		leafRow := map[string]any{
+			"attributes":       telemetrystoretypes.JSONValue{"http": "plaintext"},
+			"resources_string": map[string]string{"service.name": "api"},
+		}
+		parentRow := map[string]any{
+			"attributes":       telemetrystoretypes.JSONValue{"http": map[string]any{"route": "/a"}},
+			"resources_string": map[string]string{"service.name": "api"},
+		}
+
+		mergeSpanAttributeColumns(leafRow)
+		mergeSpanAttributeColumns(parentRow)
+
+		assert.Equal(t, "plaintext", leafRow["attributes"].(map[string]any)["http"])
+		parentAttrs := parentRow["attributes"].(map[string]any)
+		assert.Equal(t, "/a", parentAttrs["http.route"])
+		_, collapsed := parentAttrs["http"]
+		assert.False(t, collapsed, "the parent key must flatten away, not shadow the dotted leaf")
+	})
+}

pkg/statementbuilder/aistatementbuilder/statement_builder_test.go

@@ -676,7 +676,7 @@ SELECT timestamp AS __SELECT_KEY_0_timestamp, trace_id AS __SELECT_KEY_1_trace_i
     http_url AS __SELECT_KEY_17_http_url, external_http_method AS __SELECT_KEY_18_external_http_method,
     http_method AS __SELECT_KEY_19_http_method, http_host AS __SELECT_KEY_20_http_host, db_name AS __SELECT_KEY_21_db_name,
     db_operation AS __SELECT_KEY_22_db_operation, has_error AS __SELECT_KEY_23_has_error, is_remote AS __SELECT_KEY_24_is_remote,
-    attributes_string, attributes_number, attributes_bool, resources_string
+    attributes_string, attributes_number, attributes_bool, resources_string, attributes
 FROM signoz_traces.distributed_signoz_index_v3
 WHERE (((mapContains(attributes_string, 'gen_ai.request.model')
         OR mapContains(attributes_string, 'gen_ai.tool.name')

pkg/statementbuilder/tracesstatementbuilder/bulk_attributes_test.go

@@ -0,0 +1,116 @@
+package tracesstatementbuilder
+
+import (
+	"context"
+	"regexp"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// jsonAttrColRe matches the raw `attributes` JSON column in the SELECT list (a bare column, not
+// the `attributes_string`/`_number`/`_bool` maps), mid-list or as the last column before FROM.
+var jsonAttrColRe = regexp.MustCompile(`,\s*attributes\s*(,| FROM )`)
+
+func newBulkTestBuilder(t *testing.T, releaseTime time.Time) *traceQueryStatementBuilder {
+	t.Helper()
+	fl := flaggertest.New(t)
+	fm := tracestelemetryschema.NewFieldMapper(fl)
+	cb := tracestelemetryschema.NewConditionBuilder(fm, fl)
+	store := telemetrytypestest.NewMockMetadataStore()
+	store.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
+	store.KeysMap["http.route"] = []*telemetrytypes.TelemetryFieldKey{{
+		Name:          "http.route",
+		FieldContext:  telemetrytypes.FieldContextAttribute,
+		FieldDataType: telemetrytypes.FieldDataTypeString,
+		Signal:        telemetrytypes.SignalTraces,
+	}}
+	store.ColumnEvolutionMetadataMap["traces:attribute:__all__"] = tracestelemetryschema.MockAttributeEvolutionData(releaseTime)
+
+	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
+	return NewTraceQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		store, fm, cb, aggExprRewriter, nil, fl, false, 100000,
+	)
+}
+
+// TestListQuerySelectsAllAttributeHomes: the empty-selectFields list query scans every
+// attributes-bag home (the three legacy maps and the JSON column) in any window, with no
+// evolution lookup.
+func TestListQuerySelectsAllAttributeHomes(t *testing.T) {
+	releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
+	rel := releaseTime.UnixMilli()
+	day := int64(24 * time.Hour / time.Millisecond)
+
+	b := newBulkTestBuilder(t, releaseTime)
+
+	cases := []struct {
+		name    string
+		startMs uint64
+		endMs   uint64
+	}{
+		{"before rollout", uint64(rel - 2*day), uint64(rel - day)},
+		{"after rollout", uint64(rel + day), uint64(rel + 2*day)},
+		{"straddling rollout", uint64(rel - day), uint64(rel + day)},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			stmt, err := b.Build(
+				context.Background(), valuer.UUID{}, tt.startMs, tt.endMs,
+				qbtypes.RequestTypeRaw,
+				qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{Signal: telemetrytypes.SignalTraces},
+				nil,
+			)
+			require.NoError(t, err)
+			selectList := stmt.Query[:strings.Index(stmt.Query, " FROM ")]
+
+			assert.Regexp(t, jsonAttrColRe, stmt.Query, "json `attributes` column; select=%s", selectList)
+			for _, col := range []string{"attributes_string", "attributes_number", "attributes_bool", "resources_string"} {
+				assert.Contains(t, stmt.Query, col, "select=%s", selectList)
+			}
+		})
+	}
+}
+
+// TestGroupByAttributeAfterRolloutReadsJSON pins the post-dual-write guarantee at the statement
+// level: a group-by on an attribute key in a window fully after the rollout reads the JSON column,
+// never the legacy map — so aggregations keep working once map dual-write stops.
+func TestGroupByAttributeAfterRolloutReadsJSON(t *testing.T) {
+	releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
+	rel := releaseTime.UnixMilli()
+	day := int64(24 * time.Hour / time.Millisecond)
+
+	b := newBulkTestBuilder(t, releaseTime)
+
+	stmt, err := b.Build(
+		context.Background(), valuer.UUID{}, uint64(rel+day), uint64(rel+2*day),
+		qbtypes.RequestTypeTimeSeries,
+		qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
+			Signal:       telemetrytypes.SignalTraces,
+			StepInterval: qbtypes.Step{Duration: 30 * time.Second},
+			Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
+			GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
+				Name:          "http.route",
+				FieldContext:  telemetrytypes.FieldContextAttribute,
+				FieldDataType: telemetrytypes.FieldDataTypeString,
+			}}},
+			Limit: 10,
+		},
+		nil,
+	)
+	require.NoError(t, err)
+	assert.Contains(t, stmt.Query, "attributes.`http.route`::String")
+	assert.NotContains(t, stmt.Query, "attributes_string", "post-rollout group-by must not read the legacy map")
+}

pkg/statementbuilder/tracesstatementbuilder/stmt_builder_test.go

@@ -583,7 +583,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
 				Limit: 10,
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY multiIf(mapContains(attributes_string, 'user.id'), attributes_string['user.id'], NULL) desc LIMIT ?",
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY multiIf(mapContains(attributes_string, 'user.id'), attributes_string['user.id'], NULL) desc LIMIT ?",
 				Args:  []any{"redis-manual", "%service.name%", "%service.name\":\"redis-manual%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
 			},
 			expectedErr: nil,
@@ -841,7 +841,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
 				Limit:        10,
 			},
 			expected: qbtypes.Statement{
-				Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
+				Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
 			},
 			expectedErr: nil,
@@ -883,7 +883,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
 				}},
 			},
 			expected: qbtypes.Statement{
-				Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY timestamp asc LIMIT ?",
+				Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY timestamp asc LIMIT ?",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
 			},
 			expectedErr: nil,
@@ -910,7 +910,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
 				Limit: 10,
 			},
 			expected: qbtypes.Statement{
-				Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
+				Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
 				Args:  []any{"1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
 			},
 		},

pkg/statementbuilder/tracesstatementbuilder/trace_operator_cte_builder_test.go

@@ -125,7 +125,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM A_INDIR_DESC_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
+				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_INDIR_DESC_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 5},
 			},
 			expectedErr: nil,
@@ -162,7 +162,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_AND_B AS (SELECT l.* FROM A AS l INNER JOIN B AS r ON l.trace_id = r.trace_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM A_AND_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
+				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_AND_B AS (SELECT l.* FROM A AS l INNER JOIN B AS r ON l.trace_id = r.trace_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_AND_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 15},
 			},
 			expectedErr: nil,
@@ -199,7 +199,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_OR_B AS (SELECT * FROM A UNION DISTINCT SELECT * FROM B) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM A_OR_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
+				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_OR_B AS (SELECT * FROM A UNION DISTINCT SELECT * FROM B) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_OR_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 20},
 			},
 			expectedErr: nil,
@@ -236,7 +236,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_not_B AS (SELECT l.* FROM A AS l WHERE l.trace_id GLOBAL NOT IN (SELECT DISTINCT trace_id FROM B)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM A_not_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
+				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_not_B AS (SELECT l.* FROM A AS l WHERE l.trace_id GLOBAL NOT IN (SELECT DISTINCT trace_id FROM B)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_not_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
 			},
 			expectedErr: nil,
@@ -401,7 +401,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_DIR_DESC_B AS (SELECT p.* FROM A AS p INNER JOIN B AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent_span_id), __resource_filter_C AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_D AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), D AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_D) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), C_DIR_DESC_D AS (SELECT p.* FROM C AS p INNER JOIN D AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent_span_id), A_DIR_DESC_B_AND_C_DIR_DESC_D AS (SELECT l.* FROM A_DIR_DESC_B AS l INNER JOIN C_DIR_DESC_D AS r ON l.trace_id = r.trace_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM A_DIR_DESC_B_AND_C_DIR_DESC_D ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
+				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_DIR_DESC_B AS (SELECT p.* FROM A AS p INNER JOIN B AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent_span_id), __resource_filter_C AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_D AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), D AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_D) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), C_DIR_DESC_D AS (SELECT p.* FROM C AS p INNER JOIN D AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent_span_id), A_DIR_DESC_B_AND_C_DIR_DESC_D AS (SELECT l.* FROM A_DIR_DESC_B AS l INNER JOIN C_DIR_DESC_D AS r ON l.trace_id = r.trace_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_DIR_DESC_B_AND_C_DIR_DESC_D ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "auth", "%service.name%", "%service.name\":\"auth%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 5},
 			},
 			expectedErr: nil,
@@ -435,7 +435,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __return_from_B AS (SELECT * FROM B WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM __return_from_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
+				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __return_from_B AS (SELECT * FROM B WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM __return_from_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
 			},
 			expectedErr: nil,
@@ -477,7 +477,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
 				},
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __resource_filter_C AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B_AND_C AS (SELECT l.* FROM A_INDIR_DESC_B AS l INNER JOIN C AS r ON l.trace_id = r.trace_id), __return_from_C AS (SELECT * FROM C WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B_AND_C)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM __return_from_C ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
+				Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __resource_filter_C AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B_AND_C AS (SELECT l.* FROM A_INDIR_DESC_B AS l INNER JOIN C AS r ON l.trace_id = r.trace_id), __return_from_C AS (SELECT * FROM C WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B_AND_C)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM __return_from_C ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
 				Args:  []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "auth", "%service.name%", "%service.name\":\"auth%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
 			},
 			expectedErr: nil,

pkg/telemetrymetadata/body_json_metadata.go

@@ -16,6 +16,7 @@ import (
 	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
 	"github.com/SigNoz/signoz/pkg/types/ctxtypes"
 	"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
+	"github.com/SigNoz/signoz/pkg/types/promotetypes"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
 	"github.com/huandu/go-sqlbuilder"
 )
@@ -33,6 +34,10 @@ var (
 	CodeFailedToAppendPath   = errors.MustNewCode("failed_to_append_path_promoted_paths")
 )
 
+// logsBodyPromotedEntry templates the column evolution rows recorded for
+// logs body promotions.
+var logsBodyPromotedEntry = promotetypes.NewLogsBodyTarget().Entry
+
 // enrichJSONKeys enriches body-context keys with promoted path info, indexes,
 // and JSON access plans. parentTypeCache contains parent array types (ArrayJSON/ArrayDynamic)
 // pre-fetched in the main UNION query.
@@ -67,7 +72,7 @@ func (t *telemetryMetaStore) enrichJSONKeys(ctx context.Context, selectors []*te
 	}
 
 	// fetch promoted paths
-	promoted, err := t.GetPromotedPaths(ctx, paths...)
+	promoted, err := t.GetPromotedPaths(ctx, logsBodyPromotedEntry, paths...)
 	if err != nil {
 		return err
 	}
@@ -157,7 +162,7 @@ func buildListLogsJSONIndexesQuery(cluster string, filters ...string) (string, [
 }
 
 func (t *telemetryMetaStore) ListLogsJSONIndexes(ctx context.Context, filters ...string) ([]telemetrytypes.TelemetryFieldKeySkipIndex, error) {
-	ctx = withTelemetryContext(ctx, "ListLogsJSONIndexes")
+	ctx = withTelemetryContext(ctx, telemetrytypes.SignalLogs, "ListLogsJSONIndexes")
 	query, args := buildListLogsJSONIndexesQuery(t.telemetrystore.Cluster(), filters...)
 	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
 	if err != nil {
@@ -215,14 +220,14 @@ func (t *telemetryMetaStore) ListLogsJSONIndexes(ctx context.Context, filters ..
 
 // TODO(Piyush): Remove this if not used in future.
 func (t *telemetryMetaStore) ListJSONValues(ctx context.Context, path string, limit int) (*telemetrytypes.TelemetryFieldValues, bool, error) {
-	ctx = withTelemetryContext(ctx, "ListJSONValues")
+	ctx = withTelemetryContext(ctx, telemetrytypes.SignalLogs, "ListJSONValues")
 	path = CleanPathPrefixes(path)
 
 	if strings.Contains(path, telemetrytypes.ArraySep) || strings.Contains(path, telemetrytypes.ArrayAnyIndex) {
 		return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "array paths are not supported")
 	}
 
-	promoted, err := t.IsPathPromoted(ctx, path)
+	promoted, err := t.isPathPromoted(ctx, logsBodyPromotedEntry, path)
 	if err != nil {
 		return nil, false, err
 	}
@@ -376,13 +381,13 @@ func derefValue(v any) any {
 	return val.Interface()
 }
 
-// IsPathPromoted checks if a specific path is promoted (Column Evolution table: field_name for logs body).
-func (t *telemetryMetaStore) IsPathPromoted(ctx context.Context, path string) (bool, error) {
-	ctx = withTelemetryContext(ctx, "IsPathPromoted")
+// isPathPromoted checks if a specific path is promoted (Column Evolution table: field_name for the entry's column).
+func (t *telemetryMetaStore) isPathPromoted(ctx context.Context, entry telemetrytypes.EvolutionEntry, path string) (bool, error) {
+	ctx = withTelemetryContext(ctx, entry.Signal, "isPathPromoted")
 	split := strings.Split(path, telemetrytypes.ArraySep)
 	pathSegment := split[0]
 	query := fmt.Sprintf("SELECT 1 FROM %s.%s WHERE signal = ? AND column_name = ? AND field_context = ? AND field_name = ? LIMIT 1", DBName, PromotedPathsTableName)
-	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, telemetrytypes.SignalLogs, logstelemetryschema.LogsV2BodyPromotedColumn, telemetrytypes.FieldContextBody, pathSegment)
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, entry.Signal, entry.ColumnName, entry.FieldContext, pathSegment)
 	if err != nil {
 		return false, errors.WrapInternalf(err, CodeFailCheckPathPromoted, "failed to check if path %s is promoted", path)
 	}
@@ -391,14 +396,14 @@ func (t *telemetryMetaStore) IsPathPromoted(ctx context.Context, path string) (b
 	return rows.Next(), nil
 }
 
-// GetPromotedPaths returns promoted paths from the Column Evolution table (field_name for logs body).
-func (t *telemetryMetaStore) GetPromotedPaths(ctx context.Context, paths ...string) (map[string]bool, error) {
-	ctx = withTelemetryContext(ctx, "GetPromotedPaths")
+// GetPromotedPaths returns promoted paths from the Column Evolution table (field_name for the entry's column).
+func (t *telemetryMetaStore) GetPromotedPaths(ctx context.Context, entry telemetrytypes.EvolutionEntry, paths ...string) (map[string]bool, error) {
+	ctx = withTelemetryContext(ctx, entry.Signal, "GetPromotedPaths")
 	sb := sqlbuilder.Select("field_name").From(fmt.Sprintf("%s.%s", DBName, PromotedPathsTableName))
 	conditions := []string{
-		sb.Equal("signal", telemetrytypes.SignalLogs),
-		sb.Equal("column_name", logstelemetryschema.LogsV2BodyPromotedColumn),
-		sb.Equal("field_context", telemetrytypes.FieldContextBody),
+		sb.Equal("signal", entry.Signal),
+		sb.Equal("column_name", entry.ColumnName),
+		sb.Equal("field_context", entry.FieldContext),
 		sb.NotEqual("field_name", "__all__"),
 	}
 	if len(paths) > 0 {
@@ -438,9 +443,10 @@ func CleanPathPrefixes(path string) string {
 	return path
 }
 
-// PromotePaths inserts promoted paths into the Column Evolution table (same schema as signoz-otel-collector metadata_migrations).
-func (t *telemetryMetaStore) PromotePaths(ctx context.Context, paths ...string) error {
-	ctx = withTelemetryContext(ctx, "PromotePaths")
+// PromotePaths inserts promoted paths into the Column Evolution table as rows templated by entry
+// (same schema as signoz-otel-collector metadata_migrations); FieldName and ReleaseTime are set per path.
+func (t *telemetryMetaStore) PromotePaths(ctx context.Context, entry telemetrytypes.EvolutionEntry, paths ...string) error {
+	ctx = withTelemetryContext(ctx, entry.Signal, "PromotePaths")
 	batch, err := t.telemetrystore.ClickhouseDB().PrepareBatch(ctx,
 		fmt.Sprintf("INSERT INTO %s.%s (signal, column_name, column_type, field_context, field_name, version, release_time) VALUES", DBName,
 			PromotedPathsTableName))
@@ -454,7 +460,7 @@ func (t *telemetryMetaStore) PromotePaths(ctx context.Context, paths ...string)
 		if trimmed == "" {
 			continue
 		}
-		if err := batch.Append(telemetrytypes.SignalLogs, logstelemetryschema.LogsV2BodyPromotedColumn, "JSON()", telemetrytypes.FieldContextBody, trimmed, 0, releaseTime); err != nil {
+		if err := batch.Append(entry.Signal, entry.ColumnName, entry.ColumnType, entry.FieldContext, trimmed, entry.Version, releaseTime); err != nil {
 			_ = batch.Abort()
 			return errors.WrapInternalf(err, CodeFailedToAppendPath, "failed to append path")
 		}
@@ -466,9 +472,9 @@ func (t *telemetryMetaStore) PromotePaths(ctx context.Context, paths ...string)
 	return nil
 }
 
-func withTelemetryContext(ctx context.Context, functionName string) context.Context {
+func withTelemetryContext(ctx context.Context, signal telemetrytypes.Signal, functionName string) context.Context {
 	return ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
-		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
+		instrumentationtypes.TelemetrySignal:  signal.StringValue(),
 		instrumentationtypes.CodeNamespace:    "metadata",
 		instrumentationtypes.CodeFunctionName: functionName,
 	})

pkg/telemetryschema/tracestelemetryschema/const.go

@@ -447,14 +447,16 @@ var (
 		{Name: SpanIsRemoteColumn, FieldContext: telemetrytypes.FieldContextSpan},
 	}
 
-	// ContextualSpanColumns lists the typed attribute and resource columns
-	// selected raw (rather than via ColumnExpressionFor) so that consume.go
-	// can merge them into unified "attributes" and "resource" maps.
+	// ContextualSpanColumns lists the bag columns selected raw (rather than via
+	// ColumnExpressionFor) so that consume.go can merge them into unified
+	// "attributes" and "resource" maps: the legacy maps plus the attributes
+	// JSON column, merged per row with JSON winning on collision.
 	ContextualSpanColumns = []string{
 		SpanAttributesStringColumn,
 		SpanAttributesNumberColumn,
 		SpanAttributesBoolColumn,
 		SpanResourcesStringColumn,
+		SpanAttributesColumn,
 	}
 
 	DefaultFields = map[string]telemetrytypes.TelemetryFieldKey{

pkg/types/promotetypes/target.go

@@ -0,0 +1,93 @@
+package promotetypes
+
+import (
+	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
+	"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+)
+
+// Target identifies a promotion domain: the column evolution record written
+// per promoted path, the table per-path indexes are created on, and the API
+// path rules.
+type Target struct {
+	Entry              telemetrytypes.EvolutionEntry // evolution row template; FieldName and ReleaseTime are set per write
+	DBName             string                        // index DDL database, used only when IndexesSupported
+	LocalTableName     string                        // index DDL local table, used only when IndexesSupported
+	BaseColumn         string                        // column holding every path; indexes for unpromoted paths are created on it
+	RequiredPathPrefix string                        // prefix API paths must carry, stripped before storing; empty for bare names
+	IndexesSupported   bool                          // whether per-path skip indexes can be created for this domain
+}
+
+func (t Target) PromotedColumn() string { return t.Entry.ColumnName }
+
+func (t Target) BaseColumnPrefix() string { return t.BaseColumn + "." }
+
+func (t Target) PromotedColumnPrefix() string { return t.PromotedColumn() + "." }
+
+// NewTarget creates the Target for a promotion domain.
+func NewTarget(entry telemetrytypes.EvolutionEntry, dbName, localTableName, baseColumn, requiredPathPrefix string, indexesSupported bool) Target {
+	return Target{
+		Entry:              entry,
+		DBName:             dbName,
+		LocalTableName:     localTableName,
+		BaseColumn:         baseColumn,
+		RequiredPathPrefix: requiredPathPrefix,
+		IndexesSupported:   indexesSupported,
+	}
+}
+
+// NewLogsBodyTarget returns the domain for the logs body JSON column
+// (body_v2 -> body_promoted), with per-path skip index support.
+func NewLogsBodyTarget() Target {
+	return NewTarget(
+		telemetrytypes.EvolutionEntry{
+			Signal:       telemetrytypes.SignalLogs,
+			ColumnName:   logstelemetryschema.LogsV2BodyPromotedColumn,
+			ColumnType:   "JSON()",
+			FieldContext: telemetrytypes.FieldContextBody,
+		},
+		logstelemetryschema.DBName,
+		logstelemetryschema.LogsV2LocalTableName,
+		logstelemetryschema.LogsV2BodyV2Column,
+		telemetrytypes.BodyJSONStringSearchPrefix,
+		true,
+	)
+}
+
+// NewTracesAttributesTarget returns the domain for the spans attributes JSON
+// column (attributes -> attributes_promoted); promotion only for now.
+func NewTracesAttributesTarget() Target {
+	return NewTarget(
+		telemetrytypes.EvolutionEntry{
+			Signal:       telemetrytypes.SignalTraces,
+			ColumnName:   tracestelemetryschema.SpanAttributesPromotedColumn,
+			ColumnType:   "JSON()",
+			FieldContext: telemetrytypes.FieldContextAttribute,
+		},
+		tracestelemetryschema.DBName,
+		tracestelemetryschema.SpanIndexV3LocalTableName,
+		tracestelemetryschema.SpanAttributesColumn,
+		"",
+		false,
+	)
+}
+
+// Targets returns every supported promotion domain.
+func Targets() []Target {
+	return []Target{
+		NewLogsBodyTarget(),
+		NewTracesAttributesTarget(),
+	}
+}
+
+// TargetFor resolves the domain for a (signal, context) pair; ok is false
+// when no domain exists for the pair.
+func TargetFor(signal telemetrytypes.Signal, context telemetrytypes.FieldContext) (Target, bool) {
+	for _, target := range Targets() {
+		if target.Entry.Signal.StringValue() == signal.StringValue() &&
+			target.Entry.FieldContext.StringValue() == context.StringValue() {
+			return target, true
+		}
+	}
+	return Target{}, false
+}

pkg/types/promotetypes/types.go

@@ -3,7 +3,6 @@ package promotetypes
 import (
 	"strings"
 
-	"github.com/SigNoz/signoz-otel-collector/constants"
 	"github.com/SigNoz/signoz-otel-collector/pkg/keycheck"
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -23,7 +22,7 @@ type PromotePath struct {
 	Indexes []WrappedIndex `json:"indexes,omitempty"`
 }
 
-func (i *PromotePath) ValidateAndSetDefaults() error {
+func (i *PromotePath) ValidateAndSetDefaults(target Target) error {
 	if i.Path == "" {
 		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "path is required")
 	}
@@ -36,22 +35,27 @@ func (i *PromotePath) ValidateAndSetDefaults() error {
 		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "array paths can not be promoted or indexed")
 	}
 
-	if strings.HasPrefix(i.Path, constants.BodyV2ColumnPrefix) || strings.HasPrefix(i.Path, constants.BodyPromotedColumnPrefix) {
-		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "`%s`, `%s` don't add these prefixes to the path", constants.BodyV2ColumnPrefix, constants.BodyPromotedColumnPrefix)
+	if strings.HasPrefix(i.Path, target.BaseColumnPrefix()) || strings.HasPrefix(i.Path, target.PromotedColumnPrefix()) {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "`%s`, `%s` don't add these prefixes to the path", target.BaseColumnPrefix(), target.PromotedColumnPrefix())
 	}
 
-	if !strings.HasPrefix(i.Path, telemetrytypes.BodyJSONStringSearchPrefix) {
-		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "path must start with `body.`")
+	if target.RequiredPathPrefix != "" {
+		if !strings.HasPrefix(i.Path, target.RequiredPathPrefix) {
+			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "path must start with `%s`", target.RequiredPathPrefix)
+		}
+		// remove the required prefix from the path
+		i.Path = strings.TrimPrefix(i.Path, target.RequiredPathPrefix)
 	}
 
-	// remove the "body." prefix from the path
-	i.Path = strings.TrimPrefix(i.Path, telemetrytypes.BodyJSONStringSearchPrefix)
-
 	isCardinal := keycheck.IsCardinal(i.Path)
 	if isCardinal {
 		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "cardinal paths can not be promoted or indexed")
 	}
 
+	if len(i.Indexes) > 0 && !target.IndexesSupported {
+		return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "indexes are not supported for %s %s", target.Entry.Signal.StringValue(), target.Entry.FieldContext.StringValue())
+	}
+
 	for idx, index := range i.Indexes {
 		if index.Type == "" {
 			return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "index type is required")

pkg/types/promotetypes/types_test.go

@@ -0,0 +1,109 @@
+package promotetypes
+
+import (
+	"testing"
+
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	testLogsBodyTarget         = NewLogsBodyTarget()
+	testTracesAttributesTarget = NewTracesAttributesTarget()
+)
+
+func TestValidateAndSetDefaultsLogsBody(t *testing.T) {
+	t.Run("valid path gets body prefix stripped", func(t *testing.T) {
+		p := &PromotePath{Path: "body.user.name", Promote: true}
+		require.NoError(t, p.ValidateAndSetDefaults(testLogsBodyTarget))
+		assert.Equal(t, "user.name", p.Path)
+	})
+
+	t.Run("path without body prefix is rejected", func(t *testing.T) {
+		p := &PromotePath{Path: "user.name", Promote: true}
+		require.Error(t, p.ValidateAndSetDefaults(testLogsBodyTarget))
+	})
+
+	t.Run("column prefixes are rejected", func(t *testing.T) {
+		for _, path := range []string{"body_v2.user.name", "body_promoted.user.name"} {
+			p := &PromotePath{Path: path, Promote: true}
+			require.Error(t, p.ValidateAndSetDefaults(testLogsBodyTarget), path)
+		}
+	})
+
+	t.Run("empty, spaced and array paths are rejected", func(t *testing.T) {
+		for _, path := range []string{"", "body.my path", "body.users[].id", "body.users[*].id"} {
+			p := &PromotePath{Path: path, Promote: true}
+			require.Error(t, p.ValidateAndSetDefaults(testLogsBodyTarget), path)
+		}
+	})
+
+	t.Run("cardinal paths are rejected", func(t *testing.T) {
+		p := &PromotePath{Path: "body.request.550e8400-e29b-41d4-a716-446655440000", Promote: true}
+		require.Error(t, p.ValidateAndSetDefaults(testLogsBodyTarget))
+	})
+
+	t.Run("valid index gets json data type default", func(t *testing.T) {
+		p := &PromotePath{
+			Path: "body.user.name",
+			Indexes: []WrappedIndex{
+				{FieldDataType: telemetrytypes.FieldDataTypeString, Type: "ngrambf_v1(4, 1024, 2, 0)", Granularity: 1},
+			},
+		}
+		require.NoError(t, p.ValidateAndSetDefaults(testLogsBodyTarget))
+		assert.Equal(t, telemetrytypes.String, p.Indexes[0].JSONDataType)
+	})
+
+	t.Run("index with unsupported column type is rejected", func(t *testing.T) {
+		p := &PromotePath{
+			Path:    "body.user.active",
+			Indexes: []WrappedIndex{{FieldDataType: telemetrytypes.FieldDataTypeBool, Type: "minmax", Granularity: 1}},
+		}
+		require.Error(t, p.ValidateAndSetDefaults(testLogsBodyTarget))
+	})
+
+	t.Run("index without type or granularity is rejected", func(t *testing.T) {
+		p := &PromotePath{
+			Path:    "body.user.name",
+			Indexes: []WrappedIndex{{FieldDataType: telemetrytypes.FieldDataTypeString, Granularity: 1}},
+		}
+		require.Error(t, p.ValidateAndSetDefaults(testLogsBodyTarget))
+
+		p = &PromotePath{
+			Path:    "body.user.name",
+			Indexes: []WrappedIndex{{FieldDataType: telemetrytypes.FieldDataTypeString, Type: "minmax"}},
+		}
+		require.Error(t, p.ValidateAndSetDefaults(testLogsBodyTarget))
+	})
+}
+
+func TestValidateAndSetDefaultsTracesAttributes(t *testing.T) {
+	t.Run("bare attribute name is kept as is", func(t *testing.T) {
+		p := &PromotePath{Path: "http.method", Promote: true}
+		require.NoError(t, p.ValidateAndSetDefaults(testTracesAttributesTarget))
+		assert.Equal(t, "http.method", p.Path)
+	})
+
+	t.Run("column prefixes are rejected", func(t *testing.T) {
+		for _, path := range []string{"attributes.http.method", "attributes_promoted.http.method"} {
+			p := &PromotePath{Path: path, Promote: true}
+			require.Error(t, p.ValidateAndSetDefaults(testTracesAttributesTarget), path)
+		}
+	})
+
+	t.Run("empty, spaced and array paths are rejected", func(t *testing.T) {
+		for _, path := range []string{"", "my attr", "tags[].id"} {
+			p := &PromotePath{Path: path, Promote: true}
+			require.Error(t, p.ValidateAndSetDefaults(testTracesAttributesTarget), path)
+		}
+	})
+
+	t.Run("indexes are rejected for a target without index support", func(t *testing.T) {
+		p := &PromotePath{
+			Path:    "http.method",
+			Indexes: []WrappedIndex{{FieldDataType: telemetrytypes.FieldDataTypeString, Type: "ngrambf_v1(4, 1024, 2, 0)", Granularity: 1}},
+		}
+		require.Error(t, p.ValidateAndSetDefaults(testTracesAttributesTarget))
+	})
+}

pkg/types/telemetrytypes/signal.go

@@ -22,3 +22,18 @@ func (Signal) Enum() []any {
 		SignalUnspecified,
 	}
 }
+
+// SignalFromText resolves a signal word to its Signal; ok is false for an
+// unknown word.
+func SignalFromText(text string) (Signal, bool) {
+	s := Signal{valuer.NewString(text)}
+	switch s {
+	case SignalTraces:
+		return SignalTraces, true
+	case SignalLogs:
+		return SignalLogs, true
+	case SignalMetrics:
+		return SignalMetrics, true
+	}
+	return Signal{}, false
+}

pkg/types/telemetrytypes/store.go

@@ -37,11 +37,13 @@ type MetadataStore interface {
 	// ListLogsJSONIndexes lists the JSON indexes for the logs table.
 	ListLogsJSONIndexes(ctx context.Context, filters ...string) ([]TelemetryFieldKeySkipIndex, error)
 
-	// ListPromotedPaths lists the promoted paths.
-	GetPromotedPaths(ctx context.Context, paths ...string) (map[string]bool, error)
+	// GetPromotedPaths lists the promoted paths recorded in the column
+	// evolution table for the entry's signal, column and field context.
+	GetPromotedPaths(ctx context.Context, entry EvolutionEntry, paths ...string) (map[string]bool, error)
 
-	// PromotePaths promotes the paths.
-	PromotePaths(ctx context.Context, paths ...string) error
+	// PromotePaths records promoted paths in the column evolution table as
+	// rows templated by entry; FieldName and ReleaseTime are set per path.
+	PromotePaths(ctx context.Context, entry EvolutionEntry, paths ...string) error
 
 	// GetFirstSeenFromMetricMetadata gets the first seen timestamp for a metric metadata lookup key.
 	GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []MetricMetadataLookupKey) (map[MetricMetadataLookupKey]int64, error)

pkg/types/telemetrytypes/telemetrytypestest/metadata_store.go

@@ -361,7 +361,7 @@ func (m *MockMetadataStore) SetTemporality(metricName string, temporality metric
 }
 
 // PromotePaths promotes the paths.
-func (m *MockMetadataStore) PromotePaths(ctx context.Context, paths ...string) error {
+func (m *MockMetadataStore) PromotePaths(_ context.Context, _ telemetrytypes.EvolutionEntry, paths ...string) error {
 	for _, path := range paths {
 		m.PromotedPathsMap[path] = true
 	}
@@ -369,7 +369,7 @@ func (m *MockMetadataStore) PromotePaths(ctx context.Context, paths ...string) e
 }
 
 // GetPromotedPaths returns the promoted paths.
-func (m *MockMetadataStore) GetPromotedPaths(ctx context.Context, paths ...string) (map[string]bool, error) {
+func (m *MockMetadataStore) GetPromotedPaths(_ context.Context, _ telemetrytypes.EvolutionEntry, _ ...string) (map[string]bool, error) {
 	return m.PromotedPathsMap, nil
 }
 

tests/fixtures/traces.py

@@ -331,7 +331,7 @@ class Traces(ABC):
         flags: np.uint32 = 0,
         scope: dict[str, Any] = {},
         resource_write_mode: Literal["legacy_only", "dual_write"] = "dual_write",
-        attribute_write_mode: Literal["legacy_only", "dual_write"] = "dual_write",
+        attribute_write_mode: Literal["legacy_only", "dual_write", "json_only"] = "dual_write",
     ) -> None:
         if timestamp is None:
             timestamp = datetime.datetime.now()
@@ -514,8 +514,11 @@ class Traces(ABC):
 
         # Spans before the attribute JSON-evolution time populate only the legacy
         # attributes_{string,number,bool} maps; spans at or after it dual-write the
-        # native-typed `attributes` JSON column too.
+        # native-typed `attributes` JSON column too, and spans past the map-write
+        # cutoff populate only the JSON column (metadata rows are still written).
         self.attributes_json = {} if attribute_write_mode == "legacy_only" else dict(attributes)
+        if attribute_write_mode == "json_only":
+            self.attribute_string, self.attributes_number, self.attributes_bool = {}, {}, {}
 
         # Process events and derive error events. self.events holds the parsed
         # response shape; np_arr() encodes back to the DB format on insert.

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

@@ -5,11 +5,15 @@ from http import HTTPStatus
 from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.querier import (
+    BuilderQuery,
+    OrderBy,
     RequestType,
+    TelemetryFieldKey,
     assert_grouped_series,
     build_aggregation,
     build_group_by_field,
     build_traces_scalar_query,
+    get_rows,
     index_series_by_label,
     make_query_request,
 )
@@ -329,3 +333,84 @@ def test_traces_attributes_json_collision_and_map_parity(
         aggregations = (response.json()["data"]["data"]["results"][0].get("aggregations")) or []
         series = index_series_by_label(aggregations[0]["series"], "http.route") if aggregations else {}
         assert set(series.keys()) == expected, label
+
+
+def test_traces_attributes_json_list_view(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+    seed_attribute_evolution: Callable[[str, datetime], None],
+) -> None:
+    """The empty-selectFields list view reads every attribute home per row: the span before
+    the evolution carries attributes only in the legacy maps, the span past the map-write
+    cutoff carries them only in the `attributes` JSON column, and the dual-written span
+    carries both. One straddling raw query must surface each row's bag from its own home,
+    flattened to the legacy dotted-key shape with native types."""
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    evolution_time = datetime.now(tz=UTC).replace(second=0, microsecond=0) - timedelta(minutes=30)
+    seed_attribute_evolution("traces", evolution_time)
+
+    service = "list-homes-service"
+    spans = [
+        Traces(
+            timestamp=evolution_time - timedelta(minutes=10),
+            trace_id=TraceIdGenerator.trace_id(),
+            span_id=TraceIdGenerator.span_id(),
+            name="map only",
+            resources={"service.name": service},
+            attributes={"http.route": "/map", "http.retry.count": 1, "http.cache.hit": True},
+            attribute_write_mode="legacy_only",
+        ),
+        Traces(
+            timestamp=evolution_time + timedelta(minutes=5),
+            trace_id=TraceIdGenerator.trace_id(),
+            span_id=TraceIdGenerator.span_id(),
+            name="json only",
+            resources={"service.name": service},
+            attributes={"http.route": "/json", "http.retry.count": 2, "http.cache.hit": False},
+            attribute_write_mode="json_only",
+        ),
+        Traces(
+            timestamp=evolution_time + timedelta(minutes=10),
+            trace_id=TraceIdGenerator.trace_id(),
+            span_id=TraceIdGenerator.span_id(),
+            name="dual written",
+            resources={"service.name": service},
+            attributes={"http.route": "/dual", "http.retry.count": 3, "http.cache.hit": True},
+            attribute_write_mode="dual_write",
+        ),
+    ]
+    insert_traces(spans)
+
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((evolution_time - timedelta(minutes=15)).timestamp() * 1000),
+        end_ms=int((evolution_time + timedelta(minutes=15)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            BuilderQuery(
+                signal="traces",
+                name="A",
+                limit=10,
+                filter_expression=f"resource.service.name = '{service}'",
+                order=[OrderBy(TelemetryFieldKey("timestamp"), "asc")],
+            ).to_dict()
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK
+
+    rows = get_rows(response)
+    assert len(rows) == 3
+    expected = [
+        ("map only", {"http.route": "/map", "http.retry.count": 1, "http.cache.hit": True}),
+        ("json only", {"http.route": "/json", "http.retry.count": 2, "http.cache.hit": False}),
+        ("dual written", {"http.route": "/dual", "http.retry.count": 3, "http.cache.hit": True}),
+    ]
+    for row, (name, attributes) in zip(rows, expected, strict=True):
+        assert row["data"]["name"] == name
+        # int/float compare equal in Python, so the map-sourced float64 and the
+        # JSON-sourced number both match the inserted values here.
+        assert row["data"]["attributes"] == attributes, name