Introduce database abstraction layer (#132)

* Introduce database abstraction layer

README.md

@@ -56,4 +56,4 @@ Kindly take note of following options:
 *   `TESTOPTS=...`: As detailed under https://docs.ruby-lang.org/en/2.1.0/Rake/TestTask.html
 
 Example:
-`bundle exec rake test DEBUG=1 TESTOPTS="-v --name=TrivialIntegrationTests#test_logged_query_omits_columns"`
+`bundle exec rake test DEBUG=1 TESTOPTS="-v --name=TrivialIntegrationTests#test_logged_query_omits_columns"`

batch_writer.go

@@ -1,8 +1,8 @@
 package ghostferry
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 
 	"github.com/sirupsen/logrus"
 )

binlog_streamer.go

@@ -3,8 +3,9 @@ package ghostferry
 import (
 	"context"
 	"crypto/tls"
-	"database/sql"
+	sqlorig "database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"time"
 
 	"github.com/siddontang/go-mysql/mysql"
@@ -350,7 +351,7 @@ func idsOnServer(db *sql.DB) ([]uint32, error) {
 	server_ids := make([]uint32, 0)
 	for rows.Next() {
 		var server_id uint32
-		var host, port, master_id, slave_uuid sql.NullString
+		var host, port, master_id, slave_uuid sqlorig.NullString
 
 		err = rows.Scan(&server_id, &host, &port, &master_id, &slave_uuid)
 		if err != nil {

binlog_writer.go

@@ -1,8 +1,8 @@
 package ghostferry
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 
 	"github.com/sirupsen/logrus"
 )

compression_verifier.go

@@ -2,10 +2,11 @@ package ghostferry
 
 import (
 	"crypto/md5"
-	"database/sql"
+	sqlorig "database/sql"
 	"encoding/hex"
 	"errors"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"strconv"
 	"strings"
 
@@ -207,7 +208,7 @@ func NewCompressionVerifier(tableColumnCompressions TableColumnCompressionConfig
 	return compressionVerifier, nil
 }
 
-func getRows(db *sql.DB, schema, table, paginationKeyColumn string, columns []schema.TableColumn, paginationKeys []uint64) (*sql.Rows, error) {
+func getRows(db *sql.DB, schema, table, paginationKeyColumn string, columns []schema.TableColumn, paginationKeys []uint64) (*sqlorig.Rows, error) {
 	quotedPaginationKey := quoteField(paginationKeyColumn)
 	sql, args, err := rowSelector(columns, paginationKeyColumn).
 		From(QuotedTableNameFromString(schema, table)).

config.go

@@ -3,9 +3,9 @@ package ghostferry
 import (
 	"crypto/tls"
 	"crypto/x509"
-	"database/sql"
 	"errors"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"io/ioutil"
 	"time"
 

cursor.go

@@ -1,8 +1,9 @@
 package ghostferry
 
 import (
-	"database/sql"
+	sqlorig "database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"strings"
 
 	"github.com/Masterminds/squirrel"
@@ -12,7 +13,7 @@ import (
 
 // both `sql.Tx` and `sql.DB` allow a SQL query to be `Prepare`d
 type SqlPreparer interface {
-	Prepare(string) (*sql.Stmt, error)
+	Prepare(string) (*sqlorig.Stmt, error)
 }
 
 type SqlDBWithFakeRollback struct {
@@ -255,7 +256,7 @@ func (c *Cursor) Fetch(db SqlPreparer) (batch *RowBatch, paginationKeypos uint64
 	return
 }
 
-func ScanGenericRow(rows *sql.Rows, columnCount int) (RowData, error) {
+func ScanGenericRow(rows *sqlorig.Rows, columnCount int) (RowData, error) {
 	values := make(RowData, columnCount)
 	valuePtrs := make(RowData, columnCount)
 
@@ -267,7 +268,7 @@ func ScanGenericRow(rows *sql.Rows, columnCount int) (RowData, error) {
 	return values, err
 }
 
-func ScanByteRow(rows *sql.Rows, columnCount int) ([][]byte, error) {
+func ScanByteRow(rows *sqlorig.Rows, columnCount int) ([][]byte, error) {
 	values := make([][]byte, columnCount)
 	valuePtrs := make(RowData, columnCount)
 

data_iterator.go

@@ -1,8 +1,8 @@
 package ghostferry
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"math"
 	"sync"
 

ferry.go

@@ -2,10 +2,10 @@ package ghostferry
 
 import (
 	"context"
-	"database/sql"
 	"encoding/json"
 	"errors"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"math"
 	"net/http"
 	"os"

inline_verifier.go

@@ -3,8 +3,8 @@ package ghostferry
 import (
 	"bytes"
 	"context"
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"strconv"
 	"strings"
 	"sync"

iterative_verifier.go

@@ -2,9 +2,9 @@ package ghostferry
 
 import (
 	"bytes"
-	"database/sql"
 	"errors"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"math"
 	"strconv"
 	"strings"

sharding/test/trivial_integration_test.go

@@ -1,7 +1,7 @@
 package test
 
 import (
-	"database/sql"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"math/rand"
 	"testing"
 

sharding/testhelpers/unit_test_suite.go

@@ -1,8 +1,8 @@
 package testhelpers
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"net/http"
 	"net/http/httptest"
 

sqlwrapper/ghostferry_db.go

@@ -0,0 +1,88 @@
+package sqlwrapper
+
+import (
+	"context"
+	sqlorig "database/sql"
+)
+
+type DB struct {
+	*sqlorig.DB
+}
+
+type Tx struct {
+	*sqlorig.Tx
+}
+
+func Open(driverName, dataSourceName string) (*DB, error) {
+	sqlDB, err := sqlorig.Open(driverName, dataSourceName)
+	return &DB{sqlDB}, err
+}
+
+func (db DB) PrepareContext(ctx context.Context, query string) (*sqlorig.Stmt, error) {
+	return db.DB.PrepareContext(ctx, query)
+}
+
+func (db DB) ExecContext(ctx context.Context, query string, args ...interface{}) (sqlorig.Result, error) {
+	return db.DB.ExecContext(ctx, query, args...)
+}
+
+func (db DB) QueryContext(ctx context.Context, query string, args ...interface{}) (*sqlorig.Rows, error) {
+	return db.DB.QueryContext(ctx, query, args...)
+}
+
+func (db DB) Exec(query string, args ...interface{}) (sqlorig.Result, error) {
+	return db.DB.Exec(query, args...)
+}
+
+func (db DB) Prepare(query string) (*sqlorig.Stmt, error) {
+	return db.DB.Prepare(query)
+}
+
+func (db DB) Query(query string, args ...interface{}) (*sqlorig.Rows, error) {
+	return db.DB.Query(query, args...)
+}
+
+func (db DB) QueryRow(query string, args ...interface{}) *sqlorig.Row {
+	return db.DB.QueryRow(query, args...)
+}
+
+func (db DB) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sqlorig.Row {
+	return db.DB.QueryRowContext(ctx, query, args...)
+}
+
+func (db DB) Begin() (*Tx, error) {
+	tx, err := db.DB.Begin()
+	return &Tx{tx}, err
+}
+
+func (tx Tx) ExecContext(ctx context.Context, query string, args ...interface{}) (sqlorig.Result, error) {
+	return tx.Tx.ExecContext(ctx, query, args...)
+}
+
+func (tx Tx) Exec(query string, args ...interface{}) (sqlorig.Result, error) {
+	return tx.Tx.Exec(query, args...)
+}
+
+func (tx Tx) Prepare(query string) (*sqlorig.Stmt, error) {
+	return tx.Tx.Prepare(query)
+}
+
+func (tx Tx) PrepareContext(ctx context.Context, query string) (*sqlorig.Stmt, error) {
+	return tx.Tx.PrepareContext(ctx, query)
+}
+
+func (tx Tx) QueryContext(ctx context.Context, query string, args ...interface{}) (*sqlorig.Rows, error) {
+	return tx.Tx.QueryContext(ctx, query, args)
+}
+
+func (tx Tx) Query(query string, args ...interface{}) (*sqlorig.Rows, error) {
+	return tx.Tx.Query(query, args...)
+}
+
+func (tx Tx) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sqlorig.Row {
+	return tx.Tx.QueryRowContext(ctx, query, args...)
+}
+
+func (tx Tx) QueryRow(query string, args ...interface{}) *sqlorig.Row {
+	return tx.Tx.QueryRow(query, args...)
+}

table_schema_cache.go

@@ -1,8 +1,9 @@
 package ghostferry
 
 import (
-	"database/sql"
+	sqlorig "database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"strings"
 
 	sq "github.com/Masterminds/squirrel"
@@ -172,7 +173,7 @@ func LoadTables(db *sql.DB, tableFilter TableFilter, columnCompressionConfig Col
 		for _, table := range tableNames {
 			tableLog := dbLog.WithField("table", table)
 			tableLog.Debug("fetching table schema")
-			tableSchema, err := schema.NewTableFromSqlDB(db, dbname, table)
+			tableSchema, err := schema.NewTableFromSqlDB(db.DB, dbname, table)
 			if err != nil {
 				tableLog.WithError(err).Error("cannot fetch table schema from source db")
 				return tableSchemaCache, err
@@ -357,7 +358,7 @@ func maxPaginationKey(db *sql.DB, table *TableSchema) (uint64, bool, error) {
 	err = db.QueryRow(query, args...).Scan(&maxPaginationKey)
 
 	switch {
-	case err == sql.ErrNoRows:
+	case err == sqlorig.ErrNoRows:
 		return 0, false, nil
 	case err != nil:
 		return 0, false, err

test/go/binlog_streamer_test.go

@@ -1,7 +1,7 @@
 package test
 
 import (
-	"database/sql"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"strings"
 	"sync"
 	"testing"

test/go/iterative_verifier_collation_test.go

@@ -1,8 +1,8 @@
 package test
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"testing"
 
 	"github.com/Shopify/ghostferry/testhelpers"

test/go/iterative_verifier_integration_test.go

@@ -1,7 +1,7 @@
 package test
 
 import (
-	"database/sql"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"testing"
 
 	"github.com/Shopify/ghostferry"

test/go/iterative_verifier_test.go

@@ -1,8 +1,8 @@
 package test
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"sort"
 	"testing"
 	"time"

test/go/lag_throttler_test.go

@@ -2,7 +2,7 @@ package test
 
 import (
 	"context"
-	"database/sql"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"sync"
 	"testing"
 	"time"

test/go/race_conditions_integration_test.go

@@ -1,8 +1,8 @@
 package test
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"testing"
 	"time"
 

test/go/replication_config_test.go

@@ -1,8 +1,8 @@
 package test
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"math"
 	"testing"
 

test/go/trivial_integration_test.go

@@ -1,8 +1,9 @@
 package test
 
 import (
-	"database/sql"
+	sqlorig "database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"math/rand"
 	"testing"
 
@@ -106,8 +107,8 @@ func TestCopyDataWhileRenamingDatabaseAndTable(t *testing.T) {
 		targetQuery := fmt.Sprintf("CHECKSUM TABLE `%s`.`%s` EXTENDED", targetDatabaseName, targetTableName)
 
 		var tablename string
-		var sourceChecksum sql.NullInt64
-		var targetChecksum sql.NullInt64
+		var sourceChecksum sqlorig.NullInt64
+		var targetChecksum sqlorig.NullInt64
 
 		sourceRow := f.SourceDB.QueryRow(sourceQuery)
 		err := sourceRow.Scan(&tablename, &sourceChecksum)

test/go/types_integration_test.go

@@ -1,8 +1,8 @@
 package test
 
 import (
-	"database/sql"
 	"fmt"
+	sql "github.com/Shopify/ghostferry/sqlwrapper"
 	"math/rand"
 	"testing"