// ShowColumns of a table.
// Privileges: None.
//   Notes: postgres does not have a SHOW COLUMNS statement.
//          mysql only returns columns you have privileges on.
func (p *planner) ShowColumns(n *parser.ShowColumns) (planNode, error) {
	desc, err := p.getTableDesc(n.Table)
	if err != nil {
		return nil, err
	}
	if desc == nil {
		return nil, sqlbase.NewUndefinedTableError(n.Table.String())
	}
	v := &valuesNode{
		columns: []ResultColumn{
			{Name: "Field", Typ: parser.TypeString},
			{Name: "Type", Typ: parser.TypeString},
			{Name: "Null", Typ: parser.TypeBool},
			{Name: "Default", Typ: parser.TypeString},
		},
	}
	for i, col := range desc.Columns {
		defaultExpr := parser.Datum(parser.DNull)
		if e := desc.Columns[i].DefaultExpr; e != nil {
			defaultExpr = parser.NewDString(*e)
		}
		v.rows = append(v.rows, []parser.Datum{
			parser.NewDString(desc.Columns[i].Name),
			parser.NewDString(col.Type.SQLString()),
			parser.MakeDBool(parser.DBool(desc.Columns[i].Nullable)),
			defaultExpr,
		})
	}
	return v, nil
}

// getTableID retrieves the table ID for the specified table.
func getTableID(p *planner, tn *parser.TableName) (sqlbase.ID, error) {
	if err := tn.QualifyWithDatabase(p.session.Database); err != nil {
		return 0, err
	}

	virtual, err := getVirtualTableDesc(tn)
	if err != nil {
		return 0, err
	}
	if virtual != nil {
		return virtual.GetID(), nil
	}

	dbID, err := p.getDatabaseID(tn.Database())
	if err != nil {
		return 0, err
	}

	nameKey := tableKey{dbID, tn.Table()}
	key := nameKey.Key()
	gr, err := p.txn.Get(key)
	if err != nil {
		return 0, err
	}
	if !gr.Exists() {
		return 0, sqlbase.NewUndefinedTableError(parser.AsString(tn))
	}
	return sqlbase.ID(gr.ValueInt()), nil
}

// getVirtualTableEntry checks if the provided name matches a virtual database/table
// pair. The function will return the table's virtual table entry if the name matches
// a specific table. It will return an error if the name references a virtual database
// but the table is non-existent.
func getVirtualTableEntry(tn *parser.TableName) (virtualTableEntry, error) {
	if db, ok := getVirtualSchemaEntry(tn.Database()); ok {
		if t, ok := db.tables[sqlbase.NormalizeName(tn.TableName)]; ok {
			return t, nil
		}
		return virtualTableEntry{}, sqlbase.NewUndefinedTableError(tn.String())
	}
	return virtualTableEntry{}, nil
}

// getTableLease implements the SchemaAccessor interface.
func (p *planner) getTableLease(qname *parser.QualifiedName) (*sqlbase.TableDescriptor, error) {
	if log.V(2) {
		log.Infof("planner acquiring lease on table %q", qname)
	}
	if err := qname.NormalizeTableName(p.session.Database); err != nil {
		return nil, err
	}

	if qname.Database() == sqlbase.SystemDB.Name || testDisableTableLeases {
		// We don't go through the normal lease mechanism for system tables. The
		// system.lease and system.descriptor table, in particular, are problematic
		// because they are used for acquiring leases itself, creating a
		// chicken&egg problem.
		return p.mustGetTableDesc(qname)
	}

	dbID, err := p.getDatabaseID(qname.Database())
	if err != nil {
		return nil, err
	}

	// First, look to see if we already have a lease for this table.
	// This ensures that, once a SQL transaction resolved name N to id X, it will
	// continue to use N to refer to X even if N is renamed during the
	// transaction.
	var lease *LeaseState
	for _, l := range p.leases {
		if sqlbase.NormalizeName(l.Name) == sqlbase.NormalizeName(qname.Table()) &&
			l.ParentID == dbID {
			lease = l
			if log.V(2) {
				log.Infof("found lease in planner cache for table %q", qname)
			}
			break
		}
	}

	// If we didn't find a lease or the lease is about to expire, acquire one.
	if lease == nil || p.removeLeaseIfExpiring(lease) {
		var err error
		lease, err = p.leaseMgr.AcquireByName(p.txn, dbID, qname.Table())
		if err != nil {
			if err == errDescriptorNotFound {
				// Transform the descriptor error into an error that references the
				// table's name.
				return nil, sqlbase.NewUndefinedTableError(qname.String())
			}
			return nil, err
		}
		p.leases = append(p.leases, lease)
		// If the lease we just acquired expires before the txn's deadline, reduce
		// the deadline.
		p.txn.UpdateDeadlineMaybe(hlc.Timestamp{WallTime: lease.Expiration().UnixNano()})
	}
	return &lease.TableDescriptor, nil
}

// mustGetTableDesc implements the SchemaAccessor interface.
func (p *planner) mustGetTableDesc(qname *parser.QualifiedName) (*sqlbase.TableDescriptor, error) {
	desc, err := p.getTableDesc(qname)
	if err != nil {
		return nil, err
	}
	if desc == nil {
		return nil, sqlbase.NewUndefinedTableError(qname.String())
	}
	return desc, 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,
	scanVisibility scanVisibility,
) (string, error) {
	var err error

	// AS OF SYSTEM TIME queries need to fetch the table descriptor at the
	// specified time, and never lease anything. The proto transaction already
	// has its timestamps set correctly so getTableDesc will fetch with the
	// correct timestamp.
	if p.asOf {
		desc, err := p.getTableDesc(tableName)
		if err != nil {
			return "", err
		}
		if desc == nil {
			return "", sqlbase.NewUndefinedTableError(tableName.String())
		}
		n.desc = *desc
	} else {
		n.desc, err = p.getTableLease(tableName)
	}
	if err != nil {
		return "", err
	}

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

	alias := n.desc.Name

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

// mustGetTableDesc implements the SchemaAccessor interface.
func (p *planner) mustGetTableDesc(tn *parser.TableName) (*sqlbase.TableDescriptor, error) {
	desc, err := p.getTableDesc(tn)
	if err != nil {
		return nil, err
	}
	if desc == nil {
		return nil, sqlbase.NewUndefinedTableError(tn.String())
	}
	if err := filterTableState(desc); err != nil {
		return nil, err
	}
	return desc, nil
}

// getTableLeaseByID is a by-ID variant of getTableLease (i.e. uses same cache).
func (p *planner) getTableLeaseByID(tableID sqlbase.ID) (*sqlbase.TableDescriptor, error) {
	if log.V(2) {
		log.Infof(p.ctx(), "planner acquiring lease on table ID %d", tableID)
	}

	if testDisableTableLeases {
		table, err := sqlbase.GetTableDescFromID(p.txn, tableID)
		if err != nil {
			return nil, err
		}
		if err := filterTableState(table); err != nil {
			return nil, err
		}
		return table, nil
	}

	// First, look to see if we already have a lease for this table -- including
	// leases acquired via `getTableLease`.
	var lease *LeaseState
	for _, l := range p.leases {
		if l.ID == tableID {
			lease = l
			if log.V(2) {
				log.Infof(p.ctx(), "found lease in planner cache for table %d", tableID)
			}
			break
		}
	}

	// If we didn't find a lease or the lease is about to expire, acquire one.
	if lease == nil || p.removeLeaseIfExpiring(lease) {
		var err error
		lease, err = p.leaseMgr.Acquire(p.txn, tableID, 0)
		if err != nil {
			if err == sqlbase.ErrDescriptorNotFound {
				// Transform the descriptor error into an error that references the
				// table's ID.
				return nil, sqlbase.NewUndefinedTableError(fmt.Sprintf("<id=%d>", tableID))
			}
			return nil, err
		}
		p.leases = append(p.leases, lease)
		// If the lease we just acquired expires before the txn's deadline, reduce
		// the deadline.
		p.txn.UpdateDeadlineMaybe(hlc.Timestamp{WallTime: lease.Expiration().UnixNano()})
	}
	return &lease.TableDescriptor, nil
}

// ShowIndex returns all the indexes for a table.
// Privileges: None.
//   Notes: postgres does not have a SHOW INDEXES statement.
//          mysql requires some privilege for any column.
func (p *planner) ShowIndex(n *parser.ShowIndex) (planNode, error) {
	desc, err := p.getTableDesc(n.Table)
	if err != nil {
		return nil, err
	}
	if desc == nil {
		return nil, sqlbase.NewUndefinedTableError(n.Table.String())
	}

	v := &valuesNode{
		columns: []ResultColumn{
			{Name: "Table", Typ: parser.TypeString},
			{Name: "Name", Typ: parser.TypeString},
			{Name: "Unique", Typ: parser.TypeBool},
			{Name: "Seq", Typ: parser.TypeInt},
			{Name: "Column", Typ: parser.TypeString},
			{Name: "Direction", Typ: parser.TypeString},
			{Name: "Storing", Typ: parser.TypeBool},
		},
	}

	appendRow := func(index sqlbase.IndexDescriptor, colName string, sequence int,
		direction string, isStored bool) {
		v.rows = append(v.rows, []parser.Datum{
			parser.NewDString(n.Table.Table()),
			parser.NewDString(index.Name),
			parser.MakeDBool(parser.DBool(index.Unique)),
			parser.NewDInt(parser.DInt(sequence)),
			parser.NewDString(colName),
			parser.NewDString(direction),
			parser.MakeDBool(parser.DBool(isStored)),
		})
	}
	for _, index := range append([]sqlbase.IndexDescriptor{desc.PrimaryIndex}, desc.Indexes...) {
		sequence := 1
		for i, col := range index.ColumnNames {
			appendRow(index, col, sequence, index.ColumnDirections[i].String(), false)
			sequence++
		}
		for _, col := range index.StoreColumnNames {
			appendRow(index, col, sequence, "N/A", true)
			sequence++
		}
	}
	return v, nil
}

// getDescriptorsFromTargetList implements the DescriptorAccessor interface.
func (p *planner) getDescriptorsFromTargetList(targets parser.TargetList) (
	[]sqlbase.DescriptorProto, error) {
	if targets.Databases != nil {
		if len(targets.Databases) == 0 {
			return nil, errNoDatabase
		}
		descs := make([]sqlbase.DescriptorProto, 0, len(targets.Databases))
		for _, database := range targets.Databases {
			descriptor, err := p.getDatabaseDesc(database)
			if err != nil {
				return nil, err
			}
			if descriptor == nil {
				return nil, sqlbase.NewUndefinedDatabaseError(database)
			}
			descs = append(descs, descriptor)
		}
		return descs, nil
	}

	if len(targets.Tables) == 0 {
		return nil, errNoTable
	}
	descs := make([]sqlbase.DescriptorProto, 0, len(targets.Tables))
	for _, tableGlob := range targets.Tables {
		tables, err := p.expandTableGlob(tableGlob)
		if err != nil {
			return nil, err
		}
		for _, table := range tables {
			descriptor, err := p.getTableDesc(table)
			if err != nil {
				return nil, err
			}
			if descriptor == nil {
				return nil, sqlbase.NewUndefinedTableError(table.String())
			}
			descs = append(descs, descriptor)
		}
	}
	return descs, nil
}

// getTableID retrieves the table ID for the specified table.
func getTableID(p *planner, qname *parser.QualifiedName) (sqlbase.ID, error) {
	if err := qname.NormalizeTableName(p.session.Database); err != nil {
		return 0, err
	}

	dbID, err := p.getDatabaseID(qname.Database())
	if err != nil {
		return 0, err
	}

	nameKey := tableKey{dbID, qname.Table()}
	key := nameKey.Key()
	gr, err := p.txn.Get(key)
	if err != nil {
		return 0, err
	}
	if !gr.Exists() {
		return 0, sqlbase.NewUndefinedTableError(qname.String())
	}
	return sqlbase.ID(gr.ValueInt()), nil
}

// ShowConstraints returns all the constraints for a table.
// Privileges: None.
//   Notes: postgres does not have a SHOW CONSTRAINTS statement.
//          mysql requires some privilege for any column.
func (p *planner) ShowConstraints(n *parser.ShowConstraints) (planNode, error) {
	desc, err := p.getTableDesc(n.Table)
	if err != nil {
		return nil, err
	}
	if desc == nil {
		return nil, sqlbase.NewUndefinedTableError(n.Table.String())
	}

	v := &valuesNode{
		columns: []ResultColumn{
			{Name: "Table", Typ: parser.TypeString},
			{Name: "Name", Typ: parser.TypeString},
			{Name: "Type", Typ: parser.TypeString},
			{Name: "Column(s)", Typ: parser.TypeString},
			{Name: "Details", Typ: parser.TypeString},
		},
	}

	appendRow := func(name, typ, columns, details string) {
		detailsDatum := parser.DNull
		if details != "" {
			detailsDatum = parser.NewDString(details)
		}
		columnsDatum := parser.DNull
		if columns != "" {
			columnsDatum = parser.NewDString(columns)
		}
		v.rows = append(v.rows, []parser.Datum{
			parser.NewDString(n.Table.Table()),
			parser.NewDString(name),
			parser.NewDString(typ),
			columnsDatum,
			detailsDatum,
		})
	}

	for _, index := range append([]sqlbase.IndexDescriptor{desc.PrimaryIndex}, desc.Indexes...) {
		if index.ID == desc.PrimaryIndex.ID {
			appendRow(index.Name, "PRIMARY KEY", fmt.Sprintf("%+v", index.ColumnNames), "")
		} else if index.Unique {
			appendRow(index.Name, "UNIQUE", fmt.Sprintf("%+v", index.ColumnNames), "")
		}
		if index.ForeignKey != nil {
			other, err := p.getTableLeaseByID(index.ForeignKey.Table)
			if err != nil {
				return nil, errors.Errorf("error resolving table %d referenced in foreign key",
					index.ForeignKey.Table)
			}
			otherIdx, err := other.FindIndexByID(index.ForeignKey.Index)
			if err != nil {
				return nil, errors.Errorf("error resolving index %d in table %s referenced in foreign key",
					index.ForeignKey.Index, other.Name)
			}
			appendRow(index.ForeignKey.Name, "FOREIGN KEY", fmt.Sprintf("%v", index.ColumnNames),
				fmt.Sprintf("%s.%v", other.Name, otherIdx.ColumnNames))
		}
	}
	for _, c := range desc.Checks {
		appendRow(c.Name, "CHECK", "", c.Expr)
	}

	for _, c := range desc.Columns {
		if c.DefaultExprConstraintName != "" {
			appendRow(c.DefaultExprConstraintName, "DEFAULT", c.Name, *c.DefaultExpr)
		}
		if c.NullableConstraintName != "" {
			if c.Nullable {
				appendRow(c.NullableConstraintName, "NULL", c.Name, "")
			} else {
				appendRow(c.NullableConstraintName, "NOT NULL", c.Name, "")
			}
		}
	}

	// Sort the results by constraint name.
	sort := &sortNode{
		ordering: sqlbase.ColumnOrdering{{0, encoding.Ascending}, {1, encoding.Ascending}},
		columns:  v.columns,
	}
	return &selectTopNode{source: v, sort: sort}, nil
}

// ShowCreateTable returns a CREATE TABLE statement for the specified table in
// Traditional syntax.
// Privileges: None.
func (p *planner) ShowCreateTable(n *parser.ShowCreateTable) (planNode, error) {
	desc, err := p.getTableDesc(n.Table)
	if err != nil {
		return nil, err
	}
	if desc == nil {
		return nil, sqlbase.NewUndefinedTableError(n.Table.String())
	}
	v := &valuesNode{
		columns: []ResultColumn{
			{Name: "Table", Typ: parser.TypeString},
			{Name: "CreateTable", Typ: parser.TypeString},
		},
	}

	var buf bytes.Buffer
	fmt.Fprintf(&buf, "CREATE TABLE %s (", quoteNames(n.Table.String()))
	var primary string
	for i, col := range desc.VisibleColumns() {
		if i != 0 {
			buf.WriteString(",")
		}
		buf.WriteString("\n\t")
		fmt.Fprintf(&buf, "%s %s", quoteNames(col.Name), col.Type.SQLString())
		if col.Nullable {
			buf.WriteString(" NULL")
		} else {
			buf.WriteString(" NOT NULL")
		}
		if col.DefaultExpr != nil {
			fmt.Fprintf(&buf, " DEFAULT %s", *col.DefaultExpr)
		}
		if desc.PrimaryIndex.ColumnIDs[0] == col.ID {
			// Only set primary if the primary key is on a visible column (not rowid).
			primary = fmt.Sprintf(",\n\tCONSTRAINT %s PRIMARY KEY (%s)",
				quoteNames(desc.PrimaryIndex.Name),
				quoteNames(desc.PrimaryIndex.ColumnNames...),
			)
		}
	}
	buf.WriteString(primary)
	for _, idx := range desc.Indexes {
		var storing string
		if len(idx.StoreColumnNames) > 0 {
			storing = fmt.Sprintf(" STORING (%s)", quoteNames(idx.StoreColumnNames...))
		}
		fmt.Fprintf(&buf, ",\n\t%sINDEX %s (%s)%s",
			isUnique[idx.Unique],
			quoteNames(idx.Name),
			quoteNames(idx.ColumnNames...),
			storing,
		)
	}

	for _, e := range desc.Checks {
		fmt.Fprintf(&buf, ",\n\t")
		if len(e.Name) > 0 {
			fmt.Fprintf(&buf, "CONSTRAINT %s ", quoteNames(e.Name))
		}
		fmt.Fprintf(&buf, "CHECK (%s)", e.Expr)
	}

	buf.WriteString("\n)")
	v.rows = append(v.rows, []parser.Datum{
		parser.NewDString(n.Table.String()),
		parser.NewDString(buf.String()),
	})
	return v, nil
}

// getTableLease implements the SchemaAccessor interface.
func (p *planner) getTableLease(tn *parser.TableName) (*sqlbase.TableDescriptor, error) {
	if log.V(2) {
		log.Infof(p.ctx(), "planner acquiring lease on table %s", tn)
	}

	isSystemDB := tn.Database() == sqlbase.SystemDB.Name
	isVirtualDB := isVirtualDatabase(tn.Database())
	if isSystemDB || isVirtualDB || testDisableTableLeases {
		// We don't go through the normal lease mechanism for:
		// - system tables. The system.lease and system.descriptor table, in
		//   particular, are problematic because they are used for acquiring
		//   leases itself, creating a chicken&egg problem.
		// - virtual tables. These tables' descriptors are not persisted,
		//   so they cannot be leased. Instead, we simply return the static
		//   descriptor and rely on the immutability privileges set on the
		//   descriptors to cause upper layers to reject mutations statements.
		tbl, err := p.mustGetTableDesc(tn)
		if err != nil {
			return nil, err
		}
		if err := filterTableState(tbl); err != nil {
			return nil, err
		}
		return tbl, nil
	}

	dbID, err := p.getDatabaseID(tn.Database())
	if err != nil {
		return nil, err
	}

	// First, look to see if we already have a lease for this table.
	// This ensures that, once a SQL transaction resolved name N to id X, it will
	// continue to use N to refer to X even if N is renamed during the
	// transaction.
	var lease *LeaseState
	for _, l := range p.leases {
		if sqlbase.ReNormalizeName(l.Name) == sqlbase.NormalizeName(tn.TableName) &&
			l.ParentID == dbID {
			lease = l
			if log.V(2) {
				log.Infof(p.ctx(), "found lease in planner cache for table %q", tn)
			}
			break
		}
	}

	// If we didn't find a lease or the lease is about to expire, acquire one.
	if lease == nil || p.removeLeaseIfExpiring(lease) {
		var err error
		lease, err = p.leaseMgr.AcquireByName(p.txn, dbID, tn.Table())
		if err != nil {
			if err == sqlbase.ErrDescriptorNotFound {
				// Transform the descriptor error into an error that references the
				// table's name.
				return nil, sqlbase.NewUndefinedTableError(tn.String())
			}
			return nil, err
		}
		p.leases = append(p.leases, lease)
		// If the lease we just acquired expires before the txn's deadline, reduce
		// the deadline.
		p.txn.UpdateDeadlineMaybe(hlc.Timestamp{WallTime: lease.Expiration().UnixNano()})
	}
	return &lease.TableDescriptor, nil
}

// RenameColumn renames the column.
// Privileges: CREATE on table.
//   notes: postgres requires CREATE on the table.
//          mysql requires ALTER, CREATE, INSERT on the table.
func (p *planner) RenameColumn(n *parser.RenameColumn) (planNode, error) {
	newColName := string(n.NewName)
	if newColName == "" {
		return nil, errEmptyColumnName
	}

	if err := n.Table.NormalizeTableName(p.session.Database); err != nil {
		return nil, err
	}

	dbDesc, err := p.getDatabaseDesc(n.Table.Database())
	if err != nil {
		return nil, err
	}
	if dbDesc == nil {
		return nil, sqlbase.NewUndefinedDatabaseError(n.Table.Database())
	}

	// Check if table exists.
	tbKey := tableKey{dbDesc.ID, n.Table.Table()}.Key()
	gr, err := p.txn.Get(tbKey)
	if err != nil {
		return nil, err
	}
	if !gr.Exists() {
		if n.IfExists {
			// Noop.
			return &emptyNode{}, nil
		}
		// Key does not exist, but we want it to: error out.
		return nil, fmt.Errorf("table %q does not exist", n.Table.Table())
	}

	tableDesc, err := p.getTableDesc(n.Table)
	if err != nil {
		return nil, err
	}
	if tableDesc == nil {
		return nil, sqlbase.NewUndefinedTableError(n.Table.String())
	}

	colName := string(n.Name)
	status, i, err := tableDesc.FindColumnByName(colName)
	// n.IfExists only applies to table, no need to check here.
	if err != nil {
		return nil, err
	}
	var column *sqlbase.ColumnDescriptor
	if status == sqlbase.DescriptorActive {
		column = &tableDesc.Columns[i]
	} else {
		column = tableDesc.Mutations[i].GetColumn()
	}

	if err := p.checkPrivilege(tableDesc, privilege.CREATE); err != nil {
		return nil, err
	}

	if sqlbase.EqualName(colName, newColName) {
		// Noop.
		return &emptyNode{}, nil
	}

	if _, _, err := tableDesc.FindColumnByName(newColName); err == nil {
		return nil, fmt.Errorf("column name %q already exists", newColName)
	}

	preFn := func(expr parser.Expr) (err error, recurse bool, newExpr parser.Expr) {
		if qname, ok := expr.(*parser.QualifiedName); ok {
			if err := qname.NormalizeColumnName(); err != nil {
				return err, false, nil
			}
			if qname.Column() == colName {
				qname.Indirect[0] = parser.NameIndirection(newColName)
				qname.ClearString()
			}
			return nil, false, qname
		}
		return nil, true, expr
	}

	exprStrings := make([]string, len(tableDesc.Checks))
	for i, check := range tableDesc.Checks {
		exprStrings[i] = check.Expr
	}
	exprs, err := parser.ParseExprsTraditional(exprStrings)
	if err != nil {
		return nil, err
	}

	for i := range tableDesc.Checks {
		expr, err := parser.SimpleVisit(exprs[i], preFn)
		if err != nil {
			return nil, err
		}
		if after := expr.String(); after != tableDesc.Checks[i].Expr {
//.........这里部分代码省略.........

// RenameIndex renames the index.
// Privileges: CREATE on table.
//   notes: postgres requires CREATE on the table.
//          mysql requires ALTER, CREATE, INSERT on the table.
func (p *planner) RenameIndex(n *parser.RenameIndex) (planNode, error) {
	newIdxName := string(n.NewName)
	if newIdxName == "" {
		return nil, errEmptyIndexName
	}

	if err := n.Index.Table.NormalizeTableName(p.session.Database); err != nil {
		return nil, err
	}

	tableDesc, err := p.getTableDesc(n.Index.Table)
	if err != nil {
		return nil, err
	}
	if tableDesc == nil {
		return nil, sqlbase.NewUndefinedTableError(n.Index.Table.String())
	}

	idxName := string(n.Index.Index)
	status, i, err := tableDesc.FindIndexByName(idxName)
	if err != nil {
		if n.IfExists {
			// Noop.
			return &emptyNode{}, nil
		}
		// Index does not exist, but we want it to: error out.
		return nil, err
	}

	if err := p.checkPrivilege(tableDesc, privilege.CREATE); err != nil {
		return nil, err
	}

	if sqlbase.EqualName(idxName, newIdxName) {
		// Noop.
		return &emptyNode{}, nil
	}

	if _, _, err := tableDesc.FindIndexByName(newIdxName); err == nil {
		return nil, fmt.Errorf("index name %q already exists", n.NewName)
	}

	if status == sqlbase.DescriptorActive {
		tableDesc.Indexes[i].Name = newIdxName
	} else {
		tableDesc.Mutations[i].GetIndex().Name = newIdxName
	}

	if err := tableDesc.SetUpVersion(); err != nil {
		return nil, err
	}
	descKey := sqlbase.MakeDescMetadataKey(tableDesc.GetID())
	if err := tableDesc.Validate(); err != nil {
		return nil, err
	}
	if err := p.txn.Put(descKey, sqlbase.WrapDescriptor(tableDesc)); err != nil {
		return nil, err
	}
	p.notifySchemaChange(tableDesc.ID, sqlbase.InvalidMutationID)
	return &emptyNode{}, nil
}

// RenameTable renames the table.
// Privileges: DROP on source table, CREATE on destination database.
//   Notes: postgres requires the table owner.
//          mysql requires ALTER, DROP on the original table, and CREATE, INSERT
//          on the new table (and does not copy privileges over).
func (p *planner) RenameTable(n *parser.RenameTable) (planNode, error) {
	if err := n.NewName.NormalizeTableName(p.session.Database); err != nil {
		return nil, err
	}

	if n.NewName.Table() == "" {
		return nil, errEmptyTableName
	}

	if err := n.Name.NormalizeTableName(p.session.Database); err != nil {
		return nil, err
	}

	dbDesc, err := p.getDatabaseDesc(n.Name.Database())
	if err != nil {
		return nil, err
	}
	if dbDesc == nil {
		return nil, sqlbase.NewUndefinedDatabaseError(n.Name.Database())
	}

	tbKey := tableKey{dbDesc.ID, n.Name.Table()}.Key()

	// Check if table exists.
	gr, err := p.txn.Get(tbKey)
	if err != nil {
		return nil, err
	}
	if !gr.Exists() {
		if n.IfExists {
			// Noop.
			return &emptyNode{}, nil
		}
		// Key does not exist, but we want it to: error out.
		return nil, fmt.Errorf("table %q does not exist", n.Name.Table())
	}

	targetDbDesc, err := p.getDatabaseDesc(n.NewName.Database())
	if err != nil {
		return nil, err
	}
	if targetDbDesc == nil {
		return nil, sqlbase.NewUndefinedDatabaseError(n.NewName.Database())
	}

	if err := p.checkPrivilege(targetDbDesc, privilege.CREATE); err != nil {
		return nil, err
	}

	if n.Name.Database() == n.NewName.Database() && n.Name.Table() == n.NewName.Table() {
		// Noop.
		return &emptyNode{}, nil
	}

	tableDesc, err := p.getTableDesc(n.Name)
	if err != nil {
		return nil, err
	}
	if tableDesc == nil || tableDesc.State != sqlbase.TableDescriptor_PUBLIC {
		return nil, sqlbase.NewUndefinedTableError(n.Name.String())
	}

	if err := p.checkPrivilege(tableDesc, privilege.DROP); err != nil {
		return nil, err
	}

	tableDesc.SetName(n.NewName.Table())
	tableDesc.ParentID = targetDbDesc.ID

	descKey := sqlbase.MakeDescMetadataKey(tableDesc.GetID())
	newTbKey := tableKey{targetDbDesc.ID, n.NewName.Table()}.Key()

	if err := tableDesc.Validate(); err != nil {
		return nil, err
	}

	descID := tableDesc.GetID()
	descDesc := sqlbase.WrapDescriptor(tableDesc)

	if err := tableDesc.SetUpVersion(); err != nil {
		return nil, err
	}
	renameDetails := sqlbase.TableDescriptor_RenameInfo{
		OldParentID: uint32(dbDesc.ID),
		OldName:     n.Name.Table()}
	tableDesc.Renames = append(tableDesc.Renames, renameDetails)
	if err := p.writeTableDesc(tableDesc); err != nil {
		return nil, err
	}

	// We update the descriptor to the new name, but also leave the mapping of the
	// old name to the id, so that the name is not reused until the schema changer
	// has made sure it's not in use any more.
	b := client.Batch{}
	b.Put(descKey, descDesc)
//.........这里部分代码省略.........

// RenameColumn renames the column.
// Privileges: CREATE on table.
//   notes: postgres requires CREATE on the table.
//          mysql requires ALTER, CREATE, INSERT on the table.
func (p *planner) RenameColumn(n *parser.RenameColumn) (planNode, error) {
	newColName := string(n.NewName)
	if newColName == "" {
		return nil, errEmptyColumnName
	}

	if err := n.Table.NormalizeTableName(p.session.Database); err != nil {
		return nil, err
	}

	dbDesc, err := p.getDatabaseDesc(n.Table.Database())
	if err != nil {
		return nil, err
	}
	if dbDesc == nil {
		return nil, sqlbase.NewUndefinedDatabaseError(n.Table.Database())
	}

	// Check if table exists.
	tbKey := tableKey{dbDesc.ID, n.Table.Table()}.Key()
	gr, err := p.txn.Get(tbKey)
	if err != nil {
		return nil, err
	}
	if !gr.Exists() {
		if n.IfExists {
			// Noop.
			return &emptyNode{}, nil
		}
		// Key does not exist, but we want it to: error out.
		return nil, fmt.Errorf("table %q does not exist", n.Table.Table())
	}

	tableDesc, err := p.getTableDesc(n.Table)
	if err != nil {
		return nil, err
	}
	if tableDesc == nil {
		return nil, sqlbase.NewUndefinedTableError(n.Table.String())
	}

	colName := string(n.Name)
	status, i, err := tableDesc.FindColumnByName(colName)
	// n.IfExists only applies to table, no need to check here.
	if err != nil {
		return nil, err
	}
	var column *sqlbase.ColumnDescriptor
	if status == sqlbase.DescriptorActive {
		column = &tableDesc.Columns[i]
	} else {
		column = tableDesc.Mutations[i].GetColumn()
	}

	if err := p.checkPrivilege(tableDesc, privilege.CREATE); err != nil {
		return nil, err
	}

	if sqlbase.EqualName(colName, newColName) {
		// Noop.
		return &emptyNode{}, nil
	}

	if _, _, err := tableDesc.FindColumnByName(newColName); err == nil {
		return nil, fmt.Errorf("column name %q already exists", newColName)
	}

	// Rename the column in the indexes.
	tableDesc.RenameColumn(column.ID, newColName)
	column.Name = newColName
	if err := tableDesc.SetUpVersion(); err != nil {
		return nil, err
	}

	descKey := sqlbase.MakeDescMetadataKey(tableDesc.GetID())
	if err := tableDesc.Validate(); err != nil {
		return nil, err
	}
	if err := p.txn.Put(descKey, sqlbase.WrapDescriptor(tableDesc)); err != nil {
		return nil, err
	}
	p.notifySchemaChange(tableDesc.ID, sqlbase.InvalidMutationID)
	return &emptyNode{}, nil
}