o11y/pkg/telemetrylogs/stmt_builder_test.go

package telemetrylogs

import (
	"context"
	"regexp"
	"testing"
	"time"

	dsmock "github.com/hanzo-ds/mock"
	"github.com/hanzoai/o11y/pkg/errors"
	"github.com/hanzoai/o11y/pkg/flagger/flaggertest"
	"github.com/hanzoai/o11y/pkg/instrumentation/instrumentationtest"
	"github.com/hanzoai/o11y/pkg/querybuilder"
	"github.com/hanzoai/o11y/pkg/telemetrystore"
	"github.com/hanzoai/o11y/pkg/telemetrystore/telemetrystoretest"
	qbtypes "github.com/hanzoai/o11y/pkg/types/querybuildertypes/querybuildertypesv5"
	"github.com/hanzoai/o11y/pkg/types/telemetrytypes"
	"github.com/hanzoai/o11y/pkg/types/telemetrytypes/telemetrytypestest"
	"github.com/stretchr/testify/require"
)

type regexQueryMatcher struct{}

func (m *regexQueryMatcher) Match(expectedSQL, actualSQL string) error {
	re, err := regexp.Compile(expectedSQL)
	if err != nil {
		return err
	}
	if !re.MatchString(actualSQL) {
		return errors.NewInvalidInputf(errors.CodeInvalidInput, "expected query to match %s, got %s", expectedSQL, actualSQL)
	}
	return nil
}

func TestStatementBuilderTimeSeries(t *testing.T) {
	// Create a test release time
	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
	releaseTimeNano := uint64(releaseTime.UnixNano())

	cases := []struct {
		startTs     uint64
		endTs       uint64
		name        string
		requestType qbtypes.RequestType
		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
		expected    qbtypes.Statement
		expectedErr error
	}{
		{
			startTs:     releaseTimeNano + uint64(24*time.Hour.Nanoseconds()),
			endTs:       releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
			name:        "Time series with limit and count distinct on service.name",
			requestType: qbtypes.RequestTypeTimeSeries,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal:       telemetrytypes.SignalLogs,
				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
				Aggregations: []qbtypes.LogAggregation{
					{
						Expression: "count_distinct(service.name)",
					},
				},
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'cartservice'",
				},
				Limit: 10,
				GroupBy: []qbtypes.GroupByKey{
					{
						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
							Name: "service.name",
						},
					},
				},
			},
			expected: qbtypes.Statement{
				Query: "WITH __resource_filter AS (SELECT fingerprint FROM event.log_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), __limit_cte AS (SELECT toString(multiIf(notEmpty(service), service, NULL)) AS `service.name`, countDistinct(multiIf(notEmpty(service), service, NULL)) AS __result_0 FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? GROUP BY `service.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(time, INTERVAL 30 SECOND) AS ts, toString(multiIf(notEmpty(service), service, NULL)) AS `service.name`, countDistinct(multiIf(notEmpty(service), service, NULL)) AS __result_0 FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? AND (`service.name`) GLOBAL IN (SELECT `service.name` FROM __limit_cte) GROUP BY ts, `service.name`",
				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1705397400), uint64(1705485600), "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600), 10, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
			},
			expectedErr: nil,
		},
		{
			startTs:     releaseTimeNano - uint64(24*time.Hour.Nanoseconds()),
			endTs:       releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
			name:        "Time series with OR b/w resource attr and attribute filter and count distinct on service.name",
			requestType: qbtypes.RequestTypeTimeSeries,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal:       telemetrytypes.SignalTraces,
				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
				Aggregations: []qbtypes.LogAggregation{
					{
						Expression: "count_distinct(service.name)",
					},
				},
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'redis-manual' OR http.method = 'GET'",
				},
				Limit: 10,
				GroupBy: []qbtypes.GroupByKey{
					{
						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
							Name: "service.name",
						},
					},
				},
			},
			expected: qbtypes.Statement{
				Query: "WITH __limit_cte AS (SELECT toString(multiIf(notEmpty(service), service, NULL)) AS `service.name`, countDistinct(multiIf(notEmpty(service), service, NULL)) AS __result_0 FROM event.log WHERE (service = ? OR (attributes['http.method'] = ? AND mapContains(attributes, 'http.method') = ?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? GROUP BY `service.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(time, INTERVAL 30 SECOND) AS ts, toString(multiIf(notEmpty(service), service, NULL)) AS `service.name`, countDistinct(multiIf(notEmpty(service), service, NULL)) AS __result_0 FROM event.log WHERE (service = ? OR (attributes['http.method'] = ? AND mapContains(attributes, 'http.method') = ?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? AND (`service.name`) GLOBAL IN (SELECT `service.name` FROM __limit_cte) GROUP BY ts, `service.name`",
				Args:  []any{"redis-manual", "GET", true, "1705226400000000000", uint64(1705224600), "1705485600000000000", uint64(1705485600), 10, "redis-manual", "GET", true, "1705226400000000000", uint64(1705224600), "1705485600000000000", uint64(1705485600)},
			},
			expectedErr: nil,
		},
		{
			startTs:     releaseTimeNano + uint64(24*time.Hour.Nanoseconds()),
			endTs:       releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
			name:        "Time series with limit + custom order by",
			requestType: qbtypes.RequestTypeTimeSeries,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal:       telemetrytypes.SignalLogs,
				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
				Aggregations: []qbtypes.LogAggregation{
					{
						Expression: "count()",
					},
				},
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'cartservice'",
				},
				Limit: 10,
				GroupBy: []qbtypes.GroupByKey{
					{
						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
							Name: "service.name",
						},
					},
				},
				Order: []qbtypes.OrderBy{
					{
						Key: qbtypes.OrderByKey{
							TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
								Name: "service.name",
							},
						},
						Direction: qbtypes.OrderDirectionDesc,
					},
				},
			},
			expected: qbtypes.Statement{
				Query: "WITH __resource_filter AS (SELECT fingerprint FROM event.log_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), __limit_cte AS (SELECT toString(multiIf(notEmpty(service), service, NULL)) AS `service.name`, count() AS __result_0 FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? GROUP BY `service.name` ORDER BY `service.name` desc LIMIT ?) SELECT toStartOfInterval(time, INTERVAL 30 SECOND) AS ts, toString(multiIf(notEmpty(service), service, NULL)) AS `service.name`, count() AS __result_0 FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? AND (`service.name`) GLOBAL IN (SELECT `service.name` FROM __limit_cte) GROUP BY ts, `service.name` ORDER BY `service.name` desc, ts desc",
				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1705397400), uint64(1705485600), "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600), 10, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
			},
			expectedErr: nil,
		},
		{
			startTs:     releaseTimeNano + uint64(24*time.Hour.Nanoseconds()),
			endTs:       releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
			name:        "Time series with group by on materialized column",
			requestType: qbtypes.RequestTypeTimeSeries,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal:       telemetrytypes.SignalLogs,
				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
				Aggregations: []qbtypes.LogAggregation{
					{
						Expression: "count()",
					},
				},
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'cartservice'",
				},
				Limit: 10,
				GroupBy: []qbtypes.GroupByKey{
					{
						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
							Name:          "materialized.key.name",
							FieldContext:  telemetrytypes.FieldContextAttribute,
							FieldDataType: telemetrytypes.FieldDataTypeString,
						},
					},
				},
			},
			expected: qbtypes.Statement{
				Query: "WITH __resource_filter AS (SELECT fingerprint FROM event.log_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), __limit_cte AS (SELECT toString(multiIf(mapContains(attributes, 'materialized.key.name') = ?, attributes['materialized.key.name'], NULL)) AS `materialized.key.name`, count() AS __result_0 FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? GROUP BY `materialized.key.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(time, INTERVAL 30 SECOND) AS ts, toString(multiIf(mapContains(attributes, 'materialized.key.name') = ?, attributes['materialized.key.name'], NULL)) AS `materialized.key.name`, count() AS __result_0 FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? AND (`materialized.key.name`) GLOBAL IN (SELECT `materialized.key.name` FROM __limit_cte) GROUP BY ts, `materialized.key.name`",
				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1705397400), uint64(1705485600), true, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600), 10, true, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
			},
		},
		{
			startTs:     releaseTimeNano + uint64(24*time.Hour.Nanoseconds()),
			endTs:       releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
			name:        "Time series with materialised column using or with regex operator",
			requestType: qbtypes.RequestTypeTimeSeries,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal:       telemetrytypes.SignalLogs,
				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
				Aggregations: []qbtypes.LogAggregation{
					{
						Expression: "count()",
					},
				},
				Filter: &qbtypes.Filter{
					Expression: "materialized.key.name REGEXP 'redis.*' OR materialized.key.name = 'memcached'",
				},
				Limit: 10,
			},
			expected: qbtypes.Statement{
				Query: "SELECT toStartOfInterval(time, INTERVAL 30 SECOND) AS ts, count() AS __result_0 FROM event.log WHERE ((match(attributes['materialized.key.name'], ?) AND mapContains(attributes, 'materialized.key.name') = ?) OR (attributes['materialized.key.name'] = ? AND mapContains(attributes, 'materialized.key.name') = ?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? GROUP BY ts",
				Args:  []any{"redis.*", true, "memcached", true, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
			},
			expectedErr: nil,
		},
	}

	ctx := context.Background()
	fl := flaggertest.New(t)

	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
	keysMap := buildCompleteFieldKeyMap(releaseTime)

	mockMetadataStore.KeysMap = keysMap

	fm := NewFieldMapper(fl)
	cb := NewConditionBuilder(fm, fl)

	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)

	statementBuilder := NewLogQueryStatementBuilder(
		instrumentationtest.New().ToProviderSettings(),
		mockMetadataStore,
		fm,
		cb,
		aggExprRewriter,
		DefaultFullTextColumn,
		GetBodyJSONKey,
		fl,
		nil,
		false,
		100000,
	)

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {

			q, err := statementBuilder.Build(ctx, c.startTs, c.endTs, c.requestType, c.query, nil)

			if c.expectedErr != nil {
				require.Error(t, err)
				require.Contains(t, err.Error(), c.expectedErr.Error())
			} else {
				require.NoError(t, err)
				require.Equal(t, c.expected.Query, q.Query)
				require.Equal(t, c.expected.Args, q.Args)
				require.Equal(t, c.expected.Warnings, q.Warnings)
			}
		})
	}
}

func TestStatementBuilderListQuery(t *testing.T) {
	cases := []struct {
		name        string
		requestType qbtypes.RequestType
		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
		variables   map[string]qbtypes.VariableItem
		expected    qbtypes.Statement
		expectedErr error
		expectWarn  bool
	}{
		{
			name:        "default list",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'cartservice'",
				},
				Limit: 10,
			},
			expected: qbtypes.Statement{
				Query: "WITH __resource_filter AS (SELECT fingerprint FROM event.log_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 toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "list query with mat col order by",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'cartservice'",
				},
				Limit: 10,
				Order: []qbtypes.OrderBy{
					{
						Key: qbtypes.OrderByKey{
							TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
								Name:          "materialized.key.name",
								FieldContext:  telemetrytypes.FieldContextAttribute,
								FieldDataType: telemetrytypes.FieldDataTypeString,
							},
						},
						Direction: qbtypes.OrderDirectionDesc,
					},
				},
			},
			expected: qbtypes.Statement{
				Query: "WITH __resource_filter AS (SELECT fingerprint FROM event.log_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 toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? ORDER BY attributes['materialized.key.name'] AS `materialized.key.name` desc LIMIT ?",
				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "list query with mat col using or and regex operator",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "materialized.key.name REGEXP 'redis.*' OR materialized.key.name = 'memcached'",
				},
				Limit: 10,
				Order: []qbtypes.OrderBy{
					{
						Key: qbtypes.OrderByKey{
							TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
								Name:          "materialized.key.name",
								FieldContext:  telemetrytypes.FieldContextAttribute,
								FieldDataType: telemetrytypes.FieldDataTypeString,
							},
						},
						Direction: qbtypes.OrderDirectionDesc,
					},
				},
			},
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE ((match(attributes['materialized.key.name'], ?) AND mapContains(attributes, 'materialized.key.name') = ?) OR (attributes['materialized.key.name'] = ? AND mapContains(attributes, 'materialized.key.name') = ?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? ORDER BY attributes['materialized.key.name'] AS `materialized.key.name` desc LIMIT ?",
				Args:  []any{"redis.*", true, "memcached", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "filter skips entirely but emits LIKE-without-wildcards warning",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "message LIKE 'plain' OR message IN $env",
				},
				Limit: 10,
			},
			variables: map[string]qbtypes.VariableItem{
				"env": {Type: qbtypes.DynamicVariableType, Value: "__all__"},
			},
			expectedErr: nil,
			expectWarn:  true,
		},
	}

	ctx := context.Background()
	fl := flaggertest.New(t)
	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
	fm := NewFieldMapper(fl)

	// Create a test release time
	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
	cb := NewConditionBuilder(fm, fl)

	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)

	statementBuilder := NewLogQueryStatementBuilder(
		instrumentationtest.New().ToProviderSettings(),
		mockMetadataStore,
		fm,
		cb,
		aggExprRewriter,
		DefaultFullTextColumn,
		GetBodyJSONKey,
		fl,
		nil,
		false,
		100000,
	)

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {

			q, err := statementBuilder.Build(ctx, 1747947419000, 1747983448000, c.requestType, c.query, c.variables)

			if c.expectedErr != nil {
				require.Error(t, err)
				require.Contains(t, err.Error(), c.expectedErr.Error())
			} else {
				require.NoError(t, err)
				if c.expectWarn {
					require.NotEmpty(t, q.Warnings)
				} else {
					require.Equal(t, c.expected.Query, q.Query)
					require.Equal(t, c.expected.Args, q.Args)
					require.Equal(t, c.expected.Warnings, q.Warnings)
				}
			}
		})
	}
}

func TestStatementBuilderListQueryResourceTests(t *testing.T) {
	cases := []struct {
		name        string
		requestType qbtypes.RequestType
		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
		expected    qbtypes.Statement
		expectedErr error
	}{
		{
			name:        "List with full text search",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "hello",
				},
				Limit: 10,
			},
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE match(LOWER(body), LOWER(?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"hello", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "list query with mat col order by",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'cartservice' hello",
				},
				Limit: 10,
				Order: []qbtypes.OrderBy{
					{
						Key: qbtypes.OrderByKey{
							TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
								Name:          "materialized.key.name",
								FieldContext:  telemetrytypes.FieldContextAttribute,
								FieldDataType: telemetrytypes.FieldDataTypeString,
							},
						},
						Direction: qbtypes.OrderDirectionDesc,
					},
				},
			},
			expected: qbtypes.Statement{
				Query: "WITH __resource_filter AS (SELECT fingerprint FROM event.log_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 toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND match(LOWER(body), LOWER(?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? ORDER BY attributes['materialized.key.name'] AS `materialized.key.name` desc LIMIT ?",
				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "hello", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "List with json search",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "body.status = 'success'",
				},
				Limit: 10,
			},
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE (JSON_VALUE(body, '$.\"status\"') = ? AND JSON_EXISTS(body, '$.\"status\"')) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"success", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "IN operator with json search",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "body.user_names[*] IN 'john_doe'",
				},
				Limit: 10,
			},
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE ((JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(String)') = ?) AND JSON_EXISTS(body, '$.\"user_names\"[*]')) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"john_doe", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "has with json search",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "has(body.user_names[*], 'john_doe')",
				},
				Limit: 10,
			},
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE has(JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(String)'), ?) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"john_doe", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
	}

	ctx := context.Background()
	fl := flaggertest.New(t)
	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
	fm := NewFieldMapper(fl)
	// Create a test release time
	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
	cb := NewConditionBuilder(fm, fl)

	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)

	statementBuilder := NewLogQueryStatementBuilder(
		instrumentationtest.New().ToProviderSettings(),
		mockMetadataStore,
		fm,
		cb,
		aggExprRewriter,
		DefaultFullTextColumn,
		GetBodyJSONKey,
		fl,
		nil,
		false,
		100000,
	)

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {

			q, err := statementBuilder.Build(ctx, 1747947419000, 1747983448000, c.requestType, c.query, nil)

			if c.expectedErr != nil {
				require.Error(t, err)
				require.Contains(t, err.Error(), c.expectedErr.Error())
			} else {
				require.NoError(t, err)
				require.Equal(t, c.expected.Query, q.Query)
				require.Equal(t, c.expected.Args, q.Args)
				require.Equal(t, c.expected.Warnings, q.Warnings)
			}
		})
	}
}

func TestStatementBuilderTimeSeriesBodyGroupBy(t *testing.T) {
	cases := []struct {
		name                string
		requestType         qbtypes.RequestType
		query               qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
		expected            qbtypes.Statement
		expectedErrContains string
	}{
		{
			name:        "Time series with limit and body group by",
			requestType: qbtypes.RequestTypeTimeSeries,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal:       telemetrytypes.SignalLogs,
				StepInterval: qbtypes.Step{Duration: 30 * time.Second},
				Aggregations: []qbtypes.LogAggregation{
					{
						Expression: "count()",
					},
				},
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'cartservice'",
				},
				Limit: 10,
				GroupBy: []qbtypes.GroupByKey{
					{
						TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
							Name:         "status",
							FieldContext: telemetrytypes.FieldContextBody,
						},
					},
				},
			},
			expectedErrContains: "Group by/Aggregation isn't available for the body column",
		},
	}

	ctx := context.Background()
	fl := flaggertest.New(t)
	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
	fm := NewFieldMapper(fl)
	// Create a test release time
	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
	cb := NewConditionBuilder(fm, fl)

	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)

	statementBuilder := NewLogQueryStatementBuilder(
		instrumentationtest.New().ToProviderSettings(),
		mockMetadataStore,
		fm,
		cb,
		aggExprRewriter,
		DefaultFullTextColumn,
		GetBodyJSONKey,
		fl,
		nil,
		false,
		100000,
	)

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {

			q, err := statementBuilder.Build(ctx, 1747947419000, 1747983448000, c.requestType, c.query, nil)

			if c.expectedErrContains != "" {
				require.Error(t, err)
				require.Contains(t, err.Error(), c.expectedErrContains)
			} else {
				require.NoError(t, err)
				require.Equal(t, c.expected.Query, q.Query)
				require.Equal(t, c.expected.Args, q.Args)
				require.Equal(t, c.expected.Warnings, q.Warnings)
			}
		})
	}
}

func TestStatementBuilderListQueryServiceCollision(t *testing.T) {
	cases := []struct {
		name        string
		requestType qbtypes.RequestType
		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
		expected    qbtypes.Statement
		expectedErr error
		expectWarn  bool
	}{
		{
			name:        "default list",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "(service.name = 'cartservice' AND body CONTAINS 'error')",
				},
				Limit: 10,
			},
			expected: qbtypes.Statement{
				Query: "WITH __resource_filter AS (SELECT fingerprint FROM event.log_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 toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (body ILIKE ?) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "%error%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
			expectWarn:  true,
		},
		{
			name:        "list query with mat col order by",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{
					Expression: "service.name = 'cartservice' AND body CONTAINS 'error'",
				},
				Limit: 10,
				Order: []qbtypes.OrderBy{
					{
						Key: qbtypes.OrderByKey{
							TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
								Name:          "materialized.key.name",
								FieldContext:  telemetrytypes.FieldContextAttribute,
								FieldDataType: telemetrytypes.FieldDataTypeString,
							},
						},
						Direction: qbtypes.OrderDirectionDesc,
					},
				},
			},
			expected: qbtypes.Statement{
				Query: "WITH __resource_filter AS (SELECT fingerprint FROM event.log_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 toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND body ILIKE ? AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? ORDER BY attributes['materialized.key.name'] AS `materialized.key.name` desc LIMIT ?",
				Args:  []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "%error%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
			expectWarn:  true,
		},
	}

	ctx := context.Background()
	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
	fl := flaggertest.New(t)
	fm := NewFieldMapper(fl)
	mockMetadataStore.KeysMap = buildCompleteFieldKeyMapCollision()
	cb := NewConditionBuilder(fm, fl)

	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)

	statementBuilder := NewLogQueryStatementBuilder(
		instrumentationtest.New().ToProviderSettings(),
		mockMetadataStore,
		fm,
		cb,
		aggExprRewriter,
		DefaultFullTextColumn,
		GetBodyJSONKey,
		fl,
		nil,
		false,
		100000,
	)

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {

			q, err := statementBuilder.Build(ctx, 1747947419000, 1747983448000, c.requestType, c.query, nil)

			if c.expectedErr != nil {
				require.Error(t, err)
				require.Contains(t, err.Error(), c.expectedErr.Error())
			} else {
				require.NoError(t, err)
				require.Equal(t, c.expected.Query, q.Query)
				require.Equal(t, c.expected.Args, q.Args)
				if c.expectWarn {
					require.True(t, len(q.Warnings) > 0)
				}
			}
		})
	}
}

func TestAdjustKey(t *testing.T) {

	// Create a test release time
	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
	cases := []struct {
		name        string
		inputKey    telemetrytypes.TelemetryFieldKey
		keysMap     map[string][]*telemetrytypes.TelemetryFieldKey
		expectedKey telemetrytypes.TelemetryFieldKey
	}{
		{
			name: "intrinsic field with no other key match - use intrinsic",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "severity_text",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap:     buildCompleteFieldKeyMap(releaseTime),
			expectedKey: IntrinsicFields["severity_text"],
		},
		{
			name: "intrinsic field with other key match - no override",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "body",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap: map[string][]*telemetrytypes.TelemetryFieldKey{
				"body": {
					{
						Name:          "body",
						FieldContext:  telemetrytypes.FieldContextBody,
						FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
					},
					{
						Name:          "body",
						FieldContext:  telemetrytypes.FieldContextAttribute,
						FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
					},
				},
			},
			expectedKey: telemetrytypes.TelemetryFieldKey{
				Name:          "body",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
		},
		{
			name: "json field with no context specified",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "severity_number",
				FieldContext:  telemetrytypes.FieldContextBody,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap: buildCompleteFieldKeyMap(releaseTime),
			expectedKey: telemetrytypes.TelemetryFieldKey{
				Name:          "severity_number",
				FieldContext:  telemetrytypes.FieldContextBody,
				FieldDataType: telemetrytypes.FieldDataTypeNumber,
			},
		},
		{
			name: "single matching key in metadata",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "service.name",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap:     buildCompleteFieldKeyMap(releaseTime),
			expectedKey: *buildCompleteFieldKeyMap(releaseTime)["service.name"][0],
		},
		{
			name: "single matching key with incorrect context specified - no override",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "service.name",
				FieldContext:  telemetrytypes.FieldContextAttribute,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap: buildCompleteFieldKeyMap(releaseTime),
			expectedKey: telemetrytypes.TelemetryFieldKey{
				Name:          "service.name",
				FieldContext:  telemetrytypes.FieldContextAttribute,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
		},
		{
			name: "single matching key with no context specified - override",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "service.name",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap:     buildCompleteFieldKeyMap(releaseTime),
			expectedKey: *buildCompleteFieldKeyMap(releaseTime)["service.name"][0],
		},
		{
			name: "multiple matching keys - all materialized",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "multi.mat.key",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap: buildCompleteFieldKeyMap(releaseTime),
			expectedKey: telemetrytypes.TelemetryFieldKey{
				Name:          "multi.mat.key",
				FieldDataType: telemetrytypes.FieldDataTypeString,
				Materialized:  true,
			},
		},
		{
			name: "multiple matching keys - mixed materialization",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "mixed.materialization.key",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap: buildCompleteFieldKeyMap(releaseTime),
			expectedKey: telemetrytypes.TelemetryFieldKey{
				Name:          "mixed.materialization.key",
				FieldDataType: telemetrytypes.FieldDataTypeString,
				Materialized:  false,
			},
		},
		{
			name: "multiple matching keys with context specified",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "mixed.materialization.key",
				FieldContext:  telemetrytypes.FieldContextAttribute,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap:     buildCompleteFieldKeyMap(releaseTime),
			expectedKey: *buildCompleteFieldKeyMap(releaseTime)["mixed.materialization.key"][0],
		},
		{
			name: "no matching keys - unknown field",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "unknown.field",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap: buildCompleteFieldKeyMap(releaseTime),
			expectedKey: telemetrytypes.TelemetryFieldKey{
				Name:          "unknown.field",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
				Materialized:  false,
			},
		},
		{
			name: "no matching keys with context filter",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "unknown.field",
				FieldContext:  telemetrytypes.FieldContextAttribute,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap: buildCompleteFieldKeyMap(releaseTime),
			expectedKey: telemetrytypes.TelemetryFieldKey{
				Name:          "unknown.field",
				FieldContext:  telemetrytypes.FieldContextAttribute,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
				Materialized:  false,
			},
		},
		{
			name: "materialized field",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "mat.key",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap:     buildCompleteFieldKeyMap(releaseTime),
			expectedKey: *buildCompleteFieldKeyMap(releaseTime)["mat.key"][0],
		},
		{
			name: "non-materialized field",
			inputKey: telemetrytypes.TelemetryFieldKey{
				Name:          "user.id",
				FieldContext:  telemetrytypes.FieldContextUnspecified,
				FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
			},
			keysMap:     buildCompleteFieldKeyMap(releaseTime),
			expectedKey: *buildCompleteFieldKeyMap(releaseTime)["user.id"][0],
		},
	}

	fl := flaggertest.New(t)
	fm := NewFieldMapper(fl)
	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
	mockMetadataStore.KeysMap = buildCompleteFieldKeyMapCollision()
	cb := NewConditionBuilder(fm, fl)

	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)

	statementBuilder := NewLogQueryStatementBuilder(
		instrumentationtest.New().ToProviderSettings(),
		mockMetadataStore,
		fm,
		cb,
		aggExprRewriter,
		DefaultFullTextColumn,
		GetBodyJSONKey,
		fl,
		nil,
		false,
		100000,
	)

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			// Create a copy of the input key to avoid modifying the original
			key := c.inputKey

			// Call adjustKey
			statementBuilder.adjustKey(&key, c.keysMap)

			// Verify the key was adjusted as expected
			require.Equal(t, c.expectedKey.Name, key.Name, "key name should match")
			require.Equal(t, c.expectedKey.FieldContext, key.FieldContext, "field context should match")
			require.Equal(t, c.expectedKey.FieldDataType, key.FieldDataType, "field data type should match")
			require.Equal(t, c.expectedKey.Materialized, key.Materialized, "materialized should match")
			require.Equal(t, c.expectedKey.Indexes, key.Indexes, "json exists should match")
		})
	}
}

func TestStmtBuilderBodyField(t *testing.T) {
	cases := []struct {
		name              string
		requestType       qbtypes.RequestType
		query             qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
		enableUseJSONBody bool
		expected          qbtypes.Statement
		expectedErr       error
	}{
		{
			name:        "body_exists",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "body Exists"},
				Limit:  10,
			},
			enableUseJSONBody: true,
			expected: qbtypes.Statement{
				Query:    "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body_v2 as body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE body_v2.message <> ? AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:     []any{"", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
				Warnings: []string{bodySearchDefaultWarning},
			},
			expectedErr: nil,
		},
		{
			name:        "body_exists_disabled",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "body Exists"},
				Limit:  10,
			},
			enableUseJSONBody: false,
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE body <> ? AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "body_empty",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "body == ''"},
				Limit:  10,
			},
			enableUseJSONBody: true,
			expected: qbtypes.Statement{
				Query:    "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body_v2 as body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE body_v2.message = ? AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:     []any{"", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
				Warnings: []string{bodySearchDefaultWarning},
			},
			expectedErr: nil,
		},
		{
			name:        "body_empty_disabled",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "body == ''"},
				Limit:  10,
			},
			enableUseJSONBody: false,
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE body = ? AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
		{
			name:        "body_contains",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "body CONTAINS 'error'"},
				Limit:  10,
			},
			enableUseJSONBody: true,
			expected: qbtypes.Statement{
				Query:    "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body_v2 as body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE body_v2.message ILIKE ? AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:     []any{"%error%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
				Warnings: []string{bodySearchDefaultWarning},
			},
			expectedErr: nil,
		},
		{
			name:        "body_contains_disabled",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "body CONTAINS 'error'"},
				Limit:  10,
			},
			enableUseJSONBody: false,
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE body ILIKE ? AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"%error%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
	}

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			fl := flaggertest.WithUseJSONBody(t, c.enableUseJSONBody)
			fm := NewFieldMapper(fl)
			cb := NewConditionBuilder(fm, fl)
			// build the key map
			mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
			for _, field := range IntrinsicFields {
				f := field
				mockMetadataStore.KeysMap[field.Name] = append(mockMetadataStore.KeysMap[field.Name], &f)
			}
			aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)
			statementBuilder := NewLogQueryStatementBuilder(
				instrumentationtest.New().ToProviderSettings(),
				mockMetadataStore,
				fm,
				cb,
				aggExprRewriter,
				DefaultFullTextColumn,
				GetBodyJSONKey,
				fl,
				nil,
				false,
				100000,
			)

			q, err := statementBuilder.Build(context.Background(), 1747947419000, 1747983448000, c.requestType, c.query, nil)
			if c.expectedErr != nil {
				require.Error(t, err)
				require.Contains(t, err.Error(), c.expectedErr.Error())
			} else {
				if err != nil {
					_, _, _, _, _, add := errors.Unwrapb(err)
					t.Logf("error additionals: %v", add)
				}
				require.NoError(t, err)
				require.Equal(t, c.expected.Query, q.Query)
				require.Equal(t, c.expected.Args, q.Args)
				require.Equal(t, c.expected.Warnings, q.Warnings)
			}
		})
	}
}

func TestStmtBuilderBodyFullTextSearch(t *testing.T) {
	cases := []struct {
		name              string
		requestType       qbtypes.RequestType
		query             qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
		enableUseJSONBody bool
		expected          qbtypes.Statement
		expectedErr       error
	}{
		{
			name:        "fts",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "'error'"},
				Limit:  10,
			},
			enableUseJSONBody: true,
			expected: qbtypes.Statement{
				Query:    "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body_v2 as body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE match(LOWER(body_v2.message), LOWER(?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:     []any{"error", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
				Warnings: []string{querybuilder.BodyFullTextSearchDefaultWarning},
			},
			expectedErr: nil,
		},
		{
			name:        "fts_2",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "error"},
				Limit:  10,
			},
			enableUseJSONBody: true,
			expected: qbtypes.Statement{
				Query:    "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body_v2 as body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE match(LOWER(body_v2.message), LOWER(?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:     []any{"error", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
				Warnings: []string{querybuilder.BodyFullTextSearchDefaultWarning},
			},
			expectedErr: nil,
		},
		{
			name:        "fts_disabled",
			requestType: qbtypes.RequestTypeRaw,
			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
				Signal: telemetrytypes.SignalLogs,
				Filter: &qbtypes.Filter{Expression: "'error'"},
				Limit:  10,
			},
			enableUseJSONBody: false,
			expected: qbtypes.Statement{
				Query: "SELECT toUnixTimestamp64Nano(time) AS timestamp, id, trace_id, span_id, toUInt32OrZero(attributes['trace_flags']) AS trace_flags, severity_text, severity_number, attributes['scope.name'] AS scope_name, attributes['scope.version'] AS scope_version, body, attributes AS attributes_string, CAST(map() AS Map(String, Float64)) AS attributes_number, CAST(map() AS Map(String, Bool)) AS attributes_bool, map('service.name', toString(service), 'host', toString(host)) AS resources_string, CAST(map() AS Map(String, String)) AS scope_string FROM event.log WHERE match(LOWER(body), LOWER(?)) AND time >= ? AND ts_bucket_start >= ? AND time < ? AND ts_bucket_start <= ? LIMIT ?",
				Args:  []any{"error", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
			},
			expectedErr: nil,
		},
	}

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			fl := flaggertest.WithUseJSONBody(t, c.enableUseJSONBody)
			fm := NewFieldMapper(fl)
			cb := NewConditionBuilder(fm, fl)
			// build the key map
			mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
			for _, field := range IntrinsicFields {
				f := field
				mockMetadataStore.KeysMap[field.Name] = append(mockMetadataStore.KeysMap[field.Name], &f)
			}
			aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)
			statementBuilder := NewLogQueryStatementBuilder(
				instrumentationtest.New().ToProviderSettings(),
				mockMetadataStore,
				fm,
				cb,
				aggExprRewriter,
				DefaultFullTextColumn,
				GetBodyJSONKey,
				fl,
				nil,
				false,
				100000,
			)

			q, err := statementBuilder.Build(context.Background(), 1747947419000, 1747983448000, c.requestType, c.query, nil)
			if c.expectedErr != nil {
				require.Error(t, err)
				require.Contains(t, err.Error(), c.expectedErr.Error())
			} else {
				if err != nil {
					_, _, _, _, _, add := errors.Unwrapb(err)
					t.Logf("error additionals: %v", add)
				}
				require.NoError(t, err)
				require.Equal(t, c.expected.Query, q.Query)
				require.Equal(t, c.expected.Args, q.Args)
				require.Equal(t, c.expected.Warnings, q.Warnings)
			}
		})
	}
}

// TestSkipResourceFingerprintLogs exercises the three resolver outcomes for
// logs: use-CTE (count < threshold), fallback (count >= threshold), and the
// legacy path (feature disabled).
func TestSkipResourceFingerprintLogs(t *testing.T) {
	const (
		startMs   = uint64(1747947419000)
		endMs     = uint64(1747983448000)
		threshold = uint64(10)
	)

	query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
		Signal: telemetrytypes.SignalLogs,
		Filter: &qbtypes.Filter{
			Expression: "service.name = 'redis-manual'",
		},
		Limit: 5,
	}

	t.Run("disabled uses the legacy CTE", func(t *testing.T) {
		sb := newSkipResourceFingerprintLogsBuilder(t, nil, false, threshold)

		stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
		require.NoError(t, err)
		require.Contains(t, stmt.Query, "__resource_filter AS (SELECT fingerprint")
		require.Contains(t, stmt.Query, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
	})

	t.Run("CTE attached when count below threshold", func(t *testing.T) {
		mockStore := telemetrystoretest.New(telemetrystore.Config{}, &regexQueryMatcher{})
		mock := mockStore.Mock()

		mock.ExpectQueryRow(`SELECT count\(\) FROM \(SELECT fingerprint FROM event\.log_resource`).
			WillReturnRow(dsmock.NewRow([]dsmock.ColumnType{
				{Name: "count", Type: "UInt64"},
			}, []any{uint64(2)}))

		sb := newSkipResourceFingerprintLogsBuilder(t, mockStore, true, threshold)

		stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
		require.NoError(t, err)

		require.Contains(t, stmt.Query, "__resource_filter AS (SELECT fingerprint")
		require.Contains(t, stmt.Query, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")

		require.NoError(t, mock.ExpectationsWereMet())
	})

	t.Run("fallback when count at or above threshold", func(t *testing.T) {
		mockStore := telemetrystoretest.New(telemetrystore.Config{}, &regexQueryMatcher{})
		mock := mockStore.Mock()

		mock.ExpectQueryRow(`SELECT count\(\) FROM \(SELECT fingerprint FROM event\.log_resource`).
			WillReturnRow(dsmock.NewRow([]dsmock.ColumnType{
				{Name: "count", Type: "UInt64"},
			}, []any{threshold}))

		sb := newSkipResourceFingerprintLogsBuilder(t, mockStore, true, threshold)

		stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
		require.NoError(t, err)

		require.NotContains(t, stmt.Query, "__resource_filter AS")
		require.NotContains(t, stmt.Query, "resource_fingerprint")
		require.Contains(t, stmt.Query, "service.name")

		require.NoError(t, mock.ExpectationsWereMet())
	})
}

func newSkipResourceFingerprintLogsBuilder(
	t *testing.T,
	telemetryStore telemetrystore.TelemetryStore,
	skipEnable bool,
	threshold uint64,
) *logQueryStatementBuilder {
	t.Helper()

	fl := flaggertest.New(t)
	fm := NewFieldMapper(fl)
	cb := NewConditionBuilder(fm, fl)
	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
	mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC))

	aggExprRewriter := querybuilder.NewAggExprRewriter(
		instrumentationtest.New().ToProviderSettings(),
		DefaultFullTextColumn,
		fm,
		cb,
		GetBodyJSONKey,
		fl,
	)

	return NewLogQueryStatementBuilder(
		instrumentationtest.New().ToProviderSettings(),
		mockMetadataStore,
		fm,
		cb,
		aggExprRewriter,
		DefaultFullTextColumn,
		GetBodyJSONKey,
		fl,
		telemetryStore,
		skipEnable,
		threshold,
	)
}