// metadata.go 元数据查询算子：SHOW TABLES / SHOW COLUMNS / INFORMATION_SCHEMA 虚拟扫描
package executor

import (
	"fmt"
	"strings"

	"gitee.com/bigant/vds/internal/sqlerr"
	"gitee.com/bigant/vds/internal/sql/catalog"
	"gitee.com/bigant/vds/internal/sql/planner"
	"gitee.com/bigant/vds/internal/storage/engine"
)

// ── 公共辅助 ──────────────────────────────────────────────────────────────────

// asTxnScanner 将 engine.Transaction 适配为 catalog.TxnScanner
type txnScanAdapter struct{ engine.Transaction }

func (a txnScanAdapter) Scan(start, end []byte) (catalog.CatalogIterator, error) {
	return a.Transaction.Scan(start, end)
}

// ── ShowTablesOp ─────────────────────────────────────────────────────────────

// ShowTablesOp SHOW TABLES 算子：从 catalog 读取所有表名
type ShowTablesOp struct {
	cat    *catalog.Catalog
	txn    engine.Transaction
	rows   []Row
	schema *planner.Schema
	pos    int
	done   bool
}

// NewShowTablesOp 创建 ShowTablesOp
func NewShowTablesOp(cat *catalog.Catalog, txn engine.Transaction) *ShowTablesOp {
	return &ShowTablesOp{cat: cat, txn: txn}
}

func (op *ShowTablesOp) Open() error {
	op.done = false
	op.pos = 0
	op.rows = nil
	names := op.cat.ListTables()

	op.rows = make([]Row, len(names))
	for i, name := range names {
		op.rows[i] = Row{name}
	}
	op.schema = planner.ShowTablesSchema()
	return nil
}

func (op *ShowTablesOp) Next() (Row, *planner.Schema, error) {
	if op.pos >= len(op.rows) {
		return nil, nil, nil
	}
	r := op.rows[op.pos]
	op.pos++
	return r, op.schema, nil
}

func (op *ShowTablesOp) Close() error { return nil }

// ── ShowColumnsOp ─────────────────────────────────────────────────────────────

// InfoSchemaViewsOp INFORMATION_SCHEMA.VIEWS 虚拟扫描算子
type InfoSchemaViewsOp struct {
	cat    *catalog.Catalog
	txn    engine.Transaction
	rows   []Row
	schema *planner.Schema
	pos    int
}

// NewInfoSchemaViewsOp 创建 InfoSchemaViewsOp
func NewInfoSchemaViewsOp(cat *catalog.Catalog, txn engine.Transaction) *InfoSchemaViewsOp {
	return &InfoSchemaViewsOp{cat: cat, txn: txn}
}

func (op *InfoSchemaViewsOp) Open() error {
	op.pos = 0
	op.rows = nil
	names, err := op.cat.ListViews(txnScanAdapter{op.txn})
	if err != nil {
		return err
	}
	op.rows = make([]Row, 0, len(names))
	for _, name := range names {
		desc, err := op.cat.GetView(txnScanAdapter{op.txn}, name)
		if err != nil || desc == nil {
			continue
		}
		op.rows = append(op.rows, Row{"vds", name, desc.SelectSQL})
	}
	op.schema = planner.InfoSchemaViewsSchema()
	return nil
}

func (op *InfoSchemaViewsOp) Next() (Row, *planner.Schema, error) {
	if op.pos >= len(op.rows) {
		return nil, nil, nil
	}
	r := op.rows[op.pos]
	op.pos++
	return r, op.schema, nil
}

func (op *InfoSchemaViewsOp) Close() error { return nil }

// ShowColumnsOp SHOW COLUMNS FROM t 算子
type ShowColumnsOp struct {
	cat       *catalog.Catalog
	txn       engine.Transaction
	tableName string
	rows      []Row
	schema    *planner.Schema
	pos       int
}

// NewShowColumnsOp 创建 ShowColumnsOp
func NewShowColumnsOp(cat *catalog.Catalog, txn engine.Transaction, tableName string) *ShowColumnsOp {
	return &ShowColumnsOp{cat: cat, txn: txn, tableName: tableName}
}

func (op *ShowColumnsOp) Open() error {
	op.pos = 0
	op.rows = nil
	table, err := op.cat.GetTable(txnScanAdapter{op.txn}, op.tableName)
	if err != nil {
		return fmt.Errorf("show columns: %w", err)
	}
	if table == nil {
		return fmt.Errorf("show columns: table %q not found: %w", op.tableName, sqlerr.ErrTableNotFound)
	}
	op.rows = make([]Row, len(table.Cols))
	for i, col := range table.Cols {
		nullable := "YES"
		if col.NotNull {
			nullable = "NO"
		}
		key := ""
		if i == table.PKColIndex {
			key = "PRI"
		}
		def := ""
		if col.DefaultValue != nil {
			def = fmt.Sprintf("%v", col.DefaultValue)
		}
		op.rows[i] = Row{col.Name, col.Type.String(), nullable, key, def, ""}
	}
	op.schema = planner.ShowColumnsSchema()
	return nil
}

func (op *ShowColumnsOp) Next() (Row, *planner.Schema, error) {
	if op.pos >= len(op.rows) {
		return nil, nil, nil
	}
	r := op.rows[op.pos]
	op.pos++
	return r, op.schema, nil
}

func (op *ShowColumnsOp) Close() error { return nil }

// ── InfoSchemaTablesOp ────────────────────────────────────────────────────────

// InfoSchemaTablesOp INFORMATION_SCHEMA.TABLES 虚拟扫描算子
type InfoSchemaTablesOp struct {
	cat    *catalog.Catalog
	txn    engine.Transaction
	rows   []Row
	schema *planner.Schema
	pos    int
}

// NewInfoSchemaTablesOp 创建 InfoSchemaTablesOp
func NewInfoSchemaTablesOp(cat *catalog.Catalog, txn engine.Transaction) *InfoSchemaTablesOp {
	return &InfoSchemaTablesOp{cat: cat, txn: txn}
}

func (op *InfoSchemaTablesOp) Open() error {
	op.pos = 0
	op.rows = nil
	names := op.cat.ListTables()

	op.rows = make([]Row, len(names))
	for i, name := range names {
		var rowCount int64
		table, _ := op.cat.GetTable(txnScanAdapter{op.txn}, name)
		if table != nil {
			if stats, _ := op.cat.GetTableStats(txnScanAdapter{op.txn}, table.ID); stats != nil {
				rowCount = stats.RowCount
			}
		}
		op.rows[i] = Row{"vds", name, "BASE TABLE", rowCount}
	}
	op.schema = planner.InfoSchemaTablesSchema()
	return nil
}

func (op *InfoSchemaTablesOp) Next() (Row, *planner.Schema, error) {
	if op.pos >= len(op.rows) {
		return nil, nil, nil
	}
	r := op.rows[op.pos]
	op.pos++
	return r, op.schema, nil
}

func (op *InfoSchemaTablesOp) Close() error { return nil }

// ── InfoSchemaColumnsOp ───────────────────────────────────────────────────────

// InfoSchemaColumnsOp INFORMATION_SCHEMA.COLUMNS 虚拟扫描算子
type InfoSchemaColumnsOp struct {
	cat         *catalog.Catalog
	txn         engine.Transaction
	filterTable string // "" = 全部表
	rows        []Row
	schema      *planner.Schema
	pos         int
}

// NewInfoSchemaColumnsOp 创建 InfoSchemaColumnsOp
func NewInfoSchemaColumnsOp(cat *catalog.Catalog, txn engine.Transaction, filterTable string) *InfoSchemaColumnsOp {
	return &InfoSchemaColumnsOp{cat: cat, txn: txn, filterTable: filterTable}
}

func (op *InfoSchemaColumnsOp) Open() error {
	op.pos = 0
	op.rows = nil

	names := op.cat.ListTables()


	for _, name := range names {
		if op.filterTable != "" && !strings.EqualFold(name, op.filterTable) {
			continue
		}
		table, _ := op.cat.GetTable(txnScanAdapter{op.txn}, name)
		if table == nil {
			continue
		}
		for pos, col := range table.Cols {
			nullable := "YES"
			if col.NotNull {
				nullable = "NO"
			}
			def := interface{}(nil)
			if col.DefaultValue != nil {
				def = fmt.Sprintf("%v", col.DefaultValue)
			}
			op.rows = append(op.rows, Row{
				"vds",
				name,
				col.Name,
				int64(pos + 1),
				col.Type.String(),
				nullable,
				def,
			})
		}
	}
	op.schema = planner.InfoSchemaColumnsSchema()
	return nil
}

func (op *InfoSchemaColumnsOp) Next() (Row, *planner.Schema, error) {
	if op.pos >= len(op.rows) {
		return nil, nil, nil
	}
	r := op.rows[op.pos]
	op.pos++
	return r, op.schema, nil
}

func (op *InfoSchemaColumnsOp) Close() error { return nil }