func getDefaultValue(ctx context.Context, c *column.Col) (interface{}, bool, error) {
	// Check no default value flag.
	if mysql.HasNoDefaultValueFlag(c.Flag) && c.Tp != mysql.TypeEnum {
		return nil, false, errors.Errorf("Field '%s' doesn't have a default value", c.Name)
	}

	// Check and get timestamp/datetime default value.
	if c.Tp == mysql.TypeTimestamp || c.Tp == mysql.TypeDatetime {
		if c.DefaultValue == nil {
			return nil, true, nil
		}

		value, err := expression.GetTimeValue(ctx, c.DefaultValue, c.Tp, c.Decimal)
		if err != nil {
			return nil, true, errors.Errorf("Field '%s' get default value fail - %s", c.Name, errors.Trace(err))
		}

		return value, true, nil
	} else if c.Tp == mysql.TypeEnum {
		// For enum type, if no default value and not null is set,
		// the default value is the first element of the enum list
		if c.DefaultValue == nil && mysql.HasNotNullFlag(c.Flag) {
			return c.FieldType.Elems[0], true, nil
		}
	}

	return c.DefaultValue, true, nil
}

// String implements fmt.Stringer interface.
func (c *Col) String() string {
	ans := []string{c.Name.O, types.FieldTypeToStr(c.Tp, c.Charset)}
	if mysql.HasAutoIncrementFlag(c.Flag) {
		ans = append(ans, "AUTO_INCREMENT")
	}
	if mysql.HasNotNullFlag(c.Flag) {
		ans = append(ans, "NOT NULL")
	}
	return strings.Join(ans, " ")
}

func setNoDefaultValueFlag(c *column.Col, hasDefaultValue bool) {
	if hasDefaultValue {
		return
	}

	if !mysql.HasNotNullFlag(c.Flag) {
		return
	}

	// Check if it is an `AUTO_INCREMENT` field or `TIMESTAMP` field.
	if !mysql.HasAutoIncrementFlag(c.Flag) && !mysql.HasTimestampFlag(c.Flag) {
		c.Flag |= mysql.NoDefaultValueFlag
	}
}

func setTimestampDefaultValue(c *column.Col, hasDefaultValue bool, setOnUpdateNow bool) {
	if hasDefaultValue {
		return
	}

	// For timestamp Col, if is not set default value or not set null, use current timestamp.
	if mysql.HasTimestampFlag(c.Flag) && mysql.HasNotNullFlag(c.Flag) {
		if setOnUpdateNow {
			c.DefaultValue = expressions.ZeroTimestamp
		} else {
			c.DefaultValue = expressions.CurrentTimestamp
		}
	}
}

func checkDefaultValue(c *column.Col, hasDefaultValue bool) error {
	if !hasDefaultValue {
		return nil
	}

	if c.DefaultValue != nil {
		return nil
	}

	// Set not null but default null is invalid.
	if mysql.HasNotNullFlag(c.Flag) {
		return errors.Errorf("invalid default value for %s", c.Name)
	}

	return nil
}

// NewColDesc returns a new ColDesc for a column.
func NewColDesc(col *Col) *ColDesc {
	// TODO: if we have no primary key and a unique index which's columns are all not null
	// we will set these columns' flag as PriKeyFlag
	// see https://dev.mysql.com/doc/refman/5.7/en/show-columns.html
	// create table
	name := col.Name
	nullFlag := "YES"
	if mysql.HasNotNullFlag(col.Flag) {
		nullFlag = "NO"
	}
	keyFlag := ""
	if mysql.HasPriKeyFlag(col.Flag) {
		keyFlag = "PRI"
	} else if mysql.HasUniKeyFlag(col.Flag) {
		keyFlag = "UNI"
	} else if mysql.HasMultipleKeyFlag(col.Flag) {
		keyFlag = "MUL"
	}
	var defaultValue interface{}
	if !mysql.HasNoDefaultValueFlag(col.Flag) {
		defaultValue = col.DefaultValue
	}

	extra := ""
	if mysql.HasAutoIncrementFlag(col.Flag) {
		extra = "auto_increment"
	} else if mysql.HasOnUpdateNowFlag(col.Flag) {
		extra = "on update CURRENT_TIMESTAMP"
	}

	return &ColDesc{
		Field:        name.O,
		Type:         col.GetTypeDesc(),
		Collation:    col.Collate,
		Null:         nullFlag,
		Key:          keyFlag,
		DefaultValue: defaultValue,
		Extra:        extra,
		Privileges:   defaultPrivileges,
		Comment:      "",
	}
}

func (isp *InfoSchemaPlan) doStatistics(is infoschema.InfoSchema, schemas []*model.DBInfo, iterFunc plan.RowIterFunc) error {
	for _, schema := range schemas {
		for _, table := range schema.Tables {
			for _, index := range table.Indices {
				nonUnique := "1"
				if index.Unique {
					nonUnique = "0"
				}
				for i, key := range index.Columns {
					col, _ := is.ColumnByName(schema.Name, table.Name, key.Name)
					nullable := "YES"
					if mysql.HasNotNullFlag(col.Flag) {
						nullable = ""
					}
					record := []interface{}{
						catalogVal,    // TABLE_CATALOG
						schema.Name.O, // TABLE_SCHEMA
						table.Name.O,  // TABLE_NAME
						nonUnique,     // NON_UNIQUE
						schema.Name.O, // INDEX_SCHEMA
						index.Name.O,  // INDEX_NAME
						i + 1,         // SEQ_IN_INDEX
						key.Name.O,    // COLUMN_NAME
						"A",           // COLLATION
						0,             // CARDINALITY
						nil,           // SUB_PART
						nil,           // PACKED
						nullable,      // NULLABLE
						"BTREE",       // INDEX_TYPE
						"",            // COMMENT
						"",            // INDEX_COMMENT
					}
					if more, err := iterFunc(0, record); !more || err != nil {
						return err
					}
				}
			}
		}
	}
	return nil
}

func (isp *InfoSchemaPlan) fetchStatistics(is infoschema.InfoSchema, schemas []*model.DBInfo) {
	for _, schema := range schemas {
		for _, table := range schema.Tables {
			for _, index := range table.Indices {
				nonUnique := "1"
				if index.Unique {
					nonUnique = "0"
				}
				for i, key := range index.Columns {
					col, _ := is.ColumnByName(schema.Name, table.Name, key.Name)
					nullable := "YES"
					if mysql.HasNotNullFlag(col.Flag) {
						nullable = ""
					}
					record := []interface{}{
						catalogVal,    // TABLE_CATALOG
						schema.Name.O, // TABLE_SCHEMA
						table.Name.O,  // TABLE_NAME
						nonUnique,     // NON_UNIQUE
						schema.Name.O, // INDEX_SCHEMA
						index.Name.O,  // INDEX_NAME
						i + 1,         // SEQ_IN_INDEX
						key.Name.O,    // COLUMN_NAME
						"A",           // COLLATION
						0,             // CARDINALITY
						nil,           // SUB_PART
						nil,           // PACKED
						nullable,      // NULLABLE
						"BTREE",       // INDEX_TYPE
						"",            // COMMENT
						"",            // INDEX_COMMENT
					}
					isp.rows = append(isp.rows, &plan.Row{Data: record})
				}
			}
		}
	}
}

// CheckNotNull checks if nil value set to a column with NotNull flag is set.
func (c *Col) CheckNotNull(data interface{}) error {
	if mysql.HasNotNullFlag(c.Flag) && data == nil {
		return errors.Errorf("Column %s can't be null.", c.Name)
	}
	return nil
}

func (s *ShowPlan) fetchShowCreateTable(ctx context.Context) error {
	tb, err := s.getTable(ctx)
	if err != nil {
		return errors.Trace(err)
	}

	// TODO: let the result more like MySQL.
	var buf bytes.Buffer
	buf.WriteString(fmt.Sprintf("CREATE TABLE `%s` (\n", tb.TableName().O))
	for i, col := range tb.Cols() {
		buf.WriteString(fmt.Sprintf("  `%s` %s", col.Name.O, col.GetTypeDesc()))
		if mysql.HasAutoIncrementFlag(col.Flag) {
			buf.WriteString(" NOT NULL AUTO_INCREMENT")
		} else {
			if mysql.HasNotNullFlag(col.Flag) {
				buf.WriteString(" NOT NULL")
			}
			switch col.DefaultValue {
			case nil:
				buf.WriteString(" DEFAULT NULL")
			case "CURRENT_TIMESTAMP":
				buf.WriteString(" DEFAULT CURRENT_TIMESTAMP")
			default:
				buf.WriteString(fmt.Sprintf(" DEFAULT '%v'", col.DefaultValue))
			}

			if mysql.HasOnUpdateNowFlag(col.Flag) {
				buf.WriteString(" ON UPDATE CURRENT_TIMESTAMP")
			}
		}
		if i != len(tb.Cols())-1 {
			buf.WriteString(",\n")
		}
	}

	if len(tb.Indices()) > 0 {
		buf.WriteString(",\n")
	}

	for i, idx := range tb.Indices() {
		if idx.Primary {
			buf.WriteString("  PRIMARY KEY ")
		} else if idx.Unique {
			buf.WriteString(fmt.Sprintf("  UNIQUE KEY `%s` ", idx.Name.O))
		} else {
			buf.WriteString(fmt.Sprintf("  KEY `%s` ", idx.Name.O))
		}

		cols := make([]string, 0, len(idx.Columns))
		for _, c := range idx.Columns {
			cols = append(cols, c.Name.O)
		}
		buf.WriteString(fmt.Sprintf("(`%s`)", strings.Join(cols, "`,`")))
		if i != len(tb.Indices())-1 {
			buf.WriteString(",\n")
		}
	}

	buf.WriteString("\n")

	buf.WriteString(") ENGINE=InnoDB")
	if s := tb.Meta().Charset; len(s) > 0 {
		buf.WriteString(fmt.Sprintf(" DEFAULT CHARSET=%s", s))
	} else {
		buf.WriteString(" DEFAULT CHARSET=latin1")
	}

	data := []interface{}{
		tb.TableName().O,
		buf.String(),
	}

	s.rows = append(s.rows, &plan.Row{Data: data})

	return nil
}