// changeUserPasswordLocalhost changes the password for username on localhost
func (c *PasswordChanger) ChangeUserPasswordLocalhost(newPassword string) error {
	serverp, err := syscall.UTF16PtrFromString("localhost")
	if err != nil {
		return errors.Trace(err)
	}

	userp, err := syscall.UTF16PtrFromString("jujud")
	if err != nil {
		return errors.Trace(err)
	}

	passp, err := syscall.UTF16PtrFromString(newPassword)
	if err != nil {
		return errors.Trace(err)
	}

	info := netUserSetPassword{passp}

	err = netUserSetInfo(serverp, userp, changePasswordLevel, &info, nil)
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

func collectColumnsInExpr(expr *tipb.Expr, ctx *selectContext, collector map[int64]*tipb.ColumnInfo) error {
	if expr == nil {
		return nil
	}
	if expr.GetTp() == tipb.ExprType_ColumnRef {
		_, i, err := codec.DecodeInt(expr.Val)
		if err != nil {
			return errors.Trace(err)
		}
		var columns []*tipb.ColumnInfo
		if ctx.sel.TableInfo != nil {
			columns = ctx.sel.TableInfo.Columns
		} else {
			columns = ctx.sel.IndexInfo.Columns
		}
		for _, c := range columns {
			if c.GetColumnId() == i {
				collector[i] = c
				return nil
			}
		}
		return xeval.ErrInvalid.Gen("column %d not found", i)
	}
	for _, child := range expr.Children {
		err := collectColumnsInExpr(child, ctx, collector)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// handleRowData deals with raw row data:
//	1. Decodes row from raw byte slice.
//	2. Checks if it fit where condition.
//	3. Update aggregate functions.
// returns true if got a row.
func (rs *localRegion) handleRowData(ctx *selectContext, handle int64, value []byte) (bool, error) {
	columns := ctx.sel.TableInfo.Columns
	values, err := rs.getRowData(value, ctx.colTps)
	if err != nil {
		return false, errors.Trace(err)
	}
	// Fill handle and null columns.
	for _, col := range columns {
		if col.GetPkHandle() {
			var handleDatum types.Datum
			if mysql.HasUnsignedFlag(uint(col.Flag)) {
				// PK column is Unsigned
				handleDatum = types.NewUintDatum(uint64(handle))
			} else {
				handleDatum = types.NewIntDatum(handle)
			}
			handleData, err1 := codec.EncodeValue(nil, handleDatum)
			if err1 != nil {
				return false, errors.Trace(err1)
			}
			values[col.GetColumnId()] = handleData
		} else {
			_, ok := values[col.GetColumnId()]
			if !ok {
				if mysql.HasNotNullFlag(uint(col.GetFlag())) {
					return false, errors.New("Miss column")
				}
				values[col.GetColumnId()] = []byte{codec.NilFlag}
			}
		}
	}
	return rs.valuesToRow(ctx, handle, values)
}

func (*changePasswordCommand) generateOrReadPassword(ctx *cmd.Context, generate bool) (string, error) {
	if generate {
		password, err := utils.RandomPassword()
		if err != nil {
			return "", errors.Annotate(err, "failed to generate random password")
		}
		randomPasswordNotify(password)
		return password, nil
	}

	// Don't add the carriage returns before readPassword, but add
	// them directly after the readPassword so any errors are output
	// on their own lines.
	fmt.Fprint(ctx.Stdout, "password: ")
	password, err := readPassword()
	fmt.Fprint(ctx.Stdout, "\n")
	if err != nil {
		return "", errors.Trace(err)
	}
	fmt.Fprint(ctx.Stdout, "type password again: ")
	verify, err := readPassword()
	fmt.Fprint(ctx.Stdout, "\n")
	if err != nil {
		return "", errors.Trace(err)
	}
	if password != verify {
		return "", errors.New("Passwords do not match")
	}
	return password, nil
}

func (c *kvIndex) Exist(rm RetrieverMutator, indexedValues []interface{}, h int64) (bool, int64, error) {
	key, distinct, err := c.GenIndexKey(indexedValues, h)
	if err != nil {
		return false, 0, errors.Trace(err)
	}

	value, err := rm.Get(key)
	if IsErrNotFound(err) {
		return false, 0, nil
	}
	if err != nil {
		return false, 0, errors.Trace(err)
	}

	// For distinct index, the value of key is handle.
	if distinct {
		handle, err := decodeHandle(value)
		if err != nil {
			return false, 0, errors.Trace(err)
		}

		if handle != h {
			return true, handle, errors.Trace(ErrKeyExists)
		}

		return true, handle, nil
	}

	return true, h, nil
}

// ExecRestrictedSQL implements SQLHelper interface.
// This is used for executing some restricted sql statements.
func (s *session) ExecRestrictedSQL(ctx context.Context, sql string) (rset.Recordset, error) {
	if ctx.Value(&sqlexec.RestrictedSQLExecutorKeyType{}) != nil {
		// We do not support run this function concurrently.
		// TODO: Maybe we should remove this restriction latter.
		return nil, errors.New("Should not call ExecRestrictedSQL concurrently.")
	}
	statements, err := Compile(ctx, sql)
	if err != nil {
		log.Errorf("Compile %s with error: %v", sql, err)
		return nil, errors.Trace(err)
	}
	if len(statements) != 1 {
		log.Errorf("ExecRestrictedSQL only executes one statement. Too many/few statement in %s", sql)
		return nil, errors.New("Wrong number of statement.")
	}
	st := statements[0]
	// Check statement for some restriction
	// For example only support DML on system meta table.
	// TODO: Add more restrictions.
	log.Debugf("Executing %s [%s]", st.OriginText(), sql)
	ctx.SetValue(&sqlexec.RestrictedSQLExecutorKeyType{}, true)
	defer ctx.ClearValue(&sqlexec.RestrictedSQLExecutorKeyType{})
	rs, err := st.Exec(ctx)
	return rs, errors.Trace(err)
}

// If forceNew is true, GetTxn() must return a new transaction.
// In this situation, if current transaction is still in progress,
// there will be an implicit commit and create a new transaction.
func (s *session) GetTxn(forceNew bool) (kv.Transaction, error) {
	var err error
	if s.txn == nil {
		s.resetHistory()
		s.txn, err = s.store.Begin()
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !s.isAutocommit(s) {
			variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusInTrans, true)
		}
		log.Infof("New txn:%s in session:%d", s.txn, s.sid)
		return s.txn, nil
	}
	if forceNew {
		err = s.FinishTxn(false)
		if err != nil {
			return nil, errors.Trace(err)
		}
		s.txn, err = s.store.Begin()
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !s.isAutocommit(s) {
			variable.GetSessionVars(s).SetStatusFlag(mysql.ServerStatusInTrans, true)
		}
		log.Warnf("Force new txn:%s in session:%d", s.txn, s.sid)
	}
	return s.txn, nil
}

// Next implements Executor Next interface.
func (e *SortExec) Next() (*Row, error) {
	if !e.fetched {
		for {
			srcRow, err := e.Src.Next()
			if err != nil {
				return nil, errors.Trace(err)
			}
			if srcRow == nil {
				break
			}
			orderRow := &orderByRow{
				row: srcRow,
				key: make([]interface{}, len(e.ByItems)),
			}
			for i, byItem := range e.ByItems {
				orderRow.key[i], err = evaluator.Eval(e.ctx, byItem.Expr)
				if err != nil {
					return nil, errors.Trace(err)
				}
			}
			e.Rows = append(e.Rows, orderRow)
		}
		sort.Sort(e)
		e.fetched = true
	}
	if e.err != nil {
		return nil, errors.Trace(e.err)
	}
	if e.Idx >= len(e.Rows) {
		return nil, nil
	}
	row := e.Rows[e.Idx].row
	e.Idx++
	return row, nil
}

// Alloc allocs the next autoID for table with tableID.
// It gets a batch of autoIDs at a time. So it does not need to access storage for each call.
func (alloc *allocator) Alloc(tableID int64) (int64, error) {
	if tableID == 0 {
		return 0, errors.New("Invalid tableID")
	}
	metaKey := meta.AutoIDKey(tableID)
	alloc.mu.Lock()
	defer alloc.mu.Unlock()
	if alloc.base == alloc.end { // step
		err := kv.RunInNewTxn(alloc.store, true, func(txn kv.Transaction) error {
			end, err := meta.GenID(txn, []byte(metaKey), step)
			if err != nil {
				return errors.Trace(err)
			}

			alloc.end = end
			alloc.base = alloc.end - step
			return nil
		})

		if err != nil {
			return 0, errors.Trace(err)
		}
	}

	alloc.base++
	log.Infof("Alloc id %d, table ID:%d, from %p, store ID:%s", alloc.base, tableID, alloc, alloc.store.UUID())
	return alloc.base, nil
}

// ComputeBit computes the bitwise operation on two datums.
func ComputeBit(op tipb.ExprType, left, right types.Datum) (types.Datum, error) {
	var result types.Datum
	a, err := types.CoerceArithmetic(left)
	if err != nil {
		return result, errors.Trace(err)
	}

	b, err := types.CoerceArithmetic(right)
	if err != nil {
		return result, errors.Trace(err)
	}
	a, b, err = types.CoerceDatum(a, b)
	if err != nil {
		return result, errors.Trace(err)
	}
	if a.IsNull() || b.IsNull() {
		return result, nil
	}

	switch op {
	case tipb.ExprType_BitAnd:
		return types.ComputeBitAnd(a, b)
	case tipb.ExprType_BitOr:
		return types.ComputeBitOr(a, b)
	case tipb.ExprType_BitXor:
		return types.ComputeBitXor(a, b)
	case tipb.ExprType_LeftShift:
		return types.ComputeLeftShift(a, b)
	case tipb.ExprType_RighShift:
		return types.ComputeRightShift(a, b)
	default:
		return result, errors.Errorf("Unknown binop type: %v", op)
	}
}

func (txn *dbTxn) Inc(k kv.Key, step int64) (int64, error) {
	log.Debugf("Inc %q, step %d txn:%d", k, step, txn.tid)
	k = kv.EncodeKey(k)

	txn.markOrigin(k)
	val, err := txn.UnionStore.Get(k)
	if kv.IsErrNotFound(err) {
		err = txn.UnionStore.Set(k, []byte(strconv.FormatInt(step, 10)))
		if err != nil {
			return 0, errors.Trace(err)
		}

		return step, nil
	}
	if err != nil {
		return 0, errors.Trace(err)
	}

	intVal, err := strconv.ParseInt(string(val), 10, 0)
	if err != nil {
		return intVal, errors.Trace(err)
	}

	intVal += step
	err = txn.UnionStore.Set(k, []byte(strconv.FormatInt(intVal, 10)))
	if err != nil {
		return 0, errors.Trace(err)
	}
	txn.store.compactor.OnSet(k)
	return intVal, nil
}

func (e *Evaluator) checkAnyResult(cs *ast.CompareSubqueryExpr, lv types.Datum, result []types.Datum) (d types.Datum, err error) {
	hasNull := false
	for _, v := range result {
		if v.IsNull() {
			hasNull = true
			continue
		}

		comRes, err1 := lv.CompareDatum(v)
		if err1 != nil {
			return d, errors.Trace(err1)
		}

		res, err1 := getCompResult(cs.Op, comRes)
		if err1 != nil {
			return d, errors.Trace(err1)
		}
		if res {
			d.SetInt64(boolToInt64(true))
			return d, nil
		}
	}

	if hasNull {
		// If no matched but we get null, return null.
		// Like `insert t (c) values (1),(2),(null)`, then
		// `select 0 > any (select c from t)`, returns null.
		return d, nil
	}

	d.SetInt64(boolToInt64(false))
	return d, nil
}

func (e *XSelectIndexExec) doIndexRequest() (*xapi.SelectResult, error) {
	txn, err := e.ctx.GetTxn(false)
	if err != nil {
		return nil, errors.Trace(err)
	}
	selIdxReq := new(tipb.SelectRequest)
	startTs := txn.StartTS()
	selIdxReq.StartTs = &startTs
	selIdxReq.IndexInfo = xapi.IndexToProto(e.table.Meta(), e.indexPlan.Index)
	if len(e.indexPlan.FilterConditions) == 0 {
		// Push limit to index request only if there is not filter conditions.
		selIdxReq.Limit = e.indexPlan.LimitCount
	}
	if e.indexPlan.Desc {
		selIdxReq.OrderBy = append(selIdxReq.OrderBy, &tipb.ByItem{Desc: &e.indexPlan.Desc})
	}
	fieldTypes := make([]*types.FieldType, len(e.indexPlan.Index.Columns))
	for i, v := range e.indexPlan.Index.Columns {
		fieldTypes[i] = &(e.table.Cols()[v.Offset].FieldType)
	}
	selIdxReq.Ranges, err = indexRangesToPBRanges(e.indexPlan.Ranges, fieldTypes)
	if err != nil {
		return nil, errors.Trace(err)
	}
	concurrency := 1
	if e.indexPlan.OutOfOrder {
		concurrency = 10
	}
	return xapi.Select(txn.GetClient(), selIdxReq, concurrency)
}

func (s *Server) loadSchemaInfo() error {
	if err := s.parseShards(); err != nil {
		return errors.Trace(err)
	}

	if err := s.parseSchemas(); err != nil {
		return errors.Trace(err)
	}

	for _, v := range s.cfg.Schemas {
		rc := v.RouterConifg
		var overrides []tabletserver.SchemaOverride
		for _, tr := range rc.TableRule {
			or := tabletserver.SchemaOverride{Name: tr.Table}
			pks := strings.Split(tr.ShardingKey, ",")
			for _, pk := range pks {
				or.PKColumns = append(or.PKColumns, strings.TrimSpace(pk))
			}
			log.Infof("table rule:%+v", tr)
			or.Cache = &tabletserver.OverrideCacheDesc{Type: tr.RowCacheType, Prefix: or.Name, Table: or.Name}
			overrides = append(overrides, or)
		}

		//fix hard code node
		sc := s.cfg.Shards[0]
		si := tabletserver.NewSchemaInfo(s.cfg.RowCacheConf, s.cfg.Shards[0].Master, sc.User, sc.Password, v.DB, overrides)

		log.Infof("%+v", si)
		s.autoSchamas[v.DB] = si
	}

	return nil
}

func indexRangesToPBRanges(ranges []*plan.IndexRange, fieldTypes []*types.FieldType) ([]*tipb.KeyRange, error) {
	keyRanges := make([]*tipb.KeyRange, 0, len(ranges))
	for _, ran := range ranges {
		err := convertIndexRangeTypes(ran, fieldTypes)
		if err != nil {
			return nil, errors.Trace(err)
		}
		low, err := codec.EncodeKey(nil, ran.LowVal...)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if ran.LowExclude {
			low = []byte(kv.Key(low).PrefixNext())
		}
		high, err := codec.EncodeKey(nil, ran.HighVal...)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if !ran.HighExclude {
			high = []byte(kv.Key(high).PrefixNext())
		}
		keyRanges = append(keyRanges, &tipb.KeyRange{Low: low, High: high})
	}
	return keyRanges, nil
}

// getExecRet executes restricted sql and the result is one column.
// It returns a string value.
func (s *session) getExecRet(ctx context.Context, sql string) (string, error) {
	cleanTxn := s.txn == nil
	rs, err := s.ExecRestrictedSQL(ctx, sql)
	if err != nil {
		return "", errors.Trace(err)
	}
	defer rs.Close()
	row, err := rs.Next()
	if err != nil {
		return "", errors.Trace(err)
	}
	if row == nil {
		return "", terror.ExecResultIsEmpty
	}
	value, err := types.ToString(row.Data[0])
	if err != nil {
		return "", errors.Trace(err)
	}
	if cleanTxn {
		// This function has some side effect. Run select may create new txn.
		// We should make environment unchanged.
		s.txn = nil
	}
	return value, nil
}

func (e *UpdateExec) fetchRows() error {
	for {
		row, err := e.SelectExec.Next()
		if err != nil {
			return errors.Trace(err)
		}
		if row == nil {
			return nil
		}
		data := make([]types.Datum, len(e.SelectExec.Schema()))
		newData := make([]types.Datum, len(e.SelectExec.Schema()))
		for i, s := range e.SelectExec.Schema() {
			data[i], err = s.Eval(row.Data, e.ctx)
			if err != nil {
				return errors.Trace(err)
			}
			newData[i] = data[i]
			if e.OrderedList[i] != nil {
				val, err := e.OrderedList[i].Expr.Eval(row.Data, e.ctx)
				if err != nil {
					return errors.Trace(err)
				}
				newData[i] = val
			}
		}
		row.Data = data
		e.rows = append(e.rows, row)
		e.newRowsData = append(e.newRowsData, newData)
	}
}

// Run implements Command.Run.
func (c *changePasswordCommand) Run(ctx *cmd.Context) error {
	if c.api == nil {
		api, err := c.NewUserManagerAPIClient()
		if err != nil {
			return errors.Trace(err)
		}
		c.api = api
		defer c.api.Close()
	}

	password, err := c.generateOrReadPassword(ctx, c.Generate)
	if err != nil {
		return errors.Trace(err)
	}

	var writer EnvironInfoCredsWriter

	var creds configstore.APICredentials

	if c.User == "" {
		// We get the creds writer before changing the password just to
		// minimise the things that could go wrong after changing the password
		// in the server.
		if c.writer == nil {
			writer, err = c.ConnectionInfo()
			if err != nil {
				return errors.Trace(err)
			}
		} else {
			writer = c.writer
		}

		creds = writer.APICredentials()
	} else {
		creds.User = c.User
	}

	oldPassword := creds.Password
	creds.Password = password
	if err = c.api.SetPassword(creds.User, password); err != nil {
		return block.ProcessBlockedError(err, block.BlockChange)
	}

	if c.User != "" {
		return writeServerFile(c, ctx, c.User, password, c.OutPath)
	}

	writer.SetAPICredentials(creds)
	if err := writer.Write(); err != nil {
		logger.Errorf("updating the cached credentials failed, reverting to original password")
		setErr := c.api.SetPassword(creds.User, oldPassword)
		if setErr != nil {
			logger.Errorf("failed to set password back, you will need to edit your environments file by hand to specify the password: %q", password)
			return errors.Annotate(setErr, "failed to set password back")
		}
		return errors.Annotate(err, "failed to write new password to environments file")
	}
	ctx.Infof("Your password has been updated.")
	return nil
}

func (e *InsertValues) getRows(cols []*table.Column) (rows [][]types.Datum, err error) {
	// process `insert|replace ... set x=y...`
	if err = e.fillValueList(); err != nil {
		return nil, errors.Trace(err)
	}

	defaultVals, err := e.getColumnDefaultValues(e.Table.Cols())
	if err != nil {
		return nil, errors.Trace(err)
	}

	rows = make([][]types.Datum, len(e.Lists))
	length := len(e.Lists[0])
	for i, list := range e.Lists {
		if err = e.checkValueCount(length, len(list), i, cols); err != nil {
			return nil, errors.Trace(err)
		}
		e.currRow = i
		rows[i], err = e.getRow(cols, list, defaultVals)
		if err != nil {
			return nil, errors.Trace(err)
		}
	}
	return
}

// Rows implements rset.Recordset.
func (r Recordset) Rows(limit, offset int) ([][]interface{}, error) {
	var rows [][]interface{}
	defer r.Plan.Close()
	// Move to offset.
	for offset > 0 {
		row, err := r.Next()
		if row == nil || err != nil {
			return nil, errors.Trace(err)
		}
		offset--
	}
	// Negative limit means no limit.
	for limit != 0 {
		row, err := r.Next()
		if err != nil {
			return nil, errors.Trace(err)
		}
		if row == nil {
			break
		}
		rows = append(rows, row.Data)
		if limit > 0 {
			limit--
		}
	}
	return rows, nil
}