ddl: enhance validation of column names when creating table

ddl/ddl_db_test.go

@@ -162,6 +162,8 @@ func (s *testDBSuite) TestMySQLErrorCode(c *C) {
 	s.testErrorCode(c, sql, tmysql.ErrUnknownCharacterSet)
 	sql = "create table t1(a int) character set laitn1;"
 	s.testErrorCode(c, sql, tmysql.ErrUnknownCharacterSet)
+	sql = "create table t2(c1.c2 blob default null);"
+	s.testErrorCode(c, sql, tmysql.ErrWrongTableName)
 
 	// add column
 	sql = "alter table test_error_code_succ add column c1 int"

executor/executor_test.go

@@ -2524,6 +2524,21 @@ func (s *testSuite) TestIssue5341(c *C) {
 	tk.MustQuery("select * from test.t where a < 1 order by a limit 0;").Check(testkit.Rows())
 }
 
+func (s *testSuite) TestContainDotColumn(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists test.t1")
+	tk.MustExec("create table test.t1(t1.a char)")
+	tk.MustExec("drop table if exists t2")
+	tk.MustExec("create table t2(a char, t2.b int)")
+
+	tk.MustExec("drop table if exists t3")
+	_, err := tk.Exec("create table t3(s.a char);")
+	terr := errors.Trace(err).(*errors.Err).Cause().(*terror.Error)
+	c.Assert(terr.Code(), Equals, terror.ErrCode(mysql.ErrWrongTableName))
+}
+
 func (s *testSuite) TestCheckIndex(c *C) {
 	s.ctx = mock.NewContext()
 	s.ctx.Store = s.store

plan/preprocess.go

@@ -238,7 +238,7 @@ func (p *preprocessor) checkCreateTableGrammar(stmt *ast.CreateTableStmt) {
 		p.err = ddl.ErrWrongTableName.GenByArgs(tName)
 		return
 	}
-
+	p.checkContainDotColumn(stmt)
 	countPrimaryKey := 0
 	for _, colDef := range stmt.Cols {
 		if err := checkColumn(colDef); err != nil {
@@ -492,6 +492,20 @@ func isIncorrectName(name string) bool {
 	return false
 }
 
+// checkContainDotColumn checks field contains the table name.
+// for example :create table t (c1.c2 int default null).
+func (p *preprocessor) checkContainDotColumn(stmt *ast.CreateTableStmt) {
+	tName := stmt.Table.Name.String()
+
+	for _, colDef := range stmt.Cols {
+		// check table name.
+		if colDef.Name.Table.O != tName && len(colDef.Name.Table.O) != 0 {
+			p.err = ddl.ErrWrongTableName.GenByArgs(colDef.Name.Table.O)
+			return
+		}
+	}
+}
+
 func (p *preprocessor) handleTableName(tn *ast.TableName) {
 	if tn.Schema.L == "" {
 		currentDB := p.ctx.GetSessionVars().CurrentDB