// Table is a SQL table.
type Table interface {
	Nameable
	fmt.Stringer
	// Schema returns the table's schema.
	Schema() Schema
	// Collation returns the table's collation.
	Collation() CollationID
	// Partitions returns the table's partitions in an iterator.
	Partitions(*Context) (PartitionIter, error)
	// PartitionRows returns the rows in the given partition, which was returned by Partitions.
	PartitionRows(*Context, Partition) (RowIter, error)
}

// Database represents the database. Its primary job is to provide access to all tables.
type Database interface {
	Nameable
	// GetTableInsensitive retrieves a table by its case-insensitive name. To be SQL compliant, databases should not
	// allow two tables with the same case-insensitive name. Behavior is undefined when two tables have the same
	// case-insensitive name.
	GetTableInsensitive(ctx *Context, tblName string) (Table, bool, error)
	// GetTableNames returns the table names of every table in the database. It does not return the names of temporary
	// tables
	GetTableNames(ctx *Context) ([]string, error)
}

// NewFramesDBProvider creates a new FramesDBProvider with the given set of Frames.
func NewFramesDBProvider(frames data.Frames) mysql.DatabaseProvider {
	fMap := make(map[string]mysql.Table, len(frames))
	for _, frame := range frames {
		fMap[frame.RefID] = &FrameTable{Frame: frame}
	}
	return &FramesDBProvider{
		db: &framesDB{
			frames: fMap,
		},
	}
}

// framesDB is a go-mysql-server Database that provides access to a set of Frames.
type framesDB struct {
	frames map[string]mysql.Table
}

func (db *framesDB) GetTableInsensitive(_ *mysql.Context, tblName string) (mysql.Table, bool, error) {
	tbl, ok := mysql.GetTableInsensitive(tblName, db.frames)
	if !ok {
		return nil, false, nil
	}
	return tbl, ok, nil
}

// FrameTable fulfills the mysql.Table interface for a data.Frame.
type FrameTable struct {
	Frame  *data.Frame
	schema mysql.Schema
}

// Name implements the sql.Nameable interface
func (ft *FrameTable) Name() string {
	return ft.Frame.RefID
}

// String implements the fmt.Stringer interface
func (ft *FrameTable) String() string {
	return ft.Name()
}

func schemaFromFrame(frame *data.Frame) mysql.Schema {
	schema := make(mysql.Schema, len(frame.Fields))

	for i, field := range frame.Fields {
		schema[i] = &mysql.Column{
			Name:     field.Name,
			Type:     convertDataType(field.Type()),
			Nullable: field.Type().Nullable(),
			Source:   strings.ToLower(frame.RefID),
		}
	}

	return schema
}

// Schema implements the mysql.Table interface
func (ft *FrameTable) Schema() mysql.Schema {
	if ft.schema == nil {
		ft.schema = schemaFromFrame(ft.Frame)
	}
	return ft.schema
}

// PartitionRows implements the mysql.Table interface
func (ft *FrameTable) PartitionRows(ctx *mysql.Context, _ mysql.Partition) (mysql.RowIter, error) {
	return &rowIter{ft: ft, row: 0}, nil
}

type rowIter struct {
	ft  *FrameTable
	row int
}

func (ri *rowIter) Next(ctx *mysql.Context) (mysql.Row, error) {
	// We assume each field in the Frame has the same number of rows.
	numRows := 0
	if len(ri.ft.Frame.Fields) > 0 {
		numRows = ri.ft.Frame.Fields[0].Len()
	}

	// If we've already exhausted all rows, return EOF
	if ri.row >= numRows {
		return nil, io.EOF
	}

	// Construct a Row (which is []interface{} under the hood) by pulling
	// the value from each column at the current row index.
	row := make(mysql.Row, len(ri.ft.Frame.Fields))
	for colIndex, field := range ri.ft.Frame.Fields {
		if field.NilAt(ri.row) {
			continue
		}
		val, _ := field.ConcreteAt(ri.row)

		// If the field is JSON, convert json.RawMessage to types.JSONDocument
		if raw, ok := val.(json.RawMessage); ok {
			doc, inRange, err := types.JSON.Convert(ctx, raw)
			if err != nil {
				return nil, fmt.Errorf("failed to convert json.RawMessage to JSONDocument: %w", err)
			}
			if !inRange {
				return nil, fmt.Errorf("invalid JSON value detected at row %d, column %s: value required type coercion", ri.row, ri.ft.Frame.Fields[colIndex].Name)
			}
			val = doc
		}

		row[colIndex] = val
	}

	ri.row++
	return row, nil
}

// InsertableTable is a table that can process insertion of new rows.
type InsertableTable interface {
	Table
	// Inserter returns an Inserter for this table. The Inserter will get one call to Insert() for each row to be
	// inserted, and will end with a call to Close() to finalize the insert operation.
	Inserter(*Context) RowInserter
}

// IndexAddressable is a table that can be scanned through a primary index
type IndexAddressable interface {
	// IndexedAccess returns a table that can perform scans constrained to
	// an IndexLookup on the index given, or nil if the index cannot support
	// the lookup expression.
	IndexedAccess(ctx *Context, lookup IndexLookup) IndexedTable
	// GetIndexes returns an array of this table's Indexes
	GetIndexes(ctx *Context) ([]Index, error)
	// PreciseMatch returns whether an indexed access can substitute for filters
	PreciseMatch() bool
}