// rewrite function rewrites ast expr to expression.Expression.
// aggMapper maps ast.AggregateFuncExpr to the columns offset in p's output schema.
// asScalar means whether this expression must be treated as a scalar expression.
// And this function returns a result expression, a new plan that may have apply or semi-join.
func (b *planBuilder) rewrite(expr ast.ExprNode, p LogicalPlan, aggMapper map[*ast.AggregateFuncExpr]int, asScalar bool) (
	expression.Expression, LogicalPlan, error) {
	er := &expressionRewriter{
		p:        p,
		aggrMap:  aggMapper,
		b:        b,
		asScalar: asScalar,
		ctx:      b.ctx,
	}
	if p != nil {
		er.schema = p.GetSchema()
	}
	expr.Accept(er)
	if er.err != nil {
		return nil, nil, errors.Trace(er.err)
	}
	if !asScalar && len(er.ctxStack) == 0 {
		return nil, er.p, nil
	}
	if len(er.ctxStack) != 1 {
		return nil, nil, errors.Errorf("context len %v is invalid", len(er.ctxStack))
	}
	if getRowLen(er.ctxStack[0]) != 1 {
		return nil, nil, ErrOperandColumns.GenByArgs(1)
	}
	result := expression.FoldConstant(b.ctx, er.ctxStack[0])
	return result, er.p, nil
}

func convertExpr(converter *expressionConverter, expr ast.ExprNode) (expression.Expression, error) {
	expr.Accept(converter)
	if converter.err != nil {
		return nil, errors.Trace(converter.err)
	}
	return converter.exprMap[expr], nil
}

// Eval evaluates an expression to a value.
func Eval(ctx context.Context, expr ast.ExprNode) (interface{}, error) {
	e := &Evaluator{ctx: ctx}
	expr.Accept(e)
	if e.err != nil {
		return nil, errors.Trace(e.err)
	}
	return expr.GetValue(), nil
}

// EvalDatum evaluates an expression to a datum.
func EvalDatum(ctx context.Context, expr ast.ExprNode) (d types.Datum, err error) {
	e := &Evaluator{ctx: ctx}
	expr.Accept(e)
	if e.err != nil {
		return d, errors.Trace(e.err)
	}
	return *expr.GetDatum(), nil
}

func (b *planBuilder) rewrite(expr ast.ExprNode, p Plan, aggMapper map[*ast.AggregateFuncExpr]int) (newExpr expression.Expression, newPlan Plan, correlated bool, err error) {
	er := &expressionRewriter{p: p, aggrMap: aggMapper, schema: p.GetSchema(), b: b}
	expr.Accept(er)
	if er.err != nil {
		return nil, nil, false, errors.Trace(er.err)
	}
	if len(er.ctxStack) != 1 {
		return nil, nil, false, errors.Errorf("context len %v is invalid", len(er.ctxStack))
	}
	return er.ctxStack[0], er.p, er.correlated, nil
}

// attachCondition tries to attach a condition as deep as possible.
// availablePaths are paths join before this path.
// onCond represents whether the conditions is from current join's on condition. The on condition from other joins is treated as where condition.
func (p *joinPath) attachCondition(condition ast.ExprNode, availablePaths []*joinPath, onCond bool) (attached bool) {
	filterRate := guesstimateFilterRate(condition)
	// table
	if p.table != nil || p.subquery != nil {
		attacher := conditionAttachChecker{targetPath: p, availablePaths: availablePaths}
		condition.Accept(&attacher)
		if attacher.invalid {
			return false
		}
		p.conditions = append(p.conditions, condition)
		p.filterRate *= filterRate
		return true
	}
	// inner join
	if len(p.inners) > 0 {
		for _, in := range p.inners {
			if in.attachCondition(condition, availablePaths, false) {
				p.filterRate *= filterRate
				return true
			}
		}
		attacher := &conditionAttachChecker{targetPath: p, availablePaths: availablePaths}
		condition.Accept(attacher)
		if attacher.invalid {
			return false
		}
		p.conditions = append(p.conditions, condition)
		p.filterRate *= filterRate
		return true
	}

	// outer join
	if p.outer.attachCondition(condition, availablePaths, false) {
		p.filterRate *= filterRate
		return true
	}
	// can't attach any where condition
	if onCond && p.inner.attachCondition(condition, availablePaths, false) {
		p.filterRate *= filterRate
		return true
	}
	return false
}

// exprToPBExpr converts an ast.ExprNode to a tipb.Expr, if not supported, nil will be returned.
func (b *executorBuilder) exprToPBExpr(client kv.Client, expr ast.ExprNode, tn *ast.TableName) *tipb.Expr {
	switch x := expr.(type) {
	case *ast.ValueExpr, *ast.ParamMarkerExpr:
		return b.datumToPBExpr(client, *expr.GetDatum())
	case *ast.ColumnNameExpr:
		return b.columnNameToPBExpr(client, x, tn)
	case *ast.BinaryOperationExpr:
		return b.binopToPBExpr(client, x, tn)
	case *ast.ParenthesesExpr:
		return b.exprToPBExpr(client, x.Expr, tn)
	case *ast.PatternLikeExpr:
		return b.likeToPBExpr(client, x, tn)
	case *ast.UnaryOperationExpr:
		return b.unaryToPBExpr(client, x, tn)
	case *ast.PatternInExpr:
		return b.patternInToPBExpr(client, x, tn)
	case *ast.SubqueryExpr:
		return b.subqueryToPBExpr(client, x)
	case *ast.AggregateFuncExpr:
		return b.aggFuncToPBExpr(client, x, tn)
	default:
		return nil
	}
}

// rewrite function rewrites ast expr to expression.Expression.
// aggMapper maps ast.AggregateFuncExpr to the columns offset in p's output schema.
// asScalar means whether this expression must be treated as a scalar expression.
func (b *planBuilder) rewrite(expr ast.ExprNode, p LogicalPlan, aggMapper map[*ast.AggregateFuncExpr]int, asScalar bool) (
	newExpr expression.Expression, newPlan LogicalPlan, correlated bool, err error) {
	er := &expressionRewriter{
		p:        p,
		aggrMap:  aggMapper,
		schema:   p.GetSchema(),
		b:        b,
		asScalar: asScalar,
	}
	expr.Accept(er)
	if er.err != nil {
		return nil, nil, false, errors.Trace(er.err)
	}
	if !asScalar && len(er.ctxStack) == 0 {
		return nil, er.p, er.correlated, nil
	}
	if len(er.ctxStack) != 1 {
		return nil, nil, false, errors.Errorf("context len %v is invalid", len(er.ctxStack))
	}
	if getRowLen(er.ctxStack[0]) != 1 {
		return nil, nil, false, errors.New("Operand should contain 1 column(s)")
	}
	return er.ctxStack[0], er.p, er.correlated, nil
}

// AddCastToString adds a cast function to string type if the expr charset is not UTF8.
func (v *typeInferrer) addCastToString(expr ast.ExprNode) ast.ExprNode {
	if !mysql.IsUTF8Charset(expr.GetType().Charset) {
		castTp := types.NewFieldType(mysql.TypeString)
		castTp.Charset, castTp.Collate = types.DefaultCharsetForType(mysql.TypeString)
		if val, ok := expr.(*ast.ValueExpr); ok {
			newVal, err := val.Datum.ConvertTo(v.sc, castTp)
			if err != nil {
				v.err = errors.Trace(err)
			}
			expr.SetDatum(newVal)
		} else {
			castFunc := &ast.FuncCastExpr{
				Expr:         expr,
				Tp:           castTp,
				FunctionType: ast.CastFunction,
			}
			expr = castFunc
		}
		expr.SetType(castTp)
	}
	return expr
}

// Eval evaluates an expression to a datum.
func Eval(ctx context.Context, expr ast.ExprNode) (d types.Datum, err error) {
	if ast.IsEvaluated(expr) {
		return *expr.GetDatum(), nil
	}
	e := &Evaluator{ctx: ctx}
	expr.Accept(e)
	if e.err != nil {
		return d, errors.Trace(e.err)
	}
	if ast.IsPreEvaluable(expr) && (expr.GetFlag()&ast.FlagHasFunc == 0) {
		expr.SetFlag(expr.GetFlag() | ast.FlagPreEvaluated)
	}
	return *expr.GetDatum(), nil
}