package vrtxsql

// BLOB 列结果集物化测试: 行内只有 8B BlobRef, SELECT (流式) 与 JOIN (物化) 两条
// 路径都必须读出内容字节 —— 60KB 值超动态列单值上限 16360, 只有链式存储装得下,
// 物化错了 (或没接) 这两类查询当场就崩。

import (
	"database/sql"
	"strings"
	"testing"
)

// blobContent 60KB 测试内容: 超引擎动态列单值上限 (types.MaxStringSize = 16360),
// 落 BLOB 列经链页存储, 用来证明物化读回的是整条链而不是行内 8 字节引用。
const blobTestContent = "blob-payload-0123456789abcdef"

func blobFixture(t *testing.T, db *sql.DB) string {
	t.Helper()
	exec := func(q string, args ...interface{}) {
		t.Helper()
		if _, err := db.Exec(q, args...); err != nil {
			t.Fatalf("%s: %v", q, err)
		}
	}
	exec(`CREATE TABLE docs (id BIGINT PRIMARY KEY, title VARCHAR(32) NOT NULL, content BLOB)`)
	big := strings.Repeat(blobTestContent, 60000/len(blobTestContent)+1) // >16360 字节
	exec(`INSERT INTO docs (id, title, content) VALUES (?, ?, ?), (?, ?, ?), (?, ?, ?)`,
		1, "big", big,
		2, "small", "hi",
		3, "nonum", nil) // NULL blob
	return big
}

// TestBlobSelectMaterialize 单表 SELECT (流式路径): BLOB 列物化成内容字节,
// 投影列 / * 展开 / WHERE / LIMIT 都要一致; NULL blob 保持 NULL。
func TestBlobSelectMaterialize(t *testing.T) {
	db := openTestDB(t)
	big := blobFixture(t, db)

	// 投影列 + WHERE + LIMIT
	rows, err := db.Query(`SELECT title, content FROM docs WHERE id = ? LIMIT 1`, 1)
	if err != nil {
		t.Fatalf("query: %v", err)
	}
	defer rows.Close()
	if !rows.Next() {
		t.Fatal("应有一行")
	}
	var title string
	var content []byte
	if err := rows.Scan(&title, &content); err != nil {
		t.Fatalf("scan: %v", err)
	}
	if title != "big" {
		t.Fatalf("title=%q, 想要 big", title)
	}
	if string(content) != big {
		t.Fatalf("content %d 字节与写入的 %d 字节不一致", len(content), len(big))
	}
	if rows.Next() {
		t.Fatal("只应有一行")
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	rows.Close()

	// * 展开 + 全表扫描 (worker 逐行物化)
	all, err := db.Query(`SELECT * FROM docs ORDER BY id ASC`)
	if err != nil {
		t.Fatalf("select *: %v", err)
	}
	defer all.Close()
	var (
		seen  int
		nilOK bool
	)
	for all.Next() {
		var (
			id      int64
			title   string
			content []byte
		)
		if err := all.Scan(&id, &title, &content); err != nil {
			t.Fatalf("scan: %v", err)
		}
		switch id {
		case 1:
			if string(content) != big {
				t.Fatalf("id=1 content 不一致 (%d 字节)", len(content))
			}
		case 2:
			if string(content) != "hi" {
				t.Fatalf("id=2 content=%q", content)
			}
		case 3:
			if content != nil {
				t.Fatalf("id=3 (NULL blob) content 应为 nil, 实际 %v", content)
			}
			nilOK = true
		}
		seen++
	}
	if err := all.Err(); err != nil {
		t.Fatal(err)
	}
	if seen != 3 || !nilOK {
		t.Fatalf("应扫到 3 行且含 NULL 行, 实际 %d 行 nilOK=%v", seen, nilOK)
	}

	// 短值按主键点查 (行内引用 → 链页, 1 页链)
	var short []byte
	if err := db.QueryRow(`SELECT content FROM docs WHERE id = ?`, 2).Scan(&short); err != nil {
		t.Fatalf("点查: %v", err)
	}
	if string(short) != "hi" {
		t.Fatalf("短值 content=%q", short)
	}
}

// TestBlobJoinMaterialize JOIN 查询 (物化路径): 右表 BLOB 列在 scanJoinTable 里
// 当场解引用; LEFT JOIN 未匹配行的 blob 列保持 NULL。
func TestBlobJoinMaterialize(t *testing.T) {
	db := openTestDB(t)
	big := strings.Repeat(blobTestContent, 60000/len(blobTestContent)+1) // >16360 字节
	exec := func(q string, args ...interface{}) {
		t.Helper()
		if _, err := db.Exec(q, args...); err != nil {
			t.Fatalf("%s: %v", q, err)
		}
	}
	exec(`CREATE TABLE authors (id BIGINT PRIMARY KEY, name VARCHAR(32) NOT NULL)`)
	exec(`CREATE TABLE papers (id BIGINT PRIMARY KEY, author_id BIGINT, content BLOB)`)
	exec(`INSERT INTO authors (id, name) VALUES (?, ?), (?, ?)`,
		7, "ada", 8, "brian")
	exec(`INSERT INTO papers (id, author_id, content) VALUES (?, ?, ?), (?, ?, ?)`,
		1, 7, big,
		2, 7, "hi")

	// INNER JOIN + 右表条件下推: 命中行的 blob 内容物化出来
	rows, err := db.Query(`SELECT a.name, p.content FROM authors a JOIN papers p ON p.author_id = a.id WHERE a.id = ? ORDER BY p.id ASC`, 7)
	if err != nil {
		t.Fatalf("join: %v", err)
	}
	defer rows.Close()
	var got [2][]byte
	var names [2]string
	n := 0
	for rows.Next() {
		if n >= 2 {
			t.Fatal("至多两行")
		}
		if err := rows.Scan(&names[n], &got[n]); err != nil {
			t.Fatalf("scan: %v", err)
		}
		n++
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	rows.Close()
	if n != 2 || names[0] != "ada" || names[1] != "ada" {
		t.Fatalf("INNER JOIN 应出 ada 两行, 实际 %d 行 %v", n, names)
	}
	if string(got[0]) != big {
		t.Fatalf("60KB content 物化不一致 (%d 字节)", len(got[0]))
	}
	if string(got[1]) != "hi" {
		t.Fatalf("短 content=%q", got[1])
	}

	// LEFT JOIN: brian 没有论文, 右表 blob 列应为 NULL (不是空串/报错)
	lr, err := db.Query(`SELECT a.name, p.content FROM authors a LEFT JOIN papers p ON p.author_id = a.id ORDER BY a.id ASC, p.id ASC`)
	if err != nil {
		t.Fatalf("left join: %v", err)
	}
	defer lr.Close()
	type hit struct {
		name    string
		content []byte
	}
	var left []hit
	for lr.Next() {
		var h hit
		if err := lr.Scan(&h.name, &h.content); err != nil {
			t.Fatalf("scan: %v", err)
		}
		left = append(left, h)
	}
	if err := lr.Err(); err != nil {
		t.Fatal(err)
	}
	// ada 两篇 + brian 补一个 NULL 右行 = 3 行
	if len(left) != 3 {
		t.Fatalf("LEFT JOIN 应出 3 行, 实际 %d (%+v)", len(left), left)
	}
	if left[2].name != "brian" || left[2].content != nil {
		t.Fatalf("brian 行 content 应为 NULL, 实际 %v", left[2].content)
	}
}