Support gbk Encoding To Fix #532

go/base/context.go

@@ -91,6 +91,7 @@ type MigrationContext struct {
 	SkipRenamedColumns       bool
 	IsTungsten               bool
 	DiscardForeignKeys       bool
+	IncludeRiskyCharsets     bool
 
 	config            ContextConfig
 	configMutex       *sync.Mutex

go/cmd/gh-ost/main.go

@@ -120,6 +120,7 @@ func main() {
 	version := flag.Bool("version", false, "Print version & exit")
 	checkFlag := flag.Bool("check-flag", false, "Check if another flag exists/supported. This allows for cross-version scripting. Exits with 0 when all additional provided flags exist, nonzero otherwise. You must provide (dummy) values for flags that require a value. Example: gh-ost --check-flag --cut-over-lock-timeout-seconds --nice-ratio 0")
 	flag.StringVar(&migrationContext.ForceTmpTableName, "force-table-names", "", "table name prefix to be used on the temporary tables")
+	flag.BoolVar(&migrationContext.IncludeRiskyCharsets, "include-risk-charset", false, "charset includes risk encoding, e.g. gbk, gb2312, big5, cp932, sjis")
 	flag.CommandLine.SetOutput(os.Stdout)
 
 	flag.Parse()

go/logic/applier.go

@@ -69,7 +69,7 @@ func NewApplier(migrationContext *base.MigrationContext) *Applier {
 
 func (this *Applier) InitDBConnections() (err error) {
 
-	applierUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
+	applierUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName, this.migrationContext.IncludeRiskyCharsets)
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, applierUri); err != nil {
 		return err
 	}

go/logic/inspect.go

@@ -40,12 +40,12 @@ func NewInspector(migrationContext *base.MigrationContext) *Inspector {
 }
 
 func (this *Inspector) InitDBConnections() (err error) {
-	inspectorUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
+	inspectorUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName, this.migrationContext.IncludeRiskyCharsets)
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, inspectorUri); err != nil {
 		return err
 	}
 
-	informationSchemaUri := this.connectionConfig.GetDBUri("information_schema")
+	informationSchemaUri := this.connectionConfig.GetDBUri("information_schema", false)
 	if this.informationSchemaDb, _, err = mysql.GetDB(this.migrationContext.Uuid, informationSchemaUri); err != nil {
 		return err
 	}

go/logic/streamer.go

@@ -103,7 +103,7 @@ func (this *EventsStreamer) notifyListeners(binlogEvent *binlog.BinlogDMLEvent)
 }
 
 func (this *EventsStreamer) InitDBConnections() (err error) {
-	EventsStreamerUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
+	EventsStreamerUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName, this.migrationContext.IncludeRiskyCharsets)
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, EventsStreamerUri); err != nil {
 		return err
 	}

go/logic/throttler.go

@@ -183,7 +183,7 @@ func (this *Throttler) collectControlReplicasLag() {
 	)
 
 	readReplicaLag := func(connectionConfig *mysql.ConnectionConfig) (lag time.Duration, err error) {
-		dbUri := connectionConfig.GetDBUri("information_schema")
+		dbUri := connectionConfig.GetDBUri("information_schema", false)
 
 		var heartbeatValue string
 		if db, _, err := mysql.GetDB(this.migrationContext.Uuid, dbUri); err != nil {

go/mysql/connection.go

@@ -49,12 +49,18 @@ func (this *ConnectionConfig) Equals(other *ConnectionConfig) bool {
 	return this.Key.Equals(&other.Key) || this.ImpliedKey.Equals(other.ImpliedKey)
 }
 
-func (this *ConnectionConfig) GetDBUri(databaseName string) string {
+func (this *ConnectionConfig) GetDBUri(databaseName string, IncludeRiskyCharsets bool) string {
 	hostname := this.Key.Hostname
 	var ip = net.ParseIP(hostname)
 	if (ip != nil) && (ip.To4() == nil) {
 		// Wrap IPv6 literals in square brackets
 		hostname = fmt.Sprintf("[%s]", hostname)
 	}
-	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?interpolateParams=true&autocommit=true&charset=utf8mb4,utf8,latin1", this.User, this.Password, hostname, this.Key.Port, databaseName)
+	riskyCharsets := ""
+	interpolateParams := true
+	if IncludeRiskyCharsets {
+		interpolateParams = false
+		riskyCharsets = ",gbk,gb2312,big5,cp932,sjis"
+	}
+	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?interpolateParams=%t&autocommit=true&charset=utf8mb4,utf8,latin1%s", this.User, this.Password, hostname, this.Key.Port, databaseName, interpolateParams, riskyCharsets)
 }

go/mysql/connection_test.go

@@ -55,3 +55,14 @@ func TestDuplicate(t *testing.T) {
 	test.S(t).ExpectEquals(dup.User, "gromit")
 	test.S(t).ExpectEquals(dup.Password, "penguin")
 }
+func TestGetDBUri(t *testing.T) {
+	c := NewConnectionConfig()
+	c.Key = InstanceKey{Hostname: "myhost", Port: 3306}
+	c.User = "gromit"
+	c.Password = "penguin"
+
+	uri := c.GetDBUri("test", true)
+	test.S(t).ExpectEquals(uri, "gromit:penguin@tcp(myhost:3306)/test?interpolateParams=false&autocommit=true&charset=utf8mb4,utf8,latin1,gbk,gb2312,big5,cp932,sjis")
+	uri = c.GetDBUri("test", false)
+	test.S(t).ExpectEquals(uri, "gromit:penguin@tcp(myhost:3306)/test?interpolateParams=true&autocommit=true&charset=utf8mb4,utf8,latin1")
+}

go/mysql/utils.go

@@ -75,7 +75,7 @@ func GetReplicationLag(informationSchemaDb *gosql.DB, connectionConfig *Connecti
 }
 
 func GetMasterKeyFromSlaveStatus(connectionConfig *ConnectionConfig) (masterKey *InstanceKey, err error) {
-	currentUri := connectionConfig.GetDBUri("information_schema")
+	currentUri := connectionConfig.GetDBUri("information_schema", false)
 	// This function is only called once, okay to not have a cached connection pool
 	db, err := gosql.Open("mysql", currentUri)
 	if err != nil {

go/sql/encoding.go

@@ -8,6 +8,7 @@ package sql
 import (
 	"golang.org/x/text/encoding"
 	"golang.org/x/text/encoding/charmap"
+	"golang.org/x/text/encoding/simplifiedchinese"
 )
 
 type charsetEncoding map[string]encoding.Encoding
@@ -18,4 +19,5 @@ func init() {
 	charsetEncodingMap = make(map[string]encoding.Encoding)
 	// Begin mappings
 	charsetEncodingMap["latin1"] = charmap.Windows1252
+	charsetEncodingMap["gbk"] = simplifiedchinese.GBK
 }

localtests/gbk-charset/create.sql

@@ -0,0 +1,25 @@
+drop table if exists gh_ost_test;
+create table gh_ost_test (
+  id int(11) NOT NULL AUTO_INCREMENT,
+  name varchar(512) DEFAULT NULL,
+  v varchar(255) DEFAULT NULL COMMENT '添加普通列测试',
+  PRIMARY KEY (id)
+) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=gbk;
+
+insert into gh_ost_test values (null, 'gbk-test-initial', '添加普通列测试-添加普通列测试');
+insert into gh_ost_test values (null, 'gbk-test-initial', '添加普通列测试-添加普通列测试');
+
+drop event if exists gh_ost_test;
+delimiter ;;
+create event gh_ost_test
+  on schedule every 1 second
+  starts current_timestamp
+  ends current_timestamp + interval 60 second
+  on completion not preserve
+  enable
+  do
+begin
+  insert into gh_ost_test (name) values ('gbk-test-default');
+  insert into gh_ost_test values (null, 'gbk-test', '添加普通列测试-添加普通列测试');
+  update gh_ost_test set v='添加普通列测试' where v='添加普通列测试-添加普通列测试' order by id desc limit 1;
+end ;;