// Copyright (C) 2025 CardinalHQ, Inc
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, version 3.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

package queryapi

import (
	"database/sql"
	"fmt"
	"math"

	_ "github.com/marcboeker/go-duckdb/v2"

	"github.com/cardinalhq/lakerunner/logql"
	"github.com/cardinalhq/lakerunner/promql"

	"log/slog"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"
)

// -------- AGGREGATE tests ------------------------------------------------

// For worker SQL generated by BaseExpr.ToWorkerSQL(...)
// We need to replace {table}, {start}, {end}.
func replaceWorkerPlaceholders(sql string, start, end int64) string {
	sql = strings.ReplaceAll(sql, "{table}", "logs")
	sql = strings.ReplaceAll(sql, "{start}", fmt.Sprintf("%d", start))
	sql = strings.ReplaceAll(sql, "{end}", fmt.Sprintf("%d", end))
	return sql
}

// End-to-end: LogQL -> CompileLog (leaves) -> RewriteToPromQL -> parse PromQL -> Compile
// -> attach leaves -> return the compiled plan (typically 1 leaf).
func compileMetricPlanFromLogQL(t *testing.T, q string) (promql.QueryPlan, promql.RewriteResult) {
	t.Helper()

	ast, err := logql.FromLogQL(q)
	if err != nil {
		t.Fatalf("FromLogQL error: %v", err)
	}
	lplan, err := logql.CompileLog(ast)
	if err != nil {
		t.Fatalf("CompileLog error: %v", err)
	}

	rr, err := promql.RewriteToPromQL(lplan.Root)
	if err != nil {
		t.Fatalf("RewriteToPromQL error: %v", err)
	}
	promExpr, err := promql.FromPromQL(rr.PromQL)
	if err != nil {
		t.Fatalf("parse rewritten PromQL: %v (sql=%s)", err, rr.PromQL)
	}
	plan, err := promql.Compile(promExpr)
	if err != nil {
		t.Fatalf("compile rewritten PromQL: %v", err)
	}
	plan.AttachLogLeaves(rr)
	return plan, rr
}

// pull typed values (like in your existing tests)
func s(v any) string {
	switch x := v.(type) {
	case nil:
		return ""
	case string:
		return x
	case []byte:
		return string(x)
	default:
		return fmt.Sprintf("%v", x)
	}
}

func i64(v any) int64 {
	switch x := v.(type) {
	case int64:
		return x
	case int32:
		return int64(x)
	case int16:
		return int64(x)
	case int8:
		return int64(x)
	case int:
		return int64(x)
	case uint64:
		return int64(x)
	case uint32:
		return int64(x)
	case uint16:
		return int64(x)
	case uint8:
		return int64(x)
	case float64:
		return int64(x)
	case float32:
		return int64(x)
	case []byte:
		// try parsing if driver returned numeric as bytes
		if n, err := strconv.ParseInt(string(x), 10, 64); err == nil {
			return n
		}
		return 0
	default:
		// last resort: string then parse
		if n, err := strconv.ParseInt(fmt.Sprintf("%v", x), 10, 64); err == nil {
			return n
		}
		return 0
	}
}

func openDuckDB(t *testing.T) *sql.DB {
	db, err := sql.Open("duckdb", "")
	if err != nil {
		slog.Error("Error opening duckdb for local parquet", "error", err.Error())
		t.Fatalf("open duckdb: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	return db
}

func mustExec(t *testing.T, db *sql.DB, q string, args ...any) {
	t.Helper()
	if _, err := db.Exec(q, args...); err != nil {
		t.Fatalf("exec failed: %v\nsql:\n%s", err, q)
	}
}

type rowmap map[string]any

func queryAll(t *testing.T, db *sql.DB, q string) []rowmap {
	t.Helper()
	rows, err := db.Query(q)
	if err != nil {
		t.Fatalf("query failed: %v\nsql:\n%s", err, q)
	}
	defer func() { _ = rows.Close() }()

	cols, err := rows.Columns()
	if err != nil {
		t.Fatalf("columns failed: %v", err)
	}

	var out []rowmap
	for rows.Next() {
		raw := make([]any, len(cols))
		ptrs := make([]any, len(cols))
		for i := range raw {
			ptrs[i] = &raw[i]
		}
		if err := rows.Scan(ptrs...); err != nil {
			t.Fatalf("scan failed: %v", err)
		}
		m := make(rowmap, len(cols))
		for i, c := range cols {
			m[c] = raw[i]
		}
		out = append(out, m)
	}
	if err := rows.Err(); err != nil {
		t.Fatalf("rows err: %v", err)
	}
	return out
}

// --- Tests -----------------------------------------------------------------

// 1) count_over_time grouped by json field (level)
func TestRewrite_CountOverTime_ByLevel_JSON(t *testing.T) {
	db := openDuckDB(t)
	// Table must include exemplar defaults the LogQL pipeline always projects.
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Data: 3 events in two 1m buckets (0-60s, 60-120s)
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', '{"level":"INFO","user":"alice","msg":"a"}'),
	(30*1000,  'svc', '{"level":"ERROR","user":"bob","msg":"b"}'),
	(70*1000,  'svc', '{"level":"ERROR","user":"carol","msg":"c"}');
	`)

	// LogQL: count_over_time over 1m, sum by (level)
	q := `sum by (level) (count_over_time({job="svc"} | json [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	// Worker SQL for step=1m
	step := time.Minute
	sql := be.ToWorkerSQL(step)

	// Concretize placeholders for [0, 120s)
	sql = replaceWorkerPlaceholders(sql, 0, 120*1000)

	rows := queryAll(t, db, sql)

	type agg struct {
		bucket int64
		level  string
		count  int64
	}
	var got []agg
	for _, r := range rows {
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			level:  s(r["level"]),
			count:  i64(r["count"]),
		})
	}

	// Validate:
	// bucket 0: level=INFO sum=1, level=ERROR sum=1
	// bucket 60000: level=ERROR sum=1
	var b0Info, b0Err, b1Err int64
	for _, a := range got {
		switch a.bucket {
		case 0:
			switch a.level {
			case "INFO":
				b0Info += a.count
			case "ERROR":
				b0Err += a.count
			}
		case 60000:
			if a.level == "ERROR" {
				b1Err += a.count
			}
		}
	}
	if b0Info != 1 || b0Err != 1 || b1Err != 1 {
		t.Fatalf("unexpected sums: b0(INFO)=%d b0(ERROR)=%d b1(ERROR)=%d; rows=%v", b0Info, b0Err, b1Err, got)
	}
}

func TestUnwrap(t *testing.T) {
	q := `avg_over_time({app="api"} |= "request" | json | unwrap latency_ms [1m])`
	ast, err := logql.FromLogQL(q)
	if err != nil {
		t.Fatalf("FromLogQL error: %v", err)
	}
	lplan, err := logql.CompileLog(ast)
	if err != nil {
		t.Fatalf("CompileLog error: %v", err)
	}
	if len(lplan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(lplan.Leaves))
	}
}

// 2) bytes_rate grouped by logfmt field (user)
// 2) bytes_rate grouped by logfmt field (user)
func TestRewrite_BytesRate_ByUser_Logfmt(t *testing.T) {
	db := openDuckDB(t)

	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Two buckets; ensure different message lengths per user to validate bytes aggregation.
	// lengths:
	//   "ts=5 user=bob msg=\"alpha\""   -> 25
	//   "ts=40 user=carol m=\"zz\""     -> 23    (shorter: use m= instead of msg=)
	//   "ts=65 user=bob msg=\"betaa\""  -> 26    (longer: extra 'a')
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(5*1000,   'web', 'ts=5 user=bob msg="alpha"'),
	(40*1000,  'web', 'ts=40 user=carol m="zz"'),
	(65*1000,  'web', 'ts=65 user=bob msg="betaa"');
	`)

	// LogQL: bytes_rate over 1m, sum by (user)
	q := `sum by (user) (bytes_rate({job="web"} | logfmt [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	step := time.Minute
	sql := be.ToWorkerSQL(step)
	sql = replaceWorkerPlaceholders(sql, 0, 120*1000)

	rows := queryAll(t, db, sql)

	type agg struct {
		bucket int64
		user   string
		sum    int64
	}
	var got []agg
	for _, r := range rows {
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			user:   s(r["user"]),
			sum:    i64(r["sum"]),
		})
	}

	// Expect:
	//   bucket 0 (0..60s): bob=25, carol=23
	//   bucket 60s..120s:  bob=26
	var b0Bob, b0Carol, b1Bob int64
	for _, a := range got {
		switch a.bucket {
		case 0:
			if a.user == "bob" {
				b0Bob += a.sum
			}
			if a.user == "carol" {
				b0Carol += a.sum
			}
		case 60000:
			if a.user == "bob" {
				b1Bob += a.sum
			}
		}
	}
	if b0Bob != 25 || b0Carol != 23 || b1Bob != 26 {
		t.Fatalf("unexpected byte sums: b0(bob)=%d b0(carol)=%d b1(bob)=%d; rows=%v", b0Bob, b0Carol, b1Bob, got)
	}
}

// 3) rate grouped by derived label via label_format (api)
func TestRewrite_Rate_ByDerivedLabel_LabelFormat(t *testing.T) {
	db := openDuckDB(t)

	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// bucket 0: response=ErrorBadGateway, OK
	// bucket 1: response=ErrorOops
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', '{"response":"ErrorBadGateway","msg":"x"}'),
	(20*1000,  'svc', '{"response":"OK","msg":"y"}'),
	(80*1000,  'svc', '{"response":"ErrorOops","msg":"z"}');
	`)

	// Force projection of "response" before label_format so the current builder can
	// reference it inside the template (no semantic change; always-true filter).
	q := `sum by (api) (rate({job="svc"} | json | response=~".*" | label_format api=` +
		"`{{ if hasPrefix \"Error\" .response }}ERROR{{else}}{{.response}}{{end}}`" +
		` [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	step := time.Minute
	sql := be.ToWorkerSQL(step)
	sql = replaceWorkerPlaceholders(sql, 0, 120*1000)

	rows := queryAll(t, db, sql)

	type agg struct {
		bucket int64
		api    string
		sum    int64
	}
	var got []agg
	for _, r := range rows {
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			api:    s(r["api"]),
			sum:    i64(r["sum"]),
		})
	}

	// Expect grouping by derived api:
	// bucket 0: one ERROR, one OK -> sums 1 each
	// bucket 1: one ERROR -> sum 1
	var b0Err, b0OK, b1Err int64
	for _, a := range got {
		switch a.bucket {
		case 0:
			if a.api == "ERROR" {
				b0Err += a.sum
			}
			if a.api == "OK" {
				b0OK += a.sum
			}
		case 60000:
			if a.api == "ERROR" {
				b1Err += a.sum
			}
		}
	}
	if b0Err != 1 || b0OK != 1 || b1Err != 1 {
		t.Fatalf("unexpected sums: b0(ERROR)=%d b0(OK)=%d b1(ERROR)=%d; rows=%v", b0Err, b0OK, b1Err, got)
	}
}

func f64(v any) float64 {
	switch x := v.(type) {
	case float64:
		return x
	case float32:
		return float64(x)
	case int64:
		return float64(x)
	case int32:
		return float64(x)
	case int16:
		return float64(x)
	case int8:
		return float64(x)
	case int:
		return float64(x)
	case uint64:
		return float64(x)
	case uint32:
		return float64(x)
	case uint16:
		return float64(x)
	case uint8:
		return float64(x)
	case []byte:
		if n, err := strconv.ParseFloat(string(x), 64); err == nil {
			return n
		}
		return 0
	default:
		if n, err := strconv.ParseFloat(fmt.Sprintf("%v", x), 64); err == nil {
			return n
		}
		return 0
	}
}

func TestRewrite_Unwrap_Avg_JSON(t *testing.T) {
	db := openDuckDB(t)
	// Table must include exemplar defaults the LogQL pipeline always projects.
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Two 1m buckets: [0..60s) and [60..120s)
	// bucket 0: latency_ms = 100, 200   => avg = 150
	// bucket 1: latency_ms = 300        => avg = 300
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', '{"latency_ms":"100","msg":"a"}'),
	(30*1000,  'svc', '{"latency_ms":"200","msg":"b"}'),
	(70*1000,  'svc', '{"latency_ms":"300","msg":"c"}');
	`)

	// LogQL: avg_over_time over 1m on an unwrapped numeric field from JSON
	q := `avg_over_time({job="svc"} | json | unwrap latency_ms [1m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	step := time.Minute
	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(step), 0, 120*1000)

	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows returned; sql=\n%s", sql)
	}

	type agg struct {
		bucket int64
		avg    float64
	}
	var got []agg
	for _, r := range rows {
		// Prefer 'avg' if present; otherwise compute from sum/count (new behavior).
		var avgV any
		if v, ok := r["avg"]; ok && v != nil {
			avgV = v
		} else {
			sumV, okSum := r["sum"]
			cntV, okCnt := r["count"]
			if !okSum || !okCnt {
				t.Fatalf("expected sum/count columns for avg_over_time; row=%v\nsql=\n%s", r, sql)
			}
			sum := f64(sumV)
			cnt := f64(cntV)
			if cnt == 0 {
				avgV = float64(0)
			} else {
				avgV = sum / cnt
			}
		}
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			avg:    f64(avgV),
		})
	}

	var b0Avg, b1Avg float64
	for _, a := range got {
		switch a.bucket {
		case 0:
			b0Avg = a.avg
		case 60000:
			b1Avg = a.avg
		}
	}

	if b0Avg == 0 && b1Avg == 0 {
		t.Fatalf("unexpected zero avgs; rows=%v\nsql=\n%s", rows, sql)
	}

	const eps = 1e-9
	if !(b0Avg > 150.0-eps && b0Avg < 150.0+eps) || !(b1Avg > 300.0-eps && b1Avg < 300.0+eps) {
		t.Fatalf("unexpected avgs: bucket0=%v bucket1=%v; rows=%v\nsql=\n%s", b0Avg, b1Avg, rows, sql)
	}
}

func TestRewrite_Unwrap_Avg_Logfmt(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Two 1m buckets: [0..60s), [60..120s)
	// bucket 0: latency_ms = 100, 200 => avg = 150
	// bucket 1: latency_ms = 300      => avg = 300
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', 'latency_ms=100 msg=a'),
	(30*1000,  'svc', 'latency_ms=200 msg=b'),
	(70*1000,  'svc', 'latency_ms=300 msg=c');
	`)

	q := `avg_over_time({job="svc"} | logfmt | unwrap latency_ms [1m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows returned; sql=\n%s", sql)
	}

	type agg struct {
		bucket int64
		avg    float64
	}
	var got []agg
	for _, r := range rows {
		var avgV any
		if v, ok := r["avg"]; ok && v != nil {
			avgV = v
		} else {
			sumV, okSum := r["sum"]
			cntV, okCnt := r["count"]
			if !okSum || !okCnt {
				t.Fatalf("expected sum/count columns for avg_over_time; row=%v\nsql=\n%s", r, sql)
			}
			sum := f64(sumV)
			cnt := f64(cntV)
			if cnt == 0 {
				avgV = float64(0)
			} else {
				avgV = sum / cnt
			}
		}
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			avg:    f64(avgV),
		})
	}

	var b0, b1 float64
	for _, a := range got {
		switch a.bucket {
		case 0:
			b0 = a.avg
		case 60000:
			b1 = a.avg
		}
	}
	if b0 == 0 && b1 == 0 {
		t.Fatalf("unexpected zero avgs; rows=%v\nsql=\n%s", rows, sql)
	}
	const eps = 1e-9
	if !(b0 > 150.0-eps && b0 < 150.0+eps) || !(b1 > 300.0-eps && b1 < 300.0+eps) {
		t.Fatalf("unexpected avgs: bucket0=%v bucket1=%v; rows=%v\nsql=\n%s", b0, b1, rows, sql)
	}
}

// min_over_time on unwrap duration(...) from JSON (values converted to seconds)
func TestRewrite_Unwrap_Min_Duration_JSON(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// bucket 0: 100ms, 200ms -> min = 0.1
	// bucket 1: 300ms        -> min = 0.3
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', '{"lat":"100ms","msg":"a"}'),
	(30*1000,  'svc', '{"lat":"200ms","msg":"b"}'),
	(70*1000,  'svc', '{"lat":"300ms","msg":"c"}');
	`)

	q := `min_over_time({job="svc"} | json | unwrap duration(lat) [1m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows returned; sql=\n%s", sql)
	}

	type agg struct {
		bucket int64
		min    float64
	}
	var got []agg
	for _, r := range rows {
		v := r["min"]
		if v == nil {
			if x, ok := r["min_over_time"]; ok {
				v = x
			}
		}
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			min:    f64(v),
		})
	}

	var b0, b1 float64
	for _, a := range got {
		switch a.bucket {
		case 0:
			b0 = a.min
		case 60000:
			b1 = a.min
		}
	}
	const eps = 1e-9
	if !(b0 > 0.1-eps && b0 < 0.1+eps) || !(b1 > 0.3-eps && b1 < 0.3+eps) {
		t.Fatalf("unexpected mins: bucket0=%v bucket1=%v; rows=%v", b0, b1, rows)
	}
}

// max_over_time on unwrap bytes(...) from logfmt (KB are decimal *1000)
func TestRewrite_Unwrap_Max_Bytes_Logfmt(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// bucket 0: 1KB, 2KB -> max = 2000
	// bucket 1: 3KB      -> max = 3000
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', 'size=1KB path=/a'),
	(30*1000,  'svc', 'size=2KB path=/b'),
	(70*1000,  'svc', 'size=3KB path=/c');
	`)

	q := `max_over_time({job="svc"} | logfmt | unwrap bytes(size) [1m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows returned; sql=\n%s", sql)
	}

	type agg struct {
		bucket int64
		max    float64
	}
	var got []agg
	for _, r := range rows {
		v := r["max"]
		if v == nil {
			if x, ok := r["max_over_time"]; ok {
				v = x
			}
		}
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			max:    f64(v),
		})
	}

	var b0, b1 float64
	for _, a := range got {
		switch a.bucket {
		case 0:
			b0 = a.max
		case 60000:
			b1 = a.max
		}
	}
	const eps = 1e-9
	if !(b0 > 2000.0-eps && b0 < 2000.0+eps) || !(b1 > 3000.0-eps && b1 < 3000.0+eps) {
		t.Fatalf("unexpected maxes: bucket0=%v bucket1=%v; rows=%v", b0, b1, rows)
	}
}

func TestRewrite_Unwrap_Max_Duration_Regexp(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Two 1m buckets: [0..60s) and [60..120s)
	// bucket 0 messages contain 2 ms and 4 ms -> max = 4 ms = 0.004 s
	// bucket 1 message contains 3 ms         -> max = 3 ms = 0.003 s
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'kafka', '[LocalLog partition=__cluster_metadata-0, dir=/tmp/kafka-logs] Rolled new log segment at offset 111 in 2 ms.'),
	(30*1000,  'kafka', '[LocalLog partition=__cluster_metadata-0, dir=/tmp/kafka-logs] Rolled new log segment at offset 222 in 4 ms.'),
	(70*1000,  'kafka', '[LocalLog partition=__cluster_metadata-0, dir=/tmp/kafka-logs] Rolled new log segment at offset 333 in 3 ms.');
	`)

	// LogQL: max_over_time on an unwrapped duration captured by regexp
	// NOTE: unwrap duration(...) yields seconds as DOUBLE, so we expect 0.004 and 0.003.
	q := `max_over_time({job="kafka"} |= "Rolled new log segment" | regexp "in (?P<roll_dur>[0-9]+(?:\\.[0-9]+)?\\s*(?:ns|us|µs|ms|s|m|h))" | unwrap duration(roll_dur) [5m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows returned; sql=\n%s", sql)
	}

	type agg struct {
		bucket int64
		max    float64
	}
	var got []agg
	for _, r := range rows {
		v := r["max"]
		if v == nil {
			if x, ok := r["max_over_time"]; ok {
				v = x
			}
		}
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			max:    f64(v),
		})
	}

	var b0, b1 float64
	for _, a := range got {
		switch a.bucket {
		case 0:
			b0 = a.max
		case 60000:
			b1 = a.max
		}
	}

	const eps = 1e-9
	if !(b0 > 0.004-eps && b0 < 0.004+eps) || !(b1 > 0.003-eps && b1 < 0.003+eps) {
		t.Fatalf("unexpected maxes: bucket0=%v bucket1=%v; rows=%v\nsql=\n%s", b0, b1, rows, sql)
	}
}

func TestRewrite_SumByPod_Rate_ServiceFilter(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"resource_k8s_pod_name"   TEXT,
		"log_source"              TEXT
	);`)

	// Two 1m buckets: [0..60s) and [60..120s)
	// We create two streams per pod (log.source = 'a' | 'b') so sum by(pod) actually sums.
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","resource_k8s_pod_name","log_source") VALUES
	-- Pod A, bucket 0 (3)
	(10*1000,  'evt', 'api-gateway', 'api-7f', 'a'),
	(20*1000,  'evt', 'api-gateway', 'api-7f', 'b'),
	(40*1000,  'evt', 'api-gateway', 'api-7f', 'b'),

	-- Pod A, bucket 1 (6)
	(65*1000,  'evt', 'api-gateway', 'api-7f', 'a'),
	(70*1000,  'evt', 'api-gateway', 'api-7f', 'a'),
	(80*1000,  'evt', 'api-gateway', 'api-7f', 'b'),
	(90*1000,  'evt', 'api-gateway', 'api-7f', 'b'),
	(100*1000, 'evt', 'api-gateway', 'api-7f', 'b'),
	(110*1000, 'evt', 'api-gateway', 'api-7f', 'a'),

	-- Pod B, bucket 0 (2)
	(15*1000,  'evt', 'api-gateway', 'api-9x', 'a'),
	(55*1000,  'evt', 'api-gateway', 'api-9x', 'b'),

	-- Pod B, bucket 1 (1)
	(75*1000,  'evt', 'api-gateway', 'api-9x', 'a');
	`)

	// Worker SQL for: sum by (resource_k8s_pod_name)(rate({resource_service_name="api-gateway"}[5m]))
	// NOTE: Worker returns SUMs per bucket; top-level query-api converts to rate later.
	q := `sum by (resource_k8s_pod_name)(rate({resource_service_name="api-gateway"}[5m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows returned; sql=\n%s", sql)
	}

	// Pull the numeric value (worker should expose it as "sum").
	getVal := func(r map[string]any) float64 {
		if v, ok := r["sum"]; ok && v != nil {
			return f64(v)
		}
		if v, ok := r["value"]; ok && v != nil {
			return f64(v)
		}
		for k, v := range r { // fallback for variant aliases
			if strings.HasPrefix(k, "sum_") {
				return f64(v)
			}
		}
		return f64(nil)
	}

	type key struct {
		bucket int64
		pod    string
	}
	got := map[key]float64{}
	for _, r := range rows {
		b := i64(r["bucket_ts"])
		pod := ""
		if v, ok := r[`resource.k8s.pod.name`]; ok && v != nil {
			pod = fmt.Sprint(v)
		} else if v, ok := r[`resource_k8s_pod_name`]; ok && v != nil {
			pod = fmt.Sprint(v)
		} else {
			t.Fatalf("missing pod label column in row: %v\nsql=\n%s", r, sql)
		}
		got[key{bucket: b, pod: pod}] = getVal(r)
	}

	const (
		b0 = int64(0)
		b1 = int64(60000)
	)
	// EXPECT COUNTS (worker sum), NOT RATES. query-api will divide by the range later.
	exp := map[key]float64{
		{bucket: b0, pod: "api-7f"}: 3,
		{bucket: b1, pod: "api-7f"}: 6,
		{bucket: b0, pod: "api-9x"}: 2,
		{bucket: b1, pod: "api-9x"}: 1,
	}

	for k, want := range exp {
		gotv, ok := got[k]
		if !ok {
			t.Fatalf("missing bucket/pod %v in results; rows=%v\nsql=\n%s", k, rows, sql)
		}
		if gotv != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, gotv, want, rows, sql)
		}
	}
}

// --- label_format -----------------------------------------------------------
// Group by a label created via label_format: svc = {{ .resource.service.name }}
func TestRewrite_SumBy_LabelFormat_Service(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT
	);`)

	// Two buckets [0..60s), [60..120s)
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name") VALUES
	(10*1000,  'm', 'svc-a'),
	(20*1000,  'm', 'svc-a'),
	(40*1000,  'm', 'svc-b'),

	(70*1000,  'm', 'svc-a'),
	(80*1000,  'm', 'svc-b'),
	(90*1000,  'm', 'svc-b');`)

	// Group by the label created by label_format; template references a dotted base column.
	q := "sum by (svc)(count_over_time({resource_service_name=~\"svc-.*\"} | label_format svc=`{{ .resource_service_name }}` [1m]))"

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows; sql=\n%s", sql)
	}

	type key struct {
		bucket int64
		svc    string
	}
	got := map[key]int64{}
	for _, r := range rows {
		b := i64(r["bucket_ts"])
		svc := s(r["svc"])
		count := i64(r["count"])
		got[key{b, svc}] = count
	}

	exp := map[key]int64{
		{0, "svc-a"}:     2,
		{0, "svc-b"}:     1,
		{60000, "svc-a"}: 1,
		{60000, "svc-b"}: 2,
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

// --- json -------------------------------------------------------------------
// Group by JSON-extracted label: user
func TestRewrite_SumBy_JSON_User(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Two buckets:
	// bucket 0: users alice(2), bob(1)
	// bucket 1: users alice(1), carol(1)
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', '{"user":"alice","msg":"a"}'),
	(25*1000,  'svc', '{"user":"alice","msg":"b"}'),
	(50*1000,  'svc', '{"user":"bob","msg":"c"}'),

	(70*1000,  'svc', '{"user":"alice","msg":"d"}'),
	(95*1000,  'svc', '{"user":"carol","msg":"e"}');`)

	q := `sum by (user)(count_over_time({job="svc"} | json [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows; sql=\n%s", sql)
	}

	type key struct {
		bucket int64
		user   string
	}
	got := map[key]int64{}
	for _, r := range rows {
		b := i64(r["bucket_ts"])
		u := s(r["user"])
		count := i64(r["count"])
		got[key{b, u}] = count
	}

	exp := map[key]int64{
		{0, "alice"}:     2,
		{0, "bob"}:       1,
		{60000, "alice"}: 1,
		{60000, "carol"}: 1,
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

// --- regexp -----------------------------------------------------------------
// Group by a named capture created by regexp: (?P<pod>...)
func TestRewrite_SumBy_Regexp_Pod(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Messages embed "pod=<name>"
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', 'pod=api-7f src=a'),
	(20*1000,  'svc', 'pod=api-7f src=b'),
	(40*1000,  'svc', 'pod=api-9x src=b'),

	(70*1000,  'svc', 'pod=api-7f src=a'),
	(90*1000,  'svc', 'pod=api-9x src=b');`)

	// Named capture "pod"
	q := `sum by (pod)(count_over_time({job="svc"} | regexp "pod=(?P<pod>[^\\s]+)" [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows; sql=\n%s", sql)
	}

	type key struct {
		bucket int64
		pod    string
	}
	got := map[key]int64{}
	for _, r := range rows {
		b := i64(r["bucket_ts"])
		pod := s(r["pod"])
		count := i64(r["count"])
		got[key{b, pod}] = count
	}

	exp := map[key]int64{
		{0, "api-7f"}:     2,
		{0, "api-9x"}:     1,
		{60000, "api-7f"}: 1,
		{60000, "api-9x"}: 1,
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

// --- logfmt (pattern-like) --------------------------------------------------
// Group by a key parsed from logfmt: svc=<name>
func TestRewrite_SumBy_Logfmt_Svc(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Messages like: "svc=api user=alice" etc.
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', 'svc=api user=a'),
	(25*1000,  'svc', 'svc=api user=b'),
	(45*1000,  'svc', 'svc=auth user=c'),

	(70*1000,  'svc', 'svc=api user=d'),
	(95*1000,  'svc', 'svc=auth user=e');`)

	q := `sum by (svc)(count_over_time({job="svc"} | logfmt [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows; sql=\n%s", sql)
	}

	type key struct {
		bucket int64
		svc    string
	}
	got := map[key]int64{}
	for _, r := range rows {
		b := i64(r["bucket_ts"])
		svc := s(r["svc"])
		count := i64(r["count"])
		got[key{b, svc}] = count
	}

	exp := map[key]int64{
		{0, "api"}:      2,
		{0, "auth"}:     1,
		{60000, "api"}:  1,
		{60000, "auth"}: 1,
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

//  1. Group-by a BASE label (job) while a parser is present and a parser-created
//     label is used in a filter (json + user="alice")
func TestRewrite_CountOverTime_ByBaseJob_WithJSONFilter(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// bucket 0: (svc, alice), (svc, bob), (api, alice)
	// bucket 1: (svc, alice), (api, carol)
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', '{"user":"alice"}'),
	(20*1000,  'svc', '{"user":"bob"}'),
	(30*1000,  'api', '{"user":"alice"}'),
	(70*1000,  'svc', '{"user":"alice"}'),
	(90*1000,  'api', '{"user":"carol"}');`)

	// Keep only rows with user="alice" (parser-created), but group by base label "job".
	q := `sum by (job) (count_over_time({job=~".+"} | json | user="alice" [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)

	type key struct {
		bucket int64
		job    string
	}
	got := map[key]int64{}
	for _, r := range rows {
		got[key{
			bucket: i64(r["bucket_ts"]),
			job:    s(r["job"]),
		}] = i64(r["count"])
	}

	exp := map[key]int64{
		{0, "svc"}:     1, // (10s, svc, alice)
		{0, "api"}:     1, // (30s, api, alice)
		{60000, "svc"}: 1, // (70s, svc, alice)
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

//  2. Matcher on a parser-created label (json), grouped by that same label.
//     Only rows with user="alice" should remain.
func TestRewrite_CountOverTime_ByUser_JSON_WithParserMatcher(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// bucket 0: alice, bob, alice  -> alice x2
	// bucket 1: alice, bob         -> alice x1
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', '{"user":"alice"}'),
	(20*1000,  'svc', '{"user":"bob"}'),
	(40*1000,  'svc', '{"user":"alice"}'),
	(70*1000,  'svc', '{"user":"alice"}'),
	(90*1000,  'svc', '{"user":"bob"}');`)

	q := `sum by (user) (count_over_time({job="svc"} | json | user="alice" [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)

	type key struct {
		bucket int64
		user   string
	}
	got := map[key]int64{}
	for _, r := range rows {
		got[key{
			bucket: i64(r["bucket_ts"]),
			user:   s(r["user"]),
		}] = i64(r["count"])
	}

	exp := map[key]int64{
		{0, "alice"}:     2,
		{60000, "alice"}: 1,
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

//  3. Multiple group keys: sum by (user, svc) with logfmt parser.
//     Validates projection & GROUP BY for multi-key aggregation.
func TestRewrite_CountOverTime_ByUserSvc_Logfmt_MultiGroup(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// bucket 0:
	//   svc=api  user=a
	//   svc=api  user=b
	//   svc=auth user=a
	// bucket 1:
	//   svc=api  user=a
	//   svc=auth user=b
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp", job, "log_message") VALUES
	(10*1000,  'svc', 'svc=api user=a'),
	(25*1000,  'svc', 'svc=api user=b'),
	(50*1000,  'svc', 'svc=auth user=a'),
	(70*1000,  'svc', 'svc=api user=a'),
	(95*1000,  'svc', 'svc=auth user=b');`)

	q := `sum by (user, svc) (count_over_time({job="svc"} | logfmt [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)

	type key struct {
		bucket int64
		user   string
		svc    string
	}
	got := map[key]int64{}
	for _, r := range rows {
		got[key{
			bucket: i64(r["bucket_ts"]),
			user:   s(r["user"]),
			svc:    s(r["svc"]),
		}] = i64(r["count"])
	}

	exp := map[key]int64{
		{0, "a", "api"}:      1,
		{0, "b", "api"}:      1,
		{0, "a", "auth"}:     1,
		{60000, "a", "api"}:  1,
		{60000, "b", "auth"}: 1,
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

//  4. Group by a BASE column (resource.k8s.pod.name) while json is present and used
//     in a filter. With the "mark all non-base groupKeys as parser-created" assumption,
//     s0 won't project the pod column and the query will fail to bind.
//     Once fixed (only mark truly parser-created keys), this should pass.
func TestRewrite_CountOverTime_ByBasePod_WithJSONFilter(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"resource_k8s_pod_name"   TEXT
	);`)

	// Two 1m buckets: [0..60s) and [60..120s)
	// bucket 0: (pod=api-7f, alice), (pod=api-7f, bob), (pod=api-9x, alice)
	// bucket 1: (pod=api-7f, alice), (pod=api-9x, bob)
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","resource_k8s_pod_name") VALUES
	(10*1000,  '{"user":"alice"}', 'api-gateway', 'api-7f'),
	(20*1000,  '{"user":"bob"}',   'api-gateway', 'api-7f'),
	(40*1000,  '{"user":"alice"}', 'api-gateway', 'api-9x'),
	(70*1000,  '{"user":"alice"}', 'api-gateway', 'api-7f'),
	(90*1000,  '{"user":"bob"}',   'api-gateway', 'api-9x');`)

	// Keep only rows with user="alice" (parser-created), but group by the BASE pod label.
	// IMPORTANT: matcher only references resource_service_name (so pod is NOT pulled in via matchers).
	q := `sum by (resource_k8s_pod_name) (count_over_time({resource_service_name="api-gateway"} | json | user="alice" [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)

	// This call will currently FAIL with a DuckDB binder error because the column
	// "resource_k8s.pod.name" (or its underscored alias) isn't projected in s0.
	// After fixing the assumption (only mark parser-created keys as such), it will pass.
	rows := queryAll(t, db, sql)

	type key struct {
		bucket int64
		pod    string
	}
	got := map[key]int64{}
	for _, r := range rows {
		pod := ""
		if v, ok := r[`resource_k8s_pod_name`]; ok && v != nil {
			pod = fmt.Sprint(v)
		} else {
			t.Fatalf("missing pod label column in row: %v\nsql=\n%s", r, sql)
		}

		got[key{
			bucket: i64(r["bucket_ts"]),
			pod:    pod,
		}] = i64(r["count"])
	}

	exp := map[key]int64{
		{0, "api-7f"}:     1, // (10s, alice)
		{0, "api-9x"}:     1, // (40s, alice)
		{60000, "api-7f"}: 1, // (70s, alice)
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

// Group by one PARSED (json) label and one BASE label (pod)
func TestRewrite_CountOverTime_ByUserAndBasePod_JSON(t *testing.T) {
	db := openDuckDB(t)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"resource_k8s_pod_name"   TEXT
	);`)

	// Two 1m buckets: [0..60s) and [60..120s)
	// bucket 0: (pod=api-7f, alice), (pod=api-7f, bob), (pod=api-9x, alice)
	// bucket 1: (pod=api-7f, alice), (pod=api-9x, bob)
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","resource_k8s_pod_name") VALUES
	(10*1000,  '{"user":"alice"}', 'api-gateway', 'api-7f'),
	(20*1000,  '{"user":"bob"}',   'api-gateway', 'api-7f'),
	(40*1000,  '{"user":"alice"}', 'api-gateway', 'api-9x'),
	(70*1000,  '{"user":"alice"}', 'api-gateway', 'api-7f'),
	(90*1000,  '{"user":"bob"}',   'api-gateway', 'api-9x');`)

	// Group by parsed json label (user) AND base label (pod).
	q := `sum by (user, resource_k8s_pod_name) (count_over_time({resource_service_name="api-gateway"} | json [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(time.Minute), 0, 120*1000)
	rows := queryAll(t, db, sql)

	type key struct {
		bucket int64
		user   string
		pod    string
	}
	got := map[key]int64{}
	for _, r := range rows {
		// Allow either dotted or underscored naming for pod depending on rewrite
		pod := ""
		if v, ok := r[`resource_k8s_pod_name`]; ok && v != nil {
			pod = fmt.Sprint(v)
		} else {
			t.Fatalf("missing pod label column in row: %v\nsql=\n%s", r, sql)
		}
		got[key{
			bucket: i64(r["bucket_ts"]),
			user:   s(r["user"]),
			pod:    pod,
		}] = i64(r["count"])
	}

	exp := map[key]int64{
		{0, "alice", "api-7f"}:     1,
		{0, "bob", "api-7f"}:       1,
		{0, "alice", "api-9x"}:     1,
		{60000, "alice", "api-7f"}: 1,
		{60000, "bob", "api-9x"}:   1,
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum for %v: got=%v want=%v; rows=%v\nsql=\n%s", k, got[k], want, rows, sql)
		}
	}
}

// Count by pod where a JSON filter matches, two workers, Eval at the coordinator.
// The outer COUNT should row-count the inner results (presence per pod/bucket → 1).
func TestLog_CountByPod_WithJSONFilter_TwoWorkers_Eval(t *testing.T) {
	// --- Worker 1: pod api-7f -------------------------------------------------
	db1 := openDuckDB(t)
	mustExec(t, db1, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"resource_k8s_pod_name"   TEXT
	);`)
	// bucket 0: one alice (match), one bob (filtered out)
	// bucket 1: two alice matches
	mustExec(t, db1, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","resource_k8s_pod_name") VALUES
	(10*1000,  '{"user":"alice","m":"x"}', 'api-gateway', 'api-7f'),
	(20*1000,  '{"user":"bob","m":"y"}',   'api-gateway', 'api-7f'),
	(70*1000,  '{"user":"alice","m":"z"}', 'api-gateway', 'api-7f'),
	(80*1000,  '{"user":"alice","m":"w"}', 'api-gateway', 'api-7f');
	`)

	// --- Worker 2: pod api-9x -------------------------------------------------
	db2 := openDuckDB(t)
	mustExec(t, db2, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"resource_k8s_pod_name"   TEXT
	);`)
	// bucket 0: one alice (match)
	// bucket 1: one alice (match)
	mustExec(t, db2, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","resource_k8s_pod_name") VALUES
	(40*1000,  '{"user":"alice","m":"p"}', 'api-gateway', 'api-9x'),
	(100*1000, '{"user":"alice","m":"q"}', 'api-gateway', 'api-9x');
	`)

	// LogQL: count by (pod) of count_over_time(...) with a JSON filter user="alice"
	// Inner count_over_time groups by pod identity via the rewrite (BaseExpr.GroupBy keeps pod),
	// so each worker emits one row per (bucket,pod) with "sum" (the event count).
	// At Eval(), outer count by (pod) should produce 1 per pod per bucket (presence).
	q := `count by (resource_k8s_pod_name)(
			count_over_time(
				{resource_service_name="api-gateway"} | json | user="alice" [1m]
			)
		)`

	// Compile full plan via LogQL → Rewrite → PromQL.
	plan, rr := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	leaf := plan.Leaves[0]
	_ = rr // (kept here to show we used the rewrite path)

	step := time.Minute
	workerSQL := replaceWorkerPlaceholders(leaf.ToWorkerSQL(step), 0, 120*1000)

	// Run leaf SQL on both workers.
	rows1 := queryAll(t, db1, workerSQL)
	rows2 := queryAll(t, db2, workerSQL)
	if len(rows1) == 0 && len(rows2) == 0 {
		t.Fatalf("no rows from both workers; sql=\n%s", workerSQL)
	}

	// Build coordinator inputs per bucket. We pass the inner "sum" (event count),
	// but outer COUNT (not pass-through) will row-count, so the numeric value is
	// irrelevant; presence of the row is what matters.
	type bucket = int64
	perBucket := map[bucket][]promql.SketchInput{}

	addRows := func(rows []rowmap) {
		for _, r := range rows {
			b := i64(r["bucket_ts"])
			// allow dotted or underscored form
			pod := s(r[`resource_k8s_pod_name`])
			count := float64(i64(r["count"])) // inner count_over_time result (may be >0)
			if count == 0 {
				continue
			}
			perBucket[b] = append(perBucket[b], promql.SketchInput{
				ExprID:         leaf.ID,
				OrganizationID: "org-test",
				Timestamp:      b,
				Frequency:      int64(step / time.Second),
				SketchTags: promql.SketchTags{
					Tags:       map[string]any{"resource_k8s_pod_name": pod},
					SketchType: promql.SketchMAP,
					Agg:        map[string]float64{"count": count},
				},
			})
		}
	}
	addRows(rows1)
	addRows(rows2)

	// Deterministic bucket order.
	var buckets []int64
	for b := range perBucket {
		buckets = append(buckets, b)
	}
	sort.Slice(buckets, func(i, j int) bool { return buckets[i] < buckets[j] })

	// Evaluate per bucket; expect one result per (bucket,pod) with value==1.
	type key struct {
		bucket int64
		pod    string
	}
	got := map[key]int64{}

	for _, b := range buckets {
		out := plan.Root.Eval(promql.SketchGroup{
			Timestamp: b,
			Group:     map[string][]promql.SketchInput{leaf.ID: perBucket[b]},
		}, step)

		for _, er := range out {
			pod := s(er.Tags[`resource_k8s_pod_name`])
			// Outer COUNT (no pass-through) → row count = 1 per (bucket,pod)
			got[key{b, pod}] = int64(er.Value.Num + 0.5)
		}
	}

	const (
		b0 = int64(0)
		b1 = int64(60000)
	)
	exp := map[key]int64{
		{b0, "api-7f"}: 1, // worker1 had one alice in bucket0
		{b0, "api-9x"}: 1, // worker2 had one alice in bucket0
		{b1, "api-7f"}: 1, // worker1 had two alice in bucket1 (still 1 after outer count)
		{b1, "api-9x"}: 1, // worker2 had one alice in bucket1
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected count-by-pod for %v: got=%v want=%v\nrows1=%v\nrows2=%v\nsql=\n%s",
				k, got[k], want, rows1, rows2, workerSQL)
		}
	}
}

func TestLog_SumOfCountsByPod_WithJSONFilter_TwoWorkers_Eval(t *testing.T) {
	// --- Worker 1: pod api-7f -------------------------------------------------
	db1 := openDuckDB(t)
	mustExec(t, db1, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"resource_k8s_pod_name"   TEXT
	);`)
	// bucket 0: one alice (match), one bob (filtered out)
	// bucket 1: two alice matches
	mustExec(t, db1, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","resource_k8s_pod_name") VALUES
	(10*1000,  '{"user":"alice","m":"x"}', 'api-gateway', 'api-7f'),
	(20*1000,  '{"user":"bob","m":"y"}',   'api-gateway', 'api-7f'),
	(70*1000,  '{"user":"alice","m":"z"}', 'api-gateway', 'api-7f'),
	(80*1000,  '{"user":"alice","m":"w"}', 'api-gateway', 'api-7f');
	`)

	// --- Worker 2: pod api-9x -------------------------------------------------
	db2 := openDuckDB(t)
	mustExec(t, db2, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"resource_k8s_pod_name"   TEXT
	);`)
	// bucket 0: one alice (match)
	// bucket 1: one alice (match)
	mustExec(t, db2, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","resource_k8s_pod_name") VALUES
	(40*1000,  '{"user":"alice","m":"p"}', 'api-gateway', 'api-9x'),
	(100*1000, '{"user":"alice","m":"q"}', 'api-gateway', 'api-9x');
	`)

	// LogQL: sum by (pod) of count_over_time(...) with a JSON filter user="alice"
	q := `sum by (resource_k8s_pod_name)(
			count_over_time(
				{resource_service_name="api-gateway"} | json | user="alice" [1m]
			)
		)`

	// Compile full plan via LogQL → Rewrite → PromQL.
	plan, rr := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	leaf := plan.Leaves[0]
	_ = rr // show rewrite path was used

	step := time.Minute
	workerSQL := replaceWorkerPlaceholders(leaf.ToWorkerSQL(step), 0, 120*1000)

	// Run leaf SQL on both workers.
	rows1 := queryAll(t, db1, workerSQL)
	rows2 := queryAll(t, db2, workerSQL)
	if len(rows1) == 0 && len(rows2) == 0 {
		t.Fatalf("no rows from both workers; sql=\n%s", workerSQL)
	}

	// Build coordinator inputs per bucket.
	type bucket = int64
	perBucket := map[bucket][]promql.SketchInput{}

	addRows := func(rows []rowmap) {
		for _, r := range rows {
			b := i64(r["bucket_ts"])
			pod := s(r[`resource_k8s_pod_name`])
			count := float64(i64(r["count"]))
			if count == 0 {
				continue
			}
			perBucket[b] = append(perBucket[b], promql.SketchInput{
				ExprID:         leaf.ID,
				OrganizationID: "org-test",
				Timestamp:      b,
				Frequency:      int64(step / time.Second),
				SketchTags: promql.SketchTags{
					Tags:       map[string]any{"resource_k8s_pod_name": pod},
					SketchType: promql.SketchMAP,
					Agg:        map[string]float64{"count": count},
				},
			})
		}
	}
	addRows(rows1)
	addRows(rows2)

	// Deterministic bucket order.
	var buckets []int64
	for b := range perBucket {
		buckets = append(buckets, b)
	}
	sort.Slice(buckets, func(i, j int) bool { return buckets[i] < buckets[j] })

	// Evaluate per bucket; expect the actual per-pod line counts.
	type key struct {
		bucket int64
		pod    string
	}
	got := map[key]int64{}

	for _, b := range buckets {
		out := plan.Root.Eval(promql.SketchGroup{
			Timestamp: b,
			Group:     map[string][]promql.SketchInput{leaf.ID: perBucket[b]},
		}, step)

		for _, er := range out {
			pod := s(er.Tags[`resource_k8s_pod_name`])
			// sum by (pod) of the inner counts → actual #lines per pod per bucket
			got[key{b, pod}] = int64(er.Value.Num + 0.5)
		}
	}

	const (
		b0 = int64(0)
		b1 = int64(60000)
	)
	exp := map[key]int64{
		{b0, "api-7f"}: 1, // worker1 had one alice in bucket0
		{b0, "api-9x"}: 1, // worker2 had one alice in bucket0
		{b1, "api-7f"}: 2, // worker1 had two alice in bucket1
		{b1, "api-9x"}: 1, // worker2 had one alice in bucket1
	}
	for k, want := range exp {
		if got[k] != want {
			t.Fatalf("unexpected sum-of-counts for %v: got=%v want=%v\nrows1=%v\nrows2=%v\nsql=\n%s",
				k, got[k], want, rows1, rows2, workerSQL)
		}
	}
}

func TestLog_SumRateCounter_Unwrap_TwoWorkers_Eval(t *testing.T) {
	// ---------------- Worker 1 ----------------
	db1 := openDuckDB(t)
	mustExec(t, db1, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT
	);`)
	// bucket 0 (0..60s): 10 + 20
	// bucket 1 (60..120s): 30
	mustExec(t, db1, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name") VALUES
	(10*1000,  'took 10 ms', 'kafka'),
	(20*1000,  'took 20ms',  'kafka'),
	(70*1000,  'took 30 ms', 'kafka');
	`)

	// ---------------- Worker 2 ----------------
	db2 := openDuckDB(t)
	mustExec(t, db2, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT
	);`)
	// bucket 0: 40
	// bucket 1: 50
	mustExec(t, db2, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name") VALUES
	(40*1000,  'took 40ms',  'kafka'),
	(100*1000, 'took 50 ms', 'kafka');
	`)

	// Coordinator expression: sum(rate_counter(... unwrap dur [1m]))
	q := `sum(rate_counter({resource_service_name="kafka"} | regexp "(?P<dur>[0-9]+(?:\\.[0-9]+)?)\\s*(?:ns|us|µs|ms|s|m|h)" | unwrap dur [1m]))`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	leaf := plan.Leaves[0]

	step := time.Minute
	workerSQL := replaceWorkerPlaceholders(leaf.ToWorkerSQL(step), 0, 120*1000)

	// Sanity: pipeline should include regexp + unwrap and aggregate the numeric.
	if !strings.Contains(workerSQL, "regexp_extract") || !strings.Contains(workerSQL, "AS __unwrap_value") {
		t.Fatalf("expected regexp+unwrap in SQL, got:\n%s", workerSQL)
	}
	if !strings.Contains(workerSQL, "SUM(__unwrap_value)") {
		t.Fatalf("expected SUM(__unwrap_value) in SQL, got:\n%s", workerSQL)
	}

	// Run leaf SQL on both workers (per-bucket numeric SUMs from the 1m window).
	rows1 := queryAll(t, db1, workerSQL)
	rows2 := queryAll(t, db2, workerSQL)
	if len(rows1) == 0 && len(rows2) == 0 {
		t.Fatalf("no rows from both workers; sql=\n%s", workerSQL)
	}

	// Build coordinator inputs per bucket for the leaf.
	// rate_counter expects the *sum per window* so it can compute delta/step.
	type bucket = int64
	perBucket := map[bucket][]promql.SketchInput{}

	getSum := func(r rowmap) float64 {
		if v, ok := r["sum"]; ok {
			return f64(v)
		}
		t.Fatalf("sum column not found in row: %v", r)
		return 0
	}

	addRows := func(rows []rowmap) {
		for _, r := range rows {
			b := i64(r["bucket_ts"])
			sum := getSum(r)
			// No group-by labels here; we want global sum(rate_counter(...)).
			perBucket[b] = append(perBucket[b], promql.SketchInput{
				ExprID:         leaf.ID,
				OrganizationID: "org-test",
				Timestamp:      b,
				Frequency:      int64(step / time.Second),
				SketchTags: promql.SketchTags{
					Tags:       map[string]any{}, // global
					SketchType: promql.SketchMAP,
					Agg:        map[string]float64{"sum": sum},
				},
			})
		}
	}
	addRows(rows1)
	addRows(rows2)

	// Deterministic bucket order.
	var buckets []int64
	for b := range perBucket {
		buckets = append(buckets, b)
	}
	sort.Slice(buckets, func(i, j int) bool { return buckets[i] < buckets[j] })

	// Evaluate per bucket; coordinator should compute the per-second rate of the window sums,
	// then outer sum(...) (no groups → a single series).
	got := map[bucket]float64{}
	for _, b := range buckets {
		out := plan.Root.Eval(promql.SketchGroup{
			Timestamp: b,
			Group:     map[string][]promql.SketchInput{leaf.ID: perBucket[b]},
		}, step)

		if len(out) != 1 {
			t.Fatalf("expected 1 output series at bucket %d, got %d: %v", b, len(out), out)
		}
		got[b] = out["default"].Value.Num
	}

	const (
		b0 = int64(0)
		b1 = int64(60000)
	)
	exp := map[bucket]float64{
		b0: 70.0 / 60.0,
		b1: 80.0 / 60.0,
	}

	const eps = 1e-9
	for b, want := range exp {
		if math.Abs(got[b]-want) > eps {
			t.Fatalf("unexpected sum(rate_counter unwrap) at bucket %d: got=%v want=%v\nrows1=%v\nrows2=%v\nsql=\n%s",
				b, got[b], want, rows1, rows2, workerSQL)
		}
	}
}

func TestRewrite_FilterThenJSON_UnwrapNested_Max(t *testing.T) {
	db := openDuckDB(t)
	// Include the default columns s0 always projects.
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		job TEXT,
		"log_message"     TEXT
	);`)

	// Two 1m buckets: [0..60s), [60..120s)
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp","chq_id","log_level","chq_fingerprint",job,"log_message") VALUES
	(10*1000, 'id-a', '', '-4446492996171837732', 'svc', '{"req":{"url":"/foo","lat_ms":"120"},"meta":{"trace":"t1"}}'),
	(30*1000, 'id-b', '', '-4446492996171837732', 'svc', '{"req":{"url":"/bar","lat_ms":"200"},"meta":{"trace":"t2"}}'),
	(70*1000, 'id-c', '', '-4446492996171837732', 'svc', '{"req":{"url":"/foo","lat_ms":"350"},"meta":{"trace":"t3"}}');
	`)

	// Filter first, then | json mapping, then unwrap nested field, then max_over_time
	q := `max_over_time({job="svc"} |= "/foo" | json lat_ms="req.lat_ms" | unwrap lat_ms [1m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	step := time.Minute
	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(step), 0, 120*1000)
	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows returned; sql=\n%s", sql)
	}

	type agg struct {
		bucket int64
		maxv   float64
	}
	var got []agg
	for _, r := range rows {
		v := r["max"]
		if v == nil {
			if x, ok := r["max_over_time"]; ok {
				v = x
			}
		}
		got = append(got, agg{
			bucket: i64(r["bucket_ts"]),
			maxv:   f64(v),
		})
	}

	var b0, b1 float64
	for _, a := range got {
		switch a.bucket {
		case 0:
			b0 = a.maxv
		case 60000:
			b1 = a.maxv
		}
	}
	const eps = 1e-9
	if !(math.Abs(b0-120.0) < eps && math.Abs(b1-350.0) < eps) {
		t.Fatalf("unexpected maxes: bucket0=%v bucket1=%v; rows=%v\nsql=\n%s", b0, b1, rows, sql)
	}
}

func TestLog_AvgOverTime_Regexp_UnwrapBytes_TwoWorkers_Eval(t *testing.T) {
	// ---------------- Worker 1 ----------------
	db1 := openDuckDB(t)
	mustExec(t, db1, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT
	);`)
	// bucket 0 (0..60s): 100, 200  -> sum=300 count=2
	// bucket 1 (60..120s): 300     -> sum=300 count=1
	mustExec(t, db1, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name") VALUES
	(10*1000,  'processed bytes=100', 'kafka'),
	(40*1000,  'processed bytes=200', 'kafka'),
	(70*1000,  'processed bytes=300', 'kafka');
	`)

	// ---------------- Worker 2 ----------------
	db2 := openDuckDB(t)
	mustExec(t, db2, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT
	);`)
	// bucket 0: 50   -> sum=50  count=1
	// bucket 1: 400  -> sum=400 count=1
	mustExec(t, db2, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name") VALUES
	(20*1000,  'processed bytes=50',  'kafka'),
	(100*1000, 'processed bytes=400', 'kafka');
	`)

	// Coordinator expression: avg_over_time on the captured "bytes" number.
	q := `avg_over_time({resource_service_name="kafka"} | regexp "bytes=(?P<bytes>[0-9]+)" | unwrap bytes [1m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	leaf := plan.Leaves[0]

	step := time.Minute
	workerSQL := replaceWorkerPlaceholders(leaf.ToWorkerSQL(step), 0, 120*1000)

	// Sanity: with the refactor, avg_over_time worker SQL must project sum+count of __unwrap_value.
	if !strings.Contains(workerSQL, "AS __unwrap_value") {
		t.Fatalf("expected unwrap projection in SQL, got:\n%s", workerSQL)
	}
	if !strings.Contains(workerSQL, "SUM(__unwrap_value) AS sum") || !strings.Contains(workerSQL, "COUNT(__unwrap_value) AS count") {
		t.Fatalf("expected SUM(__unwrap_value) AS sum, COUNT(*) AS count in SQL, got:\n%s", workerSQL)
	}

	// Run the leaf SQL on both workers.
	rows1 := queryAll(t, db1, workerSQL)
	rows2 := queryAll(t, db2, workerSQL)
	if len(rows1) == 0 && len(rows2) == 0 {
		t.Fatalf("no rows from both workers; sql=\n%s", workerSQL)
	}

	// Build coordinator inputs per bucket for the leaf.
	// avg_over_time expects per-window sum and count so it can compute sum/count.
	type bucket = int64
	perBucket := map[bucket][]promql.SketchInput{}

	addRows := func(rows []rowmap) {
		for _, r := range rows {
			b := i64(r["bucket_ts"])
			sumV, ok := r["sum"]
			if !ok {
				t.Fatalf("missing sum column in row: %v", r)
			}
			cntV, ok := r["count"]
			if !ok {
				t.Fatalf("missing count column in row: %v", r)
			}
			perBucket[b] = append(perBucket[b], promql.SketchInput{
				ExprID:         leaf.ID,
				OrganizationID: "org-test",
				Timestamp:      b,
				Frequency:      int64(step / time.Second),
				SketchTags: promql.SketchTags{
					Tags:       map[string]any{}, // global
					SketchType: promql.SketchMAP,
					Agg: map[string]float64{
						"sum":   f64(sumV),
						"count": f64(cntV),
					},
				},
			})
		}
	}
	addRows(rows1)
	addRows(rows2)

	// Deterministic bucket order.
	var buckets []int64
	for b := range perBucket {
		buckets = append(buckets, b)
	}
	sort.Slice(buckets, func(i, j int) bool { return buckets[i] < buckets[j] })

	// Evaluate per bucket; outer avg_over_time over merged worker inputs.
	got := map[bucket]float64{}
	for _, b := range buckets {
		out := plan.Root.Eval(promql.SketchGroup{
			Timestamp: b,
			Group:     map[string][]promql.SketchInput{leaf.ID: perBucket[b]},
		}, step)

		if len(out) != 1 {
			t.Fatalf("expected 1 output series at bucket %d, got %d: %v", b, len(out), out)
		}
		got[b] = out["default"].Value.Num
	}

	const (
		b0  = int64(0)
		b1  = int64(60000)
		eps = 1e-9
	)
	// Expected: bucket0: (100+200+50)/3 = 350/3
	//           bucket1: (300+400)/2 = 350
	exp := map[bucket]float64{
		b0: 350.0 / 3.0,
		b1: 350.0,
	}

	for b, want := range exp {
		if math.Abs(got[b]-want) > eps {
			t.Fatalf("unexpected avg_over_time at bucket %d: got=%v want=%v\nrows1=%v\nrows2=%v\nsql=\n%s",
				b, got[b], want, rows1, rows2, workerSQL)
		}
	}
}

func TestLog_CountOverTime_LineRegex_TwoWorkers_Eval(t *testing.T) {
	// ---------------- Worker 1 ----------------
	db1 := openDuckDB(t)
	mustExec(t, db1, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT
	);`)
	// bucket 0 (0..60s): 2 matches
	// bucket 1 (60..120s): 1 match
	mustExec(t, db1, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name") VALUES
	(10*1000,  'topic deleted by user=alice', 'kafka'),
	(25*1000,  'partition deleted id=12',     'kafka'),
	(70*1000,  'log deleted file=0001',       'kafka');
	`)

	// ---------------- Worker 2 ----------------
	db2 := openDuckDB(t)
	mustExec(t, db2, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT
	);`)
	// bucket 0: 3 matches
	// bucket 1: 0 matches
	mustExec(t, db2, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name") VALUES
	(20*1000,  'deleted key=1',     'kafka'),
	(40*1000,  'deleted key=2',     'kafka'),
	(55*1000,  'deleted key=3',     'kafka'),
	(100*1000, 'processed key=9',   'kafka');
	`)

	// Coordinator expression: count_over_time on a regex line filter.
	q := `count_over_time({resource_service_name="kafka"} |~ "deleted" [1m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf, got %d", len(plan.Leaves))
	}
	leaf := plan.Leaves[0]

	step := time.Minute
	workerSQL := replaceWorkerPlaceholders(leaf.ToWorkerSQL(step), 0, 120*1000)

	// Sanity: worker SQL should count rows per bucket.
	if !strings.Contains(workerSQL, "COUNT(*) AS count") && !strings.Contains(workerSQL, "COUNT(*) AS COUNT") {
		t.Fatalf("expected COUNT(*) AS count in SQL, got:\n%s", workerSQL)
	}

	// Run the leaf SQL on both workers (per-bucket counts from the 1m window).
	rows1 := queryAll(t, db1, workerSQL)
	rows2 := queryAll(t, db2, workerSQL)
	if len(rows1) == 0 && len(rows2) == 0 {
		t.Fatalf("no rows from both workers; sql=\n%s", workerSQL)
	}

	// Build coordinator inputs per bucket for the leaf.
	// count_over_time expects per-window counts so it can sum across workers.
	type bucket = int64
	perBucket := map[bucket][]promql.SketchInput{}

	addRows := func(rows []rowmap) {
		for _, r := range rows {
			b := i64(r["bucket_ts"])
			cv, ok := r["count"]
			if !ok {
				t.Fatalf("missing count column in row: %v", r)
			}
			perBucket[b] = append(perBucket[b], promql.SketchInput{
				ExprID:         leaf.ID,
				OrganizationID: "org-test",
				Timestamp:      b,
				Frequency:      int64(step / time.Second),
				SketchTags: promql.SketchTags{
					Tags:       map[string]any{}, // global (no extra grouping)
					SketchType: promql.SketchMAP,
					Agg: map[string]float64{
						"count": f64(cv),
					},
				},
			})
		}
	}
	addRows(rows1)
	addRows(rows2)

	// Deterministic bucket order.
	var buckets []int64
	for b := range perBucket {
		buckets = append(buckets, b)
	}
	sort.Slice(buckets, func(i, j int) bool { return buckets[i] < buckets[j] })

	// Evaluate per bucket; coordinator should sum the counts from both workers.
	got := map[bucket]float64{}
	for _, b := range buckets {
		out := plan.Root.Eval(promql.SketchGroup{
			Timestamp: b,
			Group:     map[string][]promql.SketchInput{leaf.ID: perBucket[b]},
		}, step)

		if len(out) != 1 {
			t.Fatalf("expected 1 output series at bucket %d, got %d: %v", b, len(out), out)
		}
		got[b] = out["default"].Value.Num
	}

	const (
		b0  = int64(0)
		b1  = int64(60000)
		eps = 1e-9
	)
	// Expected totals across workers:
	//   bucket0: 2 (db1) + 3 (db2) = 5
	//   bucket1: 1 (db1) + 0 (db2) = 1
	exp := map[bucket]float64{
		b0: 5.0,
		b1: 1.0,
	}

	for b, want := range exp {
		if math.Abs(got[b]-want) > eps {
			t.Fatalf("unexpected count_over_time at bucket %d: got=%v want=%v\nrows1=%v\nrows2=%v\nsql=\n%s",
				b, got[b], want, rows1, rows2, workerSQL)
		}
	}
}

func TestToWorkerSQL_LineFormat_IndexThenJSON_TwoStage(t *testing.T) {
	db := openDuckDB(t)

	// Minimal table including defaults and the base column referenced by the first line_format:
	// - "chq_timestamp"      (required)
	// - "log_message"        (replaced by line_format)
	// - "resource_service_name"      (used by matcher)
	// - "log_@OrderResult"           (the source JSON blob to index)
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"log_@OrderResult"        TEXT
	);`)

	// Insert one row; the JSON is placed in the base column "log_@OrderResult".
	const orderJSON = `{"orderId":"O-123","shippingCost":{"currencyCode":"USD","units":"12","nanos":"500000000"}}`
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","log_@OrderResult") VALUES
	(10*1000, 'ignored', 'accounting', '`+orderJSON+`');`)

	// The exact LogQL pipeline under test.
	q := `{resource_service_name="accounting"} | line_format "{{ index . \"log.@OrderResult\" }}" | json | line_format "orderId={{.orderId}} currency={{.shippingCost.currencyCode}} units={{.shippingCost.units}} nanos={{.shippingCost.nanos}}"`

	// Compile LogQL to a log plan and pull the single leaf.
	ast, err := logql.FromLogQL(q)
	if err != nil {
		t.Fatalf("FromLogQL error: %v", err)
	}
	lplan, err := logql.CompileLog(ast)
	if err != nil {
		t.Fatalf("CompileLog error: %v", err)
	}
	if len(lplan.Leaves) != 1 {
		t.Fatalf("expected 1 log leaf, got %d", len(lplan.Leaves))
	}
	leaf := lplan.Leaves[0]

	// Build worker SQL (no explicit limit/order/extra fields) and bind placeholders.
	sql := leaf.ToWorkerSQL(0, "DESC", nil)
	sql = replaceWorkerPlaceholders(sql, 0, 60*1000)

	// Sanity checks that both line_format stages and json extraction appear.
	if !strings.Contains(sql, `log_message`) || !strings.Contains(sql, "json_extract_string") {
		t.Fatalf("expected line_format + json stages in SQL, got:\n%s", sql)
	}

	// Execute and assert the final rendered message.
	rows := queryAll(t, db, sql)
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d; sql=\n%s", len(rows), sql)
	}

	gotMsg := s(rows[0][`log_message`])
	wantMsg := "orderId=O-123 currency=USD units=12 nanos=500000000"
	if gotMsg != wantMsg {
		t.Fatalf("final log_message mismatch:\n  got:  %q\n  want: %q\nsql:\n%s", gotMsg, wantMsg, sql)
	}
}

func TestRewrite_SumOverTime_Unwrap_BaseField_LineFilter(t *testing.T) {
	db := openDuckDB(t)
	// Base table with top-level log_quantity.
	mustExec(t, db, `CREATE TABLE logs(
		"chq_timestamp"   BIGINT,
		"chq_id"          TEXT,
		"log_level"       TEXT,
		"chq_fingerprint" TEXT,
		"log_message"     TEXT,
		"resource_service_name"   TEXT,
		"log_quantity"            INTEGER
	);`)

	// Two 1m buckets: [0..60s) and [60..120s)
	// bucket 0: quantities 1 + 2  => sum = 3
	// bucket 1: quantity 3        => sum = 3
	// Include a distractor row that shouldn't match the |= "AddItem" filter.
	mustExec(t, db, `
	INSERT INTO logs("chq_timestamp","log_message","resource_service_name","log_quantity") VALUES
	(10*1000,  'AddItemAsync called with userId=U1, productId=P1, quantity=1', 'cart', 1),
	(25*1000,  'AddItem completed for userId=U2',                              'cart', 2),
	(70*1000,  'AddItem scheduled for userId=U3',                               'cart', 3),
	(50*1000,  'RemoveItem (should not match filter)',                  'cart', 999);
	`)

	q := `sum_over_time(({resource_service_name="cart"} |= "AddItem" | unwrap log_quantity)[1m])`

	plan, _ := compileMetricPlanFromLogQL(t, q)
	if len(plan.Leaves) != 1 {
		t.Fatalf("expected 1 leaf in compiled plan, got %d", len(plan.Leaves))
	}
	be := plan.Leaves[0]

	step := time.Minute
	sql := replaceWorkerPlaceholders(be.ToWorkerSQL(step), 0, 120*1000)

	// Sanity: unwrap should reference the base column as DOUBLE.
	if !strings.Contains(sql, `try_cast(log_quantity AS DOUBLE) AS __unwrap_value`) &&
		!strings.Contains(sql, `try_cast("log_quantity" AS DOUBLE) AS __unwrap_value`) {
		t.Fatalf("expected unwrap of top-level log_quantity; sql=\n%s", sql)
	}
	// Sanity: the line filter should appear as LIKE '%AddItem%'.
	if !strings.Contains(sql, `"log_message" LIKE '%AddItem%'`) {
		t.Fatalf("expected line filter on message; sql=\n%s", sql)
	}

	rows := queryAll(t, db, sql)
	if len(rows) == 0 {
		t.Fatalf("no rows returned; sql=\n%s", sql)
	}

	// Collect sums per bucket.
	got := map[int64]float64{}
	for _, r := range rows {
		b := i64(r["bucket_ts"])
		sum := f64(r["sum"])
		got[b] = sum
	}

	// Expected sums by bucket.
	want := map[int64]float64{
		0:     3, // 1 + 2 from bucket 0; distractor (999) excluded by line filter
		60000: 3, // 3 in bucket 1
	}

	const eps = 1e-9
	for b, w := range want {
		g, ok := got[b]
		if !ok {
			t.Fatalf("missing bucket %d; rows=%v\nsql=\n%s", b, rows, sql)
		}
		if math.Abs(g-w) > eps {
			t.Fatalf("bucket %d: got sum=%v want=%v; rows=%v\nsql=\n%s", b, g, w, rows, sql)
		}
	}
}
