Merge pull request #1190 from wangzihuacool/add-rocksdb-as-transactional-engine

add rocksdb as transactional engine

Author: dm-2

.github/workflows/replica-tests.yml

@@ -8,7 +8,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        version: [mysql-5.7.25,mysql-8.0.16]
+        version: [mysql-5.7.25,mysql-8.0.16,PerconaServer-8.0.21]
 
     steps:
     - uses: actions/checkout@v2

doc/command-line-flags.md

@@ -246,6 +246,18 @@ Allows `gh-ost` to connect to the MySQL servers using encrypted connections, but
 
 `--ssl-key=/path/to/ssl-key.key`: SSL private key file (in PEM format).
 
+### storage-engine
+Default is `innodb`, and `rocksdb` support is currently experimental. InnoDB and RocksDB are both transactional engines, supporting both shared and exclusive row locks.
+
+But RocksDB currently lacks a few features support compared to InnoDB:
+- Gap Locks
+- Foreign Key
+- Generated Columns
+- Spatial
+- Geometry
+
+When `--storage-engine=rocksdb`, `gh-ost` will make some changes necessary (e.g. sets isolation level to `READ_COMMITTED`) to support RocksDB.
+
 ### test-on-replica
 
 Issue the migration on a replica; do not modify data on master. Useful for validating, testing and benchmarking. See [`testing-on-replica`](testing-on-replica.md)

go/base/context.go

@@ -290,6 +290,19 @@ func NewMigrationContext() *MigrationContext {
 	}
 }
 
+func (this *MigrationContext) SetConnectionConfig(storageEngine string) error {
+	var transactionIsolation string
+	switch storageEngine {
+	case "rocksdb":
+		transactionIsolation = "READ-COMMITTED"
+	default:
+		transactionIsolation = "REPEATABLE-READ"
+	}
+	this.InspectorConnectionConfig.TransactionIsolation = transactionIsolation
+	this.ApplierConnectionConfig.TransactionIsolation = transactionIsolation
+	return nil
+}
+
 func getSafeTableName(baseName string, suffix string) string {
 	name := fmt.Sprintf("_%s_%s", baseName, suffix)
 	if len(name) <= mysql.MaxTableNameLength {
@@ -428,6 +441,10 @@ func (this *MigrationContext) IsTransactionalTable() bool {
 		{
 			return true
 		}
+	case "rocksdb":
+		{
+			return true
+		}
 	}
 	return false
 }

go/cmd/gh-ost/main.go

@@ -68,6 +68,7 @@ func main() {
 	flag.StringVar(&migrationContext.OriginalTableName, "table", "", "table name (mandatory)")
 	flag.StringVar(&migrationContext.AlterStatement, "alter", "", "alter statement (mandatory)")
 	flag.BoolVar(&migrationContext.AttemptInstantDDL, "attempt-instant-ddl", false, "Attempt to use instant DDL for this migration first")
+	storageEngine := flag.String("storage-engine", "innodb", "Specify table storage engine (default: 'innodb'). When 'rocksdb': the session transaction isolation level is changed from REPEATABLE_READ to READ_COMMITTED.")
 
 	flag.BoolVar(&migrationContext.CountTableRows, "exact-rowcount", false, "actually count table rows as opposed to estimate them (results in more accurate progress estimation)")
 	flag.BoolVar(&migrationContext.ConcurrentCountTableRows, "concurrent-rowcount", true, "(with --exact-rowcount), when true (default): count rows after row-copy begins, concurrently, and adjust row estimate later on; when false: first count rows, then start row copy")
@@ -182,6 +183,10 @@ func main() {
 		migrationContext.Log.SetLevel(log.ERROR)
 	}
 
+	if err := migrationContext.SetConnectionConfig(*storageEngine); err != nil {
+		migrationContext.Log.Fatale(err)
+	}
+
 	if migrationContext.AlterStatement == "" {
 		log.Fatal("--alter must be provided and statement must not be empty")
 	}
@@ -247,6 +252,9 @@ func main() {
 	if *replicationLagQuery != "" {
 		migrationContext.Log.Warning("--replication-lag-query is deprecated")
 	}
+	if *storageEngine == "rocksdb" {
+		migrationContext.Log.Warning("RocksDB storage engine support is experimental")
+	}
 
 	switch *cutOver {
 	case "atomic", "default", "":

go/mysql/connection.go

@@ -18,18 +18,18 @@ import (
 )
 
 const (
-	transactionIsolation = "REPEATABLE-READ"
-	TLS_CONFIG_KEY       = "ghost"
+	TLS_CONFIG_KEY = "ghost"
 )
 
 // ConnectionConfig is the minimal configuration required to connect to a MySQL server
 type ConnectionConfig struct {
-	Key        InstanceKey
-	User       string
-	Password   string
-	ImpliedKey *InstanceKey
-	tlsConfig  *tls.Config
-	Timeout    float64
+	Key                  InstanceKey
+	User                 string
+	Password             string
+	ImpliedKey           *InstanceKey
+	tlsConfig            *tls.Config
+	Timeout              float64
+	TransactionIsolation string
 }
 
 func NewConnectionConfig() *ConnectionConfig {
@@ -43,11 +43,12 @@ func NewConnectionConfig() *ConnectionConfig {
 // DuplicateCredentials creates a new connection config with given key and with same credentials as this config
 func (this *ConnectionConfig) DuplicateCredentials(key InstanceKey) *ConnectionConfig {
 	config := &ConnectionConfig{
-		Key:       key,
-		User:      this.User,
-		Password:  this.Password,
-		tlsConfig: this.tlsConfig,
-		Timeout:   this.Timeout,
+		Key:                  key,
+		User:                 this.User,
+		Password:             this.Password,
+		tlsConfig:            this.tlsConfig,
+		Timeout:              this.Timeout,
+		TransactionIsolation: this.TransactionIsolation,
 	}
 	config.ImpliedKey = &config.Key
 	return config
@@ -126,7 +127,7 @@ func (this *ConnectionConfig) GetDBUri(databaseName string) string {
 		"charset=utf8mb4,utf8,latin1",
 		"interpolateParams=true",
 		fmt.Sprintf("tls=%s", tlsOption),
-		fmt.Sprintf("transaction_isolation=%q", transactionIsolation),
+		fmt.Sprintf("transaction_isolation=%q", this.TransactionIsolation),
 		fmt.Sprintf("timeout=%fs", this.Timeout),
 		fmt.Sprintf("readTimeout=%fs", this.Timeout),
 		fmt.Sprintf("writeTimeout=%fs", this.Timeout),

go/mysql/connection_test.go

@@ -13,6 +13,10 @@ import (
 	test "github.com/openark/golib/tests"
 )
 
+const (
+	transactionIsolation = "REPEATABLE-READ"
+)
+
 func init() {
 	log.SetLevel(log.ERROR)
 }
@@ -25,6 +29,7 @@ func TestNewConnectionConfig(t *testing.T) {
 	test.S(t).ExpectEquals(c.ImpliedKey.Port, 0)
 	test.S(t).ExpectEquals(c.User, "")
 	test.S(t).ExpectEquals(c.Password, "")
+	test.S(t).ExpectEquals(c.TransactionIsolation, "")
 }
 
 func TestDuplicateCredentials(t *testing.T) {
@@ -36,6 +41,7 @@ func TestDuplicateCredentials(t *testing.T) {
 		InsecureSkipVerify: true,
 		ServerName:         "feathers",
 	}
+	c.TransactionIsolation = transactionIsolation
 
 	dup := c.DuplicateCredentials(InstanceKey{Hostname: "otherhost", Port: 3310})
 	test.S(t).ExpectEquals(dup.Key.Hostname, "otherhost")
@@ -45,13 +51,15 @@ func TestDuplicateCredentials(t *testing.T) {
 	test.S(t).ExpectEquals(dup.User, "gromit")
 	test.S(t).ExpectEquals(dup.Password, "penguin")
 	test.S(t).ExpectEquals(dup.tlsConfig, c.tlsConfig)
+	test.S(t).ExpectEquals(dup.TransactionIsolation, c.TransactionIsolation)
 }
 
 func TestDuplicate(t *testing.T) {
 	c := NewConnectionConfig()
 	c.Key = InstanceKey{Hostname: "myhost", Port: 3306}
 	c.User = "gromit"
 	c.Password = "penguin"
+	c.TransactionIsolation = transactionIsolation
 
 	dup := c.Duplicate()
 	test.S(t).ExpectEquals(dup.Key.Hostname, "myhost")
@@ -60,6 +68,7 @@ func TestDuplicate(t *testing.T) {
 	test.S(t).ExpectEquals(dup.ImpliedKey.Port, 3306)
 	test.S(t).ExpectEquals(dup.User, "gromit")
 	test.S(t).ExpectEquals(dup.Password, "penguin")
+	test.S(t).ExpectEquals(dup.TransactionIsolation, transactionIsolation)
 }
 
 func TestGetDBUri(t *testing.T) {
@@ -68,6 +77,7 @@ func TestGetDBUri(t *testing.T) {
 	c.User = "gromit"
 	c.Password = "penguin"
 	c.Timeout = 1.2345
+	c.TransactionIsolation = transactionIsolation
 
 	uri := c.GetDBUri("test")
 	test.S(t).ExpectEquals(uri, `gromit:penguin@tcp(myhost:3306)/test?autocommit=true&charset=utf8mb4,utf8,latin1&interpolateParams=true&tls=false&transaction_isolation="REPEATABLE-READ"&timeout=1.234500s&readTimeout=1.234500s&writeTimeout=1.234500s`)
@@ -80,6 +90,7 @@ func TestGetDBUriWithTLSSetup(t *testing.T) {
 	c.Password = "penguin"
 	c.Timeout = 1.2345
 	c.tlsConfig = &tls.Config{}
+	c.TransactionIsolation = transactionIsolation
 
 	uri := c.GetDBUri("test")
 	test.S(t).ExpectEquals(uri, `gromit:penguin@tcp(myhost:3306)/test?autocommit=true&charset=utf8mb4,utf8,latin1&interpolateParams=true&tls=ghost&transaction_isolation="REPEATABLE-READ"&timeout=1.234500s&readTimeout=1.234500s&writeTimeout=1.234500s`)

localtests/discard-fk/ignore_versions

@@ -0,0 +1 @@
+Percona

localtests/fail-fk-parent/ignore_versions

@@ -0,0 +1 @@
+Percona

localtests/fail-fk/ignore_versions

@@ -0,0 +1 @@
+Percona

localtests/generated-columns-add/ignore_versions

@@ -0,0 +1 @@
+Percona