// findColumn looks up the column described by a QualifiedName. The qname will be normalized.
func (s *selectNode) findColumn(qname *parser.QualifiedName) (columnRef, *roachpb.Error) {

	ref := columnRef{colIdx: invalidColIdx}

	if err := roachpb.NewError(qname.NormalizeColumnName()); err != nil {
		return ref, err
	}

	// We can't resolve stars to a single column.
	if qname.IsStar() {
		err := roachpb.NewUErrorf("qualified name \"%s\" not found", qname)
		return ref, err
	}

	// TODO(radu): when we support multiple FROMs, we will find the node with the correct alias; if
	// no alias is given, we will search for the column in all FROMs and make sure there is only
	// one.  For now we just check that the name matches (if given).
	if qname.Base == "" {
		qname.Base = parser.Name(s.from.alias)
	}
	if equalName(s.from.alias, string(qname.Base)) {
		colName := qname.Column()
		for idx, col := range s.from.columns {
			if equalName(col.Name, colName) {
				ref.from = &s.from
				ref.colIdx = idx
				return ref, nil
			}
		}
	}

	err := roachpb.NewUErrorf("qualified name \"%s\" not found", qname)
	return ref, err
}

func (p *planner) SetTimeZone(n *parser.SetTimeZone) (planNode, *roachpb.Error) {
	d, err := n.Value.Eval(p.evalCtx)
	if err != nil {
		return nil, roachpb.NewError(err)
	}
	var offset int64
	switch v := d.(type) {
	case parser.DString:
		location := string(v)
		if location == "DEFAULT" || location == "LOCAL" {
			location = "UTC"
		}
		if _, err := time.LoadLocation(location); err != nil {
			return nil, roachpb.NewUErrorf("cannot find time zone %q: %v", location, err)
		}
		p.session.Timezone = &Session_Location{Location: location}

	case parser.DInterval:
		offset = int64(v.Duration / time.Second)

	case parser.DInt:
		offset = int64(v) * 60 * 60

	case parser.DFloat:
		offset = int64(float64(v) * 60.0 * 60.0)

	default:
		return nil, roachpb.NewUErrorf("bad time zone value: %v", n.Value)
	}
	if offset != 0 {
		p.session.Timezone = &Session_Offset{Offset: offset}
	}
	p.evalCtx.GetLocation = p.session.getLocation
	return &valuesNode{}, nil
}

// getCachedDatabaseDesc looks up the database descriptor given its name in the
// descriptor cache.
func (p *planner) getCachedDatabaseDesc(name string) (*DatabaseDescriptor, *roachpb.Error) {
	if name == systemDB.Name {
		return &systemDB, nil
	}

	nameKey := databaseKey{name}
	nameVal := p.systemConfig.GetValue(nameKey.Key())
	if nameVal == nil {
		return nil, roachpb.NewUErrorf("database %q does not exist in system cache", name)
	}

	id, err := nameVal.GetInt()
	if err != nil {
		return nil, roachpb.NewError(err)
	}

	descKey := MakeDescMetadataKey(ID(id))
	descVal := p.systemConfig.GetValue(descKey)
	if descVal == nil {
		return nil, roachpb.NewUErrorf("database %q has name entry, but no descriptor in system cache", name)
	}

	desc := &Descriptor{}
	if err := descVal.GetProto(desc); err != nil {
		return nil, roachpb.NewError(err)
	}

	database := desc.GetDatabase()
	if database == nil {
		return nil, roachpb.NewErrorf("%q is not a database", name)
	}
	return database, roachpb.NewError(database.Validate())
}

// RenameDatabase renames the database.
// Privileges: "root" user.
//   Notes: postgres requires superuser, db owner, or "CREATEDB".
//          mysql >= 5.1.23 does not allow database renames.
func (p *planner) RenameDatabase(n *parser.RenameDatabase) (planNode, *roachpb.Error) {
	if n.Name == "" || n.NewName == "" {
		return nil, roachpb.NewError(errEmptyDatabaseName)
	}

	if p.user != security.RootUser {
		return nil, roachpb.NewUErrorf("only %s is allowed to rename databases", security.RootUser)
	}

	dbDesc, pErr := p.getDatabaseDesc(string(n.Name))
	if pErr != nil {
		return nil, pErr
	}

	if n.Name == n.NewName {
		// Noop.
		return &valuesNode{}, nil
	}

	// Now update the nameMetadataKey and the descriptor.
	descKey := MakeDescMetadataKey(dbDesc.GetID())
	dbDesc.SetName(string(n.NewName))

	if err := dbDesc.Validate(); err != nil {
		return nil, roachpb.NewError(err)
	}

	newKey := databaseKey{string(n.NewName)}.Key()
	oldKey := databaseKey{string(n.Name)}.Key()
	descID := dbDesc.GetID()
	descDesc := wrapDescriptor(dbDesc)

	b := client.Batch{}
	b.CPut(newKey, descID, nil)
	b.Put(descKey, descDesc)
	b.Del(oldKey)

	if pErr := p.txn.Run(&b); pErr != nil {
		if _, ok := pErr.GoError().(*roachpb.ConditionFailedError); ok {
			return nil, roachpb.NewUErrorf("the new database name %q already exists", string(n.NewName))
		}
		return nil, pErr
	}

	p.testingVerifyMetadata = func(systemConfig config.SystemConfig) error {
		if err := expectDescriptorID(systemConfig, newKey, descID); err != nil {
			return err
		}
		if err := expectDescriptor(systemConfig, descKey, descDesc); err != nil {
			return err
		}
		return expectDeleted(systemConfig, oldKey)
	}

	return &valuesNode{}, nil
}

// Values constructs a valuesNode from a VALUES expression.
func (p *planner) Values(n *parser.Values) (planNode, *roachpb.Error) {
	v := &valuesNode{
		rows: make([]parser.DTuple, 0, len(n.Tuples)),
	}

	for num, tupleOrig := range n.Tuples {
		if num == 0 {
			v.columns = make([]ResultColumn, 0, len(tupleOrig.Exprs))
		} else if a, e := len(tupleOrig.Exprs), len(v.columns); a != e {
			return nil, roachpb.NewUErrorf("VALUES lists must all be the same length, %d for %d", a, e)
		}

		// We must not modify the Values node in-place (or the changes will leak if we retry this
		// transaction).
		tuple := parser.Tuple{Exprs: parser.Exprs(append([]parser.Expr(nil), tupleOrig.Exprs...))}

		for i := range tuple.Exprs {
			var pErr *roachpb.Error
			tuple.Exprs[i], pErr = p.expandSubqueries(tuple.Exprs[i], 1)
			if pErr != nil {
				return nil, pErr
			}
			var err error
			typ, err := tuple.Exprs[i].TypeCheck(p.evalCtx.Args)
			if err != nil {
				return nil, roachpb.NewError(err)
			}
			tuple.Exprs[i], err = p.parser.NormalizeExpr(p.evalCtx, tuple.Exprs[i])
			if err != nil {
				return nil, roachpb.NewError(err)
			}
			if num == 0 {
				v.columns = append(v.columns, ResultColumn{Name: "column" + strconv.Itoa(i+1), Typ: typ})
			} else if v.columns[i].Typ == parser.DNull {
				v.columns[i].Typ = typ
			} else if typ != parser.DNull && !typ.TypeEqual(v.columns[i].Typ) {
				return nil, roachpb.NewUErrorf("VALUES list type mismatch, %s for %s", typ.Type(), v.columns[i].Typ.Type())
			}
		}
		data, err := tuple.Eval(p.evalCtx)
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		vals, ok := data.(parser.DTuple)
		if !ok {
			return nil, roachpb.NewUErrorf("expected a tuple, but found %T", data)
		}
		v.rows = append(v.rows, vals)
	}

	return v, nil
}

// limit constructs a limitNode based on the LIMIT and OFFSET clauses.
func (p *planner) limit(n *parser.Select, plan planNode) (planNode, *roachpb.Error) {
	if n.Limit == nil {
		return plan, nil
	}

	var count, offset int64

	data := []struct {
		name       string
		src        parser.Expr
		dst        *int64
		defaultVal int64
	}{
		{"LIMIT", n.Limit.Count, &count, math.MaxInt64},
		{"OFFSET", n.Limit.Offset, &offset, 0},
	}

	for _, datum := range data {
		if datum.src == nil {
			*datum.dst = datum.defaultVal
		} else {
			if parser.ContainsVars(datum.src) {
				return nil, roachpb.NewUErrorf("argument of %s must not contain variables", datum.name)
			}

			normalized, err := p.parser.NormalizeExpr(p.evalCtx, datum.src)
			if err != nil {
				return nil, roachpb.NewError(err)
			}
			dstDatum, err := normalized.Eval(p.evalCtx)
			if err != nil {
				return nil, roachpb.NewError(err)
			}

			if dstDatum == parser.DNull {
				*datum.dst = datum.defaultVal
				continue
			}

			if dstDInt, ok := dstDatum.(parser.DInt); ok {
				*datum.dst = int64(dstDInt)
				continue
			}

			return nil, roachpb.NewUErrorf("argument of %s must be type %s, not type %s", datum.name, parser.DummyInt.Type(), dstDatum.Type())
		}
	}

	return &limitNode{planNode: plan, count: count, offset: offset}, nil
}

func (p *planner) getStringVal(name string, values parser.Exprs) (string, *roachpb.Error) {
	if len(values) != 1 {
		return "", roachpb.NewUErrorf("%s: requires a single string value", name)
	}
	val, err := values[0].Eval(p.evalCtx)
	if err != nil {
		return "", roachpb.NewError(err)
	}
	s, ok := val.(parser.DString)
	if !ok {
		return "", roachpb.NewUErrorf("%s: requires a single string value: %s is a %s",
			name, values[0], val.Type())
	}
	return string(s), nil
}

// Select selects rows from a SELECT/UNION/VALUES, ordering and/or limiting them.
func (p *planner) Select(n *parser.Select, autoCommit bool) (planNode, *roachpb.Error) {
	wrapped := n.Select
	limit := n.Limit
	orderBy := n.OrderBy

	for s, ok := wrapped.(*parser.ParenSelect); ok; s, ok = wrapped.(*parser.ParenSelect) {
		wrapped = s.Select.Select
		if s.Select.OrderBy != nil {
			if orderBy != nil {
				return nil, roachpb.NewUErrorf("multiple ORDER BY clauses not allowed")
			}
			orderBy = s.Select.OrderBy
		}
		if s.Select.Limit != nil {
			if limit != nil {
				return nil, roachpb.NewUErrorf("multiple LIMIT clauses not allowed")
			}
			limit = s.Select.Limit
		}
	}

	switch s := wrapped.(type) {
	case *parser.SelectClause:
		// Select can potentially optimize index selection if it's being ordered,
		// so we allow it to do its own sorting.
		node := &selectNode{planner: p}
		return p.initSelect(node, s, orderBy, limit)
	// TODO(dan): Union can also do optimizations when it has an ORDER BY, but
	// currently expects the ordering to be done externally, so we let it fall
	// through. Instead of continuing this special casing, it may be worth
	// investigating a general mechanism for passing some context down during
	// plan node construction.
	default:
		plan, pberr := p.makePlan(s, autoCommit)
		if pberr != nil {
			return nil, pberr
		}
		sort, pberr := p.orderBy(orderBy, plan)
		if pberr != nil {
			return nil, pberr
		}
		count, offset, err := p.evalLimit(limit)
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		return p.limit(count, offset, sort.wrap(plan)), nil
	}
}

// Values constructs a valuesNode from a VALUES expression.
func (p *planner) Values(n parser.Values) (planNode, *roachpb.Error) {
	v := &valuesNode{
		rows: make([]parser.DTuple, 0, len(n)),
	}

	for num, tuple := range n {
		if num == 0 {
			v.columns = make([]resultColumn, 0, len(tuple))
		} else if a, e := len(tuple), len(v.columns); a != e {
			return nil, roachpb.NewUErrorf("VALUES lists must all be the same length, %d for %d", a, e)
		}

		for i := range tuple {
			var pErr *roachpb.Error
			tuple[i], pErr = p.expandSubqueries(tuple[i], 1)
			if pErr != nil {
				return nil, pErr
			}
			var err error
			typ, err := tuple[i].TypeCheck(p.evalCtx.Args)
			if err != nil {
				return nil, roachpb.NewError(err)
			}
			tuple[i], err = p.parser.NormalizeExpr(p.evalCtx, tuple[i])
			if err != nil {
				return nil, roachpb.NewError(err)
			}
			if num == 0 {
				v.columns = append(v.columns, resultColumn{name: "column" + strconv.Itoa(i+1), typ: typ})
			} else if v.columns[i].typ == parser.DNull {
				v.columns[i].typ = typ
			} else if typ != parser.DNull && typ != v.columns[i].typ {
				return nil, roachpb.NewUErrorf("VALUES list type mismatch, %s for %s", typ.Type(), v.columns[i].typ.Type())
			}
		}
		data, err := tuple.Eval(p.evalCtx)
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		vals, ok := data.(parser.DTuple)
		if !ok {
			return nil, roachpb.NewUErrorf("expected a tuple, but found %T", data)
		}
		v.rows = append(v.rows, vals)
	}

	return v, nil
}

func (p *planner) processColumns(tableDesc *TableDescriptor,
	node parser.QualifiedNames) ([]ColumnDescriptor, *roachpb.Error) {
	if node == nil {
		// VisibleColumns is used here to prevent INSERT INTO <table> VALUES (...)
		// (as opposed to INSERT INTO <table> (...) VALUES (...)) from writing
		// hidden columns. At present, the only hidden column is the implicit rowid
		// primary key column.
		return tableDesc.VisibleColumns(), nil
	}

	cols := make([]ColumnDescriptor, len(node))
	colIDSet := make(map[ColumnID]struct{}, len(node))
	for i, n := range node {
		// TODO(pmattis): If the name is qualified, verify the table name matches
		// tableDesc.Name.
		if err := n.NormalizeColumnName(); err != nil {
			return nil, roachpb.NewError(err)
		}
		col, pErr := tableDesc.FindActiveColumnByName(n.Column())
		if pErr != nil {
			return nil, pErr
		}
		if _, ok := colIDSet[col.ID]; ok {
			return nil, roachpb.NewUErrorf("multiple assignments to same column \"%s\"", n.Column())
		}
		colIDSet[col.ID] = struct{}{}
		cols[i] = col
	}

	return cols, nil
}

// Show a session-local variable name.
func (p *planner) Show(n *parser.Show) (planNode, *roachpb.Error) {
	name := strings.ToUpper(n.Name)

	v := &valuesNode{columns: []ResultColumn{{Name: name, Typ: parser.DummyString}}}

	switch name {
	case `DATABASE`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(p.session.Database)})
	case `TIME ZONE`:
		loc, err := p.evalCtx.GetLocation()
		if err != nil {
			return nil, roachpb.NewError(err)
		}
		v.rows = append(v.rows, []parser.Datum{parser.DString(loc.String())})
	case `SYNTAX`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(parser.Syntax(p.session.Syntax).String())})
	case `DEFAULT_TRANSACTION_ISOLATION`:
		v.rows = append(v.rows, []parser.Datum{parser.DString("SERIALIZABLE")})
	case `TRANSACTION ISOLATION LEVEL`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(p.txn.Proto.Isolation.String())})
	case `TRANSACTION PRIORITY`:
		v.rows = append(v.rows, []parser.Datum{parser.DString(p.txn.UserPriority.String())})
	default:
		return nil, roachpb.NewUErrorf("unknown variable: %q", name)
	}

	return v, nil
}

// Initializes a scanNode with a tableName. Returns the table or index name that can be used for
// fully-qualified columns if an alias is not specified.
func (n *scanNode) initTable(
	p *planner, tableName *parser.QualifiedName, indexHints *parser.IndexHints,
) (string, *roachpb.Error) {
	if n.desc, n.pErr = p.getTableLease(tableName); n.pErr != nil {
		return "", n.pErr
	}

	if err := p.checkPrivilege(&n.desc, privilege.SELECT); err != nil {
		return "", roachpb.NewError(err)
	}

	alias := n.desc.Name

	if indexHints != nil && indexHints.Index != "" {
		indexName := NormalizeName(string(indexHints.Index))
		if indexName == NormalizeName(n.desc.PrimaryIndex.Name) {
			n.specifiedIndex = &n.desc.PrimaryIndex
		} else {
			for i := range n.desc.Indexes {
				if indexName == NormalizeName(n.desc.Indexes[i].Name) {
					n.specifiedIndex = &n.desc.Indexes[i]
					break
				}
			}
			if n.specifiedIndex == nil {
				n.pErr = roachpb.NewUErrorf("index \"%s\" not found", indexName)
				return "", n.pErr
			}
		}
	}
	n.noIndexJoin = (indexHints != nil && indexHints.NoIndexJoin)
	n.initDescDefaults()
	return alias, nil
}

// getDescriptor looks up the descriptor for `plainKey`, validates it,
// and unmarshals it into `descriptor`.
func (p *planner) getDescriptor(plainKey descriptorKey, descriptor descriptorProto) *roachpb.Error {
	gr, err := p.txn.Get(plainKey.Key())
	if err != nil {
		return err
	}
	if !gr.Exists() {
		return roachpb.NewUErrorf("%s %q does not exist", descriptor.TypeName(), plainKey.Name())
	}

	descKey := MakeDescMetadataKey(ID(gr.ValueInt()))
	desc := &Descriptor{}
	if pErr := p.txn.GetProto(descKey, desc); pErr != nil {
		return pErr
	}

	switch t := descriptor.(type) {
	case *TableDescriptor:
		table := desc.GetTable()
		if table == nil {
			return roachpb.NewErrorf("%q is not a table", plainKey.Name())
		}
		*t = *table
	case *DatabaseDescriptor:
		database := desc.GetDatabase()
		if database == nil {
			return roachpb.NewErrorf("%q is not a database", plainKey.Name())
		}
		*t = *database
	}

	return roachpb.NewError(descriptor.Validate())
}

// DropTable drops a table.
// Privileges: DROP on table.
//   Notes: postgres allows only the table owner to DROP a table.
//          mysql requires the DROP privilege on the table.
func (p *planner) DropTable(n *parser.DropTable) (planNode, *roachpb.Error) {
	// TODO(XisiHuang): should do truncate and delete descriptor in
	// the same txn
	for i := range n.Names {
		droppedDesc, err := p.dropTableImpl(n.Names, i)
		if err != nil {
			return nil, err
		}
		if droppedDesc == nil {
			if n.IfExists {
				continue
			}
			// Table does not exist, but we want it to: error out.
			return nil, roachpb.NewUErrorf("table %q does not exist", n.Names[i].Table())
		}
		// Log a Drop Table event for this table.
		if pErr := MakeEventLogger(p.leaseMgr).insertEventRecord(p.txn,
			EventLogDropTable,
			int32(droppedDesc.ID),
			int32(p.evalCtx.NodeID),
			struct {
				TableName string
				Statement string
				User      string
			}{droppedDesc.Name, n.String(), p.user},
		); pErr != nil {
			return nil, pErr
		}
	}
	return &emptyNode{}, nil
}

func encodeTableKey(b []byte, val parser.Datum) ([]byte, *roachpb.Error) {
	if val == parser.DNull {
		return encoding.EncodeNull(b), nil
	}

	switch t := val.(type) {
	case parser.DBool:
		if t {
			return encoding.EncodeVarint(b, 1), nil
		}
		return encoding.EncodeVarint(b, 0), nil
	case parser.DInt:
		return encoding.EncodeVarint(b, int64(t)), nil
	case parser.DFloat:
		return encoding.EncodeFloat(b, float64(t)), nil
	case parser.DString:
		return encoding.EncodeString(b, string(t)), nil
	case parser.DBytes:
		return encoding.EncodeString(b, string(t)), nil
	case parser.DDate:
		return encoding.EncodeVarint(b, int64(t)), nil
	case parser.DTimestamp:
		return encoding.EncodeTime(b, t.Time), nil
	case parser.DInterval:
		return encoding.EncodeVarint(b, int64(t.Duration)), nil
	}
	return nil, roachpb.NewUErrorf("unable to encode table key: %T", val)
}

// CreateDatabase creates a database.
// Privileges: security.RootUser user.
//   Notes: postgres requires superuser or "CREATEDB".
//          mysql uses the mysqladmin command.
func (p *planner) CreateDatabase(n *parser.CreateDatabase) (planNode, *roachpb.Error) {
	if n.Name == "" {
		return nil, roachpb.NewError(errEmptyDatabaseName)
	}

	if p.session.User != security.RootUser {
		return nil, roachpb.NewUErrorf("only %s is allowed to create databases", security.RootUser)
	}

	desc := makeDatabaseDesc(n)

	created, err := p.createDescriptor(databaseKey{string(n.Name)}, &desc, n.IfNotExists)
	if err != nil {
		return nil, err
	}
	if created {
		// Log Create Database event.
		if pErr := MakeEventLogger(p.leaseMgr).InsertEventRecord(p.txn,
			EventLogCreateDatabase,
			int32(desc.ID),
			int32(p.evalCtx.NodeID),
			struct {
				DatabaseName string
				Statement    string
				User         string
			}{n.Name.String(), n.String(), p.session.User},
		); pErr != nil {
			return nil, pErr
		}
	}
	return &emptyNode{}, nil
}

func (p *planner) prepare(stmt parser.Statement) (planNode, *roachpb.Error) {
	p.prepareOnly = true
	switch n := stmt.(type) {
	case *parser.Delete:
		return p.Delete(n)
	case *parser.Insert:
		return p.Insert(n, false)
	case *parser.Select:
		return p.Select(n)
	case *parser.Show:
		return p.Show(n)
	case *parser.ShowColumns:
		return p.ShowColumns(n)
	case *parser.ShowDatabases:
		return p.ShowDatabases(n)
	case *parser.ShowGrants:
		return p.ShowGrants(n)
	case *parser.ShowIndex:
		return p.ShowIndex(n)
	case *parser.ShowTables:
		return p.ShowTables(n)
	case *parser.Update:
		return p.Update(n)
	default:
		return nil, roachpb.NewUErrorf("prepare statement not supported: %s", stmt.StatementTag())

		// TODO(mjibson): add support for parser.Values.
		// Broken because it conflicts with INSERT's use of VALUES.
	}
}

func (n *scanNode) initWhere(where *parser.Where) *roachpb.Error {
	if where == nil {
		return nil
	}
	n.filter, n.pErr = n.resolveQNames(where.Expr)
	if n.pErr == nil {
		var whereType parser.Datum
		var err error
		whereType, err = n.filter.TypeCheck(n.planner.evalCtx.Args)
		n.pErr = roachpb.NewError(err)
		if n.pErr == nil {
			if !(whereType == parser.DummyBool || whereType == parser.DNull) {
				n.pErr = roachpb.NewUErrorf("argument of WHERE must be type %s, not type %s", parser.DummyBool.Type(), whereType.Type())
			}
		}
	}
	if n.pErr == nil {
		// Normalize the expression (this will also evaluate any branches that are
		// constant).
		var err error
		n.filter, err = n.planner.parser.NormalizeExpr(n.planner.evalCtx, n.filter)
		n.pErr = roachpb.NewError(err)
	}
	if n.pErr == nil {
		n.filter, n.pErr = n.planner.expandSubqueries(n.filter, 1)
	}
	return n.pErr
}

// checkPrivilege verifies that p.user has `privilege` on `descriptor`.
func (p *planner) checkPrivilege(descriptor descriptorProto, privilege privilege.Kind) *roachpb.Error {
	if descriptor.GetPrivileges().CheckPrivilege(p.user, privilege) {
		return nil
	}
	return roachpb.NewUErrorf("user %s does not have %s privilege on %s %s",
		p.user, privilege, descriptor.TypeName(), descriptor.GetName())
}

// FindActiveColumnByName finds an active column with the specified name.
func (desc *TableDescriptor) FindActiveColumnByName(name string) (ColumnDescriptor, *roachpb.Error) {
	for _, c := range desc.Columns {
		if equalName(c.Name, name) {
			return c, nil
		}
	}
	return ColumnDescriptor{}, roachpb.NewUErrorf("column %q does not exist", name)
}