test: fetching logs basic

.gitignore

@@ -49,6 +49,7 @@ ee/query-service/tests/test-deploy/data/
 # local data
 *.backup
 *.db
+**/db
 /deploy/docker/clickhouse-setup/data/
 /deploy/docker-swarm/clickhouse-setup/data/
 bin/

Makefile

@@ -72,6 +72,13 @@ devenv-up: devenv-clickhouse devenv-signoz-otel-collector ## Start both clickhou
 	@echo "   - ClickHouse: http://localhost:8123"
 	@echo "   - Signoz OTel Collector: grpc://localhost:4317, http://localhost:4318"
 
+.PHONY: devenv-clickhouse-clean
+devenv-clickhouse-clean: ## Clean all ClickHouse data from filesystem
+	@echo "Removing ClickHouse data..."
+	@rm -rf .devenv/docker/clickhouse/fs/tmp/var/lib/clickhouse/*
+	@rm -rf .devenv/docker/clickhouse/fs/tmp/zookeeper/*
+	@echo "ClickHouse data cleaned!"
+
 ##############################################################
 # go commands
 ##############################################################

pkg/query-service/app/logparsingpipeline/preview.go

@@ -146,7 +146,7 @@ func SignozLogsToPLogs(logs []model.SignozLog) []plog.Logs {
 		slRecord.SetSeverityText(log.SeverityText)
 		slRecord.SetSeverityNumber(plog.SeverityNumber(log.SeverityNumber))
 
-		slRecord.Body().SetStr(log.Body)
+		slRecord.Body().FromRaw(log.Body)
 
 		slAttribs := slRecord.Attributes()
 		for k, v := range log.Attributes_int64 {

pkg/query-service/app/logs/v4/query_builder.go

@@ -404,7 +404,7 @@ func buildLogsQuery(panelType v3.PanelType, start, end, step int64, mq *v3.Build
 	// if noop create the query and return
 	if mq.AggregateOperator == v3.AggregateOperatorNoOp {
 		// with noop any filter or different order by other than ts will use new table
-		sqlSelect := constants.LogsSQLSelectV2
+		sqlSelect := constants.LogsSQLSelectV3
 		queryTmpl := sqlSelect + "from signoz_logs.%s where %s%s order by %s"
 		query := fmt.Sprintf(queryTmpl, DISTRIBUTED_LOGS_V2, timeFilter, filterSubQuery, orderBy)
 		return query, nil
@@ -488,7 +488,7 @@ func buildLogsLiveTailQuery(mq *v3.BuilderQuery) (string, error) {
 	// the reader will add the timestamp and id filters
 	switch mq.AggregateOperator {
 	case v3.AggregateOperatorNoOp:
-		query := constants.LogsSQLSelectV2 + "from signoz_logs." + DISTRIBUTED_LOGS_V2 + " where "
+		query := constants.LogsSQLSelectV3 + "from signoz_logs." + DISTRIBUTED_LOGS_V2 + " where "
 		if len(filterSubQuery) > 0 {
 			query = query + filterSubQuery + " AND "
 		}

pkg/query-service/constants/constants.go

@@ -223,6 +223,15 @@ const (
 		"attributes_bool, " +
 		"resources_string, " +
 		"scope_string "
+	// LogsSQLSelectV3 is similar to V2 but does NOT select the legacy `body` string column.
+	// It instead selects JSON/body maps so the reader can compose the final body.
+	LogsSQLSelectV3 = "SELECT " +
+		"timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2, promoted, " +
+		"attributes_string, " +
+		"attributes_number, " +
+		"attributes_bool, " +
+		"resources_string, " +
+		"scope_string "
 	TracesExplorerViewSQLSelectWithSubQuery = "(SELECT traceID, durationNano, " +
 		"serviceName, name FROM %s.%s WHERE parentSpanID = '' AND %s ORDER BY durationNano DESC LIMIT 1 BY traceID"
 	TracesExplorerViewSQLSelectBeforeSubQuery = "SELECT subQuery.serviceName as `subQuery.serviceName`, subQuery.name as `subQuery.name`, count() AS " +

pkg/query-service/model/response.go

@@ -598,13 +598,43 @@ type SignozLog struct {
 	TraceFlags         uint32             `json:"trace_flags" ch:"trace_flags"`
 	SeverityText       string             `json:"severity_text" ch:"severity_text"`
 	SeverityNumber     uint8              `json:"severity_number" ch:"severity_number"`
-	Body               string             `json:"body" ch:"body"`
+	Body               any                `json:"body" ch:"body"`
+	BodyV2             map[string]any     `json:"-" ch:"body_v2"`
+	Promoted           map[string]any     `json:"-" ch:"promoted"`
 	Resources_string   map[string]string  `json:"resources_string" ch:"resources_string"`
 	Attributes_string  map[string]string  `json:"attributes_string" ch:"attributes_string"`
 	Attributes_int64   map[string]int64   `json:"attributes_int" ch:"attributes_int64"`
 	Attributes_float64 map[string]float64 `json:"attributes_float" ch:"attributes_float64"`
 	Attributes_bool    map[string]bool    `json:"attributes_bool" ch:"attributes_bool"`
 }
+
+// MarshalJSON implements json.Marshaler for SignozLog to allow composing
+// a structured body from BodyV2 and Promoted if present.
+func (l *SignozLog) MarshalJSON() ([]byte, error) {
+	type Alias SignozLog
+
+	// Create a shallow copy to avoid mutating the receiver
+	clone := *l
+
+	// If BodyV2/Promoted are present, merge them into Body for output
+	if (clone.Body == nil) && (len(clone.BodyV2) > 0 || len(clone.Promoted) > 0) {
+		merged := map[string]any{}
+		if clone.BodyV2 != nil {
+			for k, v := range clone.BodyV2 {
+				merged[k] = v
+			}
+		}
+		if clone.Promoted != nil {
+			for k, v := range clone.Promoted {
+				merged[k] = v
+			}
+		}
+		clone.Body = merged
+	}
+
+	return json.Marshal((*Alias)(&clone))
+}
+
 type GetLogsAggregatesResponse struct {
 	Items map[int64]LogsAggregatesResponseItem `json:"items"`
 }