feature: 添加CREATE TABLE AS SELECT语法支持 (#246)

session/session.go

@@ -226,6 +226,14 @@ type session struct {
 	// 时间戳类型是否需要明确指定默认值
 	explicitDefaultsForTimestamp bool
 
+	// 强制执行GTID一致性.
+	// 当启用 enforce_gtid_consistency 功能的时候，MySQL只允许能够保障事务安全，并且能够被日志记录的SQL语句被执行，
+	// 像create table … select 和 create temporarytable语句，以及同时更新事务表和非事务表的SQL语句或事务都不允许执行
+	enforeGtidConsistency bool
+
+	// 数据库的GTID模式会影响enforce_gtid_consistency参数.
+	gtidMode string
+
 	// 判断kill操作在哪个阶段,如果是在执行阶段时,则不停止备份
 	killExecute bool
 

session/session_inception.go

@@ -1658,7 +1658,7 @@ func (s *session) mysqlServerVersion() {
 	// sql := "select @@version;"
 	sql := `show variables where Variable_name in
 	('innodb_large_prefix','version','sql_mode','lower_case_table_names','wsrep_on',
-	'explicit_defaults_for_timestamp');`
+	'explicit_defaults_for_timestamp','enforce_gtid_consistency','gtid_mode');`
 
 	rows, err := s.raw(sql)
 	if rows != nil {
@@ -1724,6 +1724,10 @@ func (s *session) mysqlServerVersion() {
 				s.isClusterNode = (value == "ON" || value == "1")
 			case "explicit_defaults_for_timestamp":
 				s.explicitDefaultsForTimestamp = (value == "ON" || value == "1")
+			case "enforce_gtid_consistency":
+				s.enforeGtidConsistency = (value == "ON" || value == "1")
+			case "gtid_mode":
+				s.gtidMode = value
 			}
 		}
 
@@ -1736,6 +1740,10 @@ func (s *session) mysqlServerVersion() {
 			}
 		}
 
+		if !s.enforeGtidConsistency && strings.HasPrefix(s.gtidMode, "ON") {
+			s.enforeGtidConsistency = true
+		}
+
 		// log.Errorf("s.innodbLargePrefix: %v ", s.innodbLargePrefix)
 	}
 
@@ -2720,8 +2728,46 @@ func (s *session) checkCreateTable(node *ast.CreateTableStmt, sql string) {
 		}
 
 		if node.Select != nil {
-			log.Error("暂不支持语法: ", sql)
-			s.appendErrorNo(ER_NOT_SUPPORTED_YET)
+			if s.enforeGtidConsistency {
+				s.appendErrorMessage("Statement violates GTID consistency: CREATE TABLE ... SELECT.")
+			} else {
+				s.checkSelectItem(node.Select, false)
+
+				if s.myRecord.ErrLevel < 2 {
+					table = &TableInfo{
+						Schema: node.Table.Schema.String(),
+						Name:   node.Table.Name.String(),
+						Fields: make([]FieldInfo, len(node.Cols)),
+						IsNew:  true,
+					}
+					if table.Schema == "" {
+						table.Schema = s.dbName
+					}
+
+					if len(node.Cols) > 0 {
+						for index, field := range node.Cols {
+							table.Fields[index] = FieldInfo{
+								Field: field.Name.String(),
+								IsNew: true,
+							}
+						}
+					} else {
+						cols := s.getSubSelectColumns(node.Select)
+						table.Fields = make([]FieldInfo, len(cols))
+						for index, field := range cols {
+							table.Fields[index] = FieldInfo{
+								Field: field,
+								IsNew: true,
+							}
+						}
+					}
+
+					s.cacheNewTable(table)
+					s.myRecord.TableInfo = table
+				}
+				// log.Error("暂不支持语法: ", sql)
+				// s.appendErrorNo(ER_NOT_SUPPORTED_YET)
+			}
 		}
 
 		if node.ReferTable != nil || len(node.Cols) > 0 {
@@ -2882,7 +2928,7 @@ func (s *session) checkColumnsMustHaveindex(table *TableInfo) {
 		}
 
 		//col_name 在表中，并且没有索引
-		if inTable == true && haveIndex == false {
+		if inTable && !haveIndex {
 			mustHaveNotHaveIndexCol = append(mustHaveNotHaveIndexCol, col_name)
 		}
 	}
@@ -4676,10 +4722,21 @@ func (s *session) checkCreateView(node *ast.CreateViewStmt, sql string) {
 			table.Schema = s.dbName
 		}
 
-		for index, field := range node.Cols {
-			table.Fields[index] = FieldInfo{
-				Field: field.String(),
-				IsNew: true,
+		if len(node.Cols) > 0 {
+			for index, field := range node.Cols {
+				table.Fields[index] = FieldInfo{
+					Field: field.String(),
+					IsNew: true,
+				}
+			}
+		} else {
+			cols := s.getSubSelectColumns(node.Select)
+			table.Fields = make([]FieldInfo, len(cols))
+			for index, field := range cols {
+				table.Fields[index] = FieldInfo{
+					Field: field,
+					IsNew: true,
+				}
 			}
 		}
 
@@ -5154,10 +5211,6 @@ func (s *session) getSubSelectColumns(node ast.ResultSetNode) []string {
 			tableList = extractTableList(sel.From.TableRefs, tableList)
 			// tableInfoList = s.getTableInfoByTableSource(tableList)
 
-			// if sel.From.TableRefs.On != nil {
-			// 	s.checkItem(sel.From.TableRefs.On.Expr, tableInfoList)
-			// }
-
 			for _, f := range sel.Fields.Fields {
 				if f.WildCard == nil {
 					// log.Infof("%#v", f)
@@ -5172,7 +5225,6 @@ func (s *session) getSubSelectColumns(node ast.ResultSetNode) []string {
 						}
 					}
 				} else {
-
 					db := f.WildCard.Schema.L
 					wildTable := f.WildCard.Table.L
 
@@ -5230,35 +5282,6 @@ func (s *session) getSubSelectColumns(node ast.ResultSetNode) []string {
 			}
 		}
 
-		// for _, t := range tableInfoList {
-		// 	log.Info(t.Name)
-		// }
-		// log.Infof("%#v", columns)
-
-		// if sel.Fields != nil {
-		// 	for _, field := range sel.Fields.Fields {
-		// 		if field.WildCard == nil {
-		// 			s.checkItem(field.Expr, tableInfoList)
-		// 		}
-		// 	}
-		// }
-
-		// if sel.GroupBy != nil {
-		// 	for _, item := range sel.GroupBy.Items {
-		// 		s.checkItem(item.Expr, tableInfoList)
-		// 	}
-		// }
-
-		// if sel.Having != nil {
-		// 	s.checkItem(sel.Having.Expr, tableInfoList)
-		// }
-
-		// if sel.OrderBy != nil {
-		// 	for _, item := range sel.OrderBy.Items {
-		// 		s.checkItem(item.Expr, tableInfoList)
-		// 	}
-		// }
-
 		return columns
 
 	default:
@@ -5267,7 +5290,6 @@ func (s *session) getSubSelectColumns(node ast.ResultSetNode) []string {
 		log.Errorf("con:%d %v", s.sessionVars.ConnectionID, sel)
 	}
 
-	// log.Infof("%#v", columns)
 	return columns
 }
 

session/session_inception_backup_test.go

@@ -68,6 +68,16 @@ func (s *testSessionIncBackupSuite) TestCreateTable(c *C) {
 	c.Assert(backup, Equals, "DROP TABLE `test_inc`.`t1`;", Commentf("%v", s.rows))
 }
 
+func (s *testSessionIncBackupSuite) TestCreateTableAsSelect(c *C) {
+	if !s.enforeGtidConsistency {
+		s.mustRunExec(c, "drop table if exists t1,t2;create table t1(id int);")
+		s.mustRunBackup(c, "create table t2 as select * from t1;")
+		row := s.rows[s.getAffectedRows()-1]
+		backup := s.query("t2", row[7].(string))
+		c.Assert(backup, Equals, "DROP TABLE `test_inc`.`t2`;", Commentf("%v", s.rows))
+	}
+}
+
 func (s *testSessionIncBackupSuite) TestCreateView(c *C) {
 	s.mustRunExec(c, "drop table if exists t1;drop view if exists v_1;create table t1(id int primary key);")
 

session/session_inception_common_test.go

@@ -86,6 +86,8 @@ type testCommon struct {
 	innodbLargePrefix bool
 	// 时间戳类型是否需要明确指定默认值
 	explicitDefaultsForTimestamp bool
+	// 强制执行GTID一致性
+	enforeGtidConsistency bool
 	// 是否忽略大小写(lower_case_table_names为1和2时忽略,否则不忽略)
 	ignoreCase bool
 
@@ -551,7 +553,8 @@ func (s *testCommon) mysqlServerVersion() error {
 
 	var name, value string
 	sql := `show variables where Variable_name in
-		('explicit_defaults_for_timestamp','innodb_large_prefix','version','sql_mode','lower_case_table_names');`
+		('explicit_defaults_for_timestamp','innodb_large_prefix',
+		'version','sql_mode','lower_case_table_names','enforce_gtid_consistency');`
 	rows, err := s.db.Raw(sql).Rows()
 	if err != nil {
 		return err
@@ -596,9 +599,9 @@ func (s *testCommon) mysqlServerVersion() error {
 				s.ignoreCase = v > 0
 			}
 		case "explicit_defaults_for_timestamp":
-			if value == "ON" {
-				s.explicitDefaultsForTimestamp = true
-			}
+			s.explicitDefaultsForTimestamp = value == "ON" || value == "1"
+		case "enforce_gtid_consistency":
+			s.enforeGtidConsistency = value == "ON" || value == "1"
 		}
 	}
 

session/session_inception_exec_test.go

@@ -273,7 +273,14 @@ primary key(id)) comment 'test';`
 		s.testErrorCode(c, sql,
 			session.NewErr(session.ER_TABLE_NOT_EXISTED_ERROR, "test_inc.T1"))
 	}
+}
 
+func (s *testSessionIncExecSuite) TestCreateTableAsSelect(c *C) {
+	if !s.enforeGtidConsistency {
+		s.mustRunExec(c, "drop table if exists t1,t2;create table t1(c1 int);")
+		sql = "create table t2 as select * from t1;"
+		s.testErrorCode(c, sql)
+	}
 }
 
 func (s *testSessionIncExecSuite) TestDropTable(c *C) {

session/session_inception_test.go

@@ -940,7 +940,17 @@ primary key(id)) comment 'test';`
 	s.testErrorCode(c, sql,
 		session.NewErr(session.ER_TABLE_PREFIX,
 			config.GetGlobalConfig().Inc.TablePrefix))
+}
 
+func (s *testSessionIncSuite) TestCreateTableAsSelect(c *C) {
+	if s.enforeGtidConsistency {
+		sql = "create table t1(id int primary key);create table t11 as select * from t1;"
+		s.testErrorCode(c, sql,
+			session.NewErrf("Statement violates GTID consistency: CREATE TABLE ... SELECT."))
+	} else {
+		sql = "create table t1(id int primary key);create table t11 as select * from t1;"
+		s.testErrorCode(c, sql)
+	}
 }
 
 func (s *testSessionIncSuite) TestDropTable(c *C) {

session/tidb_check.go

@@ -301,7 +301,7 @@ func (s *session) checkCreateTableGrammar(stmt *ast.CreateTableStmt) {
 		s.appendErrorNo(ER_MULTIPLE_PRI_KEY)
 	}
 
-	if len(stmt.Cols) == 0 && stmt.ReferTable == nil {
+	if len(stmt.Cols) == 0 && stmt.ReferTable == nil && stmt.Select == nil {
 		s.appendErrorNo(ER_MUST_AT_LEAST_ONE_COLUMN)
 	}
 }