fix: integration tests

pkg/errors/http.go

@@ -18,13 +18,17 @@ type responseerroradditional struct {
 
 func AsJSON(cause error) *JSON {
 	// See if this is an instance of the base error or not
-	_, c, m, _, u, a := Unwrapb(cause)
+	_, c, m, err, u, a := Unwrapb(cause)
 
 	rea := make([]responseerroradditional, len(a))
 	for k, v := range a {
 		rea[k] = responseerroradditional{v}
 	}
 
+	if err != nil {
+		rea = append(rea, responseerroradditional{err.Error()})
+	}
+
 	return &JSON{
 		Code:    c.String(),
 		Message: m,

pkg/query-service/rules/base_rule_test.go

@@ -658,7 +658,7 @@ func TestBaseRule_FilterNewSeries(t *testing.T) {
 			telemetryStore := telemetrystoretest.New(telemetrystore.Config{}, &queryMatcherAny{})
 
 			// Setup mock metadata store
-			mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+			mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 
 			// Create query parser
 			queryParser := queryparser.New(settings)

pkg/telemetrylogs/condition_builder.go

@@ -33,7 +33,7 @@ func (c *conditionBuilder) conditionFor(
 		return "", err
 	}
 
-	if column.Type.GetType() == schema.ColumnTypeEnumJSON && querybuilder.BodyJSONQueryEnabled && key.FieldContext != telemetrytypes.FieldContextLog {
+	if column.Type.GetType() == schema.ColumnTypeEnumJSON && querybuilder.BodyJSONQueryEnabled && !key.Materialized {
 		valueType, value := InferDataType(value, operator, key)
 		cond, err := NewJSONConditionBuilder(key, valueType).buildJSONCondition(operator, value, sb)
 		if err != nil {
@@ -52,7 +52,7 @@ func (c *conditionBuilder) conditionFor(
 	}
 
 	// Check if this is a body JSON search - either by FieldContext
-	if key.FieldContext == telemetrytypes.FieldContextBody {
+	if key.FieldContext == telemetrytypes.FieldContextBody && !querybuilder.BodyJSONQueryEnabled {
 		tblFieldName, value = GetBodyJSONKey(ctx, key, operator, value)
 	}
 

pkg/telemetrylogs/const.go

@@ -3,6 +3,7 @@ package telemetrylogs
 import (
 	"fmt"
 
+	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz-otel-collector/constants"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
@@ -39,20 +40,22 @@ const (
 
 	BodyV2ColumnPrefix       = constants.BodyV2ColumnPrefix
 	BodyPromotedColumnPrefix = constants.BodyPromotedColumnPrefix
+	MessageBodyField         = "message"
 	MessageSubColumn         = "body_v2.message"
 	bodySearchDefaultWarning = "body searches default to `body.message:string`. Use `body.<key>` to search a different field inside body"
 )
 
 var (
-	// mapping for body logical field to message sub column
-	// Here field context is log since message is a type hint and
-	// in a sense it is a direct column of log table and doesn't need any lambda expressions
-	// TODO(Piyush): Add description for detailed explanation of remapping of body to message
+	// BodyLogicalFieldJSONMapping is the canonical key for the "message" type hint.
+	// It lives under body (FieldContextBody) but Materialized=true signals the field
+	// mapper to access it as a direct sub-column (body_v2.message) rather than via
+	// dynamicElement(). Its name is the bare field name ("message"), not the column path.
 	BodyLogicalFieldJSONMapping = &telemetrytypes.TelemetryFieldKey{
-		Name:          MessageSubColumn,
+		Name:          MessageBodyField,
 		Signal:        telemetrytypes.SignalLogs,
-		FieldContext:  telemetrytypes.FieldContextLog,
+		FieldContext:  telemetrytypes.FieldContextBody,
 		FieldDataType: telemetrytypes.FieldDataTypeString,
+		Materialized:  true,
 	}
 	DefaultFullTextColumn = &telemetrytypes.TelemetryFieldKey{
 		Name:          "body",
@@ -146,7 +149,15 @@ func enrichMapsForJSONBodyEnabled() {
 	if querybuilder.BodyJSONQueryEnabled {
 		DefaultFullTextColumn = BodyLogicalFieldJSONMapping
 		IntrinsicFields["body"] = *BodyLogicalFieldJSONMapping
+
+		// Register all key names that should resolve to the message type-hint column so
+		// QB can look them up directly: bare "message" and qualified "body_v2.message".
 		IntrinsicFields[MessageSubColumn] = *BodyLogicalFieldJSONMapping
+		IntrinsicFields[MessageBodyField] = *BodyLogicalFieldJSONMapping
+		logsV2Columns[MessageSubColumn] = &schema.Column{
+			Name: MessageSubColumn,
+			Type: schema.ColumnTypeString,
+		}
 	}
 }
 

pkg/telemetrylogs/field_mapper.go

@@ -58,7 +58,6 @@ var (
 			KeyType:   schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString},
 			ValueType: schema.ColumnTypeString,
 		}},
-		MessageSubColumn: {Name: MessageSubColumn, Type: schema.ColumnTypeString},
 	}
 )
 
@@ -89,6 +88,13 @@ func (m *fieldMapper) getColumn(_ context.Context, key *telemetrytypes.Telemetry
 			return logsV2Columns["attributes_bool"], nil
 		}
 	case telemetrytypes.FieldContextBody:
+		// Type hints (Materialized=true) have a direct physical sub-column in body_v2.
+		// Return a synthetic String column so the condition builder uses the direct path
+		// instead of the JSON condition builder (which expects a JSONPlan).
+		if key.Materialized {
+			// Type hints have a direct physical sub-column in body_v2 (e.g. body_v2.message).
+			return logsV2Columns[fmt.Sprintf("%s.%s", LogsV2BodyV2Column, key.Name)], nil
+		}
 		// Body context is for JSON body fields
 		// Use body_v2 if feature flag is enabled
 		if querybuilder.BodyJSONQueryEnabled {
@@ -249,34 +255,37 @@ func (m *fieldMapper) buildFieldForJSON(key *telemetrytypes.TelemetryFieldKey) (
 		node := plan[0]
 
 		expr := fmt.Sprintf("dynamicElement(%s, '%s')", node.FieldPath(), node.TerminalConfig.ElemType.StringValue())
-		if key.Materialized {
-			if len(plan) < 2 {
-				return "", errors.Newf(errors.TypeUnexpected, CodePromotedPlanMissing,
-					"plan length is less than 2 for promoted path: %s", key.Name)
-			}
+		// TODO(Piyush): Promoted path logic commented out. Materialized now means type hint
+		// promotion will be extracted from key field evolution
+		// (direct sub-column access), not a promoted body_promoted.* column.
+		// if key.Materialized {
+		// 	if len(plan) < 2 {
+		// 		return "", errors.Newf(errors.TypeUnexpected, CodePromotedPlanMissing,
+		// 			"plan length is less than 2 for promoted path: %s", key.Name)
+		// 	}
 
-			node := plan[1]
-			promotedExpr := fmt.Sprintf(
-				"dynamicElement(%s, '%s')",
-				node.FieldPath(),
-				node.TerminalConfig.ElemType.StringValue(),
-			)
+		// 	node := plan[1]
+		// 	promotedExpr := fmt.Sprintf(
+		// 		"dynamicElement(%s, '%s')",
+		// 		node.FieldPath(),
+		// 		node.TerminalConfig.ElemType.StringValue(),
+		// 	)
 
-			// dynamicElement returns NULL for scalar types or an empty array for array types.
-			if node.TerminalConfig.ElemType.IsArray {
-				expr = fmt.Sprintf(
-					"if(length(%s) > 0, %s, %s)",
-					promotedExpr,
-					promotedExpr,
-					expr,
-				)
-			} else {
-				// promoted column first then body_json column
-				// TODO(Piyush): Change this in future for better performance
-				expr = fmt.Sprintf("coalesce(%s, %s)", promotedExpr, expr)
-			}
+		// 	// dynamicElement returns NULL for scalar types or an empty array for array types.
+		// 	if node.TerminalConfig.ElemType.IsArray {
+		// 		expr = fmt.Sprintf(
+		// 			"if(length(%s) > 0, %s, %s)",
+		// 			promotedExpr,
+		// 			promotedExpr,
+		// 			expr,
+		// 		)
+		// 	} else {
+		// 		// promoted column first then body_json column
+		// 		// TODO(Piyush): Change this in future for better performance
+		// 		expr = fmt.Sprintf("coalesce(%s, %s)", promotedExpr, expr)
+		// 	}
 
-		}
+		// }
 
 		return expr, nil
 	}

pkg/telemetrylogs/json_stmt_builder_test.go

@@ -96,6 +96,9 @@ func TestStmtBuilderTimeSeriesBodyGroupByJSON(t *testing.T) {
 	}
 }
 
+/* Promoted path tests commented out — Materialized now means type hint (direct sub-column),
+   not a body_promoted.* column. These tests assumed the old coalesce(body_promoted.x, body_v2.x) path.
+
 func TestStmtBuilderTimeSeriesBodyGroupByPromoted(t *testing.T) {
 	enable, disable := jsonQueryTestUtil(t)
 	enable()
@@ -156,6 +159,7 @@ func TestStmtBuilderTimeSeriesBodyGroupByPromoted(t *testing.T) {
 		})
 	}
 }
+*/
 
 func TestStatementBuilderListQueryBodyHas(t *testing.T) {
 	enable, disable := jsonQueryTestUtil(t)
@@ -386,8 +390,8 @@ func TestStatementBuilderListQueryBody(t *testing.T) {
 				Limit:  10,
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (arrayExists(`body_v2.education`-> (arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) BETWEEN ? AND ?, dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) BETWEEN ? AND ?, arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (arrayExists(`body_v2.education`-> (arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) BETWEEN ? AND ?, dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) BETWEEN ? AND ?, arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{float64(2), float64(4), float64(2), float64(4), uint64(1747945619), uint64(1747983448), float64(2), float64(4), float64(2), float64(4), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (arrayExists(`body_v2.education`-> (arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) BETWEEN ? AND ?, dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) BETWEEN ? AND ?, arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{uint64(1747945619), uint64(1747983448), float64(2), float64(4), float64(2), float64(4), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 			},
 			expectedErr: nil,
 		},
@@ -400,8 +404,8 @@ func TestStatementBuilderListQueryBody(t *testing.T) {
 				Limit:  10,
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (arrayExists(`body_v2.education`-> (arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) IN (?, ?), dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) IN (?, ?), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (arrayExists(`body_v2.education`-> (arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) IN (?, ?), dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) IN (?, ?), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{float64(2), float64(4), float64(2), float64(4), uint64(1747945619), uint64(1747983448), float64(2), float64(4), float64(2), float64(4), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (arrayExists(`body_v2.education`-> (arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) IN (?, ?), dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')) OR arrayExists(`body_v2.education[].awards`-> toFloat64(dynamicElement(`body_v2.education[].awards`.`semester`, 'Int64')) IN (?, ?), arrayMap(x->dynamicElement(x, 'JSON'), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))'))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Args:  []any{uint64(1747945619), uint64(1747983448), float64(2), float64(4), float64(2), float64(4), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 			},
 			expectedErr: nil,
 		},
@@ -525,6 +529,9 @@ func TestStatementBuilderListQueryBody(t *testing.T) {
 	}
 }
 
+/* Promoted path list-query tests commented out — Materialized now means type hint
+   (direct sub-column access), not a body_promoted.* column.
+
 func TestStatementBuilderListQueryBodyPromoted(t *testing.T) {
 	enable, disable := jsonQueryTestUtil(t)
 	enable()
@@ -700,6 +707,7 @@ func TestStatementBuilderListQueryBodyPromoted(t *testing.T) {
 		})
 	}
 }
+*/
 
 func TestStatementBuilderListQueryBodyMessage(t *testing.T) {
 	enable, disable := jsonQueryTestUtil(t)
@@ -707,19 +715,6 @@ func TestStatementBuilderListQueryBodyMessage(t *testing.T) {
 	defer disable()
 
 	statementBuilder := buildJSONTestStatementBuilder(t)
-	indexed := []*telemetrytypes.TelemetryFieldKey{
-		{
-			Name: "message",
-			Indexes: []telemetrytypes.JSONDataTypeIndex{
-				{
-					Type:             telemetrytypes.String,
-					ColumnExpression: "body_promoted.message",
-					IndexExpression:  "(lower(assumeNotNull(dynamicElement(body_promoted.message, 'String'))))",
-				},
-			},
-		},
-	}
-	testAddIndexedPaths(t, statementBuilder, indexed...)
 	cases := []struct {
 		name        string
 		requestType qbtypes.RequestType
@@ -732,57 +727,57 @@ func TestStatementBuilderListQueryBodyMessage(t *testing.T) {
 			requestType: qbtypes.RequestTypeRaw,
 			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
 				Signal: telemetrytypes.SignalLogs,
-				Filter: &qbtypes.Filter{Expression: "body.message Exists"},
+				Filter: &qbtypes.Filter{Expression: "message Exists"},
 				Limit:  10,
 			},
 			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (dynamicElement(body_v2.`message`, 'String') IS NOT NULL) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-			},
-			expectedErr: nil,
-		},
-		{
-			name:        "body.message equals to empty string",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Filter: &qbtypes.Filter{Expression: "body.message = ''"},
-				Limit:  10,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (dynamicElement(body_v2.`message`, 'String') = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND body_v2.message <> ? AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
 				Args:  []any{uint64(1747945619), uint64(1747983448), "", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 			},
 			expectedErr: nil,
 		},
-		{
-			name:        "body.message equals to 'Iron Award'",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Filter: &qbtypes.Filter{Expression: "body.message = 'Iron Award'"},
-				Limit:  10,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (dynamicElement(body_v2.`message`, 'String') = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{uint64(1747945619), uint64(1747983448), "Iron Award", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-			},
-			expectedErr: nil,
-		},
-		{
-			name:        "body.message contains 'Iron Award'",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Filter: &qbtypes.Filter{Expression: "body.message Contains 'Iron Award'"},
-				Limit:  10,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (LOWER(dynamicElement(body_v2.`message`, 'String')) LIKE LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{uint64(1747945619), uint64(1747983448), "%Iron Award%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
-			},
-			expectedErr: nil,
-		},
+		// {
+		// 	name:        "body.message equals to empty string",
+		// 	requestType: qbtypes.RequestTypeRaw,
+		// 	query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+		// 		Signal: telemetrytypes.SignalLogs,
+		// 		Filter: &qbtypes.Filter{Expression: "body.message = ''"},
+		// 		Limit:  10,
+		// 	},
+		// 	expected: qbtypes.Statement{
+		// 		Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND body_v2.message = ? AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+		// 		Args:  []any{uint64(1747945619), uint64(1747983448), "", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+		// 	},
+		// 	expectedErr: nil,
+		// },
+		// {
+		// 	name:        "body.message equals to 'Iron Award'",
+		// 	requestType: qbtypes.RequestTypeRaw,
+		// 	query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+		// 		Signal: telemetrytypes.SignalLogs,
+		// 		Filter: &qbtypes.Filter{Expression: "body.message = 'Iron Award'"},
+		// 		Limit:  10,
+		// 	},
+		// 	expected: qbtypes.Statement{
+		// 		Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND body_v2.message = ? AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+		// 		Args:  []any{uint64(1747945619), uint64(1747983448), "Iron Award", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+		// 	},
+		// 	expectedErr: nil,
+		// },
+		// {
+		// 	name:        "message contains 'Iron Award'",
+		// 	requestType: qbtypes.RequestTypeRaw,
+		// 	query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+		// 		Signal: telemetrytypes.SignalLogs,
+		// 		Filter: &qbtypes.Filter{Expression: "message Contains 'Iron Award'"},
+		// 		Limit:  10,
+		// 	},
+		// 	expected: qbtypes.Statement{
+		// 		Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND LOWER(body_v2.message) LIKE LOWER(?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
+		// 		Args:  []any{uint64(1747945619), uint64(1747983448), "%Iron Award%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+		// 	},
+		// 	expectedErr: nil,
+		// },
 	}
 
 	for _, c := range cases {
@@ -804,8 +799,7 @@ func TestStatementBuilderListQueryBodyMessage(t *testing.T) {
 }
 
 func buildTestTelemetryMetadataStore(t *testing.T, promotedPaths ...string) *telemetrytypestest.MockMetadataStore {
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
-
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(IntrinsicFields)
 	types, _ := telemetrytypes.TestJSONTypeSet()
 	for path, jsonTypes := range types {
 		promoted := false
@@ -842,11 +836,14 @@ func buildJSONTestStatementBuilder(t *testing.T, promotedPaths ...string) *logQu
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
 
+	resourceFilterFM := resourcefilter.NewFieldMapper()
+	resourceFilterCB := resourcefilter.NewConditionBuilder(resourceFilterFM)
+
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 	resourceFilterStmtBuilder := resourcefilter.NewLogResourceFilterStatementBuilder(
 		instrumentationtest.New().ToProviderSettings(),
-		fm,
-		cb,
+		resourceFilterFM,
+		resourceFilterCB,
 		mockMetadataStore,
 		DefaultFullTextColumn,
 		GetBodyJSONKey,
@@ -866,19 +863,6 @@ func buildJSONTestStatementBuilder(t *testing.T, promotedPaths ...string) *logQu
 	return statementBuilder
 }
 
-func testAddIndexedPaths(t *testing.T, statementBuilder *logQueryStatementBuilder, telemetryFieldKeys ...*telemetrytypes.TelemetryFieldKey) {
-	mockMetadataStore := statementBuilder.metadataStore.(*telemetrytypestest.MockMetadataStore)
-	for _, key := range telemetryFieldKeys {
-		if strings.Contains(key.Name, telemetrytypes.ArraySep) || strings.Contains(key.Name, telemetrytypes.ArrayAnyIndex) {
-			t.Fatalf("array paths are not supported: %s", key.Name)
-		}
-
-		for _, storedKey := range mockMetadataStore.KeysMap[key.Name] {
-			storedKey.Indexes = append(storedKey.Indexes, key.Indexes...)
-		}
-	}
-}
-
 func jsonQueryTestUtil(_ *testing.T) (func(), func()) {
 	querybuilder.BodyJSONQueryEnabled = true
 	base := telemetrytypes.TelemetryFieldKey{
@@ -897,6 +881,8 @@ func jsonQueryTestUtil(_ *testing.T) (func(), func()) {
 		DefaultFullTextColumn = &base
 		IntrinsicFields["body"] = base
 		delete(IntrinsicFields, MessageSubColumn)
+		delete(IntrinsicFields, MessageBodyField)
+		delete(logsV2Columns, MessageSubColumn)
 	}
 
 	return enable, disable

pkg/telemetrylogs/stmt_builder_test.go

@@ -18,7 +18,7 @@ import (
 func resourceFilterStmtBuilder() qbtypes.StatementBuilder[qbtypes.LogAggregation] {
 	fm := resourcefilter.NewFieldMapper()
 	cb := resourcefilter.NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	keysMap := buildCompleteFieldKeyMap()
 	for _, keys := range keysMap {
 		for _, key := range keys {
@@ -196,7 +196,7 @@ func TestStatementBuilderTimeSeries(t *testing.T) {
 		},
 	}
 
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
@@ -316,7 +316,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
 		},
 	}
 
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
@@ -456,7 +456,7 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
 		},
 	}
 
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
@@ -532,7 +532,7 @@ func TestStatementBuilderTimeSeriesBodyGroupBy(t *testing.T) {
 		},
 	}
 
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
@@ -627,7 +627,7 @@ func TestStatementBuilderListQueryServiceCollision(t *testing.T) {
 		},
 	}
 
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMapCollision()
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
@@ -850,7 +850,7 @@ func TestAdjustKey(t *testing.T) {
 	}
 
 	fm := NewFieldMapper()
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMapCollision()
 	cb := NewConditionBuilder(fm)
 
@@ -1005,9 +1005,7 @@ func TestStmtBuilderBodyField(t *testing.T) {
 				disable()
 			}
 			// build the key map after enabling/disabling body JSON query
-			mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
-			mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
-
+			mockMetadataStore := telemetrytypestest.NewMockMetadataStore(IntrinsicFields)
 			aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 			resourceFilterStmtBuilder := resourceFilterStmtBuilder()
 			statementBuilder := NewLogQueryStatementBuilder(

pkg/telemetrylogs/test_data.go

@@ -931,6 +931,13 @@ func buildCompleteFieldKeyMap() map[string][]*telemetrytypes.TelemetryFieldKey {
 				Materialized:  true,
 			},
 		},
+		"body": {
+			{
+				Name:          "body",
+				FieldContext:  telemetrytypes.FieldContextLog,
+				FieldDataType: telemetrytypes.FieldDataTypeString,
+			},
+		},
 	}
 
 	for _, keys := range keysMap {
@@ -939,10 +946,6 @@ func buildCompleteFieldKeyMap() map[string][]*telemetrytypes.TelemetryFieldKey {
 		}
 	}
 
-	// add intrinsic fields to the map
-	for fieldName, key := range IntrinsicFields {
-		keysMap[fieldName] = append(keysMap[fieldName], &key)
-	}
 	return keysMap
 }
 

pkg/telemetrymetadata/body_json_metadata.go

@@ -54,6 +54,7 @@ func (t *telemetryMetaStore) fetchBodyJSONPaths(ctx context.Context,
 		instrumentationtypes.CodeNamespace:    "metadata",
 		instrumentationtypes.CodeFunctionName: "fetchBodyJSONPaths",
 	})
+
 	query, args, limit := buildGetBodyJSONPathsQuery(fieldKeySelectors)
 	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
 	if err != nil {
@@ -184,6 +185,9 @@ func buildGetBodyJSONPathsQuery(fieldKeySelectors []*telemetrytypes.FieldKeySele
 		limit += fieldKeySelector.Limit
 	}
 	sb.Where(sb.Or(orClauses...))
+	// mesasge field is skipped; since it is a type hint and is handled by the field mapper
+	// TODO(Piyush): If typehints increases in future, use aftership parser to skip type hints here
+	sb.Where(sb.NotEqual("path", telemetrylogs.MessageBodyField))
 
 	// Group by path to get unique paths with aggregated types
 	sb.GroupBy("path")

pkg/telemetrymetadata/metadata.go

@@ -527,7 +527,7 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 		if err != nil {
 			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
 		}
-		key, ok := setOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
+		key, ok := setOfKeys[fieldContext.StringValue()+"."+name+":"+fieldDataType.StringValue()]
 
 		// if there is no materialised column, create a key with the field context and data type
 		if !ok {
@@ -540,7 +540,7 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 		}
 
 		mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
-		setOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()] = key
+		setOfKeys[key.Text()] = key
 	}
 
 	if rows.Err() != nil {
@@ -566,12 +566,15 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 
 		if found {
 			if field, exists := telemetrylogs.IntrinsicFields[key]; exists {
-				if _, added := setOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
-					mapOfKeys[field.Name] = append(mapOfKeys[field.Name], &field)
-					// intrinsic field can be a logical field mapped to a different physical location
-					mapOfKeys[key] = append(mapOfKeys[key], &field)
+				// Register by physical name only once and only when it differs from the
+				// logical key — if they are the same, the always-register below covers it.
+				if _, added := setOfKeys[field.Text()]; !added {
+					mapOfKeys[field.Text()] = append(mapOfKeys[field.Text()], &field)
 				}
-				continue
+				// Always register the logical key so that every alias in IntrinsicFields
+				// (e.g. "message", "body_v2.message") independently resolves to the same
+				// physical field in the keys map.
+				mapOfKeys[key] = append(mapOfKeys[key], &field)
 			}
 		}
 	}

pkg/telemetrymeter/stmt_builder_test.go

@@ -159,7 +159,7 @@ func TestStatementBuilder(t *testing.T) {
 
 	fm := telemetrymetrics.NewFieldMapper()
 	cb := telemetrymetrics.NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	keys, err := telemetrytypestest.LoadFieldKeysFromJSON("testdata/keys_map.json")
 	if err != nil {
 		t.Fatalf("failed to load field keys: %v", err)

pkg/telemetrymetrics/stmt_builder_test.go

@@ -221,7 +221,7 @@ func TestStatementBuilder(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	keys, err := telemetrytypestest.LoadFieldKeysFromJSON("testdata/keys_map.json")
 	if err != nil {
 		t.Fatalf("failed to load field keys: %v", err)

pkg/telemetrytraces/stmt_builder_test.go

@@ -18,7 +18,7 @@ import (
 func resourceFilterStmtBuilder() qbtypes.StatementBuilder[qbtypes.TraceAggregation] {
 	fm := resourcefilter.NewFieldMapper()
 	cb := resourcefilter.NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 
 	return resourcefilter.NewTraceResourceFilterStatementBuilder(
@@ -368,7 +368,7 @@ func TestStatementBuilder(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 
@@ -664,7 +664,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 
@@ -771,7 +771,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
 		t.Run(c.name, func(t *testing.T) {
 			fm := NewFieldMapper()
 			cb := NewConditionBuilder(fm)
-			mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+			mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 			mockMetadataStore.KeysMap = c.keysMap
 			if mockMetadataStore.KeysMap == nil {
 				mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
@@ -927,7 +927,7 @@ func TestStatementBuilderTraceQuery(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 
@@ -1145,7 +1145,7 @@ func TestAdjustKey(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 	resourceFilterStmtBuilder := resourceFilterStmtBuilder()
 
@@ -1420,7 +1420,7 @@ func TestAdjustKeys(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 	resourceFilterStmtBuilder := resourceFilterStmtBuilder()
 

pkg/telemetrytraces/trace_operator_cte_builder_test.go

@@ -388,7 +388,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 
@@ -504,7 +504,7 @@ func TestTraceOperatorStatementBuilderErrors(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
 

pkg/telemetrytraces/trace_time_range_test.go

@@ -19,7 +19,7 @@ func TestTraceTimeRangeOptimization(t *testing.T) {
 
 	fm := NewFieldMapper()
 	cb := NewConditionBuilder(fm)
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
+	mockMetadataStore := telemetrytypestest.NewMockMetadataStore(nil)
 
 	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
 	mockMetadataStore.KeysMap["trace_id"] = []*telemetrytypes.TelemetryFieldKey{{

pkg/types/telemetrytypes/field.go

@@ -40,7 +40,7 @@ type TelemetryFieldKey struct {
 	JSONDataType *JSONDataType       `json:"-"`
 	JSONPlan     JSONAccessPlan      `json:"-"`
 	Indexes      []JSONDataTypeIndex `json:"-"`
-	Materialized bool                `json:"-"` // refers to promoted in case of body.... fields
+	Materialized bool                `json:"-"` // refers to type hint in case of JSON column fields
 }
 
 func (f *TelemetryFieldKey) KeyNameContainsArray() bool {

pkg/types/telemetrytypes/telemetrytypestest/metadata_store.go

@@ -21,10 +21,20 @@ type MockMetadataStore struct {
 	PromotedPathsMap   map[string]bool
 	LogsJSONIndexesMap map[string][]schemamigrator.Index
 	LookupKeysMap      map[telemetrytypes.MetricMetadataLookupKey]int64
+	// StaticFields holds signal-specific intrinsic field definitions (e.g. telemetrylogs.IntrinsicFields).
+	// It is injected into GetKeys / GetKey results, mirroring what the real metadata store does when
+	// it reads telemetrylogs.IntrinsicFields directly. Callers pass their package's IntrinsicFields
+	// map at construction time to avoid a circular import.
+	StaticFields map[string]telemetrytypes.TelemetryFieldKey
 }
 
-// NewMockMetadataStore creates a new instance of MockMetadataStore with initialized maps
-func NewMockMetadataStore() *MockMetadataStore {
+// NewMockMetadataStore creates a new instance of MockMetadataStore with initialized maps.
+// Pass the signal-specific intrinsic fields (e.g. telemetrylogs.IntrinsicFields) so the mock
+// mirrors what the real metadata store does when injecting those definitions into key results.
+func NewMockMetadataStore(intrinsicFields map[string]telemetrytypes.TelemetryFieldKey) *MockMetadataStore {
+	if intrinsicFields == nil {
+		intrinsicFields = make(map[string]telemetrytypes.TelemetryFieldKey)
+	}
 	return &MockMetadataStore{
 		KeysMap:            make(map[string][]*telemetrytypes.TelemetryFieldKey),
 		RelatedValuesMap:   make(map[string][]string),
@@ -34,12 +44,13 @@ func NewMockMetadataStore() *MockMetadataStore {
 		PromotedPathsMap:   make(map[string]bool),
 		LogsJSONIndexesMap: make(map[string][]schemamigrator.Index),
 		LookupKeysMap:      make(map[telemetrytypes.MetricMetadataLookupKey]int64),
+		StaticFields:       intrinsicFields,
 	}
 }
 
 // GetKeys returns a map of field keys types.TelemetryFieldKey by name
 func (m *MockMetadataStore) GetKeys(ctx context.Context, fieldKeySelector *telemetrytypes.FieldKeySelector) (map[string][]*telemetrytypes.TelemetryFieldKey, bool, error) {
-
+	setOfKeys := make(map[string]*telemetrytypes.TelemetryFieldKey)
 	result := make(map[string][]*telemetrytypes.TelemetryFieldKey)
 
 	// If selector is nil, return all keys
@@ -47,22 +58,42 @@ func (m *MockMetadataStore) GetKeys(ctx context.Context, fieldKeySelector *telem
 		return m.KeysMap, true, nil
 	}
 
-	// Apply selector logic
+	// Apply selector logic from KeysMap
 	for name, keys := range m.KeysMap {
-		// Check if name matches
 		if matchesName(fieldKeySelector, name) {
-			filteredKeys := []*telemetrytypes.TelemetryFieldKey{}
 			for _, key := range keys {
 				if matchesKey(fieldKeySelector, key) {
-					filteredKeys = append(filteredKeys, key)
+					if _, exists := setOfKeys[key.Text()]; !exists {
+						result[name] = append(result[name], key)
+						setOfKeys[key.Text()] = key
+					}
 				}
 			}
-			if len(filteredKeys) > 0 {
-				result[name] = filteredKeys
-			}
 		}
 	}
 
+	// Inject StaticFields (e.g. IntrinsicFields), mirroring the real metadata store.
+	// StaticFields take precedence over KeysMap entries for the same logical key.
+	// Each logical key always gets its own entry (so "message" and "body.message" both
+	// resolve to the IntrinsicField independently). The physical name is registered
+	// only once to avoid duplicate entries in that slot.
+	for key, field := range m.StaticFields {
+		if !matchesName(fieldKeySelector, key) {
+			continue
+		}
+
+		// Register by physical name only once and only when it differs from the
+		// logical key — if they are the same, the always-register below covers it.
+		if _, exists := setOfKeys[field.Text()]; !exists {
+			result[field.Text()] = append(result[field.Text()], &field)
+			setOfKeys[field.Text()] = &field
+		}
+		// Always register the logical key so that every alias in IntrinsicFields
+		// (e.g. "message", "body_v2.message") independently resolves to the same
+		// physical field in the keys map.
+		result[key] = append(result[key], &field)
+	}
+
 	return result, true, nil
 }
 
@@ -109,7 +140,7 @@ func (m *MockMetadataStore) GetKey(ctx context.Context, fieldKeySelector *teleme
 
 	result := []*telemetrytypes.TelemetryFieldKey{}
 
-	// Find keys matching the selector
+	// Find keys matching the selector from KeysMap
 	for name, keys := range m.KeysMap {
 		if matchesName(fieldKeySelector, name) {
 			for _, key := range keys {
@@ -120,6 +151,14 @@ func (m *MockMetadataStore) GetKey(ctx context.Context, fieldKeySelector *teleme
 		}
 	}
 
+	// Add matching StaticFields (e.g. IntrinsicFields), same as the real metadata store does
+	for key, field := range m.StaticFields {
+		if fieldKeySelector.Name == "" || strings.Contains(key, fieldKeySelector.Name) {
+			fieldCopy := field
+			result = append(result, &fieldCopy)
+		}
+	}
+
 	return result, nil
 }
 
@@ -291,7 +330,7 @@ func (m *MockMetadataStore) FetchTemporalityMulti(ctx context.Context, queryTime
 	return result, nil
 }
 
-// FetchTemporalityMulti fetches the temporality for multiple metrics
+// FetchTemporalityAndTypeMulti fetches the temporality and type for multiple metrics
 func (m *MockMetadataStore) FetchTemporalityAndTypeMulti(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, error) {
 	temporalities := make(map[string]metrictypes.Temporality)
 	types := make(map[string]metrictypes.Type)

pkg/types/telemetrytypes/test_data.go

@@ -64,8 +64,7 @@ func TestJSONTypeSet() (map[string][]JSONDataType, MetadataStore) {
 		"interests[].entities[].reviews[].entries[].metadata[].positions[].duration": {Int64, Float64},
 		"interests[].entities[].reviews[].entries[].metadata[].positions[].unit":     {String},
 		"interests[].entities[].reviews[].entries[].metadata[].positions[].ratings":  {ArrayInt64, ArrayString},
-		"message": {String},
-		"tags":    {ArrayString},
+		"tags": {ArrayString},
 	}
 
 	return types, nil

tests/integration/conftest.py

@@ -1,5 +1,19 @@
+import json
+import os
+import tempfile
+
 import pytest
 
+if not os.environ.get("DOCKER_CONFIG"):
+    os.environ["DOCKER_CONFIG"] = tempfile.mkdtemp(prefix="docker-config-")
+os.environ.setdefault("DOCKER_CREDENTIAL_STORE", "")
+os.environ.setdefault("DOCKER_CREDENTIAL_HELPER", "")
+
+docker_config_path = os.path.join(os.environ["DOCKER_CONFIG"], "config.json")
+if not os.path.exists(docker_config_path):
+    with open(docker_config_path, "w", encoding="utf-8") as config_file:
+        json.dump({"auths": {}, "credsStore": ""}, config_file)
+
 pytest_plugins = [
     "fixtures.auth",
     "fixtures.clickhouse",
@@ -22,6 +36,7 @@ pytest_plugins = [
     "fixtures.notification_channel",
     "fixtures.alerts",
     "fixtures.cloudintegrations",
+    "fixtures.jsontypeexporter",
 ]
 
 
@@ -59,7 +74,7 @@ def pytest_addoption(parser: pytest.Parser):
     parser.addoption(
         "--clickhouse-version",
         action="store",
-        default="25.5.6",
+        default="25.8.6",
         help="clickhouse version",
     )
     parser.addoption(
@@ -74,3 +89,5 @@ def pytest_addoption(parser: pytest.Parser):
         default="v0.129.7",
         help="schema migrator version",
     )
+
+

tests/integration/fixtures/jsontypeexporter.py

@@ -0,0 +1,421 @@
+"""
+Simpler version of jsontypeexporter for test fixtures.
+This exports JSON type metadata to the path_types table by parsing JSON bodies
+and extracting all paths with their types, similar to how the real jsontypeexporter works.
+"""
+import datetime
+import json
+from abc import ABC
+from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, List, Optional, Set, Union
+
+import numpy as np
+import pytest
+
+from fixtures import types
+
+if TYPE_CHECKING:
+    from fixtures.logs import Logs
+
+
+class JSONPathType(ABC):
+    """Represents a JSON path with its type information"""
+    path: str
+    type: str
+    last_seen: np.uint64
+
+    def __init__(
+        self,
+        path: str,
+        type: str,  # pylint: disable=redefined-builtin
+        last_seen: Optional[datetime.datetime] = None,
+    ) -> None:
+        self.path = path
+        self.type = type
+        if last_seen is None:
+            last_seen = datetime.datetime.now()
+        self.last_seen = np.uint64(int(last_seen.timestamp() * 1e9))
+
+    def np_arr(self) -> np.array:
+        """Return path type data as numpy array for database insertion"""
+        return np.array([self.path, self.type, self.last_seen])
+
+
+# Constants matching jsontypeexporter
+ARRAY_SEPARATOR = "[]."  # Used in paths like "education[].name"
+ARRAY_SUFFIX = "[]"      # Used when traversing into array element objects
+
+
+def _infer_array_type_from_type_strings(types: List[str]) -> Optional[str]:
+    """
+    Infer array type from a list of pre-classified type strings.
+    Matches jsontypeexporter's inferArrayMask logic (v0.144.2+).
+
+    Type strings are: "JSON", "String", "Bool", "Float64", "Int64"
+
+    SuperTyping rules (matching Go inferArrayMask):
+    - JSON alone → Array(JSON)
+    - JSON + any primitive → Array(Dynamic)
+    - String alone → Array(Nullable(String)); String + other → Array(Dynamic)
+    - Float64 wins over Int64 and Bool
+    - Int64 wins over Bool
+    - Bool alone → Array(Nullable(Bool))
+    """
+    if len(types) == 0:
+        return None
+
+    unique = set(types)
+
+    has_json = "JSON" in unique
+    # hasPrimitive mirrors Go: (hasJSON && len(unique) > 1) || (!hasJSON && len(unique) > 0)
+    has_primitive = (has_json and len(unique) > 1) or (not has_json and len(unique) > 0)
+
+    if has_json:
+        if not has_primitive:
+            return "Array(JSON)"
+        return "Array(Dynamic)"
+
+    # ---- Primitive Type Resolution (Float > Int > Bool) ----
+    if "String" in unique:
+        if len(unique) > 1:
+            return "Array(Dynamic)"
+        return "Array(Nullable(String))"
+
+    if "Float64" in unique:
+        return "Array(Nullable(Float64))"
+    if "Int64" in unique:
+        return "Array(Nullable(Int64))"
+    if "Bool" in unique:
+        return "Array(Nullable(Bool))"
+
+    return "Array(Dynamic)"
+
+
+def _infer_array_type(elements: List[Any]) -> Optional[str]:
+    """
+    Infer array type from raw Python list elements.
+    Classifies each element then delegates to _infer_array_type_from_type_strings.
+    """
+    if len(elements) == 0:
+        return None
+
+    types = []
+    for elem in elements:
+        if elem is None:
+            continue
+        if isinstance(elem, dict):
+            types.append("JSON")
+        elif isinstance(elem, str):
+            types.append("String")
+        elif isinstance(elem, bool):  # must be before int (bool is subclass of int)
+            types.append("Bool")
+        elif isinstance(elem, float):
+            types.append("Float64")
+        elif isinstance(elem, int):
+            types.append("Int64")
+
+    return _infer_array_type_from_type_strings(types)
+
+
+def _python_type_to_clickhouse_type(value: Any) -> str:
+    """
+    Convert Python type to ClickHouse JSON type string.
+    Maps Python types to ClickHouse JSON data types.
+    """
+    if value is None:
+        return "String"  # Default for null values
+    
+    if isinstance(value, bool):
+        return "Bool"
+    elif isinstance(value, int):
+        return "Int64"
+    elif isinstance(value, float):
+        return "Float64"
+    elif isinstance(value, str):
+        return "String"
+    elif isinstance(value, list):
+        # Use the sophisticated array type inference
+        array_type = _infer_array_type(value)
+        return array_type if array_type else "Array(Dynamic)"
+    elif isinstance(value, dict):
+        return "JSON"
+    else:
+        return "String"  # Default fallback
+
+
+def _extract_json_paths(
+    obj: Any,
+    current_path: str = "",
+    path_types: Optional[Dict[str, Set[str]]] = None,
+    level: int = 0,
+) -> Dict[str, Set[str]]:
+    """
+    Recursively extract all paths and their types from a JSON object.
+    Matches jsontypeexporter's analyzePValue logic.
+    
+    Args:
+        obj: The JSON object to traverse
+        current_path: Current path being built (e.g., "user.name")
+        path_types: Dictionary mapping paths to sets of types found
+        level: Current nesting level (for depth limiting)
+    
+    Returns:
+        Dictionary mapping paths to sets of type strings
+    """
+    if path_types is None:
+        path_types = {}
+
+    if obj is None:
+        if current_path:
+            if current_path not in path_types:
+                path_types[current_path] = set()
+            path_types[current_path].add("String")  # Null defaults to String
+        return path_types
+    
+    if isinstance(obj, dict):
+        # For objects, add the object itself and recurse into keys
+        if current_path:
+            if current_path not in path_types:
+                path_types[current_path] = set()
+            path_types[current_path].add("JSON")
+        
+        for key, value in obj.items():
+            # Build the path for this key
+            if current_path:
+                new_path = f"{current_path}.{key}"
+            else:
+                new_path = key
+            
+            # Recurse into the value
+            _extract_json_paths(value, new_path, path_types, level + 1)
+    
+    elif isinstance(obj, list):
+        # Skip empty arrays
+        if len(obj) == 0:
+            return path_types
+        
+        # Collect types from array elements (matching Go: types := make([]pcommon.ValueType, 0, s.Len()))
+        types = []
+        
+        for item in obj:
+            if isinstance(item, dict):
+                # When traversing into array element objects, use ArraySuffix ([])
+                # This matches: prefix+ArraySuffix in the Go code
+                # Example: if current_path is "education", we use "education[]" to traverse into objects
+                array_prefix = current_path + ARRAY_SUFFIX if current_path else ""
+                for key, value in item.items():
+                    if array_prefix:
+                        # Use array separator: education[].name
+                        array_path = f"{array_prefix}.{key}"
+                    else:
+                        array_path = key
+                    # Recurse without increasing level (matching Go behavior)
+                    _extract_json_paths(value, array_path, path_types, level)
+                types.append("JSON")
+            elif isinstance(item, list):
+                # Arrays inside arrays are not supported - skip the whole path
+                # Matching Go: e.logger.Error("arrays inside arrays are not supported!", ...); return nil
+                return path_types
+            elif isinstance(item, str):
+                types.append("String")
+            elif isinstance(item, bool):
+                types.append("Bool")
+            elif isinstance(item, float):
+                types.append("Float64")
+            elif isinstance(item, int):
+                types.append("Int64")
+        
+        # Infer array type from collected types (matching Go: if mask := inferArrayMask(types); mask != 0)
+        if len(types) > 0:
+            array_type = _infer_array_type_from_type_strings(types)
+            if array_type and current_path:
+                if current_path not in path_types:
+                    path_types[current_path] = set()
+                path_types[current_path].add(array_type)
+    
+    else:
+        # Primitive value (string, number, bool)
+        if current_path:
+            if current_path not in path_types:
+                path_types[current_path] = set()
+            obj_type = _python_type_to_clickhouse_type(obj)
+            path_types[current_path].add(obj_type)
+    
+    return path_types
+
+
+def _parse_json_bodies_and_extract_paths(
+    json_bodies: List[str],
+    timestamp: Optional[datetime.datetime] = None,
+) -> List[JSONPathType]:
+    """
+    Parse JSON bodies and extract all paths with their types.
+    This mimics the behavior of jsontypeexporter.
+    
+    Args:
+        json_bodies: List of JSON body strings to parse
+        timestamp: Timestamp to use for last_seen (defaults to now)
+    
+    Returns:
+        List of JSONPathType objects with all discovered paths and types
+    """
+    if timestamp is None:
+        timestamp = datetime.datetime.now()
+    
+    # Aggregate all paths and their types across all JSON bodies
+    all_path_types: Dict[str, Set[str]] = {}
+    
+    for json_body in json_bodies:
+        try:
+            parsed = json.loads(json_body)
+            _extract_json_paths(parsed, "", all_path_types, level=0)
+        except (json.JSONDecodeError, TypeError):
+            # Skip invalid JSON
+            continue
+    
+    # Convert to list of JSONPathType objects
+    # Each path can have multiple types, so we create one JSONPathType per type
+    path_type_objects: List[JSONPathType] = []
+    for path, types_set in all_path_types.items():
+        for type_str in types_set:
+            path_type_objects.append(
+                JSONPathType(path=path, type=type_str, last_seen=timestamp)
+            )
+    
+    return path_type_objects
+
+
+@pytest.fixture(name="export_json_types", scope="function")
+def export_json_types(
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,  # To access migrator fixture
+) -> Generator[Callable[[Union[List[JSONPathType], List[str], List[Any]]], None], Any, None]:
+    """
+    Fixture for exporting JSON type metadata to the path_types table.
+    This is a simpler version of jsontypeexporter for test fixtures.
+    
+    The function can accept:
+    1. List of JSONPathType objects (manual specification)
+    2. List of JSON body strings (auto-extract paths)
+    3. List of Logs objects (extract from body_json field)
+    
+    Usage examples:
+        # Manual specification
+        export_json_types([
+            JSONPathType(path="user.name", type="String"),
+            JSONPathType(path="user.age", type="Int64"),
+        ])
+        
+        # Auto-extract from JSON strings
+        export_json_types([
+            '{"user": {"name": "alice", "age": 25}}',
+            '{"user": {"name": "bob", "age": 30}}',
+        ])
+        
+        # Auto-extract from Logs objects
+        export_json_types(logs_list)
+    """
+    # Ensure migrator has run to create the table
+    try:
+        request.getfixturevalue("migrator")
+    except Exception:
+        # If migrator fixture is not available, that's okay - table might already exist
+        pass
+    
+    def _export_json_types(
+        data: Union[List[JSONPathType], List[str], List[Any]],  # List[Logs] but avoiding circular import
+    ) -> None:
+        """
+        Export JSON type metadata to signoz_metadata.distributed_json_path_types table.
+        This table stores path and type information for body JSON fields.
+        """
+        path_types: List[JSONPathType] = []
+        
+        if len(data) == 0:
+            return
+        
+        # Determine input type and convert to JSONPathType list
+        first_item = data[0]
+        
+        if isinstance(first_item, JSONPathType):
+            # Already JSONPathType objects
+            path_types = data  # type: ignore
+        elif isinstance(first_item, str):
+            # List of JSON strings - parse and extract paths
+            path_types = _parse_json_bodies_and_extract_paths(data)  # type: ignore
+        else:
+            # Assume it's a list of Logs objects - extract body_v2
+            json_bodies: List[str] = []
+            for log in data:  # type: ignore
+                # Try to get body_v2 attribute
+                if hasattr(log, "body_v2") and log.body_v2:
+                    json_bodies.append(log.body_v2)
+                elif hasattr(log, "body") and log.body:
+                    # Fallback to body if body_v2 not available
+                    try:
+                        # Try to parse as JSON
+                        json.loads(log.body)
+                        json_bodies.append(log.body)
+                    except (json.JSONDecodeError, TypeError):
+                        pass
+            
+            if json_bodies:
+                path_types = _parse_json_bodies_and_extract_paths(json_bodies)
+        
+        if len(path_types) == 0:
+            return
+
+        clickhouse.conn.insert(
+            database="signoz_metadata",
+            table="distributed_json_path_types",
+            data=[path_type.np_arr() for path_type in path_types],
+            column_names=[
+                "path",
+                "type",
+                "last_seen",
+            ],
+        )
+
+    yield _export_json_types
+
+    # Cleanup - truncate the local table after tests (following pattern from logs fixture)
+    clickhouse.conn.query(
+        f"TRUNCATE TABLE signoz_metadata.json_path_types ON CLUSTER '{clickhouse.env['SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER']}' SYNC"
+    )
+
+
+@pytest.fixture(name="export_promoted_paths", scope="function")
+def export_promoted_paths(
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,  # To access migrator fixture
+) -> Generator[Callable[[List[str]], None], Any, None]:
+    """
+    Fixture for exporting promoted JSON paths to the promoted paths table.
+    """
+    # Ensure migrator has run to create the table
+    try:
+        request.getfixturevalue("migrator")
+    except Exception:
+        # If migrator fixture is not available, that's okay - table might already exist
+        pass
+
+    def _export_promoted_paths(paths: List[str]) -> None:
+        if len(paths) == 0:
+            return
+
+        now_ms = int(datetime.datetime.now().timestamp() * 1000)
+        rows = [(path, now_ms) for path in paths]
+        clickhouse.conn.insert(
+            database="signoz_metadata",
+            table="distributed_json_promoted_paths",
+            data=rows,
+            column_names=[
+                "path",
+                "created_at",
+            ],
+        )
+
+    yield _export_promoted_paths
+
+    clickhouse.conn.query(
+        f"TRUNCATE TABLE signoz_metadata.json_promoted_paths ON CLUSTER '{clickhouse.env['SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER']}' SYNC"
+    )

tests/integration/fixtures/logs.py

@@ -100,6 +100,8 @@ class Logs(ABC):
     severity_text: str
     severity_number: np.uint8
     body: str
+    body_v2: str
+    body_promoted: str
     attributes_string: dict[str, str]
     attributes_number: dict[str, np.float64]
     attributes_bool: dict[str, bool]
@@ -119,6 +121,8 @@ class Logs(ABC):
         resources: dict[str, Any] = {},
         attributes: dict[str, Any] = {},
         body: str = "default body",
+        body_v2: Optional[str] = None,
+        body_promoted: Optional[str] = None,
         severity_text: str = "INFO",
         trace_id: str = "",
         span_id: str = "",
@@ -163,6 +167,33 @@ class Logs(ABC):
         # Set body
         self.body = body
 
+        # Set body_v2 - if body is JSON, parse and stringify it, otherwise use empty string
+        # ClickHouse accepts String input for JSON column
+        if body_v2 is not None:
+            self.body_v2 = body_v2
+        else:
+            # Try to parse body as JSON; if successful use it directly,
+            # otherwise wrap as {"message": body} matching the normalize operator behavior.
+            try:
+                json.loads(body)
+                self.body_v2 = body
+            except (json.JSONDecodeError, TypeError):
+                self.body_v2 = json.dumps({"message": body})
+
+        # Set body_promoted - must be valid JSON
+        # Tests will explicitly pass promoted column's content, but we validate it
+        if body_promoted is not None:
+            # Validate that it's valid JSON
+            try:
+                json.loads(body_promoted)
+                self.body_promoted = body_promoted
+            except (json.JSONDecodeError, TypeError):
+                # If invalid, default to empty JSON object
+                self.body_promoted = "{}"
+        else:
+            # Default to empty JSON object (valid JSON)
+            self.body_promoted = "{}"
+
         # Process resources and attributes
         self.resources_string = {k: str(v) for k, v in resources.items()}
         for k, v in self.resources_string.items():
@@ -319,6 +350,8 @@ class Logs(ABC):
                 self.severity_text,
                 self.severity_number,
                 self.body,
+                self.body_v2,
+                self.body_promoted,
                 self.attributes_string,
                 self.attributes_number,
                 self.attributes_bool,
@@ -463,6 +496,8 @@ def insert_logs(
                 "severity_text",
                 "severity_number",
                 "body",
+                "body_v2",
+                "body_promoted",
                 "attributes_string",
                 "attributes_number",
                 "attributes_bool",

tests/integration/fixtures/migrator.py

@@ -20,12 +20,14 @@ def migrator(
     """
 
     def create() -> None:
-        version = request.config.getoption("--schema-migrator-version")
+        # Hardcode version for new QB tests
+        version = "v0.144.3-rc.1"
         client = docker.from_env()
 
         container = client.containers.run(
             image=f"signoz/signoz-schema-migrator:{version}",
             command=f"sync --replication=true --cluster-name=cluster --up= --dsn={clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN"]}",
+            environment={"ENABLE_LOGS_MIGRATIONS_V2": "1"},
             detach=True,
             auto_remove=False,
             network=network.id,
@@ -44,6 +46,7 @@ def migrator(
         container = client.containers.run(
             image=f"signoz/signoz-schema-migrator:{version}",
             command=f"async --replication=true --cluster-name=cluster --up= --dsn={clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN"]}",
+            environment={"ENABLE_LOGS_MIGRATIONS_V2": "1"},
             detach=True,
             auto_remove=False,
             network=network.id,

tests/integration/fixtures/signoz.py

@@ -73,6 +73,7 @@ def signoz(  # pylint: disable=too-many-arguments,too-many-positional-arguments
                 "SIGNOZ_ALERTMANAGER_SIGNOZ_POLL__INTERVAL": "5s",
                 "SIGNOZ_ALERTMANAGER_SIGNOZ_ROUTE_GROUP__WAIT": "1s",
                 "SIGNOZ_ALERTMANAGER_SIGNOZ_ROUTE_GROUP__INTERVAL": "5s",
+                "BODY_JSON_QUERY_ENABLED": "true",
             }
             | sqlstore.env
             | clickhouse.env