// Next implements Executor Next interface.
func (e *CheckTableExec) Next() (*Row, error) {
	if e.done {
		return nil, nil
	}

	dbName := model.NewCIStr(db.GetCurrentSchema(e.ctx))
	is := sessionctx.GetDomain(e.ctx).InfoSchema()

	for _, t := range e.tables {
		tb, err := is.TableByName(dbName, t.Name)
		if err != nil {
			return nil, errors.Trace(err)
		}
		for _, idx := range tb.Indices() {
			txn, err := e.ctx.GetTxn(false)
			if err != nil {
				return nil, errors.Trace(err)
			}
			err = inspectkv.CompareIndexData(txn, tb, idx)
			if err != nil {
				return nil, errors.Errorf("%v err:%v", t.Name, err)
			}
		}
	}
	e.done = true

	return nil, nil
}

// See https://dev.mysql.com/doc/refman/5.7/en/information-functions.html
func builtinDatabase(args []types.Datum, ctx context.Context) (d types.Datum, err error) {
	s := db.GetCurrentSchema(ctx)
	if s == "" {
		return d, nil
	}
	d.SetString(s)
	return d, nil
}

func (r *ShowRset) getDBName(ctx context.Context) string {
	if len(r.DBName) > 0 {
		return r.DBName
	}

	// if r.DBName is empty, we should use current db name if possible.
	return db.GetCurrentSchema(ctx)
}

// Full returns an Ident which set schema to the current schema if it is empty.
func (i Ident) Full(ctx context.Context) (full Ident) {
	full.Name = i.Name
	if i.Schema.O != "" {
		full.Schema = i.Schema
	} else {
		full.Schema = model.NewCIStr(db.GetCurrentSchema(ctx))
	}
	return
}

func (s *session) String() string {
	// TODO: how to print binded context in values appropriately?
	data := map[string]interface{}{
		"userName":   s.userName,
		"currDBName": db.GetCurrentSchema(s),
		"sid":        s.sid,
		"txn":        s.txn.String(),
	}

	b, _ := json.MarshalIndent(data, "", "  ")
	return string(b)
}

func builtinDatabase(args []interface{}, data map[interface{}]interface{}) (v interface{}, err error) {
	c, ok := data[ExprEvalArgCtx]
	if !ok {
		return nil, errors.Errorf("Missing ExprEvalArgCtx when evalue builtin")
	}
	ctx := c.(context.Context)
	d := db.GetCurrentSchema(ctx)
	if d == "" {
		return nil, nil
	}
	return d, nil
}

func (s *ShowStmt) getDBName(ctx context.Context) string {
	if len(s.DBName) > 0 {
		return s.DBName
	}

	// maybe db.table format
	if len(s.TableIdent.Schema.O) > 0 {
		return s.TableIdent.Schema.O
	}

	// try use current db name if possible.
	return db.GetCurrentSchema(ctx)
}

func (s *session) String() string {
	// TODO: how to print binded context in values appropriately?
	data := map[string]interface{}{
		"currDBName": db.GetCurrentSchema(s),
		"sid":        s.sid,
	}

	if s.txn != nil {
		// if txn is committed or rolled back, txn is nil.
		data["txn"] = s.txn.String()
	}

	b, _ := json.MarshalIndent(data, "", "  ")
	return string(b)
}

// Find the schema by dbName.
func (e *GrantExec) getTargetSchema() (*model.DBInfo, error) {
	dbName := e.Level.DBName
	if len(dbName) == 0 {
		// Grant *, use current schema
		dbName = db.GetCurrentSchema(e.ctx)
		if len(dbName) == 0 {
			return nil, errors.New("Miss DB name for grant privilege.")
		}
	}
	//check if db exists
	schema := model.NewCIStr(dbName)
	is := sessionctx.GetDomain(e.ctx).InfoSchema()
	db, ok := is.SchemaByName(schema)
	if !ok {
		return nil, errors.Errorf("Unknown schema name: %s", dbName)
	}
	return db, nil
}

// ResolveName resolves table name and column name.
// It generates ResultFields for ResultSetNode and resolves ColumnNameExpr to a ResultField.
func ResolveName(node ast.Node, info infoschema.InfoSchema, ctx context.Context) error {
	defaultSchema := db.GetCurrentSchema(ctx)
	resolver := nameResolver{Info: info, Ctx: ctx, DefaultSchema: model.NewCIStr(defaultSchema)}
	node.Accept(&resolver)
	return errors.Trace(resolver.Err)
}