// CreateTable creates a table.
// Privileges: CREATE on database.
//   Notes: postgres/mysql require CREATE on database.
func (p *planner) CreateTable(n *parser.CreateTable) (planNode, *roachpb.Error) {
	if err := n.Table.NormalizeTableName(p.session.Database); err != nil {
		return nil, roachpb.NewError(err)
	}

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

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

	desc, err := makeTableDesc(n, dbDesc.ID)
	if err != nil {
		return nil, roachpb.NewError(err)
	}
	// Inherit permissions from the database descriptor.
	desc.Privileges = dbDesc.GetPrivileges()

	if len(desc.PrimaryIndex.ColumnNames) == 0 {
		// Ensure a Primary Key exists.
		s := "unique_rowid()"
		col := ColumnDescriptor{
			Name: "rowid",
			Type: ColumnType{
				Kind: ColumnType_INT,
			},
			DefaultExpr: &s,
			Hidden:      true,
			Nullable:    false,
		}
		desc.AddColumn(col)
		idx := IndexDescriptor{
			Unique:           true,
			ColumnNames:      []string{col.Name},
			ColumnDirections: []IndexDescriptor_Direction{IndexDescriptor_ASC},
		}
		if err := desc.AddIndex(idx, true); err != nil {
			return nil, roachpb.NewError(err)
		}
	}

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

	created, pErr := p.createDescriptor(tableKey{dbDesc.ID, n.Table.Table()}, &desc, n.IfNotExists)
	if pErr != nil {
		return nil, pErr
	}

	if created {
		// Log Create Table event.
		if pErr := MakeEventLogger(p.leaseMgr).InsertEventRecord(p.txn,
			EventLogCreateTable,
			int32(desc.ID),
			int32(p.evalCtx.NodeID),
			struct {
				TableName string
				Statement string
				User      string
			}{n.Table.String(), n.String(), p.session.User},
		); pErr != nil {
			return nil, pErr
		}
	}

	return &emptyNode{}, nil
}

// MakeTableDesc creates a table descriptor from a CreateTable statement.
func MakeTableDesc(p *parser.CreateTable, parentID ID) (TableDescriptor, error) {
	desc := TableDescriptor{}
	if err := p.Table.NormalizeTableName(""); err != nil {
		return desc, err
	}
	desc.Name = p.Table.Table()
	desc.ParentID = parentID
	desc.FormatVersion = FamilyFormatVersion
	// We don't use version 0.
	desc.Version = 1

	var primaryIndexColumnSet map[parser.Name]struct{}
	for _, def := range p.Defs {
		switch d := def.(type) {
		case *parser.ColumnTableDef:
			col, idx, err := MakeColumnDefDescs(d)
			if err != nil {
				return desc, err
			}
			desc.AddColumn(*col)
			if idx != nil {
				if err := desc.AddIndex(*idx, d.PrimaryKey); err != nil {
					return desc, err
				}
			}
		case *parser.IndexTableDef:
			idx := IndexDescriptor{
				Name:             string(d.Name),
				StoreColumnNames: d.Storing,
			}
			if err := idx.FillColumns(d.Columns); err != nil {
				return desc, err
			}
			if err := desc.AddIndex(idx, false); err != nil {
				return desc, err
			}
			if d.Interleave != nil {
				return desc, util.UnimplementedWithIssueErrorf(2972, "interleaving is not yet supported")
			}
		case *parser.UniqueConstraintTableDef:
			idx := IndexDescriptor{
				Name:             string(d.Name),
				Unique:           true,
				StoreColumnNames: d.Storing,
			}
			if err := idx.FillColumns(d.Columns); err != nil {
				return desc, err
			}
			if err := desc.AddIndex(idx, d.PrimaryKey); err != nil {
				return desc, err
			}
			if d.PrimaryKey {
				primaryIndexColumnSet = make(map[parser.Name]struct{})
				for _, c := range d.Columns {
					primaryIndexColumnSet[c.Column] = struct{}{}
				}
			}
			if d.Interleave != nil {
				return desc, util.UnimplementedWithIssueErrorf(2972, "interleaving is not yet supported")
			}
		case *parser.CheckConstraintTableDef:
			// CHECK expressions seem to vary across databases. Wikipedia's entry on
			// Check_constraint (https://en.wikipedia.org/wiki/Check_constraint) says
			// that if the constraint refers to a single column only, it is possible to
			// specify the constraint as part of the column definition. Postgres allows
			// specifying them anywhere about any columns, but it moves all constraints to
			// the table level (i.e., columns never have a check constraint themselves). We
			// will adhere to the stricter definition.

			preFn := func(expr parser.Expr) (err error, recurse bool, newExpr parser.Expr) {
				qname, ok := expr.(*parser.QualifiedName)
				if !ok {
					// Not a qname, don't do anything to this node.
					return nil, true, expr
				}

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

				if qname.IsStar() {
					return fmt.Errorf("* not allowed in constraint %q", d.Expr.String()), false, nil
				}
				col, err := desc.FindActiveColumnByName(qname.Column())
				if err != nil {
					return fmt.Errorf("column %q not found for constraint %q", qname.String(), d.Expr.String()), false, nil
				}
				// Convert to a dummy datum of the correct type.
				return nil, false, col.Type.ToDatumType()
			}

			expr, err := parser.SimpleVisit(d.Expr, preFn)
			if err != nil {
				return desc, err
			}

			if err := SanitizeVarFreeExpr(expr, parser.TypeBool, "CHECK"); err != nil {
				return desc, err
			}
//.........这里部分代码省略.........