percona/percona-toolkit
1
0
Fork 0
mirror of https://github.com/percona/percona-toolkit.git synced 2025-10-19 08:56:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert to r243 to undo pt-table-sync changes.

Browse Source
This commit is contained in:
Daniel Nichter
2011-12-22 12:06:25 -07:00
parent 2760322c87
commit 1d8da408fc
10 changed files with 4800 additions and 4285 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6689
bin/pt-table-sync
View File

File diff suppressed because it is too large Load Diff

34
lib/Cxn.pm
View File

@@ -54,7 +54,6 @@ use constant PERCONA_TOOLKIT_TEST_USE_DSN_NAMES => $ENV{PERCONA_TOOLKIT_TEST_USE
# Optional Arguments:
# dbh - Pre-created, uninitialized dbh
# set - Callback to set vars on dbh when dbh is first connected
# aux - Create a secondy (auxiliary) dbh, get with <aux_dbh()>.
#
# Returns:
# Cxn object
@@ -102,8 +101,6 @@ sub new {
dsn_name => $dp->as_string($dsn, [qw(h P S)]),
hostname => '',
set => $args{set},
aux => $args{aux},
dbh_opts => $args{dbh_opts} || {AutoCommit => 1},
dbh_set => 0,
OptionParser => $o,
DSNParser => $dp,
@@ -125,33 +122,13 @@ sub connect {
$dsn->{p} = OptionParser::prompt_noecho("Enter MySQL password: ");
$self->{asked_for_pass} = 1;
}
$dbh = $dp->get_dbh($dp->get_cxn_params($dsn), $self->{dbh_opts});
$dbh = $dp->get_dbh($dp->get_cxn_params($dsn), { AutoCommit => 1 });
}
MKDEBUG && _d($dbh, 'Connected dbh to', $self->{name});
if ( $self->{aux} && (!$self->{aux_dbh} || !$self->{aux_dbh}->ping()) ) {
my $aux_dbh = $dp->get_dbh($dp->get_cxn_params($dsn), {AutoCommit => 1});
MKDEBUG && _d($aux_dbh, 'Connected aux dbh to', $self->{name});
$dbh->{FetchHashKeyName} = 'NAME_lc';
$self->{aux_dbh} = $aux_dbh;
}
return $self->set_dbh($dbh);
}
sub disconnect {
my ($self) = @_;
if ( $self->{dbh} ) {
MKDEBUG && _d('Disconnecting dbh', $self->{dbh}, $self->{name});
$self->{dbh}->disconnect();
}
if ( $self->{aux_dbh} ) {
MKDEBUG && _d('Disconnecting aux dbh', $self->{aux_dbh});
$self->{aux_dbh}->disconnect();
}
return;
}
sub set_dbh {
my ($self, $dbh) = @_;
@@ -199,11 +176,6 @@ sub dbh {
return $self->{dbh};
}
sub aux_dbh {
my ($self) = @_;
return $self->{aux_dbh};
}
# Sub: dsn
# Return the cxn's dsn.
sub dsn {
@@ -225,10 +197,6 @@ sub DESTROY {
MKDEBUG && _d('Disconnecting dbh', $self->{dbh}, $self->{name});
$self->{dbh}->disconnect();
}
if ( $self->{aux_dbh} ) {
MKDEBUG && _d('Disconnecting aux dbh', $self->{aux_dbh});
$self->{aux_dbh}->disconnect();
}
return;
}

65
lib/NibbleIterator.pm
View File

@@ -64,11 +64,8 @@ sub new {
my $where = $o->get('where');
my ($row_est, $mysql_index) = get_row_estimate(%args, where => $where);
my $chunk_size_limit = $o->has('chunk-size-limit')
? $o->get('chunk-size-limit')
: 1;
my $one_nibble = !defined $args{one_nibble} || $args{one_nibble}
? $row_est <= $chunk_size * $chunk_size_limit
? $row_est <= $chunk_size * $o->get('chunk-size-limit')
: 0;
MKDEBUG && _d('One nibble:', $one_nibble ? 'yes' : 'no');
@@ -84,11 +81,6 @@ sub new {
if ( !$index && !$one_nibble ) {
die "There is no good index and the table is oversized.";
}
my ($index_cols, $order_by);
if ( $index ) {
$index_cols = $tbl->{tbl_struct}->{keys}->{$index}->{cols};
$order_by = join(', ', map {$q->quote($_)} @{$index_cols});
}
my $tbl_struct = $tbl->{tbl_struct};
my $ignore_col = $o->get('ignore-columns') || {};
@@ -98,20 +90,20 @@ sub new {
if ( $one_nibble ) {
# If the chunk size is >= number of rows in table, then we don't
# need to chunk; we can just select all rows, in order, at once.
my $cols = ($args{select} ? $args{select}
: join(', ', map { $q->quote($_) } @cols));
my $from = $q->quote(@{$tbl}{qw(db tbl)});
my $nibble_sql
= ($args{dml} ? "$args{dml} " : "SELECT ")
. $cols
. " FROM $from "
. ($args{select} ? $args{select}
: join(', ', map { $q->quote($_) } @cols))
. " FROM " . $q->quote(@{$tbl}{qw(db tbl)})
. ($where ? " AND ($where)" : '')
. " /*checksum table*/";
MKDEBUG && _d('One nibble statement:', $nibble_sql);
my $explain_nibble_sql
= "EXPLAIN SELECT $cols FROM $from"
= "EXPLAIN SELECT "
. ($args{select} ? $args{select}
: join(', ', map { $q->quote($_) } @cols))
. " FROM " . $q->quote(@{$tbl}{qw(db tbl)})
. ($where ? " AND ($where)" : '')
. " /*explain checksum table*/";
MKDEBUG && _d('Explain one nibble statement:', $explain_nibble_sql);
@@ -122,15 +114,11 @@ sub new {
limit => 0,
nibble_sql => $nibble_sql,
explain_nibble_sql => $explain_nibble_sql,
sql => {
columns => $cols,
from => $from,
where => $where,
order_by => $order_by,
},
};
}
else {
my $index_cols = $tbl->{tbl_struct}->{keys}->{$index}->{cols};
# Figure out how to nibble the table with the index.
my $asc = $args{TableNibbler}->generate_asc_stmt(
%args,
@@ -145,6 +133,7 @@ sub new {
# ORDER BY are the same for all statements. FORCE IDNEX and ORDER BY
# are needed to ensure deterministic nibbling.
my $from = $q->quote(@{$tbl}{qw(db tbl)}) . " FORCE INDEX(`$index`)";
my $order_by = join(', ', map {$q->quote($_)} @{$index_cols});
# The real first row in the table. Usually we start nibbling from
# this row. Called once in _get_bounds().
@@ -238,6 +227,7 @@ sub new {
$self = {
%args,
index => $index,
limit => $limit,
first_lb_sql => $first_lb_sql,
last_ub_sql => $last_ub_sql,
@@ -256,13 +246,11 @@ sub new {
};
}
$self->{index} = $index;
$self->{row_est} = $row_est;
$self->{nibbleno} = 0;
$self->{have_rows} = 0;
$self->{rowno} = 0;
$self->{oktonibble} = 1;
$self->{no_more_boundaries} = 0;
return bless $self, $class;
}
@@ -319,21 +307,12 @@ sub next {
if ( $self->{have_rows} ) {
# Return rows in nibble. sth->{Active} is always true with
# DBD::mysql v3, so we track the status manually.
my $row = $self->{fetch_hashref}
? $self->{nibble_sth}->fetchrow_hashref()
: $self->{nibble_sth}->fetchrow_arrayref();
my $row = $self->{nibble_sth}->fetchrow_arrayref();
if ( $row ) {
$self->{rowno}++;
MKDEBUG && _d('Row', $self->{rowno}, 'in nibble',$self->{nibbleno},
'from', $self->{Cxn}->name());
MKDEBUG && _d('Row', $self->{rowno}, 'in nibble',$self->{nibbleno});
# fetchrow_arraryref re-uses an internal arrayref, so we must copy.
return $self->{fetch_hashref} ? $row : [ @$row ];
}
else {
MKDEBUG && _d('No row in nibble');
if ( $self->{empty_results} ) {
return $self->{fetch_hashref} ? {} : [];
}
return [ @$row ];
}
}
@@ -434,15 +413,6 @@ sub more_boundaries {
return !$self->{no_more_boundaries};
}
sub no_more_rows {
my ($self) = @_;
$self->{nibble_sth}->finish() if $self->{nibble_sth};
$self->{have_rows} = 0;
$self->{rowno} = 0;
MKDEBUG && _d('No more rows');
return;
}
sub row_estimate {
my ($self) = @_;
return $self->{row_est};
@@ -646,8 +616,7 @@ sub _next_boundaries {
# which will cause us to nibble further ahead and maybe get a new lower
# boundary that isn't identical, but we can't detect this, and in any
# case, if there's one infinite loop there will probably be others.
if ( !$self->{manual_nibble}
&& $self->identical_boundaries($self->{lower}, $self->{next_lower}) ) {
if ( $self->identical_boundaries($self->{lower}, $self->{next_lower}) ) {
MKDEBUG && _d('Infinite loop detected');
my $tbl = $self->{tbl};
my $index = $tbl->{tbl_struct}->{keys}->{$self->{index}};
@@ -679,8 +648,6 @@ sub _next_boundaries {
}
}
return 1 if $self->{manual_nibble};
MKDEBUG && _d($self->{ub_sth}->{Statement}, 'params:',
join(', ', @{$self->{lower}}), $self->{limit});
$self->{ub_sth}->execute(@{$self->{lower}}, $self->{limit});

2
lib/RowChecksum.pm
View File

@@ -93,7 +93,7 @@ sub make_row_checksum {
}
if ( uc $func ne 'FNV_64' && uc $func ne 'FNV1A_64' ) {
my $sep = ($o->has('separator') && $o->get('separator')) || '#';
my $sep = $o->get('separator') || '#';
$sep =~ s/'//g;
$sep ||= '#';

104
lib/RowSyncer.pm
View File

@@ -1,104 +0,0 @@
# This program is copyright 2011 Percona Inc.
# Feedback and improvements are welcome.
#
# THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
# WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
# MERCHANTIBILITY AND FITNESS FOR A PARTICULAR PURPOSE.
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2; OR the Perl Artistic License. On UNIX and similar
# systems, you can issue `man perlgpl' or `man perlartistic' to read these
# licenses.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place, Suite 330, Boston, MA 02111-1307 USA.
# ###########################################################################
# RowSyncer package
# ###########################################################################
{
# Package: RowSyncer
# RowSyncer syncs a destination row to a source row.
package RowSyncer;
use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant MKDEBUG => $ENV{MKDEBUG} || 0;
use Data::Dumper;
$Data::Dumper::Indent = 1;
$Data::Dumper::Sortkeys = 1;
$Data::Dumper::Quotekeys = 0;
sub new {
my ( $class, %args ) = @_;
my @required_args = qw(ChangeHandler);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless defined $args{$arg};
}
my $self = {
crc_col => 'crc',
%args,
};
return bless $self, $class;
}
sub set_crc_col {
my ($self, $crc_col) = @_;
$self->{crc_col} = $crc_col;
return;
}
sub set_key_cols {
my ($self, $key_cols) = @_;
$self->{key_cols} = $key_cols;
return;
}
sub key_cols {
my ($self) = @_;
return $self->{key_cols};
}
sub same_row {
my ($self, %args) = @_;
my ($lr, $rr) = @args{qw(lr rr)};
if ( $lr->{$self->{crc_col}} ne $rr->{$self->{crc_col}} ) {
$self->{ChangeHandler}->change('UPDATE', $lr, $self->key_cols());
}
return;
}
sub not_in_right {
my ( $self, %args ) = @_;
# Row isn't in the dest, re-insert it in the source.
$self->{ChangeHandler}->change('INSERT', $args{lr}, $self->key_cols());
return;
}
sub not_in_left {
my ( $self, %args ) = @_;
# Row isn't in source, delete it from the dest.
$self->{ChangeHandler}->change('DELETE', $args{rr}, $self->key_cols());
return;
}
sub done_with_rows {
return;
}
sub _d {
my ($package, undef, $line) = caller 0;
@_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
map { defined $_ ? $_ : 'undef' }
@_;
print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}
1;
}
# ###########################################################################
# End RowSyncer package
# ###########################################################################

251
lib/RowSyncerBidirectional.pm
View File

@@ -1,251 +0,0 @@
# This program is copyright 2011 Percona Inc.
# Feedback and improvements are welcome.
#
# THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
# WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
# MERCHANTIBILITY AND FITNESS FOR A PARTICULAR PURPOSE.
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2; OR the Perl Artistic License. On UNIX and similar
# systems, you can issue `man perlgpl' or `man perlartistic' to read these
# licenses.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place, Suite 330, Boston, MA 02111-1307 USA.
# ###########################################################################
# RowSyncerBidirectional package
# ###########################################################################
{
# Package: RowSyncerBidirectional
# RowSyncerBidirectional syncs a destination row to a source row.
package RowSyncerBidirectional;
use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant MKDEBUG => $ENV{MKDEBUG} || 0;
use Data::Dumper;
$Data::Dumper::Indent = 1;
$Data::Dumper::Sortkeys = 1;
$Data::Dumper::Quotekeys = 0;
use constant UPDATE_LEFT => -1;
use constant UPDATE_RIGHT => 1;
use constant UPDATE_NEITHER => 0; # neither value equals/matches
use constant FAILED_THRESHOLD => 2; # failed to exceed threshold
sub new {
my ( $class, %args ) = @_;
my @required_args = qw(OptionParser ChangeHandler);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless defined $args{$arg};
}
my $self = {
crc_col => 'crc',
%args,
};
return bless $self, $class;
}
sub set_crc_col {
my ($self, $crc_col) = @_;
$self->{crc_col} = $crc_col;
return;
}
sub set_key_cols {
my ($self, $key_cols) = @_;
$self->{key_cols} = $key_cols;
return;
}
sub key_cols {
my ($self) = @_;
return $self->{key_cols};
}
# Sub: cmd_conflict_col
# Compare --conflict-column values for --bidirectional. This sub is
# used as a callback in <set_bidirectional_callbacks()>.
#
# Parameters:
# $left_val - Column value from left (usually the source host)
# $right_val - Column value from right (usually the destination host)
# $cmp - Type of conflict comparison, --conflict-comparison
# $val - Value for certain types of comparisons, --conflict-value
# $thr - Threshold for certain types of comparisons,
# --conflict-threshold
#
# Returns:
# One of the constants above, UPDATE_* or FAILED_THRESHOLD
sub cmp_conflict_col {
my ( $left_val, $right_val, $cmp, $val, $thr ) = @_;
MKDEBUG && _d('Compare', @_);
my $res;
if ( $cmp eq 'newest' || $cmp eq 'oldest' ) {
$res = $cmp eq 'newest' ? ($left_val || '') cmp ($right_val || '')
: ($right_val || '') cmp ($left_val || '');
if ( $thr ) {
$thr = Transformers::time_to_secs($thr);
my $lts = Transformers::any_unix_timestamp($left_val);
my $rts = Transformers::any_unix_timestamp($right_val);
my $diff = abs($lts - $rts);
MKDEBUG && _d('Check threshold, lts rts thr abs-diff:',
$lts, $rts, $thr, $diff);
if ( $diff < $thr ) {
MKDEBUG && _d("Failed threshold");
return FAILED_THRESHOLD;
}
}
}
elsif ( $cmp eq 'greatest' || $cmp eq 'least' ) {
$res = $cmp eq 'greatest' ? (($left_val ||0) > ($right_val ||0) ? 1 : -1)
: (($left_val ||0) < ($right_val ||0) ? 1 : -1);
$res = 0 if ($left_val || 0) == ($right_val || 0);
if ( $thr ) {
my $diff = abs($left_val - $right_val);
MKDEBUG && _d('Check threshold, abs-diff:', $diff);
if ( $diff < $thr ) {
MKDEBUG && _d("Failed threshold");
return FAILED_THRESHOLD;
}
}
}
elsif ( $cmp eq 'equals' ) {
$res = ($left_val || '') eq $val ? 1
: ($right_val || '') eq $val ? -1
: 0;
}
elsif ( $cmp eq 'matches' ) {
$res = ($left_val || '') =~ m/$val/ ? 1
: ($right_val || '') =~ m/$val/ ? -1
: 0;
}
else {
# Should happen; caller should have verified this.
die "Invalid comparison: $cmp";
}
return $res;
}
sub same_row {
my ($self, %args) = @_;
my ($lr, $rr, $syncer) = @args{qw(lr rr syncer)};
return unless $lr->{$self->{crc_col}} ne $rr->{$self->{crc_col}};
my $ch = $self->{ChangeHandler};
my $action = 'UPDATE';
my $auth_row = $lr;
my $change_dbh;
my $err;
my $o = $self->{OptionParser};
my $col = $o->get('conflict-column');
my $cmp = $o->get('conflict-comparison');
my $val = $o->get('conflict-value');
my $thr = $o->get('conflict-threshold');
my $left_val = $lr->{$col} || '';
my $right_val = $rr->{$col} || '';
MKDEBUG && _d('left', $col, 'value:', $left_val);
MKDEBUG && _d('right', $col, 'value:', $right_val);
my $res = cmp_conflict_col($left_val, $right_val, $cmp, $val, $thr);
if ( $res == UPDATE_LEFT ) {
MKDEBUG && _d("right dbh $args{right_dbh} $cmp; "
. "update left dbh $args{left_dbh}");
$ch->set_src('right', $args{right_dbh});
$auth_row = $args{rr};
$change_dbh = $args{left_dbh};
}
elsif ( $res == UPDATE_RIGHT ) {
MKDEBUG && _d("left dbh $args{left_dbh} $cmp; "
. "update right dbh $args{right_dbh}");
$ch->set_src('left', $args{left_dbh});
$auth_row = $args{lr};
$change_dbh = $args{right_dbh};
}
elsif ( $res == UPDATE_NEITHER ) {
if ( $cmp eq 'equals' || $cmp eq 'matches' ) {
$err = "neither `$col` value $cmp $val";
}
else {
$err = "`$col` values are the same"
}
}
elsif ( $res == FAILED_THRESHOLD ) {
$err = "`$col` values do not differ by the threhold, $thr."
}
else {
# Shouldn't happen.
die "cmp_conflict_col() returned an invalid result: $res."
}
if ( $err ) {
$action = undef; # skip change in case we just warn
my $where = $ch->make_where_clause($lr, $self->key_cols());
$err = "# Cannot resolve conflict WHERE $where: $err\n";
# die here is caught in sync_a_table(). We're deeply nested:
# sync_a_table > sync_table > compare_sets > syncer > here
my $print_err = $o->get('conflict-error');
$print_err =~ m/warn/i ? warn $err
: $print_err =~ m/die/i ? die $err
: $print_err =~ m/ignore/i ? MKDEBUG && _d("Conflict error:", $err)
: die "Invalid --conflict-error: $print_err";
return;
}
return $ch->change(
$action, # Execute the action
$auth_row, # with these row values
$self->key_cols(), # identified by these key cols
$change_dbh, # on this dbh
);
}
sub not_in_right {
my ( $self, %args ) = @_;
$self->{ChangeHandler}->set_src('left', $args{left_dbh});
return $self->{ChangeHandler}->change(
'INSERT', # Execute the action
$args{lr}, # with these row values
$self->key_cols(), # identified by these key cols
$args{right_dbh}, # on this dbh
);
}
sub not_in_left {
my ( $self, %args ) = @_;
$self->{ChangeHandler}->set_src('right', $args{right_dbh});
return $self->{ChangeHandler}->change(
'INSERT', # Execute the action
$args{rr}, # with these row values
$self->key_cols(), # identified by these key cols
$args{left_dbh}, # on this dbh
);
}
sub done_with_rows {
return;
}
sub _d {
my ($package, undef, $line) = caller 0;
@_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
map { defined $_ ? $_ : 'undef' }
@_;
print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}
1;
}
# ###########################################################################
# End RowSyncerBidirectional package
# ###########################################################################

759
lib/TableSyncer.pm
View File

@@ -1,4 +1,4 @@
# This program is copyright 2011 Percona Inc.
# This program is copyright 2007-2011 Baron Schwartz, 2011 Percona Inc.
# Feedback and improvements are welcome.
#
# THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
@@ -32,10 +32,16 @@ $Data::Dumper::Indent = 1;
$Data::Dumper::Sortkeys = 1;
$Data::Dumper::Quotekeys = 0;
# Arguments:
# * MasterSlave A MasterSlave module
# * Quoter A Quoter module
# * VersionParser A VersionParser module
# * TableChecksum A TableChecksum module
# * Retry A Retry module
# * DSNParser (optional)
sub new {
my ( $class, %args ) = @_;
my @required_args = qw(MasterSlave OptionParser Quoter TableParser
TableNibbler RowChecksum RowDiff Retry);
my @required_args = qw(MasterSlave Quoter VersionParser TableChecksum Retry);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless defined $args{$arg};
}
@@ -43,6 +49,29 @@ sub new {
return bless $self, $class;
}
# Return the first plugin from the arrayref of TableSync* plugins
# that can sync the given table struct. plugin->can_sync() usually
# returns a hashref that it wants back when plugin->prepare_to_sync()
# is called. Or, it may return nothing (false) to say that it can't
# sync the table.
sub get_best_plugin {
my ( $self, %args ) = @_;
foreach my $arg ( qw(plugins tbl_struct) ) {
die "I need a $arg argument" unless $args{$arg};
}
MKDEBUG && _d('Getting best plugin');
foreach my $plugin ( @{$args{plugins}} ) {
MKDEBUG && _d('Trying plugin', $plugin->name);
my ($can_sync, %plugin_args) = $plugin->can_sync(%args);
if ( $can_sync ) {
MKDEBUG && _d('Can sync with', $plugin->name, Dumper(\%plugin_args));
return $plugin, %plugin_args;
}
}
MKDEBUG && _d('No plugin can sync the table');
return;
}
# Required arguments:
# * plugins Arrayref of TableSync* modules, in order of preference
# * src Hashref with source (aka left) dbh, db, tbl
@@ -72,371 +101,340 @@ sub new {
# * wait locking
sub sync_table {
my ( $self, %args ) = @_;
my @required_args = qw(src dst RowSyncer ChangeHandler);
my @required_args = qw(plugins src dst tbl_struct cols chunk_size
RowDiff ChangeHandler);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless $args{$arg};
}
my ($src, $dst, $row_syncer, $changer) = @args{@required_args};
MKDEBUG && _d('Syncing table with args:',
map { "$_: " . Dumper($args{$_}) }
qw(plugins src dst tbl_struct cols chunk_size));
my $diffs = $args{diffs};
my $changing_src = $args{changing_src};
my ($plugins, $src, $dst, $tbl_struct, $cols, $chunk_size, $rd, $ch)
= @args{@required_args};
my $dp = $self->{DSNParser};
$args{trace} = 1 unless defined $args{trace};
if ( $args{bidirectional} && $args{ChangeHandler}->{queue} ) {
# This should be checked by the caller but just in case...
die "Queueing does not work with bidirectional syncing";
}
$args{index_hint} = 1 unless defined $args{index_hint};
$args{lock} ||= 0;
$args{wait} ||= 0;
$args{transaction} ||= 0;
$args{timeout_ok} ||= 0;
my $o = $self->{OptionParser};
my $q = $self->{Quoter};
my $row_diff = $self->{RowDiff};
my $row_checksum = $self->{RowChecksum};
my $vp = $self->{VersionParser};
# USE db on src and dst for cases like when replicate-do-db is being used.
foreach my $host ( $src, $dst ) {
$host->{Cxn}->dbh()->do("USE " . $q->quote($host->{tbl}->{db}));
}
# ########################################################################
# Get and prepare the first plugin that can sync this table.
# ########################################################################
my ($plugin, %plugin_args) = $self->get_best_plugin(%args);
die "No plugin can sync $src->{db}.$src->{tbl}" unless $plugin;
my $trace;
if ( !defined $args{trace} || $args{trace} ) {
chomp(my $hostname = `hostname`);
$trace = "src_host:" . $src->{Cxn}->name()
. " src_tbl:" . join('.', @{$src->{tbl}}{qw(db tbl)})
. " dst_host:" . $dst->{Cxn}->name()
. " dst_tbl:" . join('.', @{$dst->{tbl}}{qw(db tbl)})
. " changing_src:" . ($changing_src ? "yes" : "no")
. " diffs:" . ($diffs ? scalar @$diffs : "0")
. " " . join(" ", map { "$_:" . ($o->get($_) ? "yes" : "no") }
qw(lock transaction replicate bidirectional))
. " pid:$PID "
. ($ENV{USER} ? "user:$ENV{USER} " : "")
. ($hostname ? "host:$hostname" : "");
MKDEBUG && _d("Binlog trace message:", $trace);
}
# Make NibbleIterator for checksumming chunks of rows to see if
# there are any diffs.
my %crc_args = $row_checksum->get_crc_args(dbh => $src->{Cxn}->dbh());
my $chunk_cols;
if ( $diffs ) {
$chunk_cols = "0 AS cnt, '' AS crc";
}
else {
$chunk_cols = $row_checksum->make_chunk_checksum(
dbh => $src->{Cxn}->dbh(),
tbl => $src->{tbl},
%crc_args
);
}
if ( !defined $src->{select_lock} || !defined $dst->{select_lock} ) {
if ( $o->get('transaction') ) {
if ( $o->get('bidirectional') ) {
# Making changes on src and dst.
$src->{select_lock} = 'FOR UPDATE';
$dst->{select_lock} = 'FOR UPDATE';
}
elsif ( $changing_src ) {
# Making changes on master (src) which replicate to slave (dst).
$src->{select_lock} = 'FOR UPDATE';
$dst->{select_lock} = 'LOCK IN SHARE MODE';
}
else {
# Making changes on slave (dst).
$src->{select_lock} = 'LOCK IN SHARE MODE';
$dst->{select_lock} = 'FOR UPDATE';
}
}
else {
$src->{select_lock} = '';
$dst->{select_lock} = '';
}
MKDEBUG && _d('SELECT lock:', $src->{select_lock});
MKDEBUG && _d('SELECT lock:', $dst->{select_lock});
}
my $user_where = $o->get('where');
my ($src_nibble_iter, $dst_nibble_iter);
foreach my $host ($src, $dst) {
my $callbacks = {
init => sub {
my (%args) = @_;
my $cxn = $args{Cxn};
my $tbl = $args{tbl};
my $nibble_iter = $args{NibbleIterator};
my $sths = $nibble_iter->statements();
my $oktonibble = 1;
if ( $o->get('explain') ) {
# --explain level 1: print the checksum and next boundary
# statements.
print "--\n"
. "-- "
. ($cxn->{is_source} ? "Source" : "Destination")
. " " . $cxn->name()
. " " . "$tbl->{db}.$tbl->{tbl}\n"
. "--\n\n";
my $statements = $nibble_iter->statements();
foreach my $sth ( sort keys %$statements ) {
next if $sth =~ m/^explain/;
if ( $statements->{$sth} ) {
print $statements->{$sth}->{Statement}, "\n\n";
}
}
if ( $o->get('explain') < 2 ) {
$oktonibble = 0; # don't nibble table; next table
}
}
else {
if ( $o->get('buffer-to-client') ) {
$host->{sth}->{mysql_use_result} = 1;
}
# Lock the table.
$self->lock_and_wait(
lock_level => 2,
host => $host,
src => $src,
changing_src => $changing_src,
);
}
return $oktonibble;
},
next_boundaries => sub {
my (%args) = @_;
my $tbl = $args{tbl};
if ( my $diff = $tbl->{diff} ) {
my $nibble_iter = $args{NibbleIterator};
my $boundary = $nibble_iter->boundaries();
$nibble_iter->set_boundary(
'upper', [ split ',', $diff->{upper_boundary} ]);
$nibble_iter->set_boundary(
'lower', [ split ',', $diff->{lower_boundary} ]);
}
return 1;
},
exec_nibble => sub {
my (%args) = @_;
my $tbl = $args{tbl};
my $nibble_iter = $args{NibbleIterator};
my $sths = $nibble_iter->statements();
my $boundary = $nibble_iter->boundaries();
# --explain level 2: print chunk,lower boundary values,upper
# boundary values.
if ( $o->get('explain') > 1 ) {
my $lb_quoted = join(',', @{$boundary->{lower} || []});
my $ub_quoted = join(',', @{$boundary->{upper} || []});
my $chunk = $nibble_iter->nibble_number();
printf "%d %s %s\n",
$chunk,
(defined $lb_quoted ? $lb_quoted : '1=1'),
(defined $ub_quoted ? $ub_quoted : '1=1');
if ( !$nibble_iter->more_boundaries() ) {
print "\n"; # blank line between this table and the next table
}
return 0; # next boundary
}
# Lock the chunk.
$self->lock_and_wait(
%args,
lock_level => 1,
host => $host,
src => $src,
changing_src => $changing_src,
);
# Execute the chunk checksum statement.
# The nibble iter will return the row.
MKDEBUG && _d('nibble', $args{Cxn}->name());
$sths->{nibble}->execute(@{$boundary->{lower}}, @{$boundary->{upper}});
return 1;
},
};
my $nibble_iter = new NibbleIterator(
Cxn => $host->{Cxn},
tbl => $host->{tbl},
chunk_size => $o->get('chunk-size'),
chunk_index => $diffs ? $diffs->[0]->{chunk_index}
: $o->get('chunk-index'),
manual_nibble => $diffs ? 1 : 0,
empty_results => 1,
select => $chunk_cols,
select_lock => $host->{select_lock},
callbacks => $callbacks,
fetch_hashref => 1,
one_nibble => $args{one_nibble},
OptionParser => $self->{OptionParser},
Quoter => $self->{Quoter},
TableNibbler => $self->{TableNibbler},
TableParser => $self->{TableParser},
RowChecksum => $self->{RowChecksum},
);
if ( $host->{Cxn}->{is_source} ) {
$src_nibble_iter = $nibble_iter;
}
else {
$dst_nibble_iter = $nibble_iter;
}
}
my $index = $src_nibble_iter->nibble_index();
my $key_cols = $index ? $src->{tbl}->{tbl_struct}->{keys}->{$index}->{cols}
: $src->{tbl}->{tbl_struct}->{cols};
$row_syncer->set_key_cols($key_cols);
my $crc_col = 'crc';
while ( $src->{tbl}->{tbl_struct}->{is_col}->{$crc_col} ) {
# The row-level (state 2) checksums use __crc, so the table can't use that.
my $crc_col = '__crc';
while ( $tbl_struct->{is_col}->{$crc_col} ) {
$crc_col = "_$crc_col"; # Prepend more _ until not a column.
}
$row_syncer->set_crc_col($crc_col);
MKDEBUG && _d('CRC column:', $crc_col);
my $rows_sql;
my $row_cols = $row_checksum->make_row_checksum(
dbh => $src->{Cxn}->dbh(),
# Make an index hint for either the explicitly given chunk_index
# or the chunk_index chosen by the plugin if index_hint is true.
my $index_hint;
my $hint = ($vp->version_ge($src->{dbh}, '4.0.9')
&& $vp->version_ge($dst->{dbh}, '4.0.9') ? 'FORCE' : 'USE')
. ' INDEX';
if ( $args{chunk_index} ) {
MKDEBUG && _d('Using given chunk index for index hint');
$index_hint = "$hint (" . $q->quote($args{chunk_index}) . ")";
}
elsif ( $plugin_args{chunk_index} && $args{index_hint} ) {
MKDEBUG && _d('Using chunk index chosen by plugin for index hint');
$index_hint = "$hint (" . $q->quote($plugin_args{chunk_index}) . ")";
}
MKDEBUG && _d('Index hint:', $index_hint);
eval {
$plugin->prepare_to_sync(
%args,
%plugin_args,
dbh => $src->{dbh},
db => $src->{db},
tbl => $src->{tbl},
%crc_args,
crc_col => $crc_col,
index_hint => $index_hint,
);
my $sql_clause = $src_nibble_iter->sql();
foreach my $host ($src, $dst) {
if ( $src_nibble_iter->one_nibble() ) {
$rows_sql
= 'SELECT /*rows in nibble*/ '
. ($self->{buffer_in_mysql} ? 'SQL_BUFFER_RESULT ' : '')
. "$row_cols AS $crc_col"
. " FROM " . $q->quote(@{$host->{tbl}}{qw(db tbl)})
. " WHERE 1=1 "
. ($user_where ? " AND ($user_where)" : '')
. ($sql_clause->{order_by} ? " ORDER BY " . $sql_clause->{order_by}
: "");
};
if ( $EVAL_ERROR ) {
# At present, no plugin should fail to prepare, but just in case...
die 'Failed to prepare TableSync', $plugin->name, ' plugin: ',
$EVAL_ERROR;
}
else {
$rows_sql
= 'SELECT /*rows in nibble*/ '
. ($self->{buffer_in_mysql} ? 'SQL_BUFFER_RESULT ' : '')
. "$row_cols AS $crc_col"
. " FROM " . $q->quote(@{$host->{tbl}}{qw(db tbl)})
. " WHERE " . $sql_clause->{boundaries}->{'>='} # lower boundary
. " AND " . $sql_clause->{boundaries}->{'<='} # upper boundary
. ($user_where ? " AND ($user_where)" : '')
. " ORDER BY " . $sql_clause->{order_by};
# Some plugins like TableSyncChunk use checksum queries, others like
# TableSyncGroupBy do not. For those that do, make chunk (state 0)
# and row (state 2) checksum queries.
if ( $plugin->uses_checksum() ) {
eval {
my ($chunk_sql, $row_sql) = $self->make_checksum_queries(%args);
$plugin->set_checksum_queries($chunk_sql, $row_sql);
};
if ( $EVAL_ERROR ) {
# This happens if src and dst are really different and the same
# checksum algo and hash func can't be used on both.
die "Failed to make checksum queries: $EVAL_ERROR";
}
$host->{rows_sth} = $host->{Cxn}->dbh()->prepare($rows_sql);
}
# ########################################################################
# Plugin is ready, return now if this is a dry run.
# ########################################################################
if ( $args{dry_run} ) {
return $ch->get_changes(), ALGORITHM => $plugin->name;
}
# ########################################################################
# Start syncing the table.
# ########################################################################
while ( $src_nibble_iter->more_boundaries()
|| $dst_nibble_iter->more_boundaries() ) {
if ( $diffs ) {
my $diff = shift @$diffs;
if ( !$diff ) {
MKDEBUG && _d('No more checksum diffs');
last;
}
MKDEBUG && _d('Syncing checksum diff', Dumper($diff));
$src->{tbl}->{diff} = $diff;
$dst->{tbl}->{diff} = $diff;
# USE db on src and dst for cases like when replicate-do-db is being used.
eval {
$src->{dbh}->do("USE `$src->{db}`");
$dst->{dbh}->do("USE `$dst->{db}`");
};
if ( $EVAL_ERROR ) {
# This shouldn't happen, but just in case. (The db and tbl on src
# and dst should be checked before calling this sub.)
die "Failed to USE database on source or destination: $EVAL_ERROR";
}
my $src_chunk = $src_nibble_iter->next();
my $dst_chunk = $dst_nibble_iter->next();
MKDEBUG && _d('Got chunk');
# For bidirectional syncing it's important to know on which dbh
# changes are made or rows are fetched. This identifies the dbhs,
# then you can search for each one by its address like
# "dbh DBI::db=HASH(0x1028b38)".
MKDEBUG && _d('left dbh', $src->{dbh});
MKDEBUG && _d('right dbh', $dst->{dbh});
if ( $diffs
|| ($src_chunk->{cnt} || 0) != ($dst_chunk->{cnt} || 0)
|| ($src_chunk->{crc} || '') ne ($dst_chunk->{crc} || '') ) {
MKDEBUG && _d("Chunks differ");
my $boundary = $src_nibble_iter->boundaries();
foreach my $host ($src, $dst) {
MKDEBUG && _d($host->{Cxn}->name(), $host->{rows_sth}->{Statement},
'params:', @{$boundary->{lower}}, @{$boundary->{upper}});
$host->{rows_sth}->execute(
@{$boundary->{lower}}, @{$boundary->{upper}});
}
$row_diff->compare_sets(
left_dbh => $src->{Cxn}->dbh(),
left_sth => $src->{rows_sth},
right_dbh => $dst->{Cxn}->dbh(),
right_sth => $dst->{rows_sth},
tbl_struct => $src->{tbl}->{tbl_struct},
syncer => $row_syncer,
chomp(my $hostname = `hostname`);
my $trace_msg
= $args{trace} ? "src_db:$src->{db} src_tbl:$src->{tbl} "
. ($dp && $src->{dsn} ? "src_dsn:".$dp->as_string($src->{dsn}) : "")
. " dst_db:$dst->{db} dst_tbl:$dst->{tbl} "
. ($dp && $dst->{dsn} ? "dst_dsn:".$dp->as_string($dst->{dsn}) : "")
. " " . join(" ", map { "$_:" . ($args{$_} || 0) }
qw(lock transaction changing_src replicate bidirectional))
. " pid:$PID "
. ($ENV{USER} ? "user:$ENV{USER} " : "")
. ($hostname ? "host:$hostname" : "")
: "";
MKDEBUG && _d("Binlog trace message:", $trace_msg);
$self->lock_and_wait(%args, lock_level => 2); # per-table lock
my $callback = $args{callback};
my $cycle = 0;
while ( !$plugin->done() ) {
# Do as much of the work as possible before opening a transaction or
# locking the tables.
MKDEBUG && _d('Beginning sync cycle', $cycle);
my $src_sql = $plugin->get_sql(
database => $src->{db},
table => $src->{tbl},
where => $args{where},
);
$changer->process_rows(1, $trace);
foreach my $host ($src, $dst) {
$host->{rows_sth}->finish();
}
if ( !$o->get('lock') && $o->get('transaction') ) {
foreach my $host ($src, $dst) {
$host->{Cxn}->dbh()->commit()
unless $host->{Cxn}->dbh()->{AutoCommit};
}
}
}
# Get next chunks.
$src_nibble_iter->no_more_rows();
$dst_nibble_iter->no_more_rows();
}
$changer->process_rows(0, $trace);
# Unlock the table.
foreach my $host ($src, $dst) {
$self->unlock(
lock_level => 2,
host => $host,
OptionParser => $o,
my $dst_sql = $plugin->get_sql(
database => $dst->{db},
table => $dst->{tbl},
where => $args{where},
);
if ( $args{transaction} ) {
if ( $args{bidirectional} ) {
# Making changes on src and dst.
$src_sql .= ' FOR UPDATE';
$dst_sql .= ' FOR UPDATE';
}
elsif ( $args{changing_src} ) {
# Making changes on master (src) which replicate to slave (dst).
$src_sql .= ' FOR UPDATE';
$dst_sql .= ' LOCK IN SHARE MODE';
}
else {
# Making changes on slave (dst).
$src_sql .= ' LOCK IN SHARE MODE';
$dst_sql .= ' FOR UPDATE';
}
}
MKDEBUG && _d('src:', $src_sql);
MKDEBUG && _d('dst:', $dst_sql);
# Give callback a chance to do something with the SQL statements.
$callback->($src_sql, $dst_sql) if $callback;
# Prepare each host for next sync cycle. This does stuff
# like reset/init MySQL accumulator vars, etc.
$plugin->prepare_sync_cycle($src);
$plugin->prepare_sync_cycle($dst);
# Prepare SQL statements on host. These aren't real prepared
# statements (i.e. no ? placeholders); we just need sths to
# pass to compare_sets(). Also, we can control buffering
# (mysql_use_result) on the sths.
my $src_sth = $src->{dbh}->prepare($src_sql);
my $dst_sth = $dst->{dbh}->prepare($dst_sql);
if ( $args{buffer_to_client} ) {
$src_sth->{mysql_use_result} = 1;
$dst_sth->{mysql_use_result} = 1;
}
return $changer->get_changes();
# The first cycle should lock to begin work; after that, unlock only if
# the plugin says it's OK (it may want to dig deeper on the rows it
# currently has locked).
my $executed_src = 0;
if ( !$cycle || !$plugin->pending_changes() ) {
# per-sync cycle lock
$executed_src
= $self->lock_and_wait(%args, src_sth => $src_sth, lock_level => 1);
}
sub lock_table {
# The source sth might have already been executed by lock_and_wait().
$src_sth->execute() unless $executed_src;
$dst_sth->execute();
# Compare rows in the two sths. If any differences are found
# (same_row, not_in_left, not_in_right), the appropriate $syncer
# methods are called to handle them. Changes may be immediate, or...
$rd->compare_sets(
left_sth => $src_sth,
right_sth => $dst_sth,
left_dbh => $src->{dbh},
right_dbh => $dst->{dbh},
syncer => $plugin,
tbl_struct => $tbl_struct,
);
# ...changes may be queued and executed now.
$ch->process_rows(1, $trace_msg);
MKDEBUG && _d('Finished sync cycle', $cycle);
$cycle++;
}
$ch->process_rows(0, $trace_msg);
$self->unlock(%args, lock_level => 2);
return $ch->get_changes(), ALGORITHM => $plugin->name;
}
sub make_checksum_queries {
my ( $self, %args ) = @_;
my @required_args = qw(host mode);
my @required_args = qw(src dst tbl_struct);
foreach my $arg ( @required_args ) {
die "I need a $arg argument" unless $args{$arg};
}
my ($host, $mode) = @args{@required_args};
my $q = $self->{Quoter};
my $sql = "LOCK TABLES "
. $q->quote(@{$host->{tbl}}{qw(db tbl)})
. " $mode";
MKDEBUG && _d($host->{Cxn}->name(), $sql);
$host->{Cxn}->dbh()->do($sql);
return;
my ($src, $dst, $tbl_struct) = @args{@required_args};
my $checksum = $self->{TableChecksum};
# Decide on checksumming strategy and store checksum query prototypes for
# later.
my $src_algo = $checksum->best_algorithm(
algorithm => 'BIT_XOR',
dbh => $src->{dbh},
where => 1,
chunk => 1,
count => 1,
);
my $dst_algo = $checksum->best_algorithm(
algorithm => 'BIT_XOR',
dbh => $dst->{dbh},
where => 1,
chunk => 1,
count => 1,
);
if ( $src_algo ne $dst_algo ) {
die "Source and destination checksum algorithms are different: ",
"$src_algo on source, $dst_algo on destination"
}
MKDEBUG && _d('Chosen algo:', $src_algo);
my $src_func = $checksum->choose_hash_func(dbh => $src->{dbh}, %args);
my $dst_func = $checksum->choose_hash_func(dbh => $dst->{dbh}, %args);
if ( $src_func ne $dst_func ) {
die "Source and destination hash functions are different: ",
"$src_func on source, $dst_func on destination";
}
MKDEBUG && _d('Chosen hash func:', $src_func);
# Since the checksum algo and hash func are the same on src and dst
# it doesn't matter if we use src_algo/func or dst_algo/func.
my $crc_wid = $checksum->get_crc_wid($src->{dbh}, $src_func);
my ($crc_type) = $checksum->get_crc_type($src->{dbh}, $src_func);
my $opt_slice;
if ( $src_algo eq 'BIT_XOR' && $crc_type !~ m/int$/ ) {
$opt_slice = $checksum->optimize_xor(
dbh => $src->{dbh},
function => $src_func
);
}
my $chunk_sql = $checksum->make_checksum_query(
%args,
db => $src->{db},
tbl => $src->{tbl},
algorithm => $src_algo,
function => $src_func,
crc_wid => $crc_wid,
crc_type => $crc_type,
opt_slice => $opt_slice,
replicate => undef, # replicate means something different to this sub
); # than what we use it for; do not pass it!
MKDEBUG && _d('Chunk sql:', $chunk_sql);
my $row_sql = $checksum->make_row_checksum(
%args,
function => $src_func,
);
MKDEBUG && _d('Row sql:', $row_sql);
return $chunk_sql, $row_sql;
}
sub lock_table {
my ( $self, $dbh, $where, $db_tbl, $mode ) = @_;
my $query = "LOCK TABLES $db_tbl $mode";
MKDEBUG && _d($query);
$dbh->do($query);
MKDEBUG && _d('Acquired table lock on', $where, 'in', $mode, 'mode');
}
# Doesn't work quite the same way as lock_and_wait. It will unlock any LOWER
# priority lock level, not just the exact same one.
sub unlock {
my ( $self, %args ) = @_;
my @required_args = qw(lock_level host);
foreach my $arg ( @required_args ) {
foreach my $arg ( qw(src dst lock transaction lock_level) ) {
die "I need a $arg argument" unless defined $args{$arg};
}
my ($lock_level, $host) = @args{@required_args};
my $o = $self->{OptionParser};
my $src = $args{src};
my $dst = $args{dst};
my $lock = $o->get('lock');
return unless $lock && $lock <= $lock_level;
MKDEBUG && _d('Unlocking level', $lock);
return unless $args{lock} && $args{lock} <= $args{lock_level};
if ( $o->get('transaction') ) {
MKDEBUG && _d('Committing', $host->{Cxn}->name());
$host->{Cxn}->dbh()->commit();
# First, unlock/commit.
foreach my $dbh ( $src->{dbh}, $dst->{dbh} ) {
if ( $args{transaction} ) {
MKDEBUG && _d('Committing', $dbh);
$dbh->commit();
}
else {
my $sql = 'UNLOCK TABLES';
MKDEBUG && _d($host->{Cxn}->name(), $sql);
$host->{Cxn}->dbh()->do($sql);
MKDEBUG && _d($dbh, $sql);
$dbh->do($sql);
}
}
return;
@@ -460,86 +458,73 @@ sub unlock {
# $src_sth was executed.
sub lock_and_wait {
my ( $self, %args ) = @_;
my @required_args = qw(lock_level host src);
foreach my $arg ( @required_args ) {
my $result = 0;
foreach my $arg ( qw(src dst lock lock_level) ) {
die "I need a $arg argument" unless defined $args{$arg};
}
my ($lock_level, $host, $src) = @args{@required_args};
my $o = $self->{OptionParser};
my $src = $args{src};
my $dst = $args{dst};
my $lock = $o->get('lock');
return unless $lock && $lock == $lock_level;
return unless $args{lock} && $args{lock} == $args{lock_level};
MKDEBUG && _d('lock and wait, lock level', $args{lock});
# First, commit/unlock the previous transaction/lock.
if ( $o->get('transaction') ) {
MKDEBUG && _d('Committing', $host->{Cxn}->name());
$host->{Cxn}->dbh()->commit();
foreach my $dbh ( $src->{dbh}, $dst->{dbh} ) {
if ( $args{transaction} ) {
MKDEBUG && _d('Committing', $dbh);
$dbh->commit();
}
else {
my $sql = 'UNLOCK TABLES';
MKDEBUG && _d($host->{Cxn}->name(), $sql);
$host->{Cxn}->dbh()->do($sql);
MKDEBUG && _d($dbh, $sql);
$dbh->do($sql);
}
}
# Lock/start xa.
return $host->{Cxn}->{is_source} ? $self->_lock_src(%args)
: $self->_lock_dst(%args);
}
sub _lock_src {
my ( $self, %args ) = @_;
my @required_args = qw(lock_level host src);
my ($lock_level, $host, $src) = @args{@required_args};
my $o = $self->{OptionParser};
my $lock = $o->get('lock');
MKDEBUG && _d('Locking', $host->{Cxn}->name(), 'level', $lock);
if ( $lock == 3 ) {
# User wants us to lock for consistency. But lock only on source initially;
# might have to wait for the slave to catch up before locking on the dest.
if ( $args{lock} == 3 ) {
my $sql = 'FLUSH TABLES WITH READ LOCK';
MKDEBUG && _d($host->{Cxn}->name(), $sql);
$host->{Cxn}->dbh()->do($sql);
MKDEBUG && _d($src->{dbh}, $sql);
$src->{dbh}->do($sql);
}
else {
# Lock level 2 (per-table) or 1 (per-chunk).
if ( $o->get('transaction') ) {
# Lock level 2 (per-table) or 1 (per-sync cycle)
if ( $args{transaction} ) {
if ( $args{src_sth} ) {
# Execute the $src_sth on the source, so LOCK IN SHARE MODE/FOR
# UPDATE will lock the rows examined.
MKDEBUG && _d('Executing statement on source to lock rows');
my $sql = "START TRANSACTION /*!40108 WITH CONSISTENT SNAPSHOT */";
MKDEBUG && _d($host->{Cxn}->name(), $sql);
$host->{Cxn}->dbh()->do($sql);
MKDEBUG && _d($src->{dbh}, $sql);
$src->{dbh}->do($sql);
$args{src_sth}->execute();
$result = 1;
}
}
else {
$self->lock_table(
host => $host,
mode => $args{changing_src} ? 'WRITE' : 'READ',
);
$self->lock_table($src->{dbh}, 'source',
$self->{Quoter}->quote($src->{db}, $src->{tbl}),
$args{changing_src} ? 'WRITE' : 'READ');
}
}
return;
}
sub _lock_dst {
my ( $self, %args ) = @_;
my @required_args = qw(lock_level host src);
my ($lock_level, $host, $src) = @args{@required_args};
my $o = $self->{OptionParser};
my $lock = $o->get('lock');
MKDEBUG && _d('Locking', $host->{Cxn}->name(), 'level', $lock);
# Wait for the dest to catchup to the source, then lock the dest.
# If there is any error beyond this point, we need to unlock/commit.
eval {
if ( my $timeout = $o->get('wait') ) {
if ( my $timeout = $args{wait} ) {
my $ms = $self->{MasterSlave};
my $wait;
my $tries = $args{wait_retry_args}->{tries} || 3;
my $wait;
$self->{Retry}->retry(
tries => $tries,
wait => sub { sleep 5; },
wait => sub { sleep $args{wait_retry_args}->{wait} || 10 },
try => sub {
my ( %args ) = @_;
# Be careful using $args{...} in this callback! %args in
# here are the passed-in args, not the args to the sub.
# here are the passed-in args, not the args to lock_and_wait().
if ( $args{tryno} > 1 ) {
warn "Retrying MASTER_POS_WAIT() for --wait $timeout...";
@@ -549,8 +534,8 @@ sub _lock_dst {
# because the main dbh might be in use due to executing
# $src_sth.
$wait = $ms->wait_for_master(
master_status => $ms->get_master_status($src->{Cxn}->aux_dbh()),
slave_dbh => $host->{Cxn}->dbh(),
master_status => $ms->get_master_status($src->{misc_dbh}),
slave_dbh => $dst->{dbh},
timeout => $timeout,
);
if ( defined $wait->{result} && $wait->{result} != -1 ) {
@@ -607,24 +592,27 @@ sub _lock_dst {
'(syncing via replication or sync-to-master)');
}
else {
if ( $lock == 3 ) {
if ( $args{lock} == 3 ) {
my $sql = 'FLUSH TABLES WITH READ LOCK';
MKDEBUG && _d($host->{Cxn}->name(), $sql);
$host->{Cxn}->dbh()->do($sql);
MKDEBUG && _d($dst->{dbh}, ',', $sql);
$dst->{dbh}->do($sql);
}
elsif ( !$o->get('transaction') ) {
$self->lock_table(
host => $host,
mode => 'READ', # $args{execute} ? 'WRITE' : 'READ')
);
elsif ( !$args{transaction} ) {
$self->lock_table($dst->{dbh}, 'dest',
$self->{Quoter}->quote($dst->{db}, $dst->{tbl}),
$args{execute} ? 'WRITE' : 'READ');
}
}
};
if ( $EVAL_ERROR ) {
# Must abort/unlock/commit so that we don't interfere with any further
# tables we try to do.
foreach my $dbh ( $host->{Cxn}->dbh(), $src->{Cxn}->dbh() ) {
MKDEBUG && _d('Caught error, unlocking/committing', $dbh);
if ( $args{src_sth}->{Active} ) {
$args{src_sth}->finish();
}
foreach my $dbh ( $src->{dbh}, $dst->{dbh}, $src->{misc_dbh} ) {
next unless $dbh;
MKDEBUG && _d('Caught error, unlocking/committing on', $dbh);
$dbh->do('UNLOCK TABLES');
$dbh->commit() unless $dbh->{AutoCommit};
}
@@ -632,7 +620,7 @@ sub _lock_dst {
die $EVAL_ERROR;
}
return;
return $result;
}
# This query will check all needed privileges on the table without actually
@@ -663,7 +651,6 @@ sub have_all_privs {
return 0;
}
sub _d {
my ($package, undef, $line) = caller 0;
@_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }

922
t/lib/TableSyncer.t
View File

File diff suppressed because it is too large Load Diff

51
t/pt-table-sync/basics.t
View File

@@ -28,7 +28,7 @@ elsif ( !$slave_dbh ) {
plan skip_all => 'Cannot connect to sandbox slave';
}
else {
plan tests => 9;
plan tests => 17;
}
$sb->wipe_clean($master_dbh);
@@ -82,17 +82,56 @@ is_deeply(
'Synced OK with no alg'
);
$sb->load_file('master', 't/pt-table-sync/samples/before.sql');
$output = run('test1', 'test2', '--algorithms Stream --no-bin-log');
is($output, "INSERT INTO `test`.`test2`(`a`, `b`) VALUES ('1', 'en');
INSERT INTO `test`.`test2`(`a`, `b`) VALUES ('2', 'ca');", 'Basic Stream sync');
is_deeply(
query_slave('select * from test.test2'),
[ { a => 1, b => 'en' }, { a => 2, b => 'ca' } ],
'Synced OK with Stream'
);
$sb->load_file('master', 't/pt-table-sync/samples/before.sql');
$output = run('test1', 'test2', '--algorithms GroupBy --no-bin-log');
is($output, "INSERT INTO `test`.`test2`(`a`, `b`) VALUES ('1', 'en');
INSERT INTO `test`.`test2`(`a`, `b`) VALUES ('2', 'ca');", 'Basic GroupBy sync');
is_deeply(
query_slave('select * from test.test2'),
[ { a => 1, b => 'en' }, { a => 2, b => 'ca' } ],
'Synced OK with GroupBy'
);
$sb->load_file('master', 't/pt-table-sync/samples/before.sql');
$output = run('test1', 'test2', '--algorithms Chunk,GroupBy --no-bin-log');
is($output, "INSERT INTO `test`.`test2`(`a`, `b`) VALUES ('1', 'en');
INSERT INTO `test`.`test2`(`a`, `b`) VALUES ('2', 'ca');", 'Basic Chunk sync');
is_deeply(
query_slave('select * from test.test2'),
[ { a => 1, b => 'en' }, { a => 2, b => 'ca' } ],
'Synced OK with Chunk'
);
$sb->load_file('master', 't/pt-table-sync/samples/before.sql');
$output = run('test1', 'test2', '--algorithms Nibble --no-bin-log');
is($output, "INSERT INTO `test`.`test2`(`a`, `b`) VALUES ('1', 'en');
INSERT INTO `test`.`test2`(`a`, `b`) VALUES ('2', 'ca');", 'Basic Nibble sync');
is_deeply(
query_slave('select * from test.test2'),
[ { a => 1, b => 'en' }, { a => 2, b => 'ca' } ],
'Synced OK with Nibble'
);
# Save original MKDEBUG env because we modify it below.
my $dbg = $ENV{MKDEBUG};
$sb->load_file('master', 't/pt-table-sync/samples/before.sql');
$ENV{MKDEBUG} = 1;
$output = run_cmd('test1', 'test2', '--no-bin-log --chunk-size 1 --transaction --lock 1');
$output = run_cmd('test1', 'test2', '--algorithms Nibble --no-bin-log --chunk-size 1 --transaction --lock 1');
delete $ENV{MKDEBUG};
# TODO: rewrite this poor test
like(
$output,
qr/START TRANSACTION/,
qr/Executing statement on source/,
'Nibble with transactions and locking'
);
is_deeply(
@@ -117,12 +156,12 @@ like(
);
like(
$output,
qr/2\s+\S+\s+\S+\s+2\s+test.test3/,
qr/2 Chunk\s+\S+\s+\S+\s+2\s+test.test3/,
'Right number of rows to update',
);
# Sync a table with Nibble and a chunksize in data size, not number of rows
$output = run('test3', 'test4', '--chunk-size 1k --print --no-bin-log --verbose --function MD5');
$output = run('test3', 'test4', '--algorithms Nibble --chunk-size 1k --print --verbose --function MD5');
# If it lived, it's OK.
ok($output, 'Synced with Nibble and data-size chunksize');

6
t/pt-table-sync/option_sanity.t
View File

@@ -19,13 +19,13 @@ my $output;
# #############################################################################
# Issue 111: Make mk-table-sync require --print or --execute or --explain
# Issue 111: Make mk-table-sync require --print or --execute or --dry-run
# #############################################################################
# This test reuses the test.message table created above for issue 22.
$output = `$trunk/bin/pt-table-sync h=127.1,P=12345,u=msandbox,p=msandbox,D=test,t=messages P=12346`;
like($output, qr/Specify at least one of --print, --execute or --explain/,
'Requires --print, --execute or --explain');
like($output, qr/Specify at least one of --print, --execute or --dry-run/,
'Requires --print, --execute or --dry-run');
# #############################################################################
# Don't let people try to restrict syncing with D=foo