fathom/internal/query/query_test.go

package query

import (
	"fmt"
	"strings"
	"testing"
	"time"

	"git.z.gd/zzg/fathom/internal/model"
	"git.z.gd/zzg/fathom/internal/store"
)

func TestParseMinimal(t *testing.T) {
	q, err := Parse("SELECT name FROM spans WHERE time >= 100 AND time <= 200")
	if err != nil {
		t.Fatal(err)
	}
	if q.Table != "spans" || q.MinTS != 100 || q.MaxTS != 200 {
		t.Errorf("parsed %+v, want spans/100/200", q)
	}
	if len(q.Selects) != 1 || q.Selects[0].Column != "name" {
		t.Errorf("selects = %+v", q.Selects)
	}
	if q.Limit != 100 {
		t.Errorf("default limit = %d, want 100", q.Limit)
	}
}

func TestParseFull(t *testing.T) {
	q, err := Parse(`SELECT name, count(*), avg(duration) AS mean, p95(duration)
	                 FROM spans
	                 WHERE time >= 100 AND time <= 200 AND kind = tool_call AND name contains 'bash'
	                 GROUP BY name
	                 ORDER BY count(*) DESC
	                 LIMIT 25`)
	if err != nil {
		t.Fatal(err)
	}
	if len(q.Selects) != 4 {
		t.Fatalf("got %d selects, want 4", len(q.Selects))
	}
	if q.Selects[2].Label() != "mean" {
		t.Errorf("alias not applied: label = %q", q.Selects[2].Label())
	}
	if q.Selects[1].Label() != "count(*)" {
		t.Errorf("label = %q, want count(*)", q.Selects[1].Label())
	}
	if len(q.Conds) != 2 {
		t.Errorf("got %d non-time conditions, want 2", len(q.Conds))
	}
	if !q.Desc || q.OrderBy != "count(*)" {
		t.Errorf("order by = %q desc=%v", q.OrderBy, q.Desc)
	}
	if q.Limit != 25 {
		t.Errorf("limit = %d", q.Limit)
	}
}

func TestParseRejects(t *testing.T) {
	cases := []struct{ src, wants string }{
		{"SELECT name FROM spans", "time range"},
		{"SELECT name FROM spans WHERE name = 'x'", "time"},
		{"SELECT name FROM spans WHERE time >= 1 AND time <= 2 OR name = 'x'", "OR"},
		{"SELECT nope(x) FROM spans WHERE time >= 1 AND time <= 2", "unknown aggregate"},
		{"SELECT name, count(*) FROM spans WHERE time >= 1 AND time <= 2", "GROUP BY"},
		{"SELECT name FROM spans WHERE time >= 1 AND time <= 2 GROUP BY name", "aggregate"},
		{"SELECT count(*) FROM spans WHERE time >= 200 AND time <= 100", "inverted"},
		{"SELECT name FROM spans WHERE time >= 1 AND time <= 2 LIMIT x", "number"},
		{"SELECT name FROM spans WHERE time >= 1 AND time <= 2 trailing", "trailing"},
		{"SELECT sum(*) FROM spans WHERE time >= 1 AND time <= 2", "not meaningful"},
		{"", "SELECT"},
	}
	for _, tc := range cases {
		_, err := Parse(tc.src)
		if err == nil {
			t.Errorf("Parse(%q) succeeded, want an error about %q", tc.src, tc.wants)
			continue
		}
		if !strings.Contains(err.Error(), tc.wants) {
			t.Errorf("Parse(%q) error = %q, want it to mention %q", tc.src, err, tc.wants)
		}
	}
}

func TestParseStringEscapes(t *testing.T) {
	q, err := Parse(`SELECT name FROM spans WHERE time >= 1 AND time <= 2 AND name = 'it\'s here'`)
	if err != nil {
		t.Fatal(err)
	}
	if q.Conds[0].Value != "it's here" {
		t.Errorf("escaped quote mishandled: %q", q.Conds[0].Value)
	}
}

func TestLikeMatch(t *testing.T) {
	cases := []struct {
		s, p string
		want bool
	}{
		{"claude_code.tool", "claude%", true},
		{"claude_code.tool", "%tool", true},
		{"claude_code.tool", "%code%", true},
		{"claude_code.tool", "claude_code.too_", true},
		{"claude_code.tool", "%nope%", false},
		{"abc", "a%%c", true},
		{"abc", "___", true},
		{"abc", "____", false},
		{"ABC", "abc", true}, // case-insensitive
		{"", "%", true},
		{"", "", true},
	}
	for _, tc := range cases {
		if got := likeMatch(tc.s, tc.p); got != tc.want {
			t.Errorf("likeMatch(%q, %q) = %v, want %v", tc.s, tc.p, got, tc.want)
		}
	}
}

// --- execution ---

func execStore(t *testing.T) (*store.Store, *Engine, int64) {
	t.Helper()
	s, err := store.Open(store.Options{
		Dir: t.TempDir(), FlushInterval: time.Hour, FlushRows: 1 << 20,
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { s.Close() })

	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()
	mk := func(id, name string, kind model.Kind, st model.Status, offset, dur int64, attrs model.Attrs) model.Span {
		return model.Span{
			TraceID: "t1", SpanID: id, Name: name, Service: "claude-code",
			Kind: kind, Status: st,
			StartTS: base + offset, EndTS: base + offset + dur,
			SampleRate: 1, Attrs: attrs,
		}
	}
	spans := []model.Span{
		mk("s1", "claude_code.llm_request", model.KindLLMCall, model.StatusOK, 0, 3000,
			model.Attrs{"model": model.StringValue("opus"), "input_tokens": model.IntValue(100)}),
		mk("s2", "claude_code.llm_request", model.KindLLMCall, model.StatusOK, 4000, 5000,
			model.Attrs{"model": model.StringValue("opus"), "input_tokens": model.IntValue(300)}),
		mk("s3", "claude_code.tool", model.KindToolCall, model.StatusOK, 10000, 1000,
			model.Attrs{"tool_name": model.StringValue("Bash")}),
		mk("s4", "claude_code.tool", model.KindToolCall, model.StatusError, 12000, 2000,
			model.Attrs{"tool_name": model.StringValue("Read")}),
	}
	if err := s.AddSpans(spans); err != nil {
		t.Fatal(err)
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}
	return s, NewEngine(s), base
}

func run(t *testing.T, e *Engine, src string) *Result {
	t.Helper()
	q, err := Parse(src)
	if err != nil {
		t.Fatalf("Parse(%q): %v", src, err)
	}
	res, err := e.Run(q)
	if err != nil {
		t.Fatalf("Run(%q): %v", src, err)
	}
	return res
}

func TestExecRowListing(t *testing.T) {
	_, e, base := execStore(t)
	res := run(t, e, fmt.Sprintf(
		"SELECT name, duration FROM spans WHERE time >= %d AND time <= %d", base-1, base+100000))
	if len(res.Rows) != 4 {
		t.Fatalf("got %d rows, want 4", len(res.Rows))
	}
	if res.Stats.RowsScanned != 4 {
		t.Errorf("stats say %d rows scanned, want 4", res.Stats.RowsScanned)
	}
}

func TestExecTimeRangeFiltersWithinShard(t *testing.T) {
	// The shard granularity is a day; a narrower range must still filter rows.
	_, e, base := execStore(t)
	res := run(t, e, fmt.Sprintf(
		"SELECT name FROM spans WHERE time >= %d AND time <= %d", base+9000, base+100000))
	if len(res.Rows) != 2 {
		t.Errorf("got %d rows, want the 2 inside the narrowed range", len(res.Rows))
	}
}

func TestExecFilterByEnumName(t *testing.T) {
	_, e, base := execStore(t)
	res := run(t, e, fmt.Sprintf(
		"SELECT name FROM spans WHERE time >= %d AND time <= %d AND status = error", base-1, base+100000))
	if len(res.Rows) != 1 {
		t.Errorf("status = error matched %d rows, want 1", len(res.Rows))
	}

	res = run(t, e, fmt.Sprintf(
		"SELECT name FROM spans WHERE time >= %d AND time <= %d AND kind = tool_call", base-1, base+100000))
	if len(res.Rows) != 2 {
		t.Errorf("kind = tool_call matched %d rows, want 2", len(res.Rows))
	}
}

func TestExecFilterByAttribute(t *testing.T) {
	_, e, base := execStore(t)
	res := run(t, e, fmt.Sprintf(
		"SELECT name FROM spans WHERE time >= %d AND time <= %d AND tool_name = 'Bash'", base-1, base+100000))
	if len(res.Rows) != 1 {
		t.Errorf("attribute filter matched %d rows, want 1", len(res.Rows))
	}

	// Numeric comparison against an integer attribute.
	res = run(t, e, fmt.Sprintf(
		"SELECT name FROM spans WHERE time >= %d AND time <= %d AND input_tokens > 200", base-1, base+100000))
	if len(res.Rows) != 1 {
		t.Errorf("input_tokens > 200 matched %d rows, want 1", len(res.Rows))
	}
}

func TestExecAggregateGroupBy(t *testing.T) {
	_, e, base := execStore(t)
	res := run(t, e, fmt.Sprintf(
		`SELECT kind, count(*), avg(duration), max(duration)
		 FROM spans WHERE time >= %d AND time <= %d
		 GROUP BY kind ORDER BY count(*) DESC`, base-1, base+100000))

	if len(res.Rows) != 2 {
		t.Fatalf("got %d groups, want 2 (llm_call and tool_call)", len(res.Rows))
	}
	byKind := map[string][]any{}
	for _, r := range res.Rows {
		byKind[fmt.Sprint(r[0])] = r
	}
	llm, ok := byKind["llm_call"]
	if !ok {
		t.Fatalf("no llm_call group; got %v", res.Rows)
	}
	if llm[1].(float64) != 2 {
		t.Errorf("llm_call count = %v, want 2", llm[1])
	}
	if llm[2].(float64) != 4000 {
		t.Errorf("llm_call avg duration = %v, want 4000", llm[2])
	}
	if llm[3].(float64) != 5000 {
		t.Errorf("llm_call max duration = %v, want 5000", llm[3])
	}
}

func TestExecPercentile(t *testing.T) {
	_, e, base := execStore(t)
	res := run(t, e, fmt.Sprintf(
		"SELECT p50(duration) FROM spans WHERE time >= %d AND time <= %d", base-1, base+100000))
	if len(res.Rows) != 1 {
		t.Fatalf("got %d rows, want 1", len(res.Rows))
	}
	// Durations are 3000, 5000, 1000, 2000 -> sorted 1000 2000 3000 5000.
	// p50 lands on the 2nd value with this definition.
	if got := res.Rows[0][0].(float64); got != 2000 && got != 3000 {
		t.Errorf("p50 = %v, want 2000 or 3000", got)
	}
}

func TestExecLimit(t *testing.T) {
	_, e, base := execStore(t)
	res := run(t, e, fmt.Sprintf(
		"SELECT name FROM spans WHERE time >= %d AND time <= %d LIMIT 2", base-1, base+100000))
	if len(res.Rows) != 2 {
		t.Errorf("LIMIT 2 returned %d rows", len(res.Rows))
	}
}

func TestExecEmptyRangeReturnsNoRows(t *testing.T) {
	_, e, _ := execStore(t)
	res := run(t, e, "SELECT name FROM spans WHERE time >= 1 AND time <= 2")
	if len(res.Rows) != 0 {
		t.Errorf("got %d rows for a range with no data", len(res.Rows))
	}
	if res.Rows == nil {
		t.Error("Rows should be an empty slice, not nil, so JSON renders [] not null")
	}
}

func TestExecUnknownTable(t *testing.T) {
	_, e, _ := execStore(t)
	q, _ := Parse("SELECT name FROM nope WHERE time >= 1 AND time <= 2")
	if _, err := e.Run(q); err == nil {
		t.Error("querying an unknown table succeeded")
	}
}

func TestSampleRateCompensation(t *testing.T) {
	s, err := store.Open(store.Options{Dir: t.TempDir(), FlushInterval: time.Hour, FlushRows: 1 << 20})
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()

	// One row recorded at 1-in-10 sampling stands for ten rows.
	if err := s.AddSpans([]model.Span{{
		TraceID: "t", SpanID: "s", Name: "x", StartTS: base, EndTS: base + 1000,
		SampleRate: 0.1,
	}}); err != nil {
		t.Fatal(err)
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}

	e := NewEngine(s)
	res := run(t, e, fmt.Sprintf("SELECT count(*) FROM spans WHERE time >= %d AND time <= %d", base-1, base+10000))
	if got := res.Rows[0][0].(float64); got != 10 {
		t.Errorf("count of a 0.1-sampled row = %v, want 10 (Scuba-style compensation)", got)
	}
}

func TestChunkSkipOnAbsentLiteral(t *testing.T) {
	// An equality literal absent from a chunk's dictionary means no row in that
	// chunk can match, so the scan should skip it wholesale.
	_, e, base := execStore(t)
	res := run(t, e, fmt.Sprintf(
		"SELECT name FROM spans WHERE time >= %d AND time <= %d AND name = 'nonexistent.span'", base-1, base+100000))
	if len(res.Rows) != 0 {
		t.Errorf("got %d rows for a literal that is not in the data", len(res.Rows))
	}
}

func TestLogsTableQueryable(t *testing.T) {
	s, err := store.Open(store.Options{Dir: t.TempDir(), FlushInterval: time.Hour, FlushRows: 1 << 20})
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	base := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC).UnixMicro()

	if err := s.AddLogs([]model.LogRecord{
		{TraceID: "t", SpanID: "s", EventName: "tool_result", TS: base},
		{TraceID: "t", SpanID: "s", EventName: "api_request", TS: base + 100},
		{TraceID: "t", SpanID: "s", EventName: "tool_result", TS: base + 200},
	}); err != nil {
		t.Fatal(err)
	}
	if err := s.Flush(); err != nil {
		t.Fatal(err)
	}

	e := NewEngine(s)
	// The logs table uses ts, not start_ts, as its time column; `time` must
	// resolve correctly without the query having to know that.
	res := run(t, e, fmt.Sprintf(
		`SELECT event_name, count(*) FROM logs WHERE time >= %d AND time <= %d
		 GROUP BY event_name ORDER BY count(*) DESC`, base-1, base+10000))
	if len(res.Rows) != 2 {
		t.Fatalf("got %d groups, want 2", len(res.Rows))
	}
	if fmt.Sprint(res.Rows[0][0]) != "tool_result" || res.Rows[0][1].(float64) != 2 {
		t.Errorf("top group = %v, want tool_result with 2", res.Rows[0])
	}
}