// newReturningHelper creates a new returningHelper for use by an
// insert/update node.
func (p *planner) newReturningHelper(
	r parser.ReturningExprs,
	desiredTypes []parser.Type,
	alias string,
	tablecols []sqlbase.ColumnDescriptor,
) (*returningHelper, error) {
	rh := &returningHelper{
		p: p,
	}
	if len(r) == 0 {
		return rh, nil
	}

	for _, e := range r {
		if err := p.parser.AssertNoAggregationOrWindowing(
			e.Expr, "RETURNING", p.session.SearchPath,
		); err != nil {
			return nil, err
		}
	}

	rh.columns = make(ResultColumns, 0, len(r))
	aliasTableName := parser.TableName{TableName: parser.Name(alias)}
	rh.source = newSourceInfoForSingleTable(aliasTableName, makeResultColumns(tablecols))
	rh.exprs = make([]parser.TypedExpr, 0, len(r))
	ivarHelper := parser.MakeIndexedVarHelper(rh, len(tablecols))
	for i, target := range r {
		// Pre-normalize VarNames at the top level so that checkRenderStar can see stars.
		if err := target.NormalizeTopLevelVarName(); err != nil {
			return nil, err
		}

		if isStar, cols, typedExprs, err := checkRenderStar(target, rh.source, ivarHelper); err != nil {
			return nil, err
		} else if isStar {
			rh.exprs = append(rh.exprs, typedExprs...)
			rh.columns = append(rh.columns, cols...)
			continue
		}

		// When generating an output column name it should exactly match the original
		// expression, so determine the output column name before we perform any
		// manipulations to the expression.
		outputName := getRenderColName(target)

		desired := parser.TypeAny
		if len(desiredTypes) > i {
			desired = desiredTypes[i]
		}

		typedExpr, err := rh.p.analyzeExpr(target.Expr, multiSourceInfo{rh.source}, ivarHelper, desired, false, "")
		if err != nil {
			return nil, err
		}
		rh.exprs = append(rh.exprs, typedExpr)
		rh.columns = append(rh.columns, ResultColumn{Name: outputName, Typ: typedExpr.ResolvedType()})
	}
	return rh, nil
}

func makeSelectNode(t *testing.T) *selectNode {
	desc := testTableDesc()
	sel := testInitDummySelectNode(desc)
	if err := desc.AllocateIDs(); err != nil {
		t.Fatal(err)
	}
	numColumns := len(sel.sourceInfo[0].sourceColumns)
	sel.ivarHelper = parser.MakeIndexedVarHelper(sel, numColumns)
	sel.curSourceRow = make(parser.DTuple, numColumns)
	return sel
}

func (eh *exprHelper) init(
	expr Expression, types []sqlbase.ColumnType_Kind, evalCtx *parser.EvalContext,
) error {
	if expr.Expr == "" {
		return nil
	}
	eh.types = types
	eh.evalCtx = evalCtx
	eh.vars = parser.MakeIndexedVarHelper(eh, len(types))
	var err error
	eh.expr, err = processExpression(expr, &eh.vars)
	return err
}

func newAggregator(
	ctx *FlowCtx, spec *AggregatorSpec, input RowSource, output RowReceiver,
) (*aggregator, error) {
	ag := &aggregator{
		input:       input,
		output:      output,
		ctx:         log.WithLogTag(ctx.Context, "Agg", nil),
		rows:        &RowBuffer{},
		buckets:     make(map[string]struct{}),
		inputCols:   make(columns, len(spec.Exprs)),
		outputTypes: make([]*sqlbase.ColumnType, len(spec.Exprs)),
		groupCols:   make(columns, len(spec.GroupCols)),
	}

	inputTypes := make([]*sqlbase.ColumnType, len(spec.Exprs))
	for i, expr := range spec.Exprs {
		ag.inputCols[i] = expr.ColIdx
		inputTypes[i] = spec.Types[expr.ColIdx]
	}
	copy(ag.groupCols, spec.GroupCols)

	// Loop over the select expressions and extract any aggregate functions --
	// non-aggregation functions are replaced with parser.NewIdentAggregate,
	// (which just returns the last value added to them for a bucket) to provide
	// grouped-by values for each bucket.  ag.funcs is updated to contain all
	// the functions which need to be fed values.
	eh := &exprHelper{types: inputTypes}
	eh.vars = parser.MakeIndexedVarHelper(eh, len(eh.types))
	for i, expr := range spec.Exprs {
		fn, retType, err := ag.extractFunc(expr, eh)
		if err != nil {
			return nil, err
		}
		ag.funcs = append(ag.funcs, fn)

		// The aggregate function extracted is an identity function, the return
		// type of this therefore being the i-th input type.
		if retType == nil {
			ag.outputTypes[i] = inputTypes[i]
		} else {
			typ := sqlbase.DatumTypeToColumnType(retType)
			ag.outputTypes[i] = &typ
		}
	}

	return ag, nil
}

func (eh *exprHelper) init(
	expr Expression, types []sqlbase.ColumnType_Kind, evalCtx *parser.EvalContext,
) error {
	if expr.Expr == "" {
		return nil
	}
	eh.types = types
	eh.evalCtx = evalCtx
	eh.vars = parser.MakeIndexedVarHelper(eh, len(types))
	var err error
	eh.expr, err = processExpression(expr, &eh.vars)
	if err != nil {
		return err
	}
	var p parser.Parser
	if p.AggregateInExpr(eh.expr, evalCtx.SearchPath) {
		return errors.Errorf("expression '%s' has aggregate", eh.expr)
	}
	return nil
}

func TestProcessExpression(t *testing.T) {
	defer leaktest.AfterTest(t)()

	e := Expression{Expr: "@1 * (@2 + @3) + @1"}
	h := parser.MakeIndexedVarHelper(testVarContainer{}, 4)
	expr, err := processExpression(e, &h)
	if err != nil {
		t.Fatal(err)
	}

	if !h.IndexedVarUsed(0) || !h.IndexedVarUsed(1) || !h.IndexedVarUsed(2) || h.IndexedVarUsed(3) {
		t.Errorf("invalid IndexedVarUsed results %t %t %t %t (expected false false false true)",
			h.IndexedVarUsed(0), h.IndexedVarUsed(1), h.IndexedVarUsed(2), h.IndexedVarUsed(3))
	}

	str := expr.String()
	expectedStr := "(var0 * (var1 + var2)) + var0"
	if str != expectedStr {
		t.Errorf("invalid expression string '%s', expected '%s'", str, expectedStr)
	}
}

// newReturningHelper creates a new returningHelper for use by an
// insert/update node.
func (p *planner) newReturningHelper(
	r parser.ReturningExprs,
	desiredTypes []parser.Type,
	alias string,
	tablecols []sqlbase.ColumnDescriptor,
) (*returningHelper, error) {
	rh := &returningHelper{
		p: p,
	}
	if len(r) == 0 {
		return rh, nil
	}

	for _, e := range r {
		if err := p.parser.AssertNoAggregationOrWindowing(
			e.Expr, "RETURNING", p.session.SearchPath,
		); err != nil {
			return nil, err
		}
	}

	rh.columns = make(ResultColumns, 0, len(r))
	aliasTableName := parser.TableName{TableName: parser.Name(alias)}
	rh.source = newSourceInfoForSingleTable(aliasTableName, makeResultColumns(tablecols))
	rh.exprs = make([]parser.TypedExpr, 0, len(r))
	ivarHelper := parser.MakeIndexedVarHelper(rh, len(tablecols))
	for _, target := range r {
		cols, typedExprs, _, err := p.computeRender(target, parser.TypeAny, rh.source, ivarHelper, true)
		if err != nil {
			return nil, err
		}
		rh.columns = append(rh.columns, cols...)
		rh.exprs = append(rh.exprs, typedExprs...)
	}
	return rh, nil
}

// SelectClause selects rows from a single table. Select is the workhorse of the
// SQL statements. In the slowest and most general case, select must perform
// full table scans across multiple tables and sort and join the resulting rows
// on arbitrary columns. Full table scans can be avoided when indexes can be
// used to satisfy the where-clause. scanVisibility controls which columns are
// visible to the select.
//
// NB: This is passed directly to planNode only when there is no ORDER BY,
// LIMIT, or parenthesis in the parsed SELECT. See `sql/parser.Select` and
// `sql/parser.SelectStatement`.
//
// Privileges: SELECT on table
//   Notes: postgres requires SELECT. Also requires UPDATE on "FOR UPDATE".
//          mysql requires SELECT.
func (p *planner) SelectClause(
	parsed *parser.SelectClause,
	orderBy parser.OrderBy,
	limit *parser.Limit,
	desiredTypes []parser.Type,
	scanVisibility scanVisibility,
) (planNode, error) {
	s := &selectNode{planner: p}

	if err := s.initFrom(parsed, scanVisibility); err != nil {
		return nil, err
	}

	s.ivarHelper = parser.MakeIndexedVarHelper(s, len(s.sourceInfo[0].sourceColumns))

	if err := s.initTargets(parsed.Exprs, desiredTypes); err != nil {
		return nil, err
	}

	if err := s.initWhere(parsed.Where); err != nil {
		return nil, err
	}

	// NB: orderBy, window, and groupBy are passed and can modify the selectNode,
	// but must do so in that order.
	sort, err := p.orderBy(orderBy, s)
	if err != nil {
		return nil, err
	}
	window, err := p.window(parsed, s)
	if err != nil {
		return nil, err
	}
	group, err := p.groupBy(parsed, s)
	if err != nil {
		return nil, err
	}

	if s.filter != nil && group != nil {
		// Allow the group-by to add an implicit "IS NOT NULL" filter.
		s.filter = group.isNotNullFilter(s.filter)
	}

	limitPlan, err := p.Limit(limit)
	if err != nil {
		return nil, err
	}
	distinctPlan := p.Distinct(parsed)

	result := &selectTopNode{
		source:   s,
		group:    group,
		window:   window,
		sort:     sort,
		distinct: distinctPlan,
		limit:    limitPlan,
	}
	s.top = result
	limitPlan.setTop(result)
	distinctPlan.setTop(result)

	return result, nil
}

func (p *planner) makeUpsertHelper(
	tn *parser.TableName,
	tableDesc *sqlbase.TableDescriptor,
	insertCols []sqlbase.ColumnDescriptor,
	updateCols []sqlbase.ColumnDescriptor,
	updateExprs parser.UpdateExprs,
	upsertConflictIndex *sqlbase.IndexDescriptor,
) (*upsertHelper, error) {
	defaultExprs, err := makeDefaultExprs(updateCols, &p.parser, &p.evalCtx)
	if err != nil {
		return nil, err
	}

	untupledExprs := make(parser.Exprs, 0, len(updateExprs))
	i := 0
	for _, updateExpr := range updateExprs {
		if updateExpr.Tuple {
			if t, ok := updateExpr.Expr.(*parser.Tuple); ok {
				for _, e := range t.Exprs {
					typ := updateCols[i].Type.ToDatumType()
					e := fillDefault(e, typ, i, defaultExprs)
					untupledExprs = append(untupledExprs, e)
					i++
				}
			}
		} else {
			typ := updateCols[i].Type.ToDatumType()
			e := fillDefault(updateExpr.Expr, typ, i, defaultExprs)
			untupledExprs = append(untupledExprs, e)
			i++
		}
	}

	sourceInfo := newSourceInfoForSingleTable(*tn, makeResultColumns(tableDesc.Columns))
	excludedSourceInfo := newSourceInfoForSingleTable(upsertExcludedTable, makeResultColumns(insertCols))

	helper := &upsertHelper{
		p:                  p,
		sourceInfo:         sourceInfo,
		excludedSourceInfo: excludedSourceInfo,
	}

	var evalExprs []parser.TypedExpr
	ivarHelper := parser.MakeIndexedVarHelper(helper, len(sourceInfo.sourceColumns)+len(excludedSourceInfo.sourceColumns))
	sources := multiSourceInfo{sourceInfo, excludedSourceInfo}
	for _, expr := range untupledExprs {
		normExpr, err := p.analyzeExpr(expr, sources, ivarHelper, parser.NoTypePreference, false, "")
		if err != nil {
			return nil, err
		}
		evalExprs = append(evalExprs, normExpr)
	}
	helper.evalExprs = evalExprs

	helper.allExprsIdentity = true
	for i, expr := range evalExprs {
		// analyzeExpr above has normalized all direct column names to ColumnItems.
		c, ok := expr.(*parser.ColumnItem)
		if !ok {
			helper.allExprsIdentity = false
			break
		}
		if len(c.Selector) > 0 ||
			!c.TableName.TableName.Equal(upsertExcludedTable.TableName) ||
			c.ColumnName.Normalize() != parser.ReNormalizeName(updateCols[i].Name) {
			helper.allExprsIdentity = false
			break
		}
	}

	return helper, nil
}