octoleo/gh-ost
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' into nm-refactor-migration-context

Browse Source
This commit is contained in:
Shlomi Noach 2017-12-07 13:59:12 +02:00 committed by GitHub
commit 0530b09700
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 175 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.travis.yml
View File

@ -1,7 +1,7 @@
# http://docs.travis-ci.com/user/languages/go/
language: go
go: 1.8
go: 1.9
os:
- linux

6
build.sh
View File

@ -10,6 +10,11 @@ function build {
GOOS=$3
GOARCH=$4
if [[ $(go version | egrep "go1[.][012345678]") ]]; then
echo "go version is too low. Must use 1.9 or above"
exit 1
fi
echo "Building ${osname} binary"
export GOOS
export GOARCH
@ -27,7 +32,6 @@ buildpath=/tmp/gh-ost
target=gh-ost
timestamp=$(date "+%Y%m%d%H%M%S")
ldflags="-X main.AppVersion=${RELEASE_VERSION}"
export GO15VENDOREXPERIMENT=1
mkdir -p ${buildpath}
build macOS osx darwin amd64

153
go/logic/applier.go
View File

@ -24,6 +24,28 @@ const (
atomicCutOverMagicHint = "ghost-cut-over-sentry"
)
type dmlBuildResult struct {
query string
args []interface{}
rowsDelta int64
err error
}
func newDmlBuildResult(query string, args []interface{}, rowsDelta int64, err error) *dmlBuildResult {
return &dmlBuildResult{
query: query,
args: args,
rowsDelta: rowsDelta,
err: err,
}
}
func newDmlBuildResultError(err error) *dmlBuildResult {
return &dmlBuildResult{
err: err,
}
}
// Applier connects and writes the the applier-server, which is the server where migration
// happens. This is typically the master, but could be a replica when `--test-on-replica` or
// `--execute-on-replica` are given.
@ -904,79 +926,103 @@ func (this *Applier) ShowStatusVariable(variableName string) (result int64, err
return result, nil
}
// updateModifiesUniqueKeyColumns checks whether a UPDATE DML event actually
// modifies values of the migration's unique key (the iterated key). This will call
// for special handling.
func (this *Applier) updateModifiesUniqueKeyColumns(dmlEvent *binlog.BinlogDMLEvent) (modifiedColumn string, isModified bool) {
for _, column := range this.migrationContext.UniqueKey.Columns.Columns() {
tableOrdinal := this.migrationContext.OriginalTableColumns.Ordinals[column.Name]
whereColumnValue := dmlEvent.WhereColumnValues.AbstractValues()[tableOrdinal]
newColumnValue := dmlEvent.NewColumnValues.AbstractValues()[tableOrdinal]
if newColumnValue != whereColumnValue {
return column.Name, true
}
}
return "", false
}
// buildDMLEventQuery creates a query to operate on the ghost table, based on an intercepted binlog
// event entry on the original table.
func (this *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) (query string, args []interface{}, rowsDelta int64, err error) {
func (this *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) (results [](*dmlBuildResult)) {
switch dmlEvent.DML {
case binlog.DeleteDML:
{
query, uniqueKeyArgs, err := sql.BuildDMLDeleteQuery(dmlEvent.DatabaseName, this.migrationContext.GetGhostTableName(), this.migrationContext.OriginalTableColumns, &this.migrationContext.UniqueKey.Columns, dmlEvent.WhereColumnValues.AbstractValues())
return query, uniqueKeyArgs, -1, err
return append(results, newDmlBuildResult(query, uniqueKeyArgs, -1, err))
}
case binlog.InsertDML:
{
query, sharedArgs, err := sql.BuildDMLInsertQuery(dmlEvent.DatabaseName, this.migrationContext.GetGhostTableName(), this.migrationContext.OriginalTableColumns, this.migrationContext.SharedColumns, this.migrationContext.MappedSharedColumns, dmlEvent.NewColumnValues.AbstractValues())
return query, sharedArgs, 1, err
return append(results, newDmlBuildResult(query, sharedArgs, 1, err))
}
case binlog.UpdateDML:
{
if _, isModified := this.updateModifiesUniqueKeyColumns(dmlEvent); isModified {
dmlEvent.DML = binlog.DeleteDML
results = append(results, this.buildDMLEventQuery(dmlEvent)...)
dmlEvent.DML = binlog.InsertDML
results = append(results, this.buildDMLEventQuery(dmlEvent)...)
return results
}
query, sharedArgs, uniqueKeyArgs, err := sql.BuildDMLUpdateQuery(dmlEvent.DatabaseName, this.migrationContext.GetGhostTableName(), this.migrationContext.OriginalTableColumns, this.migrationContext.SharedColumns, this.migrationContext.MappedSharedColumns, &this.migrationContext.UniqueKey.Columns, dmlEvent.NewColumnValues.AbstractValues(), dmlEvent.WhereColumnValues.AbstractValues())
args := sqlutils.Args()
args = append(args, sharedArgs...)
args = append(args, uniqueKeyArgs...)
return query, args, 0, err
return append(results, newDmlBuildResult(query, args, 0, err))
}
}
return "", args, 0, fmt.Errorf("Unknown dml event type: %+v", dmlEvent.DML)
return append(results, newDmlBuildResultError(fmt.Errorf("Unknown dml event type: %+v", dmlEvent.DML)))
}
// ApplyDMLEventQuery writes an entry to the ghost table, in response to an intercepted
// original-table binlog event
func (this *Applier) ApplyDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) error {
query, args, rowDelta, err := this.buildDMLEventQuery(dmlEvent)
if err != nil {
return err
}
// TODO The below is in preparation for transactional writes on the ghost tables.
// Such writes would be, for example:
// - prepended with sql_mode setup
// - prepended with time zone setup
// - prepended with SET SQL_LOG_BIN=0
// - prepended with SET FK_CHECKS=0
// etc.
//
// a known problem: https://github.com/golang/go/issues/9373 -- bitint unsigned values, not supported in database/sql
// is solved by silently converting unsigned bigints to string values.
//
err = func() error {
tx, err := this.db.Begin()
if err != nil {
return err
for _, buildResult := range this.buildDMLEventQuery(dmlEvent) {
if buildResult.err != nil {
return buildResult.err
}
sessionQuery := `SET
// TODO The below is in preparation for transactional writes on the ghost tables.
// Such writes would be, for example:
// - prepended with sql_mode setup
// - prepended with time zone setup
// - prepended with SET SQL_LOG_BIN=0
// - prepended with SET FK_CHECKS=0
// etc.
//
// a known problem: https://github.com/golang/go/issues/9373 -- bitint unsigned values, not supported in database/sql
// is solved by silently converting unsigned bigints to string values.
//
err := func() error {
tx, err := this.db.Begin()
if err != nil {
return err
}
sessionQuery := `SET
SESSION time_zone = '+00:00',
sql_mode = CONCAT(@@session.sql_mode, ',STRICT_ALL_TABLES')
`
if _, err := tx.Exec(sessionQuery); err != nil {
return err
}
if _, err := tx.Exec(query, args...); err != nil {
return err
}
if err := tx.Commit(); err != nil {
return err
}
return nil
}()
if _, err := tx.Exec(sessionQuery); err != nil {
return err
}
if _, err := tx.Exec(buildResult.query, buildResult.args...); err != nil {
return err
}
if err := tx.Commit(); err != nil {
return err
}
return nil
}()
if err != nil {
err = fmt.Errorf("%s; query=%s; args=%+v", err.Error(), query, args)
return log.Errore(err)
}
// no error
atomic.AddInt64(&this.migrationContext.TotalDMLEventsApplied, 1)
if this.migrationContext.CountTableRows {
atomic.AddInt64(&this.migrationContext.RowsDeltaEstimate, rowDelta)
if err != nil {
err = fmt.Errorf("%s; query=%s; args=%+v", err.Error(), buildResult.query, buildResult.args)
return log.Errore(err)
}
// no error
atomic.AddInt64(&this.migrationContext.TotalDMLEventsApplied, 1)
if this.migrationContext.CountTableRows {
atomic.AddInt64(&this.migrationContext.RowsDeltaEstimate, buildResult.rowsDelta)
}
}
return nil
}
@ -1005,15 +1051,16 @@ func (this *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent))
return rollback(err)
}
for _, dmlEvent := range dmlEvents {
query, args, rowDelta, err := this.buildDMLEventQuery(dmlEvent)
if err != nil {
return rollback(err)
for _, buildResult := range this.buildDMLEventQuery(dmlEvent) {
if buildResult.err != nil {
return rollback(buildResult.err)
}
if _, err := tx.Exec(buildResult.query, buildResult.args...); err != nil {
err = fmt.Errorf("%s; query=%s; args=%+v", err.Error(), buildResult.query, buildResult.args)
return rollback(err)
}
totalDelta += buildResult.rowsDelta
}
if _, err := tx.Exec(query, args...); err != nil {
err = fmt.Errorf("%s; query=%s; args=%+v", err.Error(), query, args)
return rollback(err)
}
totalDelta += rowDelta
}
if err := tx.Commit(); err != nil {
return err

52
localtests/fail-update-pk-column/create.sql Normal file
View File

@ -0,0 +1,52 @@
drop table if exists gh_ost_test;
create table gh_ost_test (
id int auto_increment,
i int not null,
primary key(id)
) auto_increment=1;
insert into gh_ost_test values (null, 101);
insert into gh_ost_test values (null, 102);
insert into gh_ost_test values (null, 103);
insert into gh_ost_test values (null, 104);
insert into gh_ost_test values (null, 105);
insert into gh_ost_test values (null, 106);
insert into gh_ost_test values (null, 107);
insert into gh_ost_test values (null, 108);
insert into gh_ost_test values (null, 109);
insert into gh_ost_test values (null, 110);
insert into gh_ost_test values (null, 111);
insert into gh_ost_test values (null, 112);
insert into gh_ost_test values (null, 113);
insert into gh_ost_test values (null, 114);
insert into gh_ost_test values (null, 115);
insert into gh_ost_test values (null, 116);
insert into gh_ost_test values (null, 117);
insert into gh_ost_test values (null, 118);
insert into gh_ost_test values (null, 119);
insert into gh_ost_test values (null, 120);
insert into gh_ost_test values (null, 121);
insert into gh_ost_test values (null, 122);
insert into gh_ost_test values (null, 123);
insert into gh_ost_test values (null, 124);
insert into gh_ost_test values (null, 125);
insert into gh_ost_test values (null, 126);
insert into gh_ost_test values (null, 127);
insert into gh_ost_test values (null, 128);
insert into gh_ost_test values (null, 129);
drop event if exists gh_ost_test;
delimiter ;;
create event gh_ost_test
on schedule every 1 second
starts current_timestamp + interval 3 second
ends current_timestamp + interval 60 second
on completion not preserve
enable
do
begin
update gh_ost_test set id=-2 where id=21;
update gh_ost_test set id=55 where id=22;
update gh_ost_test set id=23 where id=23;
update gh_ost_test set i=5024 where id=24;
end ;;

5
localtests/test.sh
View File

@ -170,7 +170,12 @@ test_single() {
build_binary() {
echo "Building"
rm -f $ghost_binary
go build -o $ghost_binary go/cmd/gh-ost/main.go
if [ $? -ne 0 ] ; then
echo "Build failure"
exit 1
fi
}
test_all() {

21
script/ensure-go-installed
View File

@ -1,19 +1,20 @@
#!/bin/bash
GO_VERSION=go1.7
PREFERRED_GO_VERSION=go1.9.2
SUPPORTED_GO_VERSIONS='go1.[89]'
GO_PKG_DARWIN=${GO_VERSION}.darwin-amd64.pkg
GO_PKG_DARWIN_SHA=e7089843bc7148ffcc147759985b213604d22bb9fd19bd930b515aa981bf1b22
GO_PKG_DARWIN=${PREFERRED_GO_VERSION}.darwin-amd64.pkg
GO_PKG_DARWIN_SHA=73fd5840d55f5566d8db6c0ffdd187577e8ebe650c783f68bd27cbf95bde6743
GO_PKG_LINUX=${GO_VERSION}.linux-amd64.tar.gz
GO_PKG_LINUX_SHA=702ad90f705365227e902b42d91dd1a40e48ca7f67a2f4b2fd052aaa4295cd95
GO_PKG_LINUX=${PREFERRED_GO_VERSION}.linux-amd64.tar.gz
GO_PKG_LINUX_SHA=de874549d9a8d8d8062be05808509c09a88a248e77ec14eb77453530829ac02b
export ROOTDIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )/.." && pwd )"
cd $ROOTDIR
# If Go isn't installed globally, setup environment variables for local install.
if [ -z "$(which go)" ] || [ -z "$(go version | grep $GO_VERSION)" ]; then
GODIR="$ROOTDIR/.vendor/go17"
if [ -z "$(which go)" ] || [ -z "$(go version | grep "$SUPPORTED_GO_VERSIONS")" ]; then
GODIR="$ROOTDIR/.vendor/go19"
if [ $(uname -s) = "Darwin" ]; then
export GOROOT="$GODIR/usr/local/go"
@ -25,7 +26,7 @@ if [ -z "$(which go)" ] || [ -z "$(go version | grep $GO_VERSION)" ]; then
fi
# Check if local install exists, and install otherwise.
if [ -z "$(which go)" ] || [ -z "$(go version | grep $GO_VERSION)" ]; then
if [ -z "$(which go)" ] || [ -z "$(go version | grep "$SUPPORTED_GO_VERSIONS")" ]; then
[ -d "$GODIR" ] && rm -rf $GODIR
mkdir -p "$GODIR"
cd "$GODIR";
@ -42,7 +43,9 @@ if [ -z "$(which go)" ] || [ -z "$(go version | grep $GO_VERSION)" ]; then
fi
# Prove we did something right
echo "$GO_VERSION installed in $GODIR: Go Binary: $(which go)"
echo "$(go version) installed in $GODIR: Go Binary: $(which go)"
else
echo "$(go version) found in $GODIR: Go Binary: $(which go)"
fi
cd $ROOTDIR