File: mk-table-sync

package info (click to toggle)
maatkit 1877-1
links: PTS, VCS
area: main
in suites: lenny
size: 1,428 kB
ctags: 1,190
sloc: perl: 27,341; cpp: 70; makefile: 38
file content (5103 lines) | stat: -rwxr-xr-x 161,312 bytes
#!/usr/bin/perl

# This program synchronizes data efficiently between two MySQL tables, which
# can be on different servers.
#
# This program is copyright (c) 2007 Baron Schwartz.
# 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.

# TODO
# * make it more efficient.  Ideas: profile the Perl, use prepared statements,
#   don't use Perl to quote values, use arrays instead of hashes.
# * Add back top-down, bottom-up, sja algorithms for efficiency when locking
#   isn't an issue.
# * Add a feature to warn you if you're making changes on a slave.

our $VERSION = '1.0.6';
our $DISTRIB = '1877';
our $SVN_REV = sprintf("%d", map { $_ || 0 } q$Revision: 1870 $ =~ m/(\d+)/g);

use strict;
use warnings FATAL => 'all';

# ###########################################################################
# OptionParser package 1844
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package OptionParser;

use Getopt::Long;
use List::Util qw(max);
use English qw(-no_match_vars);

sub new {
   my ( $class, @opts ) = @_;
   my %key_seen;
   my %long_seen;
   my %key_for;
   my %defaults;
   my @mutex;
   my @atleast1;
   my %long_for;
   my %disables;
   my %copyfrom;
   unshift @opts,
      { s => 'help',    d => 'Show this help message' },
      { s => 'version', d => 'Output version information and exit' };
   foreach my $opt ( @opts ) {
      if ( ref $opt ) {
         my ( $long, $short ) = $opt->{s} =~ m/^([\w-]+)(?:\|([^!+=]*))?/;
         $opt->{k} = $short || $long;
         $key_for{$long} = $opt->{k};
         $long_for{$opt->{k}} = $long;
         $long_for{$long} = $long;
         $opt->{l} = $long;
         die "Duplicate option $opt->{k}" if $key_seen{$opt->{k}}++;
         die "Duplicate long option $opt->{l}" if $long_seen{$opt->{l}}++;
         $opt->{t} = $short;
         $opt->{n} = $opt->{s} =~ m/!/;
         $opt->{g} ||= 'o';
         if ( (my ($y) = $opt->{s} =~ m/=([mdHhAaz])/) ) {
            $ENV{MKDEBUG} && _d("Option $opt->{k} type: $y");
            $opt->{y} = $y;
            $opt->{s} =~ s/=./=s/;
         }
         if ( $opt->{d} =~ m/required/ ) {
            $opt->{r} = 1;
            $ENV{MKDEBUG} && _d("Option $opt->{k} is required");
         }
         if ( (my ($def) = $opt->{d} =~ m/default\b(?: ([^)]+))?/) ) {
            $defaults{$opt->{k}} = defined $def ? $def : 1;
            $ENV{MKDEBUG} && _d("Option $opt->{k} has a default");
         }
         if ( (my ($dis) = $opt->{d} =~ m/(disables .*)/) ) {
            $disables{$opt->{k}} = [ $class->get_participants($dis) ];
            $ENV{MKDEBUG} && _d("Option $opt->{k} $dis");
         }
      }
      else { # It's an instruction.

         if ( $opt =~ m/at least one|mutually exclusive|one and only one/ ) {
            my @participants = map {
                  die "No such option '$_' in $opt" unless $long_for{$_};
                  $long_for{$_};
               } $class->get_participants($opt);
            if ( $opt =~ m/mutually exclusive|one and only one/ ) {
               push @mutex, \@participants;
               $ENV{MKDEBUG} && _d(@participants, ' are mutually exclusive');
            }
            if ( $opt =~ m/at least one|one and only one/ ) {
               push @atleast1, \@participants;
               $ENV{MKDEBUG} && _d(@participants, ' require at least one');
            }
         }
         elsif ( $opt =~ m/default to/ ) {
            my @participants = map {
                  die "No such option '$_' in $opt" unless $long_for{$_};
                  $key_for{$_};
               } $class->get_participants($opt);
            $copyfrom{$participants[0]} = $participants[1];
            $ENV{MKDEBUG} && _d(@participants, ' copy from each other');
         }

      }
   }

   if ( $ENV{MKDEBUG} ) {
      my $text = do {
         local $RS = undef;
         open my $fh, "<", $PROGRAM_NAME
            or die "Can't open $PROGRAM_NAME: $OS_ERROR";
         <$fh>;
      };
      my %used = map { $_ => 1 } $text =~ m/\$opts\{'?([\w-]+)'?\}/g;
      my @unused;
      my @undefined;
      my %option_exists;
      foreach my $opt ( @opts ) {
         next unless ref $opt;
         my $key = $opt->{k};
         $option_exists{$key}++;
         next if $opt->{l} =~ m/^(?:help|version|defaults-file|database|charset
                                    |password|port|socket|user|host)$/x
              || $disables{$key};
         push @unused, $key unless $used{$key};
      }
      foreach my $key ( keys %used ) {
         push @undefined, $key unless $option_exists{$key};
      }
      if ( @unused || @undefined ) {
         die "The following command-line options are unused: "
            . join(',', @unused)
            . ' The following are undefined: '
            . join(',', @undefined);
      }
   }

   foreach my $dis ( keys %disables ) {
      $disables{$dis} = [ map {
            die "No such option '$_' while processing $dis" unless $long_for{$_};
            $long_for{$_};
         } @{$disables{$dis}} ];
   }

   return bless {
      specs => [ grep { ref $_ } @opts ],
      notes => [],
      instr => [ grep { !ref $_ } @opts ],
      mutex => \@mutex,
      defaults => \%defaults,
      long_for => \%long_for,
      atleast1 => \@atleast1,
      disables => \%disables,
      key_for  => \%key_for,
      copyfrom => \%copyfrom,
      strict   => 1,
      groups   => [ { k => 'o', d => 'Options' } ],
   }, $class;
}

sub get_participants {
   my ( $self, $str ) = @_;
   my @participants;
   foreach my $thing ( $str =~ m/(--?[\w-]+)/g ) {
      if ( (my ($long) = $thing =~ m/--(.+)/) ) {
         push @participants, $long;
      }
      else {
         foreach my $short ( $thing =~ m/([^-])/g ) {
            push @participants, $short;
         }
      }
   }
   $ENV{MKDEBUG} && _d("Participants for $str: ", @participants);
   return @participants;
}

sub parse {
   my ( $self, %defaults ) = @_;
   my @specs = @{$self->{specs}};
   my %factor_for = (k => 1_024, M => 1_048_576, G => 1_073_741_824);

   my %opt_seen;
   my %vals = %{$self->{defaults}};
   @vals{keys %defaults} = values %defaults;
   foreach my $spec ( @specs ) {
      $vals{$spec->{k}} = undef unless defined $vals{$spec->{k}};
      $opt_seen{$spec->{k}} = 1;
   }

   foreach my $key ( keys %defaults ) {
      die "Cannot set default for non-existent option '$key'\n"
         unless $opt_seen{$key};
   }

   Getopt::Long::Configure('no_ignore_case', 'bundling');
   GetOptions( map { $_->{s} => \$vals{$_->{k}} } @specs )
      or $self->error('Error parsing options');

   if ( $vals{version} ) {
      my $prog = $self->prog;
      printf("%s  Ver %s Distrib %s Changeset %s\n",
         $prog, $main::VERSION, $main::DISTRIB, $main::SVN_REV);
      exit(0);
   }

   if ( @ARGV && $self->{strict} ) {
      $self->error("Unrecognized command-line options @ARGV");
   }

   foreach my $dis ( grep { defined $vals{$_} } keys %{$self->{disables}} ) {
      my @disses = map { $self->{key_for}->{$_} } @{$self->{disables}->{$dis}};
      $ENV{MKDEBUG} && _d("Unsetting options: ", @disses);
      @vals{@disses} = map { undef } @disses;
   }

   foreach my $spec ( grep { $_->{r} } @specs ) {
      if ( !defined $vals{$spec->{k}} ) {
         $self->error("Required option --$spec->{l} must be specified");
      }
   }

   foreach my $mutex ( @{$self->{mutex}} ) {
      my @set = grep { defined $vals{$self->{key_for}->{$_}} } @$mutex;
      if ( @set > 1 ) {
         my $note = join(', ',
            map { "--$self->{long_for}->{$_}" }
                @{$mutex}[ 0 .. scalar(@$mutex) - 2] );
         $note .= " and --$self->{long_for}->{$mutex->[-1]}"
               . " are mutually exclusive.";
         $self->error($note);
      }
   }

   foreach my $required ( @{$self->{atleast1}} ) {
      my @set = grep { defined $vals{$self->{key_for}->{$_}} } @$required;
      if ( !@set ) {
         my $note = join(', ',
            map { "--$self->{long_for}->{$_}" }
                @{$required}[ 0 .. scalar(@$required) - 2] );
         $note .= " or --$self->{long_for}->{$required->[-1]}";
         $self->error("Specify at least one of $note");
      }
   }

   foreach my $spec ( grep { $_->{y} && defined $vals{$_->{k}} } @specs ) {
      my $val = $vals{$spec->{k}};
      if ( $spec->{y} eq 'm' ) {
         my ( $num, $suffix ) = $val =~ m/(\d+)([a-z])?$/;
         if ( !$suffix ) {
            my ( $s ) = $spec->{d} =~ m/\(suffix (.)\)/;
            $suffix = $s || 's';
            $ENV{MKDEBUG} && _d("No suffix given; using $suffix for $spec->{k} "
               . "(value: '$val')");
         }
         if ( $suffix =~ m/[smhd]/ ) {
            $val = $suffix eq 's' ? $num            # Seconds
                 : $suffix eq 'm' ? $num * 60       # Minutes
                 : $suffix eq 'h' ? $num * 3600     # Hours
                 :                  $num * 86400;   # Days
            $vals{$spec->{k}} = $val;
            $ENV{MKDEBUG} && _d("Setting option $spec->{k} to $val");
         }
         else {
            $self->error("Invalid --$spec->{l} argument");
         }
      }
      elsif ( $spec->{y} eq 'd' ) {
         $ENV{MKDEBUG} && _d("Parsing option $spec->{y} as a DSN");
         my $from_key = $self->{copyfrom}->{$spec->{k}};
         my $default = {};
         if ( $from_key ) {
            $ENV{MKDEBUG} && _d("Option $spec->{y} DSN copies from option $from_key");
            $default = $self->{dsn}->parse($self->{dsn}->as_string($vals{$from_key}));
         }
         $vals{$spec->{k}} = $self->{dsn}->parse($val, $default);
      }
      elsif ( $spec->{y} eq 'z' ) {
         my ($pre, $num, $factor) = $val =~ m/^([+-])?(\d+)([kMG])?$/;
         if ( defined $num ) {
            if ( $factor ) {
               $num *= $factor_for{$factor};
               $ENV{MKDEBUG} && _d("Setting option $spec->{y} to num * factor");
            }
            $vals{$spec->{k}} = ($pre || '') . $num;
         }
         else {
            $self->error("Invalid --$spec->{l} argument");
         }
      }
   }

   foreach my $spec ( grep { $_->{y} } @specs ) {
      $ENV{MKDEBUG} && _d("Treating option $spec->{k} as a list");
      my $val = $vals{$spec->{k}};
      if ( $spec->{y} eq 'H' || (defined $val && $spec->{y} eq 'h') ) {
         $vals{$spec->{k}} = { map { $_ => 1 } split(',', ($val || '')) };
      }
      elsif ( $spec->{y} eq 'A' || (defined $val && $spec->{y} eq 'a') ) {
         $vals{$spec->{k}} = [ split(',', ($val || '')) ];
      }
   }

   return %vals;
}

sub error {
   my ( $self, $note ) = @_;
   $self->{__error__} = 1;
   push @{$self->{notes}}, $note;
}

sub prog {
   (my $prog) = $PROGRAM_NAME =~ m/([.A-Za-z-]+)$/;
   return $prog || $PROGRAM_NAME;
}

sub prompt {
   my ( $self ) = @_;
   my $prog   = $self->prog;
   my $prompt = $self->{prompt} || '<options>';
   return "Usage: $prog $prompt\n";
}

sub descr {
   my ( $self ) = @_;
   my $prog = $self->prog;
   my $descr  = $prog . ' ' . ($self->{descr} || '')
          . "  For more details, please use the --help option, "
          . "or try 'perldoc $prog' for complete documentation.";
   $descr = join("\n", $descr =~ m/(.{0,80})(?:\s+|$)/g);
   $descr =~ s/ +$//mg;
   return $descr;
}

sub usage_or_errors {
   my ( $self, %opts ) = @_;
   if ( $opts{help} ) {
      print $self->usage(%opts);
      exit(0);
   }
   elsif ( $self->{__error__} ) {
      print $self->errors();
      exit(0);
   }
}

sub errors {
   my ( $self ) = @_;
   my $usage = $self->prompt() . "\n";
   if ( (my @notes = @{$self->{notes}}) ) {
      $usage .= join("\n  * ", 'Errors in command-line arguments:', @notes) . "\n";
   }
   return $usage . "\n" . $self->descr();
}

sub usage {
   my ( $self, %vals ) = @_;
   my @specs = @{$self->{specs}};

   my $maxl = max(map { length($_->{l}) + ($_->{n} ? 4 : 0)} @specs);

   my $maxs = max(0,
      map { length($_->{l}) + ($_->{n} ? 4 : 0)}
      grep { $_->{t} } @specs);

   my $lcol = max($maxl, ($maxs + 3));
   my $rcol = 80 - $lcol - 6;
   my $rpad = ' ' x ( 80 - $rcol );

   $maxs = max($lcol - 3, $maxs);

   my $usage = $self->descr() . "\n" . $self->prompt();
   foreach my $g ( @{$self->{groups}} ) {
      $usage .= "\n$g->{d}:\n";
      foreach my $spec (
         sort { $a->{l} cmp $b->{l} } grep { $_->{g} eq $g->{k} } @specs )
      {
         my $long  = $spec->{n} ? "[no]$spec->{l}" : $spec->{l};
         my $short = $spec->{t};
         my $desc  = $spec->{d};
         if ( $spec->{y} && $spec->{y} eq 'm' ) {
            my ($s) = $desc =~ m/\(suffix (.)\)/;
            $s    ||= 's';
            $desc =~ s/\s+\(suffix .\)//;
            $desc .= ".  Optional suffix s=seconds, m=minutes, h=hours, "
                   . "d=days; if no suffix, $s is used.";
         }
         $desc = join("\n$rpad", grep { $_ } $desc =~ m/(.{0,$rcol})(?:\s+|$)/g);
         $desc =~ s/ +$//mg;
         if ( $short ) {
            $usage .= sprintf("  --%-${maxs}s -%s  %s\n", $long, $short, $desc);
         }
         else {
            $usage .= sprintf("  --%-${lcol}s  %s\n", $long, $desc);
         }
      }
   }

   if ( (my @instr = @{$self->{instr}}) ) {
      $usage .= join("\n", map { "  $_" } @instr) . "\n";
   }
   if ( $self->{dsn} ) {
      $usage .= "\n" . $self->{dsn}->usage();
   }
   $usage .= "\nOptions and values after processing arguments:\n";
   foreach my $spec ( sort { $a->{l} cmp $b->{l} } @specs ) {
      my $val   = $vals{$spec->{k}};
      my $type  = $spec->{y} || '';
      my $bool  = $spec->{s} =~ m/^[\w-]+(?:\|[\w-])?!?$/;
      $val      = $bool                     ? ( $val ? 'TRUE' : 'FALSE' )
                : !defined $val             ? '(No value)'
                : $type eq 'd'              ? $self->{dsn}->as_string($val)
                : $type =~ m/H|h/           ? join(',', sort keys %$val)
                : $type =~ m/A|a/           ? join(',', @$val)
                :                             $val;
      $usage .= sprintf("  --%-${lcol}s  %s\n", $spec->{l}, $val);
   }
   return $usage;
}

sub prompt_noecho {
   shift @_ if ref $_[0] eq __PACKAGE__;
   my ( $prompt ) = @_;
   local $OUTPUT_AUTOFLUSH = 1;
   print $prompt;
   my $response;
   eval {
      require Term::ReadKey;
      Term::ReadKey::ReadMode('noecho');
      chomp($response = <STDIN>);
      Term::ReadKey::ReadMode('normal');
      print "\n";
   };
   if ( $EVAL_ERROR ) {
      die "Cannot read response; is Term::ReadKey installed? $EVAL_ERROR";
   }
   return $response;
}

sub groups {
   my ( $self, @groups ) = @_;
   push @{$self->{groups}}, @groups;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# OptionParser:$line $PID ", @_, "\n";
}

if ( $ENV{MKDEBUG} ) {
   print '# ', $^X, ' ', $], "\n";
   my $uname = `uname -a`;
   if ( $uname ) {
      $uname =~ s/\s+/ /g;
      print "# $uname\n";
   }
   printf("# %s  Ver %s Distrib %s Changeset %s line %d\n",
      $PROGRAM_NAME, ($main::VERSION || ''), ($main::DISTRIB || ''),
      ($main::SVN_REV || ''), __LINE__);
}

1;

# ###########################################################################
# End OptionParser package
# ###########################################################################

# ###########################################################################
# Quoter package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package Quoter;

use English qw(-no_match_vars);

sub new {
   my ( $class ) = @_;
   bless {}, $class;
}

sub quote {
   my ( $self, @vals ) = @_;
   foreach my $val ( @vals ) {
      $val =~ s/`/``/g;
   }
   return join('.', map { '`' . $_ . '`' } @vals);
}

sub quote_val {
   my ( $self, @vals ) = @_;
   return join(', ',
      map {
         if ( defined $_ ) {
            $_ =~ s/(['\\])/\\$1/g;
            $_ eq '' || $_ =~ m/^0|\D/ ? "'$_'" : $_;
         }
         else {
            'NULL';
         }
      } @vals
   );
}

1;

# ###########################################################################
# End Quoter package
# ###########################################################################

# ###########################################################################
# DSNParser package 1868
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package DSNParser;

use DBI;
use Data::Dumper;
$Data::Dumper::Indent    = 0;
$Data::Dumper::Quotekeys = 0;
use English qw(-no_match_vars);

sub new {
   my ( $class, @opts ) = @_;
   my $self = {
      opts => {
         A => {
            desc => 'Default character set',
            dsn  => 'charset',
            copy => 1,
         },
         D => {
            desc => 'Database to use',
            dsn  => 'database',
            copy => 1,
         },
         F => {
            desc => 'Only read default options from the given file',
            dsn  => 'mysql_read_default_file',
            copy => 1,
         },
         h => {
            desc => 'Connect to host',
            dsn  => 'host',
            copy => 1,
         },
         p => {
            desc => 'Password to use when connecting',
            dsn  => 'password',
            copy => 1,
         },
         P => {
            desc => 'Port number to use for connection',
            dsn  => 'port',
            copy => 1,
         },
         S => {
            desc => 'Socket file to use for connection',
            dsn  => 'mysql_socket',
            copy => 1,
         },
         u => {
            desc => 'User for login if not current user',
            dsn  => 'user',
            copy => 1,
         },
      },
   };
   foreach my $opt ( @opts ) {
      $ENV{MKDEBUG} && _d('Adding extra property ' . $opt->{key});
      $self->{opts}->{$opt->{key}} = { desc => $opt->{desc}, copy => $opt->{copy} };
   }
   return bless $self, $class;
}

sub prop {
   my ( $self, $prop, $value ) = @_;
   if ( @_ > 2 ) {
      $ENV{MKDEBUG} && _d("Setting $prop property");
      $self->{$prop} = $value;
   }
   return $self->{$prop};
}

sub parse {
   my ( $self, $dsn, $prev, $defaults ) = @_;
   if ( !$dsn ) {
      $ENV{MKDEBUG} && _d('No DSN to parse');
      return;
   }
   $ENV{MKDEBUG} && _d("Parsing $dsn");
   $prev     ||= {};
   $defaults ||= {};
   my %vals;
   my %opts = %{$self->{opts}};
   if ( $dsn !~ m/=/ && (my $p = $self->prop('autokey')) ) {
      $ENV{MKDEBUG} && _d("Interpreting $dsn as $p=$dsn");
      $dsn = "$p=$dsn";
   }
   my %hash = map { m/^(.)=(.*)$/g } split(/,/, $dsn);
   foreach my $key ( keys %opts ) {
      $ENV{MKDEBUG} && _d("Finding value for $key");
      $vals{$key} = $hash{$key};
      if ( !defined $vals{$key} && defined $prev->{$key} && $opts{$key}->{copy} ) {
         $vals{$key} = $prev->{$key};
         $ENV{MKDEBUG} && _d("Copying value for $key from previous DSN");
      }
      if ( !defined $vals{$key} ) {
         $vals{$key} = $defaults->{$key};
         $ENV{MKDEBUG} && _d("Copying value for $key from defaults");
      }
   }
   foreach my $key ( keys %hash ) {
      die "Unrecognized DSN part '$key' in '$dsn'\n"
         unless exists $opts{$key};
   }
   if ( (my $required = $self->prop('required')) ) {
      foreach my $key ( keys %$required ) {
         die "Missing DSN part '$key' in '$dsn'\n" unless $vals{$key};
      }
   }
   return \%vals;
}

sub as_string {
   my ( $self, $dsn ) = @_;
   return $dsn unless ref $dsn;
   return join(',',
      map  { "$_=" . ($_ eq 'p' ? '...' : $dsn->{$_}) }
      grep { defined $dsn->{$_} && $self->{opts}->{$_} }
      sort keys %$dsn );
}

sub usage {
   my ( $self ) = @_;
   my $usage
      = "DSN syntax is key=value[,key=value...]  Allowable DSN keys:\n"
      . "  KEY  COPY  MEANING\n"
      . "  ===  ====  =============================================\n";
   my %opts = %{$self->{opts}};
   foreach my $key ( sort keys %opts ) {
      $usage .= "  $key    "
             .  ($opts{$key}->{copy} ? 'yes   ' : 'no    ')
             .  ($opts{$key}->{desc} || '[No description]')
             . "\n";
   }
   if ( (my $key = $self->prop('autokey')) ) {
      $usage .= "  If the DSN is a bareword, the word is treated as the '$key' key.\n";
   }
   return $usage;
}

sub get_cxn_params {
   my ( $self, $info ) = @_;
   my $dsn;
   my %opts = %{$self->{opts}};
   my $driver = $self->prop('dbidriver') || '';
   if ( $driver eq 'Pg' ) {
      $dsn = 'DBI:Pg:dbname=' . ( $info->{D} || '' ) . ';'
         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
                     grep { defined $info->{$_} }
                     qw(h P));
   }
   else {
      $dsn = 'DBI:mysql:' . ( $info->{D} || '' ) . ';'
         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
                     grep { defined $info->{$_} }
                     qw(F h P S A))
         . ';mysql_read_default_group=mysql';
   }
   $ENV{MKDEBUG} && _d($dsn);
   return ($dsn, $info->{u}, $info->{p});
}

sub get_dbh {
   my ( $self, $cxn_string, $user, $pass, $opts ) = @_;
   $opts ||= {};
   my $defaults = {
      AutoCommit        => 0,
      RaiseError        => 1,
      PrintError        => 0,
      mysql_enable_utf8 => ($cxn_string =~ m/charset=utf8/ ? 1 : 0),
   };
   @{$defaults}{ keys %$opts } = values %$opts;
   $ENV{MKDEBUG} && _d($cxn_string, ' ', $user, ' ', $pass, ' {',
      join(', ', map { "$_=>$defaults->{$_}" } keys %$defaults ), '}');
   my $dbh = DBI->connect($cxn_string, $user, $pass, $defaults);
   if ( my ($charset) = $cxn_string =~ m/charset=(\w+)/ ) {
      my $sql = "/*!40101 SET NAMES $charset*/";
      $ENV{MKDEBUG} && _d("$dbh: $sql");
      $dbh->do($sql);
      $ENV{MKDEBUG} && _d('Enabling charset for STDOUT');
      if ( $charset eq 'utf8' ) {
         binmode(STDOUT, ':utf8')
            or die "Can't binmode(STDOUT, ':utf8'): $OS_ERROR";
      }
      else {
         binmode(STDOUT) or die "Can't binmode(STDOUT): $OS_ERROR";
      }
   }
   my $setvars = $self->prop('setvars');
   if ( $setvars ) {
      my $sql = "SET $setvars";
      $ENV{MKDEBUG} && _d("$dbh: $sql");
      $dbh->do($sql);
   }
   $ENV{MKDEBUG} && _d('DBH info: ',
      $dbh,
      Dumper($dbh->selectrow_hashref(
         'SELECT DATABASE(), CONNECTION_ID(), VERSION()/*!50038 , @@hostname*/')),
      ' Connection info: ', ($dbh->{mysql_hostinfo} || 'undef'),
      ' Character set info: ',
      Dumper($dbh->selectall_arrayref(
         'SHOW VARIABLES LIKE "character_set%"', { Slice => {}})),
      ' $DBD::mysql::VERSION: ', $DBD::mysql::VERSION,
      ' $DBI::VERSION: ', $DBI::VERSION,
   );
   return $dbh;
}

sub get_hostname {
   my ( $self, $dbh ) = @_;
   if ( my ($host) = ($dbh->{mysql_hostinfo} || '') =~ m/^(\w+) via/ ) {
      return $host;
   }
   my ( $hostname, $one ) = $dbh->selectrow_array(
      'SELECT /*!50038 @@hostname, */ 1');
   return $hostname;
}

sub disconnect {
   my ( $self, $dbh ) = @_;
   $ENV{MKDEBUG} && $self->print_active_handles($dbh);
   $dbh->disconnect;
}

sub print_active_handles {
   my ( $self, $thing, $level ) = @_;
   $level ||= 0;
   printf("# Active %sh: %s %s %s\n", $thing->{Type}, "\t" x $level,
      $thing, ($thing->{Type} eq 'st' ? $thing->{Statement} || '' : ''));
   foreach my $handle ( grep {defined} @{ $thing->{ChildHandles} } ) {
      $self->print_active_handles->( $handle, $level + 1 );
   }
}

sub _d {
   my ( $line ) = (caller(0))[2];
   @_ = map { defined $_ ? $_ : 'undef' } @_;
   print "# DSNParser:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End DSNParser package
# ###########################################################################

# ###########################################################################
# VersionParser package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package VersionParser;

use English qw(-no_match_vars);

sub new {
   my ( $class ) = @_;
   bless {}, $class;
}

sub parse {
   my ( $self, $str ) = @_;
   my $result = sprintf('%03d%03d%03d', $str =~ m/(\d+)/g);
   $ENV{MKDEBUG} && _d("$str parses to $result");
   return $result;
}

sub version_ge {
   my ( $self, $dbh, $target ) = @_;
   if ( !$self->{$dbh} ) {
      $self->{$dbh} = $self->parse(
         $dbh->selectrow_array('SELECT VERSION()'));
   }
   my $result = $self->{$dbh} ge $self->parse($target) ? 1 : 0;
   $ENV{MKDEBUG} && _d("$self->{$dbh} ge $target: $result");
   return $result;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# VersionParser:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End VersionParser package
# ###########################################################################

# ###########################################################################
# TableSyncStream package 1755
# ###########################################################################

use strict;
use warnings FATAL => 'all';

package TableSyncStream;

use English qw(-no_match_vars);

sub new {
   my ( $class, %args ) = @_;
   foreach my $arg ( qw(handler cols) ) {
      die "I need a $arg argument" unless defined $args{$arg};
   }
   return bless { %args }, $class;
}

sub get_sql {
   my ( $self, %args ) = @_;
   return "SELECT "
      . join(', ', map { $args{quoter}->quote($_) } @{$self->{cols}})
      . ' FROM ' . $args{quoter}->quote(@args{qw(database table)})
      . ' WHERE ' . ( $args{where} || '1=1' );
}

sub same_row {
   my ( $self, $lr, $rr ) = @_;
}

sub not_in_right {
   my ( $self, $lr ) = @_;
   $self->{handler}->change('INSERT', $lr, $self->key_cols());
}

sub not_in_left {
   my ( $self, $rr ) = @_;
   $self->{handler}->change('DELETE', $rr, $self->key_cols());
}

sub done_with_rows {
   my ( $self ) = @_;
   $self->{done} = 1;
}

sub done {
   my ( $self ) = @_;
   return $self->{done};
}

sub key_cols {
   my ( $self ) = @_;
   return $self->{cols};
}

sub prepare {
   my ( $self, $dbh ) = @_;
}

sub pending_changes {
   my ( $self ) = @_;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# TableSyncStream:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End TableSyncStream package
# ###########################################################################

# ###########################################################################
# TableParser package 1841
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package TableParser;

use English qw(-no_match_vars);

sub new {
   bless {}, shift;
}

sub parse {
   my ( $self, $ddl, $opts ) = @_;

   if ( ref $ddl eq 'ARRAY' ) {
      if ( lc $ddl->[0] eq 'table' ) {
         $ddl = $ddl->[1];
      }
      else {
         return {
            engine => 'VIEW',
         };
      }
   }

   if ( $ddl !~ m/CREATE (?:TEMPORARY )?TABLE `/ ) {
      die "Cannot parse table definition; is ANSI quoting "
         . "enabled or SQL_QUOTE_SHOW_CREATE disabled?";
   }

   $ddl =~ s/(`[^`]+`)/\L$1/g;

   my ( $engine ) = $ddl =~ m/\) (?:ENGINE|TYPE)=(\w+)/;
   $ENV{MKDEBUG} && _d('Storage engine: ', $engine);

   my @defs = $ddl =~ m/^(\s+`.*?),?$/gm;
   my @cols = map { $_ =~ m/`([^`]+)`/g } @defs;
   $ENV{MKDEBUG} && _d('Columns: ' . join(', ', @cols));

   my %def_for;
   @def_for{@cols} = @defs;

   my (@nums, @null);
   my (%type_for, %is_nullable, %is_numeric, %is_autoinc);
   foreach my $col ( @cols ) {
      my $def = $def_for{$col};
      my ( $type ) = $def =~ m/`[^`]+`\s([a-z]+)/;
      die "Can't determine column type for $def" unless $type;
      $type_for{$col} = $type;
      if ( $type =~ m/(?:(?:tiny|big|medium|small)?int|float|double|decimal|year)/ ) {
         push @nums, $col;
         $is_numeric{$col} = 1;
      }
      if ( $def !~ m/NOT NULL/ ) {
         push @null, $col;
         $is_nullable{$col} = 1;
      }
      $is_autoinc{$col} = $def =~ m/AUTO_INCREMENT/i ? 1 : 0;
   }

   my %keys;
   foreach my $key ( $ddl =~ m/^  ((?:[A-Z]+ )?KEY .*)$/gm ) {

      if ( $engine !~ m/MEMORY|HEAP/ ) {
         $key =~ s/USING HASH/USING BTREE/;
      }

      my ( $type, $cols ) = $key =~ m/(?:USING (\w+))? \((.+)\)/;
      my ( $special ) = $key =~ m/(FULLTEXT|SPATIAL)/;
      $type = $type || $special || 'BTREE';
      if ( $opts->{mysql_version} && $opts->{mysql_version} lt '004001000'
         && $engine =~ m/HEAP|MEMORY/i )
      {
         $type = 'HASH'; # MySQL pre-4.1 supports only HASH indexes on HEAP
      }

      my ($name) = $key =~ m/(PRIMARY|`[^`]*`)/;
      my $unique = $key =~ m/PRIMARY|UNIQUE/ ? 1 : 0;
      my @cols   = grep { m/[^,]/ } split('`', $cols);
      $name      =~ s/`//g;
      $ENV{MKDEBUG} && _d("Index $name columns: " . join(', ', @cols));

      $keys{$name} = {
         colnames    => $cols,
         cols        => \@cols,
         unique      => $unique,
         is_col      => { map { $_ => 1 } @cols },
         is_nullable => scalar(grep { $is_nullable{$_} } @cols),
         type        => $type,
         name        => $name,
      };
   }

   return {
      cols           => \@cols,
      col_posn       => { map { $cols[$_] => $_ } 0..$#cols },
      is_col         => { map { $_ => 1 } @cols },
      null_cols      => \@null,
      is_nullable    => \%is_nullable,
      is_autoinc     => \%is_autoinc,
      keys           => \%keys,
      defs           => \%def_for,
      numeric_cols   => \@nums,
      is_numeric     => \%is_numeric,
      engine         => $engine,
      type_for       => \%type_for,
   };
}

sub sort_indexes {
   my ( $self, $tbl ) = @_;
   my @indexes
      = sort {
         (($a ne 'PRIMARY') <=> ($b ne 'PRIMARY'))
         || ( !$tbl->{keys}->{$a}->{unique} <=> !$tbl->{keys}->{$b}->{unique} )
         || ( $tbl->{keys}->{$a}->{is_nullable} <=> $tbl->{keys}->{$b}->{is_nullable} )
         || ( scalar(@{$tbl->{keys}->{$a}->{cols}}) <=> scalar(@{$tbl->{keys}->{$b}->{cols}}) )
      }
      grep {
         $tbl->{keys}->{$_}->{type} eq 'BTREE'
      }
      sort keys %{$tbl->{keys}};
   $ENV{MKDEBUG} && _d('Indexes sorted best-first: ' . join(', ', @indexes));
   return @indexes;
}

sub find_best_index {
   my ( $self, $tbl, $index ) = @_;
   my $best;
   if ( $index ) {
      ($best) = grep { uc $_ eq uc $index } keys %{$tbl->{keys}};
   }
   if ( !$best ) {
      if ( $index ) {
         die "Index '$index' does not exist in table";
      }
      else {
         ($best) = $self->sort_indexes($tbl);
      }
   }
   $ENV{MKDEBUG} && _d("Best index found is " . ($best || 'undef'));
   return $best;
}

sub find_possible_keys {
   my ( $self, $dbh, $database, $table, $quoter, $where ) = @_;
   return () unless $where;
   my $sql = 'EXPLAIN SELECT * FROM ' . $quoter->quote($database, $table)
      . ' WHERE ' . $where;
   $ENV{MKDEBUG} && _d($sql);
   my $expl = $dbh->selectrow_hashref($sql);
   $expl = { map { lc($_) => $expl->{$_} } keys %$expl };
   if ( $expl->{possible_keys} ) {
      $ENV{MKDEBUG} && _d("possible_keys=$expl->{possible_keys}");
      my @candidates = split(',', $expl->{possible_keys});
      my %possible   = map { $_ => 1 } @candidates;
      if ( $expl->{key} ) {
         $ENV{MKDEBUG} && _d("MySQL chose $expl->{key}");
         unshift @candidates, grep { $possible{$_} } split(',', $expl->{key});
         $ENV{MKDEBUG} && _d('Before deduping: ' . join(', ', @candidates));
         my %seen;
         @candidates = grep { !$seen{$_}++ } @candidates;
      }
      $ENV{MKDEBUG} && _d('Final list: ' . join(', ', @candidates));
      return @candidates;
   }
   else {
      $ENV{MKDEBUG} && _d('No keys in possible_keys');
      return ();
   }
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# TableParser:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End TableParser package
# ###########################################################################

# ###########################################################################
# RowDiff package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package RowDiff;

use English qw(-no_match_vars);

sub new {
   my ( $class, %args ) = @_;
   die "I need a dbh" unless $args{dbh};
   my $self = \%args;
   return bless $self, $class;
}

sub compare_sets {
   my ( $self, %args ) = @_;
   my ( $left, $right, $syncer, $tbl )
      = @args{qw(left right syncer tbl)};

   my ($lr, $rr);       # Current row from the left/right sources.

   do {

      if ( !$lr && $left->{Active} ) {
         $ENV{MKDEBUG} && _d('Fetching row from left');
         $lr = $left->fetchrow_hashref;
      }
      if ( !$rr && $right->{Active} ) {
         $ENV{MKDEBUG} && _d('Fetching row from right');
         $rr = $right->fetchrow_hashref;
      }

      my $cmp;
      if ( $lr && $rr ) {
         $cmp = $self->key_cmp($lr, $rr, $syncer->key_cols(), $tbl);
         $ENV{MKDEBUG} && _d('Key comparison on left and right: '
            . (defined $cmp ? $cmp : 'undef'));
      }
      if ( $lr || $rr ) {
         if ( $lr && $rr && defined $cmp && $cmp == 0 ) {
            $ENV{MKDEBUG} && _d('Left and right have the same key');
            $syncer->same_row($lr, $rr);
            $lr = $rr = undef; # Fetch another row from each side.
         }
         elsif ( !$rr || ( defined $cmp && $cmp < 0 ) ) {
            $ENV{MKDEBUG} && _d('Left is not in right');
            $syncer->not_in_right($lr);
            $lr = undef;
         }
         else {
            $ENV{MKDEBUG} && _d('Right is not in left');
            $syncer->not_in_left($rr);
            $rr = undef;
         }
      }
   } while ( $left->{Active} || $right->{Active} );
   $ENV{MKDEBUG} && _d('No more rows');
   $syncer->done_with_rows();
}

sub key_cmp {
   my ( $self, $lr, $rr, $key_cols, $tbl ) = @_;
   $ENV{MKDEBUG} && _d("Comparing keys using columns " . join(',', @$key_cols));
   foreach my $col ( @$key_cols ) {
      my $l = $lr->{$col};
      my $r = $rr->{$col};
      if ( !defined $l || !defined $r ) {
         $ENV{MKDEBUG} && _d("$col is not defined in both rows");
         return defined $l || -1;
      }
      else {
         if ($tbl->{is_numeric}->{$col} ) {   # Numeric column
            $ENV{MKDEBUG} && _d("$col is numeric");
            my $cmp = $l <=> $r;
            return $cmp unless $cmp == 0;
         }
         elsif ( $l ne $r ) {
            my $cmp;
            my $coll = $tbl->{collation_for}->{$col};
            if ( $coll && ( $coll ne 'latin1_swedish_ci'
                           || $l =~ m/[^\040-\177]/ || $r =~ m/[^\040-\177]/) ) {
               $ENV{MKDEBUG} && _d("Comparing $col via MySQL");
               $cmp = $self->db_cmp($coll, $l, $r);
            }
            else {
               $ENV{MKDEBUG} && _d("Comparing $col in lowercase");
               $cmp = lc $l cmp lc $r;
            }
            return $cmp unless $cmp == 0;
         }
      }
   }
   return 0;
}

sub db_cmp {
   my ( $self, $collation, $l, $r ) = @_;
   if ( !$self->{sth}->{$collation} ) {
      if ( !$self->{charset_for} ) {
         $ENV{MKDEBUG} && _d("Fetching collations from MySQL");
         my @collations = @{$self->{dbh}->selectall_arrayref(
            'SHOW COLLATION', {Slice => { collation => 1, charset => 1 }})};
         foreach my $collation ( @collations ) {
            $self->{charset_for}->{$collation->{collation}}
               = $collation->{charset};
         }
      }
      my $sql = "SELECT STRCMP(_$self->{charset_for}->{$collation}? COLLATE $collation, "
         . "_$self->{charset_for}->{$collation}? COLLATE $collation) AS res";
      $ENV{MKDEBUG} && _d($sql);
      $self->{sth}->{$collation} = $self->{dbh}->prepare($sql);
   }
   my $sth = $self->{sth}->{$collation};
   $sth->execute($l, $r);
   return $sth->fetchall_arrayref()->[0]->[0];
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# RowDiff:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End RowDiff package
# ###########################################################################

# ###########################################################################
# MySQLDump package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package MySQLDump;

use English qw(-no_match_vars);

( our $before = <<'EOF') =~ s/^   //gm;
   /*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
   /*!40101 SET @OLD_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS */;
   /*!40101 SET @OLD_COLLATION_CONNECTION=@@COLLATION_CONNECTION */;
   /*!40101 SET NAMES utf8 */;
   /*!40103 SET @OLD_TIME_ZONE=@@TIME_ZONE */;
   /*!40103 SET TIME_ZONE='+00:00' */;
   /*!40014 SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0 */;
   /*!40014 SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0 */;
   /*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;
   /*!40111 SET @OLD_SQL_NOTES=@@SQL_NOTES, SQL_NOTES=0 */;
EOF

( our $after = <<'EOF') =~ s/^   //gm;
   /*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;
   /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
   /*!40014 SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS */;
   /*!40014 SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS */;
   /*!40101 SET CHARACTER_SET_CLIENT=@OLD_CHARACTER_SET_CLIENT */;
   /*!40101 SET CHARACTER_SET_RESULTS=@OLD_CHARACTER_SET_RESULTS */;
   /*!40101 SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION */;
   /*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;
EOF

sub new {
   my ( $class ) = @_;
   my $self = bless {}, $class;
   return $self;
}

sub dump {
   my ( $self, $dbh, $quoter, $db, $tbl, $what ) = @_;

   if ( $what eq 'table' ) {
      my $ddl = $self->get_create_table($dbh, $quoter, $db, $tbl);
      if ( $ddl->[0] eq 'table' ) {
         return $before
            . 'DROP TABLE IF EXISTS ' . $quoter->quote($tbl) . ";\n"
            . $ddl->[1] . ";\n";
      }
      else {
         return 'DROP TABLE IF EXISTS ' . $quoter->quote($tbl) . ";\n"
            . '/*!50001 DROP VIEW IF EXISTS '
            . $quoter->quote($tbl) . "*/;\n/*!50001 "
            . $self->get_tmp_table($dbh, $quoter, $db, $tbl) . "*/;\n";
      }
   }
   elsif ( $what eq 'triggers' ) {
      my $trgs = $self->get_triggers($dbh, $quoter, $db, $tbl);
      if ( $trgs && @$trgs ) {
         my $result = $before . "\nDELIMITER ;;\n";
         foreach my $trg ( @$trgs ) {
            if ( $trg->{sql_mode} ) {
               $result .= "/*!50003 SET SESSION SQL_MODE=\"$trg->{sql_mode}\" */;;\n";
            }
            $result .= "/*!50003 CREATE */ ";
            if ( $trg->{definer} ) {
               my ( $user, $host )
                  = map { s/'/''/g; "'$_'"; }
                    split('@', $trg->{definer}, 2);
               $result .= "/*!50017 DEFINER=$user\@$host */ ";
            }
            $result .= sprintf("/*!50003 TRIGGER %s %s %s ON %s\nFOR EACH ROW %s */;;\n\n",
               $quoter->quote($trg->{trigger}),
               @{$trg}{qw(timing event)},
               $quoter->quote($trg->{table}),
               $trg->{statement});
         }
         $result .= "DELIMITER ;\n\n/*!50003 SET SESSION SQL_MODE=\@OLD_SQL_MODE */;\n\n";
         return $result;
      }
      else {
         return undef;
      }
   }
   elsif ( $what eq 'view' ) {
      my $ddl = $self->get_create_table($dbh, $quoter, $db, $tbl);
      return '/*!50001 DROP TABLE IF EXISTS ' . $quoter->quote($tbl) . "*/;\n"
         . '/*!50001 DROP VIEW IF EXISTS ' . $quoter->quote($tbl) . "*/;\n"
         . '/*!50001 ' . $ddl->[1] . "*/;\n";
   }
   else {
      die "You didn't say what to dump.";
   }
}

sub _use_db {
   my ( $self, $dbh, $quoter, $new ) = @_;
   if ( !$new ) {
      $ENV{MKDEBUG} && _d('No new DB to use');
      return;
   }
   my $sql = 'SELECT DATABASE()';
   $ENV{MKDEBUG} && _d($sql);
   my $curr = $dbh->selectrow_array($sql);
   if ( $curr && $new && $curr eq $new ) {
      $ENV{MKDEBUG} && _d('Current and new DB are the same');
      return $curr;
   }
   $sql = 'USE ' . $quoter->quote($new);
   $ENV{MKDEBUG} && _d($sql);
   $dbh->do($sql);
   return $curr;
}

sub get_create_table {
   my ( $self, $dbh, $quoter, $db, $tbl ) = @_;
   if ( !$self->{tables}->{$db}->{$tbl} ) {
      my $sql = '/*!40101 SET @OLD_SQL_MODE := @@SQL_MODE, '
         . '@@SQL_MODE := REPLACE(REPLACE(@@SQL_MODE, "ANSI_QUOTES", ""), ",,", ","), '
         . '@OLD_QUOTE := @@SQL_QUOTE_SHOW_CREATE, '
         . '@@SQL_QUOTE_SHOW_CREATE := 1 */';
      $ENV{MKDEBUG} && _d($sql);
      $dbh->do($sql);
      my $curr_db = $self->_use_db($dbh, $quoter, $db);
      $sql = "SHOW CREATE TABLE " . $quoter->quote($db, $tbl);
      $ENV{MKDEBUG} && _d($sql);
      my $href = $dbh->selectrow_hashref($sql);
      $self->_use_db($dbh, $quoter, $curr_db);
      $sql = '/*!40101 SET @@SQL_MODE := @OLD_SQL_MODE, '
         . '@@SQL_QUOTE_SHOW_CREATE := @OLD_QUOTE */';
      $ENV{MKDEBUG} && _d($sql);
      $dbh->do($sql);
      my ($key) = grep { m/create table/i } keys %$href;
      if ( $key ) {
         $ENV{MKDEBUG} && _d('This table is a base table');
         $self->{tables}->{$db}->{$tbl} = [ 'table', $href->{$key} ];
      }
      else {
         $ENV{MKDEBUG} && _d('This table is a view');
         ($key) = grep { m/create view/i } keys %$href;
         $self->{tables}->{$db}->{$tbl} = [ 'view', $href->{$key} ];
      }
   }
   return $self->{tables}->{$db}->{$tbl};
}

sub get_columns {
   my ( $self, $dbh, $quoter, $db, $tbl ) = @_;
   $ENV{MKDEBUG} && _d("Get columns for $db.$tbl");
   if ( !$self->{columns}->{$db}->{$tbl} ) {
      my $curr_db = $self->_use_db($dbh, $quoter, $db);
      my $sql = "SHOW COLUMNS FROM " . $quoter->quote($db, $tbl);
      $ENV{MKDEBUG} && _d($sql);
      my $cols = $dbh->selectall_arrayref($sql, { Slice => {} });
      $self->_use_db($dbh, $quoter, $curr_db);
      $self->{columns}->{$db}->{$tbl} = [
         map {
            my %row;
            @row{ map { lc $_ } keys %$_ } = values %$_;
            \%row;
         } @$cols
      ];
   }
   return $self->{columns}->{$db}->{$tbl};
}

sub get_tmp_table {
   my ( $self, $dbh, $quoter, $db, $tbl ) = @_;
   my $result = 'CREATE TABLE ' . $quoter->quote($tbl) . " (\n";
   $result .= join(",\n",
      map { '  ' . $quoter->quote($_->{field}) . ' ' . $_->{type} }
      @{$self->get_columns($dbh, $quoter, $db, $tbl)});
   $result .= "\n)";
   $ENV{MKDEBUG} && _d($result);
   return $result;
}

sub get_triggers {
   my ( $self, $dbh, $quoter, $db, $tbl ) = @_;
   if ( !$self->{triggers}->{$db} ) {
      $self->{triggers}->{$db} = {};
      my $sql = '/*!40101 SET @OLD_SQL_MODE := @@SQL_MODE, '
         . '@@SQL_MODE := REPLACE(REPLACE(@@SQL_MODE, "ANSI_QUOTES", ""), ",,", ","), '
         . '@OLD_QUOTE := @@SQL_QUOTE_SHOW_CREATE, '
         . '@@SQL_QUOTE_SHOW_CREATE := 1 */';
      $ENV{MKDEBUG} && _d($sql);
      $dbh->do($sql);
      $sql = "SHOW TRIGGERS FROM " . $quoter->quote($db);
      $ENV{MKDEBUG} && _d($sql);
      my $sth = $dbh->prepare($sql);
      $sth->execute();
      if ( $sth->rows ) {
         my $trgs = $sth->fetchall_arrayref({});
         foreach my $trg (@$trgs) {
            my %trg;
            @trg{ map { lc $_ } keys %$trg } = values %$trg;
            push @{ $self->{triggers}->{$db}->{ $trg{table} } }, \%trg;
         }
      }
      $sql = '/*!40101 SET @@SQL_MODE := @OLD_SQL_MODE, '
         . '@@SQL_QUOTE_SHOW_CREATE := @OLD_QUOTE */';
      $ENV{MKDEBUG} && _d($sql);
      $dbh->do($sql);
   }
   return $self->{triggers}->{$db}->{$tbl};
}

sub get_databases {
   my ( $self, $dbh, $quoter, $like ) = @_;
   if ( !$self->{databases} || $like ) {
      my $sql = 'SHOW DATABASES';
      my @params;
      if ( $like ) {
         $sql .= ' LIKE ?';
         push @params, $like;
      }
      my $sth = $dbh->prepare($sql);
      $ENV{MKDEBUG} && _d($sql, @params);
      $sth->execute( @params );
      my @dbs = map { $_->[0] } @{$sth->fetchall_arrayref()};
      $self->{databases} = \@dbs unless $like;
      return @dbs;
   }
   return @{$self->{databases}};
}

sub get_table_status {
   my ( $self, $dbh, $quoter, $db, $like ) = @_;
   if ( !$self->{table_status}->{$db} || $like ) {
      my $sql = "SHOW TABLE STATUS FROM " . $quoter->quote($db);
      my @params;
      if ( $like ) {
         $sql .= ' LIKE ?';
         push @params, $like;
      }
      $ENV{MKDEBUG} && _d($sql, @params);
      my $sth = $dbh->prepare($sql);
      $sth->execute(@params);
      my @tables = @{$sth->fetchall_arrayref({})};
      @tables = map {
         my %tbl; # Make a copy with lowercased keys
         @tbl{ map { lc $_ } keys %$_ } = values %$_;
         $tbl{engine} ||= $tbl{type} || $tbl{comment};
         delete $tbl{type};
         \%tbl;
      } @tables;
      $self->{table_status}->{$db} = \@tables unless $like;
      return @tables;
   }
   return @{$self->{table_status}->{$db}};
}

sub get_table_list {
   my ( $self, $dbh, $quoter, $db, $like ) = @_;
   if ( !$self->{table_list}->{$db} || $like ) {
      my $sql = "SHOW /*!50002 FULL*/ TABLES FROM " . $quoter->quote($db);
      my @params;
      if ( $like ) {
         $sql .= ' LIKE ?';
         push @params, $like;
      }
      $ENV{MKDEBUG} && _d($sql, @params);
      my $sth = $dbh->prepare($sql);
      $sth->execute(@params);
      my @tables = @{$sth->fetchall_arrayref()};
      @tables = map {
         my %tbl = (
            name   => $_->[0],
            engine => ($_->[1] || '') eq 'VIEW' ? 'VIEW' : '',
         );
         \%tbl;
      } @tables;
      $self->{table_list}->{$db} = \@tables unless $like;
      return @tables;
   }
   return @{$self->{table_list}->{$db}};
}


sub _d {
   my ( $line ) = (caller(0))[2];
   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; } @_;
   print "# MySQLDump:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End MySQLDump package
# ###########################################################################

# ###########################################################################
# ChangeHandler package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package ChangeHandler;

use English qw(-no_match_vars);

my $DUPE_KEY  = qr/Duplicate entry/;
our @ACTIONS  = qw(DELETE REPLACE INSERT UPDATE);

sub new {
   my ( $class, %args ) = @_;
   foreach my $arg ( qw(quoter database table sdatabase stable replace queue)
   ) {
      die "I need a $arg argument" unless defined $args{$arg};
   }
   my $self = { %args, map { $_ => [] } @ACTIONS };
   $self->{db_tbl}  = $self->{quoter}->quote(@args{qw(database table)});
   $self->{sdb_tbl} = $self->{quoter}->quote(@args{qw(sdatabase stable)});
   $self->{changes} = { map { $_ => 0 } @ACTIONS };
   return bless $self, $class;
}

sub fetch_back {
   my ( $self, $dbh ) = @_;
   $self->{fetch_back} = $dbh;
   $ENV{MKDEBUG} && _d('Will fetch rows from source when updating destination');
}

sub take_action {
   my ( $self, @sql ) = @_;
   $ENV{MKDEBUG} && _d('Calling subroutines on ', @sql);
   foreach my $action ( @{$self->{actions}} ) {
      $action->(@sql);
   }
}

sub change {
   my ( $self, $action, $row, $cols ) = @_;
   $ENV{MKDEBUG} && _d("$action where ", $self->make_where_clause($row, $cols));
   $self->{changes}->{
      $self->{replace} && $action ne 'DELETE' ? 'REPLACE' : $action
   }++;
   if ( $self->{queue} ) {
      $self->__queue($action, $row, $cols);
   }
   else {
      eval {
         my $func = "make_$action";
         $self->take_action($self->$func($row, $cols));
      };
      if ( $EVAL_ERROR =~ m/$DUPE_KEY/ ) {
         $ENV{MKDEBUG} && _d('Duplicate key violation; will queue and rewrite');
         $self->{queue}++;
         $self->{replace} = 1;
         $self->__queue($action, $row, $cols);
      }
      elsif ( $EVAL_ERROR ) {
         die $EVAL_ERROR;
      }
   }
}

sub __queue {
   my ( $self, $action, $row, $cols ) = @_;
   $ENV{MKDEBUG} && _d('Queueing change for later');
   if ( $self->{replace} ) {
      $action = $action eq 'DELETE' ? $action : 'REPLACE';
   }
   push @{$self->{$action}}, [ $row, $cols ];
}

sub process_rows {
   my ( $self, $queue_level ) = @_;
   my $error_count = 0;
   TRY: {
      if ( $queue_level && $queue_level < $self->{queue} ) { # see redo below!
         $ENV{MKDEBUG} && _d("Not processing now $queue_level<$self->{queue}");
         return;
      }

      my ($row, $cur_act);
      eval {
         foreach my $action ( @ACTIONS ) {
            my $func = "make_$action";
            my $rows = $self->{$action};
            $ENV{MKDEBUG} && _d(scalar(@$rows) . " to $action");
            $cur_act = $action;
            while ( @$rows ) {
               $row = shift @$rows;
               $self->take_action($self->$func(@$row));
            }
         }
         $error_count = 0;
      };
      if ( !$error_count++ && $EVAL_ERROR =~ m/$DUPE_KEY/ ) {
         $ENV{MKDEBUG}
            && _d('Duplicate key violation; re-queueing and rewriting');
         $self->{queue}++; # Defer rows to the very end
         $self->{replace} = 1;
         $self->__queue($cur_act, @$row);
         redo TRY;
      }
      elsif ( $EVAL_ERROR ) {
         die $EVAL_ERROR;
      }
   }
}

sub make_DELETE {
   my ( $self, $row, $cols ) = @_;
   return "DELETE FROM $self->{db_tbl} WHERE "
      . $self->make_where_clause($row, $cols)
      . ' LIMIT 1';
}

sub make_UPDATE {
   my ( $self, $row, $cols ) = @_;
   if ( $self->{replace} ) {
      return $self->make_row('REPLACE', $row, $cols);
   }
   my %in_where = map { $_ => 1 } @$cols;
   my $where = $self->make_where_clause($row, $cols);
   if ( my $dbh = $self->{fetch_back} ) {
      my $sql = "SELECT * FROM $self->{sdb_tbl} WHERE $where LIMIT 1";
      $ENV{MKDEBUG} && _d("Fetching data for UPDATE: $sql");
      my $res = $dbh->selectrow_hashref($sql);
      @{$row}{keys %$res} = values %$res;
      $cols = [sort keys %$res];
   }
   else {
      $cols = [ sort keys %$row ];
   }
   return "UPDATE $self->{db_tbl} SET "
      . join(', ', map {
            $self->{quoter}->quote($_)
            . '=' .  $self->{quoter}->quote_val($row->{$_})
         } grep { !$in_where{$_} } @$cols)
      . " WHERE $where LIMIT 1";
}

sub make_INSERT {
   my ( $self, $row, $cols ) = @_;
   if ( $self->{replace} ) {
      return $self->make_row('REPLACE', $row, $cols);
   }
   return $self->make_row('INSERT', $row, $cols);
}

sub make_REPLACE {
   my ( $self, $row, $cols ) = @_;
   return $self->make_row('REPLACE', $row, $cols);
}

sub make_row {
   my ( $self, $verb, $row, $cols ) = @_;
   my @cols = sort keys %$row;
   if ( my $dbh = $self->{fetch_back} ) {
      my $where = $self->make_where_clause($row, $cols);
      my $sql = "SELECT * FROM $self->{sdb_tbl} WHERE $where LIMIT 1";
      $ENV{MKDEBUG} && _d("Fetching data for UPDATE: $sql");
      my $res = $dbh->selectrow_hashref($sql);
      @{$row}{keys %$res} = values %$res;
      @cols = sort keys %$res;
   }
   return "$verb INTO $self->{db_tbl}("
      . join(', ', map { $self->{quoter}->quote($_) } @cols)
      . ') VALUES ('
      . $self->{quoter}->quote_val( @{$row}{@cols} )
      . ')';
}

sub make_where_clause {
   my ( $self, $row, $cols ) = @_;
   my @clauses = map {
      my $val = $row->{$_};
      my $sep = defined $val ? '=' : ' IS ';
      $self->{quoter}->quote($_) . $sep . $self->{quoter}->quote_val($val);
   } @$cols;
   return join(' AND ', @clauses);
}

sub get_changes {
   my ( $self ) = @_;
   return %{$self->{changes}};
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# ChangeHandler:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End ChangeHandler package
# ###########################################################################

# ###########################################################################
# TableChunker package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package TableChunker;

use English qw(-no_match_vars);
use POSIX qw(ceil);
use List::Util qw(min max);

sub new {
   my ( $class, %args ) = @_;
   die "I need a quoter" unless $args{quoter};
   bless { %args }, $class;
}

my $EPOCH      = '1970-01-01';
my %int_types  = map { $_ => 1 }
   qw( bigint date datetime int mediumint smallint time timestamp tinyint year );
my %real_types = map { $_ => 1 }
   qw( decimal double float );

sub find_chunk_columns {
   my ( $self, $table, $opts ) = @_;
   $opts ||= {};

   my %prefer;
   if ( $opts->{possible_keys} && @{$opts->{possible_keys}} ) {
      my $i = 1;
      %prefer = map { $_ => $i++ } @{$opts->{possible_keys}};
      $ENV{MKDEBUG} && _d("Preferred indexes for chunking: "
         . join(', ', @{$opts->{possible_keys}}));
   }

   my @candidate_cols;

   my @possible_keys = grep { $_->{type} eq 'BTREE' } values %{$table->{keys}};
   @possible_keys = sort {
      ($prefer{$a->{name}} || 9999) <=> ($prefer{$b->{name}} || 9999)
   } @possible_keys;
   $ENV{MKDEBUG} && _d('Possible keys in order: '
      . join(', ', map { $_->{name} } @possible_keys));

   my $can_chunk_exact = 0;
   if ($opts->{exact}) {
      @candidate_cols =
         grep {
            $int_types{$table->{type_for}->{$_}}
            || $real_types{$table->{type_for}->{$_}}
         }
         map  { $_->{cols}->[0] }
         grep { $_->{unique} && @{$_->{cols}} == 1 }
              @possible_keys;
      if ( @candidate_cols ) {
         $can_chunk_exact = 1;
      }
      $ENV{MKDEBUG} && _d('Exact chunkable: ' . join(', ', @candidate_cols));
   }

   if ( !@candidate_cols ) {
      @candidate_cols =
         grep {
            $int_types{$table->{type_for}->{$_}}
            || $real_types{$table->{type_for}->{$_}}
         }
         map { $_->{cols}->[0] }
         @possible_keys;
      $ENV{MKDEBUG} && _d('Inexact chunkable: ' . join(', ', @candidate_cols));
   }

   my @result;
   if ( !%prefer ) {
      $ENV{MKDEBUG} && _d('Ordering columns by order in tbl, PK first');
      if ( $table->{keys}->{PRIMARY} ) {
         my $pk_first_col = $table->{keys}->{PRIMARY}->{cols}->[0];
         @result = grep { $_ eq $pk_first_col } @candidate_cols;
         @candidate_cols = grep { $_ ne $pk_first_col } @candidate_cols;
      }
      my $i = 0;
      my %col_pos = map { $_ => $i++ } @{$table->{cols}};
      push @result, sort { $col_pos{$a} <=> $col_pos{$b} } @candidate_cols;
   }
   else {
      @result = @candidate_cols;
   }
   $ENV{MKDEBUG} && _d('Chunkable columns: ' . join(', ', @result));
   $ENV{MKDEBUG} && _d("Can chunk exactly: $can_chunk_exact");

   return ($can_chunk_exact, \@result);
}

sub calculate_chunks {
   my ( $self, %args ) = @_;
   foreach my $arg ( qw(table col min max rows_in_range size dbh) ) {
      die "Required argument $arg not given or undefined"
         unless defined $args{$arg};
   }
   $ENV{MKDEBUG} && _d("Arguments: "
      . join(', ',
         map { "$_=" . (defined $args{$_} ? $args{$_} : 'undef') } keys %args));

   my @chunks;
   my ($range_func, $start_point, $end_point);
   my $col_type = $args{table}->{type_for}->{$args{col}};
   $ENV{MKDEBUG} && _d("Chunking on $args{col} ($col_type)");


   if ( $col_type =~ m/(?:int|year|float|double|decimal)$/ ) {
      $start_point = $args{min};
      $end_point   = $args{max};
      $range_func  = 'range_num';
   }
   elsif ( $col_type eq 'timestamp' ) {
      my $sql = "SELECT UNIX_TIMESTAMP('$args{min}'), UNIX_TIMESTAMP('$args{max}')";
      $ENV{MKDEBUG} && _d($sql);
      ($start_point, $end_point) = $args{dbh}->selectrow_array($sql);
      $range_func  = 'range_timestamp';
   }
   elsif ( $col_type eq 'date' ) {
      my $sql = "SELECT TO_DAYS('$args{min}'), TO_DAYS('$args{max}')";
      $ENV{MKDEBUG} && _d($sql);
      ($start_point, $end_point) = $args{dbh}->selectrow_array($sql);
      $range_func  = 'range_date';
   }
   elsif ( $col_type eq 'time' ) {
      my $sql = "SELECT TIME_TO_SEC('$args{min}'), TIME_TO_SEC('$args{max}')";
      $ENV{MKDEBUG} && _d($sql);
      ($start_point, $end_point) = $args{dbh}->selectrow_array($sql);
      $range_func  = 'range_time';
   }
   elsif ( $col_type eq 'datetime' ) {
      $start_point = $self->timestampdiff($args{dbh}, $args{min});
      $end_point   = $self->timestampdiff($args{dbh}, $args{max});
      $range_func  = 'range_datetime';
   }
   else {
      die "I don't know how to chunk $col_type\n";
   }

   if ( !defined $start_point ) {
      $ENV{MKDEBUG} && _d('Start point is undefined');
      $start_point = 0;
   }
   if ( !defined $end_point || $end_point < $start_point ) {
      $ENV{MKDEBUG} && _d('End point is undefined or before start point');
      $end_point = 0;
   }
   $ENV{MKDEBUG} && _d("Start and end of chunk range: $start_point, $end_point");

   my $interval = $args{size} * ($end_point - $start_point) / $args{rows_in_range};
   if ( $int_types{$col_type} ) {
      $interval = ceil($interval);
   }
   $interval ||= $args{size};
   if ( $args{exact} ) {
      $interval = $args{size};
   }
   $ENV{MKDEBUG} && _d("Chunk interval: $interval units");

   my $col = "`$args{col}`";
   if ( $start_point < $end_point ) {
      my ( $beg, $end );
      my $iter = 0;
      for ( my $i = $start_point; $i < $end_point; $i += $interval ) {
         ( $beg, $end ) = $self->$range_func($args{dbh}, $i, $interval, $end_point);

         if ( $iter++ == 0 ) {
            push @chunks, "$col < " . $self->quote($end);
         }
         else {
            push @chunks, "$col >= " . $self->quote($beg) . " AND $col < " . $self->quote($end);
         }
      }

      my $nullable = $args{table}->{is_nullable}->{$args{col}};
      pop @chunks;
      if ( @chunks ) {
         push @chunks, "$col >= " . $self->quote($beg);
      }
      else {
         push @chunks, $nullable ? "$col IS NOT NULL" : '1=1';
      }
      if ( $nullable ) {
         push @chunks, "$col IS NULL";
      }

   }
   else {
      push @chunks, '1=1';
   }

   return @chunks;
}

sub get_first_chunkable_column {
   my ( $self, $table, $opts ) = @_;
   my ($exact, $cols) = $self->find_chunk_columns($table, $opts);
   return $cols->[0];
}

sub size_to_rows {
   my ( $self, $dbh, $db, $tbl, $size, $dumper ) = @_;
  
   my ( $num, $suffix ) = $size =~ m/^(\d+)([MGk])?$/;
   if ( $suffix ) { # Convert to bytes.
      $size = $suffix eq 'k' ? $num * 1_024
            : $suffix eq 'M' ? $num * 1_024 * 1_024
            :                  $num * 1_024 * 1_024 * 1_024;
   }
   elsif ( $num ) {
      return $num;
   }
   else {
      die "Invalid size spec $size; must be an integer with optional suffix kMG";
   }

   my @status = $dumper->get_table_status($dbh, $self->{quoter}, $db);
   my ($status) = grep { $_->{name} eq $tbl } @status;
   my $avg_row_length = $status->{avg_row_length};
   return $avg_row_length ? ceil($size / $avg_row_length) : undef;
}

sub get_range_statistics {
   my ( $self, $dbh, $db, $tbl, $col, $where ) = @_;
   my $q = $self->{quoter};
   my $sql = "SELECT MIN(" . $q->quote($col) . "), MAX(" . $q->quote($col)
      . ") FROM " . $q->quote($db, $tbl)
      . ($where ? " WHERE $where" : '');
   $ENV{MKDEBUG} && _d($sql);
   my ( $min, $max ) = $dbh->selectrow_array($sql);
   $sql = "EXPLAIN SELECT * FROM " . $q->quote($db, $tbl)
      . ($where ? " WHERE $where" : '');
   $ENV{MKDEBUG} && _d($sql);
   my $expl = $dbh->selectrow_hashref($sql);
   return (
      min           => $min,
      max           => $max,
      rows_in_range => $expl->{rows},
   );
}

sub quote {
   my ( $self, $val ) = @_;
   return $val =~ m/\d[:-]/ ? qq{"$val"} : $val;
}

sub inject_chunks {
   my ( $self, %args ) = @_;
   foreach my $arg ( qw(database table chunks chunk_num query) ) {
      die "$arg is required" unless defined $args{$arg};
   }
   $ENV{MKDEBUG} && _d("Injecting chunk $args{chunk_num}");
   my $comment = sprintf("/*%s.%s:%d/%d*/",
      $args{database}, $args{table},
      $args{chunk_num} + 1, scalar @{$args{chunks}});
   $args{query} =~ s!/\*PROGRESS_COMMENT\*/!$comment!;
   my $where = "WHERE (" . $args{chunks}->[$args{chunk_num}] . ')';
   if ( $args{where} ) {
      $where .= " AND ($args{where})";
   }
   $args{query} =~ s!/\*WHERE\*/! $where!;
   my $db_tbl = $self->{quoter}->quote(@args{qw(database table)});
   $args{query} =~ s!/\*DB_TBL\*/!$db_tbl!;
   $args{query} =~ s!/\*CHUNK_NUM\*/! $args{chunk_num} AS chunk_num,!;
   return $args{query};
}

sub range_num {
   my ( $self, $dbh, $start, $interval, $max ) = @_;
   my $end = min($max, $start + $interval);
   $start =~ s/\.(\d{5}).*$/.$1/;
   $end   =~ s/\.(\d{5}).*$/.$1/;
   if ( $end > $start ) {
      return ( $start, $end );
   }
   else {
      die "Chunk size is too small: $end !> $start\n";
   }
}

sub range_time {
   my ( $self, $dbh, $start, $interval, $max ) = @_;
   my $sql = "SELECT SEC_TO_TIME($start), SEC_TO_TIME(LEAST($max, $start + $interval))";
   $ENV{MKDEBUG} && _d($sql);
   return $dbh->selectrow_array($sql);
}

sub range_date {
   my ( $self, $dbh, $start, $interval, $max ) = @_;
   my $sql = "SELECT FROM_DAYS($start), FROM_DAYS(LEAST($max, $start + $interval))";
   $ENV{MKDEBUG} && _d($sql);
   return $dbh->selectrow_array($sql);
}

sub range_datetime {
   my ( $self, $dbh, $start, $interval, $max ) = @_;
   my $sql = "SELECT DATE_ADD('$EPOCH', INTERVAL $start SECOND), "
       . "DATE_ADD('$EPOCH', INTERVAL LEAST($max, $start + $interval) SECOND)";
   $ENV{MKDEBUG} && _d($sql);
   return $dbh->selectrow_array($sql);
}

sub range_timestamp {
   my ( $self, $dbh, $start, $interval, $max ) = @_;
   my $sql = "SELECT FROM_UNIXTIME($start), FROM_UNIXTIME(LEAST($max, $start + $interval))";
   $ENV{MKDEBUG} && _d($sql);
   return $dbh->selectrow_array($sql);
}

sub timestampdiff {
   my ( $self, $dbh, $time ) = @_;
   my $sql = "SELECT (TO_DAYS('$time') * 86400 + TIME_TO_SEC('$time')) "
      . "- TO_DAYS('$EPOCH 00:00:00') * 86400";
   my ( $diff ) = $dbh->selectrow_array($sql);
   $sql = "SELECT DATE_ADD('$EPOCH', INTERVAL $diff SECOND)";
   $ENV{MKDEBUG} && _d($sql);
   my ( $check ) = $dbh->selectrow_array($sql);
   die <<"   EOF"
   Incorrect datetime math: given $time, calculated $diff but checked to $check.
   This is probably because you are using a version of MySQL that overflows on
   large interval values to DATE_ADD().  If not, please report this as a bug.
   EOF
      unless $check eq $time;
   return $diff;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# TableChunker:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End TableChunker package
# ###########################################################################

# ###########################################################################
# TableChecksum package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package TableChecksum;

use English qw(-no_match_vars);
use List::Util qw(max);

our %ALGOS = (
   CHECKSUM => { pref => 0, hash => 0 },
   BIT_XOR  => { pref => 2, hash => 1 },
   ACCUM    => { pref => 3, hash => 1 },
);

sub new {
   bless {}, shift;
}

sub get_crc_wid {
   my ( $self, $dbh, $func ) = @_;
   my $crc_wid = 16;
   if ( uc $func ne 'FNV_64' ) {
      eval {
         my ($val) = $dbh->selectrow_array("SELECT $func('a')");
         $crc_wid = max(16, length($val));
      };
   }
   return $crc_wid;
}

sub best_algorithm {
   my ( $self, %args ) = @_;
   my ($alg, $vp, $dbh) = @args{ qw(algorithm vp dbh) };
   my @choices = sort { $ALGOS{$a}->{pref} <=> $ALGOS{$b}->{pref} } keys %ALGOS;
   die "Invalid checksum algorithm $alg"
      if $alg && !$ALGOS{$alg};

   if (
      $args{where} || $args{chunk}        # CHECKSUM does whole table
      || $args{replicate}                 # CHECKSUM can't do INSERT.. SELECT
      || !$vp->version_ge($dbh, '4.1.1')) # CHECKSUM doesn't exist
   {
      $ENV{MKDEBUG} && _d('Cannot use CHECKSUM algorithm');
      @choices = grep { $_ ne 'CHECKSUM' } @choices;
   }

   if ( !$vp->version_ge($dbh, '4.1.1') ) {
      $ENV{MKDEBUG} && _d('Cannot use BIT_XOR algorithm');
      @choices = grep { $_ ne 'BIT_XOR' } @choices;
   }

   if ( $alg && grep { $_ eq $alg } @choices ) {
      $ENV{MKDEBUG} && _d("User requested $alg algorithm");
      return $alg;
   }

   if ( $args{count} && grep { $_ ne 'CHECKSUM' } @choices ) {
      $ENV{MKDEBUG} && _d('Not using CHECKSUM algorithm because COUNT desired');
      @choices = grep { $_ ne 'CHECKSUM' } @choices;
   }

   $ENV{MKDEBUG} && _d('Algorithms, in order: ', @choices);
   return $choices[0];
}

sub is_hash_algorithm {
   my ( $self, $algorithm ) = @_;
   return $ALGOS{$algorithm} && $ALGOS{$algorithm}->{hash};
}

sub choose_hash_func {
   my ( $self, %args ) = @_;
   my @funcs = qw(FNV_64 MD5 SHA1);
   if ( $args{func} ) {
      unshift @funcs, $args{func};
   }
   my ($result, $error);
   do {
      my $func;
      eval {
         $func = shift(@funcs);
         my $sql = "SELECT $func('test-string')";
         $ENV{MKDEBUG} && _d($sql);
         $args{dbh}->do($sql);
         $result = $func;
      };
      if ( $EVAL_ERROR && $EVAL_ERROR =~ m/failed: (.*?) at \S+ line/ ) {
         $error .= qq{$func cannot be used because "$1"\n};
         $ENV{MKDEBUG} && _d("$func cannot be used because $1");
      }
   } while ( @funcs && !$result );

   die $error unless $result;
   return $result;
}

sub optimize_xor {
   my ( $self, %args ) = @_;
   my ( $dbh, $func ) = @args{qw(dbh func)};

   die "FNV_64 never needs the BIT_XOR optimization" if uc $func eq 'FNV_64';

   my $opt_slice = 0;
   my $unsliced  = uc $dbh->selectall_arrayref("SELECT $func('a')")->[0]->[0];
   my $sliced    = '';
   my $start     = 1;
   my $crc_wid   = length($unsliced) < 16 ? 16 : length($unsliced);

   do { # Try different positions till sliced result equals non-sliced.
      $ENV{MKDEBUG} && _d("Trying slice $opt_slice");
      $dbh->do('SET @crc := "", @cnt := 0');
      my $slices = $self->make_xor_slices(
         query     => "\@crc := $func('a')",
         crc_wid   => $crc_wid,
         opt_slice => $opt_slice,
      );

      my $sql = "SELECT CONCAT($slices) AS TEST FROM (SELECT NULL) AS x";
      $sliced = ($dbh->selectrow_array($sql))[0];
      if ( $sliced ne $unsliced ) {
         $ENV{MKDEBUG} && _d("Slice $opt_slice does not work");
         $start += 16;
         ++$opt_slice;
      }
   } while ( $start < $crc_wid && $sliced ne $unsliced );

   if ( $sliced eq $unsliced ) {
      $ENV{MKDEBUG} && _d("Slice $opt_slice works");
      return $opt_slice;
   }
   else {
      $ENV{MKDEBUG} && _d("No slice works");
      return undef;
   }
}

sub make_xor_slices {
   my ( $self, %args ) = @_;
   my ( $query, $crc_wid, $opt_slice )
      = @args{qw(query crc_wid opt_slice)};

   my @slices;
   for ( my $start = 1; $start <= $crc_wid; $start += 16 ) {
      my $len = $crc_wid - $start + 1;
      if ( $len > 16 ) {
         $len = 16;
      }
      push @slices,
         "LPAD(CONV(BIT_XOR("
         . "CAST(CONV(SUBSTRING(\@crc, $start, $len), 16, 10) AS UNSIGNED))"
         . ", 10, 16), $len, '0')";
   }

   if ( defined $opt_slice && $opt_slice < @slices ) {
      $slices[$opt_slice] =~ s/\@crc/\@crc := $query/;
   }
   else {
      map { s/\@crc/$query/ } @slices;
   }

   return join(', ', @slices);
}

sub make_row_checksum {
   my ( $self, %args ) = @_;
   my ( $table, $quoter, $func )
      = @args{ qw(table quoter func) };

   my $sep = $args{sep} || '#';
   $sep =~ s/'//g;
   $sep ||= '#';

   my %cols = map { $_ => 1 } ($args{cols} ? @{$args{cols}} : @{$table->{cols}});
   my @cols =
      map {
         my $type = $table->{type_for}->{$_};
         my $result = $quoter->quote($_);
         if ( $type eq 'timestamp' ) {
            $result .= ' + 0';
         }
         elsif ( $type =~ m/float|double/ && $args{precision} ) {
            $result = "ROUND($result, $args{precision})";
         }
         $result;
      }
      grep {
         $cols{$_}
      }
      @{$table->{cols}};

   my $query;
   if ( uc $func ne 'FNV_64' ) {
      my @nulls = grep { $cols{$_} } @{$table->{null_cols}};
      if ( @nulls ) {
         my $bitmap = "CONCAT("
            . join(', ', map { 'ISNULL(' . $quoter->quote($_) . ')' } @nulls)
            . ")";
         push @cols, $bitmap;
      }

      $query = @cols > 1
             ? "$func(CONCAT_WS('$sep', " . join(', ', @cols) . '))'
             : "$func($cols[0])";
   }
   else {
      $query = 'FNV_64(' . join(', ', @cols) . ')';
   }

   return $query;
}

sub make_checksum_query {
   my ( $self, %args ) = @_;
   my ( $dbname, $tblname, $table, $quoter, $algorithm,
        $func, $crc_wid, $opt_slice )
      = @args{ qw(dbname tblname table quoter algorithm
        func crc_wid opt_slice) };
   die "Invalid or missing checksum algorithm"
      unless $algorithm && $ALGOS{$algorithm};

   my $result;

   if ( $algorithm eq 'CHECKSUM' ) {
      return "CHECKSUM TABLE " . $quoter->quote($dbname, $tblname);
   }

   my $expr = $self->make_row_checksum(%args);

   if ( $algorithm eq 'BIT_XOR' ) {
      if ( uc $func eq 'FNV_64' ) {
         $result = "LOWER(CONV(BIT_XOR(CAST($expr AS UNSIGNED)), 10, 16)) AS crc ";
      }
      else {
         my $slices = $self->make_xor_slices( query => $expr, %args );
         $result = "LOWER(CONCAT($slices)) AS crc ";
      }
   }
   else {
      if ( uc $func eq 'FNV_64' ) {
         $result = "RIGHT(MAX("
            . "\@crc := CONCAT(LPAD(\@cnt := \@cnt + 1, 16, '0'), "
            . "CONV(CAST(FNV_64(CONCAT(\@crc, $expr)) AS UNSIGNED), 10, 16))"
            . "), $crc_wid) AS crc ";
      }
      else {
         $result = "RIGHT(MAX("
            . "\@crc := CONCAT(LPAD(\@cnt := \@cnt + 1, 16, '0'), "
            . "$func(CONCAT(\@crc, $expr)))"
            . "), $crc_wid) AS crc ";
      }
   }
   if ( $args{replicate} ) {
      $result = "REPLACE /*PROGRESS_COMMENT*/ INTO $args{replicate} "
         . "(db, tbl, chunk, boundaries, this_cnt, this_crc) "
         . "SELECT ?, ?, /*CHUNK_NUM*/ ?, COUNT(*) AS cnt, $result";
   }
   else {
      $result = "SELECT /*PROGRESS_COMMENT*//*CHUNK_NUM*/ COUNT(*) AS cnt, $result";
   }
   return $result . "FROM /*DB_TBL*//*WHERE*/";
}

sub find_replication_differences {
   my ( $self, $dbh, $table ) = @_;

   (my $sql = <<"   EOF") =~ s/\s+/ /gm;
      SELECT db, tbl, chunk, boundaries,
         COALESCE(this_cnt-master_cnt, 0) AS cnt_diff,
         COALESCE(
            this_crc <> master_crc OR ISNULL(master_crc) <> ISNULL(this_crc),
            0
         ) AS crc_diff
      FROM $table
      WHERE master_cnt <> this_cnt OR master_crc <> this_crc
      OR ISNULL(master_crc) <> ISNULL(this_crc)
   EOF

   $ENV{MKDEBUG} && _d($sql);
   my $diffs = $dbh->selectall_arrayref($sql, { Slice => {} });
   return @$diffs;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# TableChecksum:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End TableChecksum package
# ###########################################################################

# ###########################################################################
# TableSyncChunk package 1868
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package TableSyncChunk;

use English qw(-no_match_vars);
use List::Util qw(max);
use Data::Dumper;
$Data::Dumper::Indent    = 0;
$Data::Dumper::Quotekeys = 0;

sub new {
   my ( $class, %args ) = @_;
   foreach my $arg ( qw(dbh database table handler chunker quoter struct
                        checksum cols vp chunksize where possible_keys
                        dumper) ) {
      die "I need a $arg argument" unless defined $args{$arg};
   }

   $args{crc_col} = '__crc';
   while ( $args{struct}->{is_col}->{$args{crc_col}} ) {
      $args{crc_col} = "_$args{crc_col}"; # Prepend more _ until not a column.
   }
   $ENV{MKDEBUG} && _d('CRC column will be named ' . $args{crc_col});

   my @chunks;
   my $col = $args{chunker}->get_first_chunkable_column(
      $args{struct}, { possible_keys => $args{possible_keys} });
   if ( $col ) {
      my %params = $args{chunker}->get_range_statistics(
         $args{dbh}, $args{database}, $args{table}, $col,
         $args{where});
      if ( !grep { !defined $params{$_} }
            qw(min max rows_in_range) )
      {
         @chunks = $args{chunker}->calculate_chunks(
            dbh      => $args{dbh},
            table    => $args{struct},
            col      => $col,
            size     => $args{chunksize},
            %params,
         );
      }
      else {
         @chunks = '1=1';
      }
      $args{chunk_col} = $col;
   }
   die "Cannot chunk $args{database}.$args{table}" unless @chunks;
   $args{chunks}     = \@chunks;
   $args{chunk_num}  = 0;

   $args{algorithm} = $args{checksum}->best_algorithm(
      algorithm   => 'BIT_XOR',
      vp          => $args{vp},
      dbh         => $args{dbh},
      where       => 1,
      chunk       => 1,
      count       => 1,
   );
   $args{func} = $args{checksum}->choose_hash_func(
      func => $args{func},
      dbh  => $args{dbh},
   );
   $args{crc_wid} = $args{checksum}->get_crc_wid($args{dbh}, $args{func});
   if ( $args{algorithm} eq 'BIT_XOR' && uc $args{func} ne 'FNV_64' ) {
      $args{opt_slice}
         = $args{checksum}->optimize_xor(dbh => $args{dbh}, func => $args{func});
   }
   $args{chunk_sql} ||= $args{checksum}->make_checksum_query(
      dbname    => $args{database},
      tblname   => $args{table},
      table     => $args{struct},
      quoter    => $args{quoter},
      algorithm => $args{algorithm},
      func      => $args{func},
      crc_wid   => $args{crc_wid},
      opt_slice => $args{opt_slice},
      cols      => $args{cols},
   );
   $args{row_sql} ||= $args{checksum}->make_row_checksum(
      table     => $args{struct},
      quoter    => $args{quoter},
      func      => $args{func},
      cols      => $args{cols},
   );

   $args{state} = 0;
   $args{handler}->fetch_back($args{dbh});
   return bless { %args }, $class;
}

sub get_sql {
   my ( $self, %args ) = @_;
   if ( $self->{state} ) {
      return 'SELECT '
         . join(', ', map { $self->{quoter}->quote($_) } @{$self->key_cols()})
         . ', ' . $self->{row_sql} . " AS $self->{crc_col}"
         . ' FROM ' . $self->{quoter}->quote(@args{qw(database table)})
         . ' WHERE (' . $self->{chunks}->[$self->{chunk_num}] . ')'
         . ($args{where} ? " AND ($args{where})" : '');
   }
   else {
      return $self->{chunker}->inject_chunks(
         database  => $args{database},
         table     => $args{table},
         chunks    => $self->{chunks},
         chunk_num => $self->{chunk_num},
         query     => $self->{chunk_sql},
         where     => $args{where},
         quoter    => $self->{quoter},
      );
   }
}

sub prepare {
   my ( $self, $dbh ) = @_;
   $dbh->do(q{SET @crc := ''});
}

sub same_row {
   my ( $self, $lr, $rr ) = @_;
   if ( $self->{state} ) {
      if ( $lr->{$self->{crc_col}} ne $rr->{$self->{crc_col}} ) {
         $self->{handler}->change('UPDATE', $lr, $self->key_cols());
      }
   }
   elsif ( $lr->{cnt} != $rr->{cnt} || $lr->{crc} ne $rr->{crc} ) {
      $ENV{MKDEBUG} && _d('Rows: ', Dumper($lr, $rr));
      $ENV{MKDEBUG} && _d('Will examine this chunk before moving to next');
      $self->{state} = 1; # Must examine this chunk row-by-row
   }
}

sub not_in_right {
   my ( $self, $lr ) = @_;
   die "Called not_in_right in state 0" unless $self->{state};
   $self->{handler}->change('INSERT', $lr, $self->key_cols());
}

sub not_in_left {
   my ( $self, $rr ) = @_;
   die "Called not_in_left in state 0" unless $self->{state};
   $self->{handler}->change('DELETE', $rr, $self->key_cols());
}

sub done_with_rows {
   my ( $self ) = @_;
   if ( $self->{state} == 1 ) {
      $self->{state} = 2;
      $ENV{MKDEBUG} && _d("Setting state=$self->{state}");
   }
   else {
      $self->{state} = 0;
      $self->{chunk_num}++;
      $ENV{MKDEBUG}
         && _d("Setting state=$self->{state}, chunk_num=$self->{chunk_num}");
   }
}

sub done {
   my ( $self ) = @_;
   $ENV{MKDEBUG}
      && _d("Done with $self->{chunk_num} of "
       . scalar(@{$self->{chunks}}) . ' chunks');
   $ENV{MKDEBUG} && $self->{state} && _d('Chunk differs; must examine rows');
   return $self->{state} == 0
      && $self->{chunk_num} >= scalar(@{$self->{chunks}})
}

sub pending_changes {
   my ( $self ) = @_;
   if ( $self->{state} ) {
      $ENV{MKDEBUG} && _d('There are pending changes');
      return 1;
   }
   else {
      $ENV{MKDEBUG} && _d('No pending changes');
      return 0;
   }
}

sub key_cols {
   my ( $self ) = @_;
   my @cols;
   if ( $self->{state} == 0 ) {
      @cols = qw(chunk_num);
   }
   else {
      @cols = $self->{chunk_col};
   }
   $ENV{MKDEBUG} && _d("State $self->{state}, key cols " . join(', ', @cols));
   return \@cols;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# TableSyncChunk:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End TableSyncChunk package
# ###########################################################################

# ###########################################################################
# TableSyncNibble package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package TableSyncNibble;

use English qw(-no_match_vars);
use List::Util qw(max);
use Data::Dumper;
$Data::Dumper::Indent    = 0;
$Data::Dumper::Quotekeys = 0;

sub new {
   my ( $class, %args ) = @_;
   foreach my $arg ( qw(dbh database table handler nibbler quoter struct
                        parser checksum cols vp chunksize where chunker
                        versionparser possible_keys) ) {
      die "I need a $arg argument" unless defined $args{$arg};
   }

   $args{crc_col} = '__crc';
   while ( $args{struct}->{is_col}->{$args{crc_col}} ) {
      $args{crc_col} = "_$args{crc_col}"; # Prepend more _ until not a column.
   }
   $ENV{MKDEBUG} && _d('CRC column will be named ' . $args{crc_col});

   $args{sel_stmt} = $args{nibbler}->generate_asc_stmt(
      parser   => $args{parser},
      tbl      => $args{struct},
      index    => $args{possible_keys}->[0],
      quoter   => $args{quoter},
      asconly  => 1,
   );

   die "No suitable index found"
      unless $args{sel_stmt}->{index}
         && $args{struct}->{keys}->{$args{sel_stmt}->{index}}->{unique};
   $args{key_cols} = $args{struct}->{keys}->{$args{sel_stmt}->{index}}->{cols};

   $args{algorithm} = $args{checksum}->best_algorithm(
      algorithm   => 'BIT_XOR',
      vp          => $args{vp},
      dbh         => $args{dbh},
      where       => 1,
      chunk       => 1,
      count       => 1,
   );
   $args{func} = $args{checksum}->choose_hash_func(
      dbh  => $args{dbh},
      func => $args{func},
   );
   $args{crc_wid} = $args{checksum}->get_crc_wid($args{dbh}, $args{func});
   if ( $args{algorithm} eq 'BIT_XOR' && uc $args{func} ne 'FNV_64' ) {
      $args{opt_slice}
         = $args{checksum}->optimize_xor(dbh => $args{dbh}, func => $args{func});
   }

   $args{nibble_sql} ||= $args{checksum}->make_checksum_query(
      dbname    => $args{database},
      tblname   => $args{table},
      table     => $args{struct},
      quoter    => $args{quoter},
      algorithm => $args{algorithm},
      func      => $args{func},
      crc_wid   => $args{crc_wid},
      opt_slice => $args{opt_slice},
      cols      => $args{cols},
   );
   $args{row_sql} ||= $args{checksum}->make_row_checksum(
      table     => $args{struct},
      quoter    => $args{quoter},
      func      => $args{func},
      cols      => $args{cols},
   );

   $args{state}  = 0;
   $args{nibble} = 0;
   $args{handler}->fetch_back($args{dbh});
   return bless { %args }, $class;
}

sub get_sql {
   my ( $self, %args ) = @_;
   if ( $self->{state} ) {
      return 'SELECT '
         . join(', ', map { $self->{quoter}->quote($_) } @{$self->key_cols()})
         . ', ' . $self->{row_sql} . " AS $self->{crc_col}"
         . ' FROM ' . $self->{quoter}->quote(@args{qw(database table)})
         . ' WHERE (' . $self->__get_boundaries() . ')'
         . ($args{where} ? " AND ($args{where})" : '');
   }
   else {
      my $where = $self->__get_boundaries();
      return $self->{chunker}->inject_chunks(
         database  => $args{database},
         table     => $args{table},
         chunks    => [$where],
         chunk_num => 0,
         query     => $self->{nibble_sql},
         where     => $args{where},
         quoter    => $self->{quoter},
      );
   }
}

sub __get_boundaries {
   my ( $self ) = @_;

   if ( $self->{cached_boundaries} ) {
      $ENV{MKDEBUG} && _d('Using cached boundaries');
      return $self->{cached_boundaries};
   }

   my $q = $self->{quoter};
   my $s = $self->{sel_stmt};
   my $row;
   my $lb; # Lower boundaries
   if ( $self->{cached_row} && $self->{cached_nibble} == $self->{nibble} ) {
      $ENV{MKDEBUG} && _d('Using cached row for boundaries');
      $row = $self->{cached_row};
   }
   else {
      my $sql      = 'SELECT '
         . join(',', map { $q->quote($_) } @{$s->{cols}})
         . " FROM " . $q->quote($self->{database}, $self->{table})
         . ($self->{versionparser}->version_ge($self->{dbh}, '4.0.9')
            ? " FORCE" : " USE")
         . " INDEX(" . $q->quote($s->{index}) . ")";
      if ( $self->{nibble} ) {
         my $tmp = $self->{cached_row};
         my $i   = 0;
         ($lb = $s->{boundaries}->{'>'})
            =~ s{([=><]) \?}
                {"$1 " . $q->quote_val($tmp->{$s->{scols}->[$i++]})}eg;
         $sql .= ' WHERE ' . $lb;
      }
      $sql .= ' LIMIT ' . ($self->{chunksize} - 1) . ', 1';
      $ENV{MKDEBUG} && _d($sql);
      $row = $self->{dbh}->selectrow_hashref($sql);
   }

   my $where;
   if ( $row ) {
      my $i = 0;
      ($where = $s->{boundaries}->{'<='})
         =~ s{([=><]) \?}{"$1 " . $q->quote_val($row->{$s->{scols}->[$i++]})}eg;
   }
   else {
      $where = '1=1';
   }

   if ( $lb ) {
      $where = "($lb AND $where)";
   }

   $self->{cached_row}        = $row;
   $self->{cached_nibble}     = $self->{nibble};
   $self->{cached_boundaries} = $where;

   $ENV{MKDEBUG} && _d('WHERE clause: ', $where);
   return $where;
}

sub prepare {
   my ( $self, $dbh ) = @_;
   $dbh->do(q{SET @crc := ''});
}

sub same_row {
   my ( $self, $lr, $rr ) = @_;
   if ( $self->{state} ) {
      if ( $lr->{$self->{crc_col}} ne $rr->{$self->{crc_col}} ) {
         $self->{handler}->change('UPDATE', $lr, $self->key_cols());
      }
   }
   elsif ( $lr->{cnt} != $rr->{cnt} || $lr->{crc} ne $rr->{crc} ) {
      $ENV{MKDEBUG} && _d('Rows: ', Dumper($lr, $rr));
      $ENV{MKDEBUG} && _d('Will examine this nibble before moving to next');
      $self->{state} = 1; # Must examine this nibble row-by-row
   }
}

sub not_in_right {
   my ( $self, $lr ) = @_;
   die "Called not_in_right in state 0" unless $self->{state};
   $self->{handler}->change('INSERT', $lr, $self->key_cols());
}

sub not_in_left {
   my ( $self, $rr ) = @_;
   die "Called not_in_left in state 0" unless $self->{state};
   $self->{handler}->change('DELETE', $rr, $self->key_cols());
}

sub done_with_rows {
   my ( $self ) = @_;
   if ( $self->{state} == 1 ) {
      $self->{state} = 2;
      $ENV{MKDEBUG} && _d("Setting state=$self->{state}");
   }
   else {
      $self->{state} = 0;
      $self->{nibble}++;
      delete $self->{cached_boundaries};
      $ENV{MKDEBUG}
         && _d("Setting state=$self->{state}, nibble=$self->{nibble}");
   }
}

sub done {
   my ( $self ) = @_;
   $ENV{MKDEBUG}
      && _d("Done with nibble $self->{nibble}");
   $ENV{MKDEBUG} && $self->{state} && _d('Nibble differs; must examine rows');
   return $self->{state} == 0 && $self->{nibble} && !$self->{cached_row};
}

sub pending_changes {
   my ( $self ) = @_;
   if ( $self->{state} ) {
      $ENV{MKDEBUG} && _d('There are pending changes');
      return 1;
   }
   else {
      $ENV{MKDEBUG} && _d('No pending changes');
      return 0;
   }
}

sub key_cols {
   my ( $self ) = @_;
   my @cols;
   if ( $self->{state} == 0 ) {
      @cols = qw(chunk_num);
   }
   else {
      @cols = @{$self->{key_cols}};
   }
   $ENV{MKDEBUG} && _d("State $self->{state}, key cols " . join(', ', @cols));
   return \@cols;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# TableSyncNibble:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End TableSyncNibble package
# ###########################################################################

# ###########################################################################
# TableSyncer package 1869
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package TableSyncer;

use English qw(-no_match_vars);

our %ALGOS = map { lc $_ => $_ } qw(Stream Chunk Nibble);

sub new {
   bless {}, shift;
}

sub best_algorithm {
   my ( $self, %args ) = @_;
   foreach my $arg ( qw(tbl_struct parser nibbler chunker) ) {
      die "I need a $arg argument" unless $args{$arg};
   }
   my $result;

   my ($exact, $cols) = $args{chunker}
      ->find_chunk_columns($args{tbl_struct}, { exact => 1 });
   if ( $exact ) {
      $ENV{MKDEBUG} && _d("Chunker says $cols->[0] supports chunking exactly");
      $result = 'Chunk';
   }
   else {
      my ($idx) = $args{parser}->find_best_index($args{tbl_struct});
      if ( $idx ) {
         $ENV{MKDEBUG} && _d("Parser found best index $idx, so Nibbler will work");
         $result = 'Nibble';
      }
      else {
         $ENV{MKDEBUG} && _d("No primary or unique non-null key in table");
         $result = 'Stream';
      }
   }
   $ENV{MKDEBUG} && _d("Algorithm: $result");
   return $result;
}

sub sync_table {
   my ( $self, %args ) = @_;
   foreach my $arg ( qw(
      buffer checksum chunker chunksize dst_db dst_dbh dst_tbl execute lock
      misc_dbh quoter replace replicate src_db src_dbh src_tbl test tbl_struct
      timeoutok transaction versionparser wait where possible_keys cols
      nibbler parser master_slave func dumper) )
   {
      die "I need a $arg argument" unless defined $args{$arg};
   }
   $ENV{MKDEBUG} && _d("Syncing table with args "
      . join(', ',
         map { "$_=" . (defined $args{$_} ? $args{$_} : 'undef') }
         sort keys %args));

   my $update_func;
   my $change_dbh;
   if ( $args{execute} ) {
      if ( $args{replicate} ) {
         $change_dbh = $args{src_dbh};
         $self->check_permissions(@args{qw(src_dbh src_db src_tbl quoter)});
      }
      else {
         $change_dbh = $args{dst_dbh};
         $self->check_permissions(@args{qw(dst_dbh dst_db dst_tbl quoter)});
      }
      $ENV{MKDEBUG} && _d('Will make changes via ' . $change_dbh);
      $update_func = sub {  map { $change_dbh->do($_) } @_ };
   }

   my $ch = new ChangeHandler(
      queue     => $args{buffer} ? 0 : 1,
      quoter    => $args{quoter},
      database  => $args{dst_db},
      table     => $args{dst_tbl},
      sdatabase => $args{src_db},
      stable    => $args{src_tbl},
      replace   => $args{replace} || $args{replicate},
      actions   => [
         ( $update_func ? $update_func            : () ),
         ( $args{print} ? sub { print @_, ";\n" } : () ),
      ],
   );
   my $rd = new RowDiff( dbh => $args{misc_dbh} );

   $args{algorithm} ||= $self->best_algorithm(
      map { $_ => $args{$_} } qw(tbl_struct parser nibbler chunker));

   if ( !$ALGOS{ lc $args{algorithm} } ) {
      die "No such algorithm $args{algorithm}; try one of "
         . join(', ', values %ALGOS) . "\n";
   }
   $args{algorithm} = $ALGOS{ lc $args{algorithm} };

   if ( $args{test} ) {
      return ($ch->get_changes(), ALGORITHM => $args{algorithm});
   }

   my $chunksize = $args{chunker}->size_to_rows(
         @args{qw(src_dbh src_db src_tbl chunksize dumper)}),

   my $class  = "TableSync$args{algorithm}";
   my $plugin = $class->new(
      handler   => $ch,
      cols      => $args{cols},
      dbh       => $args{src_dbh},
      database  => $args{src_db},
      dumper    => $args{dumper},
      table     => $args{src_tbl},
      chunker   => $args{chunker},
      nibbler   => $args{nibbler},
      parser    => $args{parser},
      struct    => $args{tbl_struct},
      checksum  => $args{checksum},
      vp        => $args{versionparser},
      quoter    => $args{quoter},
      chunksize => $chunksize,
      where     => $args{where},
      possible_keys => [],
      versionparser => $args{versionparser},
      func          => $args{func},
   );

   $self->lock_and_wait(%args, lock_level => 2);

   my $cycle = 0;
   while ( !$plugin->done ) {

      $ENV{MKDEBUG} && _d("Beginning sync cycle $cycle");
      my $src_sql = $plugin->get_sql(
         quoter   => $args{quoter},
         database => $args{src_db},
         table    => $args{src_tbl},
         where    => $args{where},
      );
      my $dst_sql = $plugin->get_sql(
         quoter   => $args{quoter},
         database => $args{dst_db},
         table    => $args{dst_tbl},
         where    => $args{where},
      );
      if ( $args{transaction} ) {
         if ( $change_dbh && $change_dbh eq $args{src_dbh} ) {
            $src_sql .= ' FOR UPDATE';
            $dst_sql .= ' LOCK IN SHARE MODE';
         }
         elsif ( $change_dbh ) {
            $src_sql .= ' LOCK IN SHARE MODE';
            $dst_sql .= ' FOR UPDATE';
         }
         else {
            $src_sql .= ' LOCK IN SHARE MODE';
            $dst_sql .= ' LOCK IN SHARE MODE';
         }
      }
      $plugin->prepare($args{src_dbh});
      $plugin->prepare($args{dst_dbh});
      $ENV{MKDEBUG} && _d("src: " . $src_sql);
      $ENV{MKDEBUG} && _d("dst: " . $dst_sql);
      my $src_sth = $args{src_dbh}
         ->prepare( $src_sql, { mysql_use_result => !$args{buffer} } );
      my $dst_sth = $args{dst_dbh}
         ->prepare( $dst_sql, { mysql_use_result => !$args{buffer} } );

      my $executed_src = 0;
      if ( !$cycle || !$plugin->pending_changes() ) {
         $executed_src
            = $self->lock_and_wait(%args, src_sth => $src_sth, lock_level => 1);
      }

      $src_sth->execute() unless $executed_src;
      $dst_sth->execute();

      $rd->compare_sets(
         left   => $src_sth,
         right  => $dst_sth,
         syncer => $plugin,
         tbl    => $args{tbl_struct},
      );
      $ENV{MKDEBUG} && _d("Finished sync cycle $cycle");
      $ch->process_rows(1);

      $cycle++;
   }

   $ch->process_rows();

   $self->unlock(%args, lock_level => 2);

   return ($ch->get_changes(), ALGORITHM => $args{algorithm});
}

sub check_permissions {
   my ( $self, $dbh, $db, $tbl, $quoter ) = @_;
   my $db_tbl = $quoter->quote($db, $tbl);
   my $sql = "REPLACE INTO $db_tbl SELECT * FROM $db_tbl LIMIT 0";
   $ENV{MKDEBUG} && _d('Permissions check: ', $sql);
   $dbh->do($sql);
}

sub lock_table {
   my ( $self, $dbh, $where, $db_tbl, $mode ) = @_;
   my $query = "LOCK TABLES $db_tbl $mode";
   $ENV{MKDEBUG} && _d($query);
   $dbh->do($query);
   $ENV{MKDEBUG} && _d("Acquired table lock on $where in $mode mode");
}

sub unlock {
   my ( $self, %args ) = @_;

   foreach my $arg ( qw(
      dst_db dst_dbh dst_tbl lock quoter replicate src_db src_dbh src_tbl
      timeoutok transaction wait lock_level) )
   {
      die "I need a $arg argument" unless defined $args{$arg};
   }

   return unless $args{lock} && $args{lock} <= $args{lock_level};

   foreach my $dbh( @args{qw(src_dbh dst_dbh)} ) {
      if ( $args{transaction} ) {
         $ENV{MKDEBUG} && _d("Committing $dbh");
         $dbh->commit;
      }
      else {
         my $sql = 'UNLOCK TABLES';
         $ENV{MKDEBUG} && _d($dbh, $sql);
         $dbh->do($sql);
      }
   }
}

sub lock_and_wait {
   my ( $self, %args ) = @_;
   my $result = 0;

   foreach my $arg ( qw(
      dst_db dst_dbh dst_tbl lock quoter replicate src_db src_dbh src_tbl
      timeoutok transaction wait lock_level misc_dbh master_slave) )
   {
      die "I need a $arg argument" unless defined $args{$arg};
   }

   return unless $args{lock} && $args{lock} == $args{lock_level};

   foreach my $dbh( @args{qw(src_dbh dst_dbh)} ) {
      if ( $args{transaction} ) {
         $ENV{MKDEBUG} && _d("Committing $dbh");
         $dbh->commit;
      }
      else {
         my $sql = 'UNLOCK TABLES';
         $ENV{MKDEBUG} && _d($dbh, $sql);
         $dbh->do($sql);
      }
   }

   if ( $args{lock} == 3 ) {
      my $sql = 'FLUSH TABLES WITH READ LOCK';
      $ENV{MKDEBUG} && _d("$args{src_dbh}, $sql");
      $args{src_dbh}->do($sql);
   }
   else {
      if ( $args{transaction} ) {
         if ( $args{src_sth} ) {
            $ENV{MKDEBUG} && _d('Executing statement on source to lock rows');
            $args{src_sth}->execute();
            $result = 1;
         }
      }
      else {
         $self->lock_table($args{src_dbh}, 'source',
            $args{quoter}->quote($args{src_db}, $args{src_tbl}),
            $args{replicate} ? 'WRITE' : 'READ');
      }
   }

   if ( $args{wait} ) {
      $args{master_slave}->wait_for_master(
         $args{misc_dbh}, $args{dst_dbh}, $args{wait}, $args{timeoutok});
   }

   if ( $args{replicate} ) {
      $ENV{MKDEBUG}
         && _d('Not locking destination because syncing via replication');
   }
   else {
      if ( $args{lock} == 3 ) {
         my $sql = 'FLUSH TABLES WITH READ LOCK';
         $ENV{MKDEBUG} && _d("$args{dst_dbh}, $sql");
         $args{dst_dbh}->do($sql);
      }
      elsif ( !$args{transaction} ) {
         $self->lock_table($args{dst_dbh}, 'dest',
            $args{quoter}->quote($args{dst_db}, $args{dst_tbl}),
            $args{execute} ? 'WRITE' : 'READ');
      }
   }

   return $result;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# TableSyncer:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End TableSyncer package
# ###########################################################################

# ###########################################################################
# MySQLFind package 1831
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package MySQLFind;

use Data::Dumper;
$Data::Dumper::Indent    = 0;
$Data::Dumper::Quotekeys = 0;
use English qw(-no_match_vars);


use English qw(-no_match_vars);

sub new {
   my ( $class, %args ) = @_;
   my $self = bless \%args, $class;
   map { die "I need a $_ argument" unless defined $args{$_} } qw(dumper quoter);
   die "Do not pass me a dbh argument" if $args{dbh};
   $self->{engines}->{views} = 1 unless defined $self->{engines}->{views};
   if ( $args{useddl} ) {
      $ENV{MKDEBUG} && _d('Will prefer DDL');
   }
   return $self;
}

sub init_timestamp {
   my ( $self, $dbh ) = @_;
   return if $self->{timestamp}->{$dbh}->{now};
   my $sql = 'SELECT CURRENT_TIMESTAMP';
   $ENV{MKDEBUG} && _d($sql);
   ($self->{timestamp}->{$dbh}->{now}) = $dbh->selectrow_array($sql);
   $ENV{MKDEBUG} && _d("Current timestamp: $self->{timestamp}->{$dbh}->{now}");
}

sub find_databases {
   my ( $self, $dbh ) = @_;
   return grep {
      $_ !~ m/^(information_schema|lost\+found)$/i
   }  $self->_filter('databases', sub { $_[0] },
         $self->{dumper}->get_databases(
            $dbh,
            $self->{quoter},
            $self->{databases}->{like}));
}

sub find_tables {
   my ( $self, $dbh, %args ) = @_;
   my $views = $self->{engines}->{views};
   my @tables 
      = $self->_filter('engines', sub { $_[0]->{engine} },
         $self->_filter('tables', sub { $_[0]->{name} },
            $self->_fetch_tbl_list($dbh, %args)));
   @tables = grep {
         ( $views || ($_->{engine} ne 'VIEW') )
      } @tables;
   map { $_->{name} =~ s/^[^.]*\.// } @tables; # <database>.<table> => <table> 
   foreach my $crit ( @{$self->{tables}->{status}} ) {
      my ($key, $test) = %$crit;
      @tables
         = grep {
            $self->_test_date($_, $key, $test, $dbh)
         } @tables;
   }
   return map { $_->{name} } @tables;
}

sub find_views {
   my ( $self, $dbh, %args ) = @_;
   my @tables = $self->_fetch_tbl_list($dbh, %args);
   @tables = grep { $_->{engine} eq 'VIEW' } @tables;
   map { $_->{name} =~ s/^[^.]*\.// } @tables; # <database>.<table> => <table> 
   return map { $_->{name} } @tables;
}

sub _use_db {
   my ( $self, $dbh, $new ) = @_;
   if ( !$new ) {
      $ENV{MKDEBUG} && _d('No new DB to use');
      return;
   }
   my $sql = 'SELECT DATABASE()';
   $ENV{MKDEBUG} && _d($sql);
   my $curr = $dbh->selectrow_array($sql);
   if ( $curr && $new && $curr eq $new ) {
      $ENV{MKDEBUG} && _d('Current and new DB are the same');
      return $curr;
   }
   $sql = 'USE ' . $self->{quoter}->quote($new);
   $ENV{MKDEBUG} && _d($sql);
   $dbh->do($sql);
   return $curr;
}

sub _fetch_tbl_list {
   my ( $self, $dbh, %args ) = @_;
   die "database is required" unless $args{database};
   my $curr_db = $self->_use_db($dbh, $args{database});
   my $need_engine = $self->{engines}->{permit}
        || $self->{engines}->{reject}
        || $self->{engines}->{regexp};
   my $need_status = $self->{tables}->{status};
   if ( $need_status || ($need_engine && !$self->{useddl}) ) {
      my @tables = $self->{dumper}->get_table_status(
         $dbh,
         $self->{quoter},
         $args{database},
         $self->{tables}->{like});
      @tables = map {
         my %hash = %$_;
         $hash{name} = join('.', $args{database}, $hash{name});
         \%hash;
      } @tables;
      return @tables;
   }
   else {
      my @result;
      my @tables = $self->{dumper}->get_table_list(
         $dbh,
         $self->{quoter},
         $args{database},
         $self->{tables}->{like});
      foreach my $tbl ( @tables ) {
         if ( $need_engine && !$tbl->{engine} ) {
            my $struct = $self->{parser}->parse(
               $self->{dumper}->get_create_table(
                  $dbh, $self->{quoter}, $args{database}, $tbl->{name}));
            $tbl->{engine} = $struct->{engine};
         }
         push @result,
         {  name   => join('.', $args{database}, $tbl->{name}),
            engine => $tbl->{engine},
         }
      }
      return @result;
   }
   $self->_use_db($dbh, $curr_db);
}

sub _filter {
   my ( $self, $thing, $sub, @vals ) = @_;
   $ENV{MKDEBUG} && _d("Filtering $thing list on ", Dumper($self->{$thing}));
   my $permit = $self->{$thing}->{permit};
   my $reject = $self->{$thing}->{reject};
   my $regexp = $self->{$thing}->{regexp};
   return grep {
      my $val = $sub->($_);
      $val = '' unless defined $val;
      if ( $thing eq 'tables' ) {
         (my $tbl = $val) =~ s/^.*\.//;
         ( !$reject || (!$reject->{$val} && !$reject->{$tbl}) )
            && ( !$permit || $permit->{$val} || $permit->{$tbl} )
            && ( !$regexp || $val =~ m/$regexp/ )
      }
      else {
         ( !$reject || !$reject->{$val} )
            && ( !$permit || $permit->{$val} )
            && ( !$regexp || $val =~ m/$regexp/ )
      }
   } @vals;
}

sub _test_date {
   my ( $self, $table, $prop, $test, $dbh ) = @_;
   $prop = lc $prop;
   if ( !defined $table->{$prop} ) {
      $ENV{MKDEBUG} && _d("$prop is not defined");
      return $self->{nullpass};
   }
   my ( $equality, $num ) = $test =~ m/^([+-])?(\d+)$/;
   die "Invalid date test $test for $prop" unless defined $num;
   $self->init_timestamp($dbh);
   my $sql = "SELECT DATE_SUB('$self->{timestamp}->{$dbh}->{now}', "
           . "INTERVAL $num SECOND)";
   $ENV{MKDEBUG} && _d($sql);
   ($self->{timestamp}->{$dbh}->{$num}) ||= $dbh->selectrow_array($sql);
   my $time = $self->{timestamp}->{$dbh}->{$num};
   return 
         ( $equality eq '-' && $table->{$prop} gt $time )
      || ( $equality eq '+' && $table->{$prop} lt $time )
      || (                     $table->{$prop} eq $time );
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# MySQLFind:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End MySQLFind package
# ###########################################################################

# ###########################################################################
# TableNibbler package 1755
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package TableNibbler;

use English qw(-no_match_vars);

sub new {
   bless {}, shift;
}

sub generate_asc_stmt {
   my ( $self, %args ) = @_;

   my $tbl  = $args{tbl};
   my @cols = $args{cols} ? @{$args{cols}} : @{$tbl->{cols}};
   my $q    = $args{quoter};

   my @asc_cols;
   my @asc_slice;

   my $index = $args{parser}->find_best_index($tbl, $args{index});
   die "Cannot find an ascendable index in table" unless $index;

   @asc_cols = @{$tbl->{keys}->{$index}->{cols}};
   $ENV{MKDEBUG} && _d("Will ascend index $index");
   $ENV{MKDEBUG} && _d("Will ascend columns " . join(', ', @asc_cols));
   if ( $args{ascfirst} ) {
      @asc_cols = $asc_cols[0];
      $ENV{MKDEBUG} && _d("Ascending only first column");
   }

   my %col_posn = do { my $i = 0; map { $_ => $i++ } @cols };
   foreach my $col ( @asc_cols ) {
      if ( !exists $col_posn{$col} ) {
         push @cols, $col;
         $col_posn{$col} = $#cols;
      }
      push @asc_slice, $col_posn{$col};
   }
   $ENV{MKDEBUG}
      && _d('Will ascend, in ordinal position: ' . join(', ', @asc_slice));

   my $asc_stmt = {
      cols  => \@cols,
      index => $index,
      where => '',
      slice => [],
      scols => [],
   };

   if ( @asc_slice ) {
      my $cmp_where;
      foreach my $cmp ( qw(< <= >= >) ) {
         $cmp_where = $self->generate_cmp_where(
            type        => $cmp,
            slice       => \@asc_slice,
            cols        => \@cols,
            quoter      => $q,
            is_nullable => $tbl->{is_nullable},
         );
         $asc_stmt->{boundaries}->{$cmp} = $cmp_where->{where};
      }
      my $cmp = $args{asconly} ? '>' : '>=';
      $asc_stmt->{where} = $asc_stmt->{boundaries}->{$cmp};
      $asc_stmt->{slice} = $cmp_where->{slice};
      $asc_stmt->{scols} = $cmp_where->{scols};
   }

   return $asc_stmt;
}

sub generate_cmp_where {
   my ( $self, %args ) = @_;
   foreach my $arg ( qw(type slice cols quoter is_nullable) ) {
      die "I need a $arg arg" unless defined $args{$arg};
   }

   my @slice       = @{$args{slice}};
   my @cols        = @{$args{cols}};
   my $q           = $args{quoter};
   my $is_nullable = $args{is_nullable};
   my $type        = $args{type};

   (my $cmp = $type) =~ s/=//;

   my @r_slice;    # Resulting slice columns, by ordinal
   my @r_scols;    # Ditto, by name

   my @clauses;
   foreach my $i ( 0 .. $#slice ) {
      my @clause;

      foreach my $j ( 0 .. $i - 1 ) {
         my $ord = $slice[$j];
         my $col = $cols[$ord];
         my $quo = $q->quote($col);
         if ( $is_nullable->{$col} ) {
            push @clause, "((? IS NULL AND $quo IS NULL) OR ($quo = ?))";
            push @r_slice, $ord, $ord;
            push @r_scols, $col, $col;
         }
         else {
            push @clause, "$quo = ?";
            push @r_slice, $ord;
            push @r_scols, $col;
         }
      }

      my $ord = $slice[$i];
      my $col = $cols[$ord];
      my $quo = $q->quote($col);
      my $end = $i == $#slice; # Last clause of the whole group.
      if ( $is_nullable->{$col} ) {
         if ( $type =~ m/=/ && $end ) {
            push @clause, "(? IS NULL OR $quo $type ?)";
         }
         elsif ( $type =~ m/>/ ) {
            push @clause, "((? IS NULL AND $quo IS NOT NULL) OR ($quo $cmp ?))";
         }
         else { # If $type =~ m/</ ) {
            push @clause, "((? IS NOT NULL AND $quo IS NULL) OR ($quo $cmp ?))";
         }
         push @r_slice, $ord, $ord;
         push @r_scols, $col, $col;
      }
      else {
         push @r_slice, $ord;
         push @r_scols, $col;
         push @clause, ($type =~ m/=/ && $end ? "$quo $type ?" : "$quo $cmp ?");
      }

      push @clauses, '(' . join(' AND ', @clause) . ')';
   }
   my $result = '(' . join(' OR ', @clauses) . ')';
   return {
      slice => \@r_slice,
      scols => \@r_scols,
      where => $result,
   };
}

sub generate_del_stmt {
   my ( $self, %args ) = @_;

   my $tbl  = $args{tbl};
   my @cols = $args{cols} ? @{$args{cols}} : ();
   my $q    = $args{quoter};

   my @del_cols;
   my @del_slice;

   my $index = $args{parser}->find_best_index($tbl, $args{index});
   die "Cannot find an ascendable index in table" unless $index;

   if ( $index ) {
      @del_cols = @{$tbl->{keys}->{$index}->{cols}};
   }
   else {
      @del_cols = @{$tbl->{cols}};
   }
   $ENV{MKDEBUG} && _d('Columns needed for DELETE: ' . join(', ', @del_cols));

   my %col_posn = do { my $i = 0; map { $_ => $i++ } @cols };
   foreach my $col ( @del_cols ) {
      if ( !exists $col_posn{$col} ) {
         push @cols, $col;
         $col_posn{$col} = $#cols;
      }
      push @del_slice, $col_posn{$col};
   }
   $ENV{MKDEBUG} && _d('Ordinals needed for DELETE: ' . join(', ', @del_slice));

   my $del_stmt = {
      cols  => \@cols,
      index => $index,
      where => '',
      slice => [],
      scols => [],
   };

   my @clauses;
   foreach my $i ( 0 .. $#del_slice ) {
      my $ord = $del_slice[$i];
      my $col = $cols[$ord];
      my $quo = $q->quote($col);
      if ( $tbl->{is_nullable}->{$col} ) {
         push @clauses, "((? IS NULL AND $quo IS NULL) OR ($quo = ?))";
         push @{$del_stmt->{slice}}, $ord, $ord;
         push @{$del_stmt->{scols}}, $col, $col;
      }
      else {
         push @clauses, "$quo = ?";
         push @{$del_stmt->{slice}}, $ord;
         push @{$del_stmt->{scols}}, $col;
      }
   }

   $del_stmt->{where} = '(' . join(' AND ', @clauses) . ')';

   return $del_stmt;
}

sub generate_ins_stmt {
   my ( $self, %args ) = @_;

   my $tbl  = $args{tbl};
   my @cols = @{$args{cols}};

   die "You didn't specify any columns" unless @cols;

   my %col_posn = do { my $i = 0; map { $_ => $i++ } @{$tbl->{cols}} };
   my @ins_cols;
   my @ins_slice;

   foreach my $col ( @cols ) {
      if ( exists $col_posn{$col} ) {
         push @ins_cols, $col;
         push @ins_slice, $col_posn{$col};
      }
   }

   my $ins_stmt = {
      cols  => \@ins_cols,
      slice => \@ins_slice,
   };

   return $ins_stmt;
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# TableNibbler:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End TableNibbler package
# ###########################################################################

# ###########################################################################
# MasterSlave package 1823
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package MasterSlave;

use English qw(-no_match_vars);
use List::Util qw(min max);

sub new {
   bless {}, shift;
}

sub recurse_to_slaves {
   my ( $self, $args, $level ) = @_;
   $level ||= 0;
   my $dp   = $args->{dsn_parser};
   my $dsn  = $args->{dsn};

   my $dbh;
   eval {
      $dbh = $args->{dbh} || $dp->get_dbh(
         $dp->get_cxn_params($dsn), { AutoCommit => 1 });
      $ENV{MKDEBUG} && _d('Connected to ', $dp->as_string($dsn));
   };
   if ( $EVAL_ERROR ) {
      print STDERR "Cannot connect to ", $dp->as_string($dsn), "\n";
      return;
   }

   my $sql  = 'SELECT @@SERVER_ID';
   $ENV{MKDEBUG} && _d($sql);
   my ($id) = $dbh->selectrow_array($sql);
   $ENV{MKDEBUG} && _d('Working on server ID ', $id);
   my $master_thinks_i_am = $dsn->{server_id};
   if ( !defined $id
       || ( defined $master_thinks_i_am && $master_thinks_i_am != $id )
       || $args->{server_ids_seen}->{$id}++
   ) {
      $ENV{MKDEBUG} && _d('Server ID seen, or not what master said');
      if ( $args->{skip_callback} ) {
         $args->{skip_callback}->($dsn, $dbh, $level, $args->{parent});
      }
      return;
   }

   $args->{callback}->($dsn, $dbh, $level, $args->{parent});

   if ( !defined $args->{recurse} || $level < $args->{recurse} ) {

      my @slaves =
         grep { !$_->{master_id} || $_->{master_id} == $id } # Only my slaves.
         $self->find_slave_hosts($dp, $dbh, $dsn, $args->{method});

      foreach my $slave ( @slaves ) {
         $ENV{MKDEBUG} && _d('Recursing from ',
            $dp->as_string($dsn), ' to ', $dp->as_string($slave));
         $self->recurse_to_slaves(
            { %$args, dsn => $slave, dbh => undef, parent => $dsn }, $level + 1 );
      }
   }
}

sub find_slave_hosts {
   my ( $self, $dsn_parser, $dbh, $dsn, $method ) = @_;
   $method ||= '';
   $ENV{MKDEBUG} && _d('Looking for slaves on ', $dsn_parser->as_string($dsn));

   my @slaves;

   if ( (!$method && ($dsn->{P}||3306) == 3306) || $method eq 'processlist' ) {

      my $proc =
         grep { m/ALL PRIVILEGES.*?\*\.\*|PROCESS/ }
         @{$dbh->selectcol_arrayref('SHOW GRANTS')};
      if ( !$proc ) {
         die "You do not have the PROCESS privilege";
      }

      my $sql = 'SHOW PROCESSLIST';
      $ENV{MKDEBUG} && _d($sql);
      @slaves =
         map  {
            my $slave        = $dsn_parser->parse("h=$_", $dsn);
            $slave->{source} = 'processlist';
            $slave;
         }
         grep { $_ }
         map  {
            my ( $host ) = $_->{host} =~ m/^([^:]+):/;
            if ( $host eq 'localhost' ) {
               $host = '127.0.0.1'; # Replication never uses sockets.
            }
            $host;
         }
         grep { $_->{command} =~ m/Binlog Dump/i }
         map  {
            my %hash;
            @hash{ map { lc $_ } keys %$_ } = values %$_;
            \%hash;
         }
         @{$dbh->selectall_arrayref($sql, { Slice => {} })};
   }

   if ( !@slaves ) {
      my $sql = 'SHOW SLAVE HOSTS';
      $ENV{MKDEBUG} && _d($sql);
      @slaves = @{$dbh->selectall_arrayref($sql, { Slice => {} })};

      if ( @slaves ) {
         $ENV{MKDEBUG} && _d('Found some SHOW SLAVE HOSTS info');
         @slaves = map {
            my %hash;
            @hash{ map { lc $_ } keys %$_ } = values %$_;
            my $spec = "h=$hash{host},P=$hash{port}"
               . ( $hash{user} ? ",u=$hash{user}" : '')
               . ( $hash{password} ? ",p=$hash{password}" : '');
            my $dsn           = $dsn_parser->parse($spec, $dsn);
            $dsn->{server_id} = $hash{server_id};
            $dsn->{master_id} = $hash{master_id};
            $dsn->{source}    = 'hosts';
            $dsn;
         } @slaves;
      }
   }

   $ENV{MKDEBUG} && _d('Found ', scalar(@slaves), ' slaves');
   return @slaves;
}

sub get_master_dsn {
   my ( $self, $dbh, $dsn, $dsn_parser ) = @_;
   my $master = $self->get_slave_status($dbh) or return undef;
   my $spec   = "h=$master->{master_host},P=$master->{master_port}";
   return       $dsn_parser->parse($spec, $dsn);
}

sub get_slave_status {
   my ( $self, $dbh ) = @_;
   if ( !$self->{not_a_slave}->{$dbh} ) {
      my $sth = $self->{sths}->{$dbh}->{SLAVE_STATUS}
            ||= $dbh->prepare('SHOW SLAVE STATUS');
      $ENV{MKDEBUG} && _d('SHOW SLAVE STATUS');
      $sth->execute();
      my ($ss) = @{$sth->fetchall_arrayref({})};

      if ( $ss && %$ss ) {
         $ss = { map { lc($_) => $ss->{$_} } keys %$ss }; # lowercase the keys
         return $ss;
      }

      $ENV{MKDEBUG} && _d('This server returns nothing for SHOW SLAVE STATUS');
      $self->{not_a_slave}->{$dbh}++;
   }
}

sub get_master_status {
   my ( $self, $dbh ) = @_;
   if ( !$self->{not_a_master}->{$dbh} ) {
      my $sth = $self->{sths}->{$dbh}->{MASTER_STATUS}
            ||= $dbh->prepare('SHOW MASTER STATUS');
      $ENV{MKDEBUG} && _d('SHOW MASTER STATUS');
      $sth->execute();
      my ($ms) = @{$sth->fetchall_arrayref({})};

      if ( $ms && %$ms ) {
         $ms = { map { lc($_) => $ms->{$_} } keys %$ms }; # lowercase the keys
         if ( $ms->{file} && $ms->{position} ) {
            return $ms;
         }
      }

      $ENV{MKDEBUG} && _d('This server returns nothing for SHOW MASTER STATUS');
      $self->{not_a_master}->{$dbh}++;
   }
}

sub wait_for_master {
   my ( $self, $master, $slave, $time, $timeoutok, $ms ) = @_;
   my $result;
   $ENV{MKDEBUG} && _d('Waiting for slave to catch up to master');
   $ms ||= $self->get_master_status($master);
   if ( $ms ) {
      my $query = "SELECT MASTER_POS_WAIT('$ms->{file}', $ms->{position}, $time)";
      $ENV{MKDEBUG} && _d($query);
      ($result) = $slave->selectrow_array($query);
      my $stat = defined $result ? $result : 'NULL';
      if ( $stat eq 'NULL' || $stat < 0 && !$timeoutok ) {
         die "MASTER_POS_WAIT returned $stat";
      }
      $ENV{MKDEBUG} && _d("Result of waiting: $stat");
   }
   else {
      $ENV{MKDEBUG} && _d("Not waiting: this server is not a master");
   }
   return $result;
}

sub stop_slave {
   my ( $self, $dbh ) = @_;
   my $sth = $self->{sths}->{$dbh}->{STOP_SLAVE}
         ||= $dbh->prepare('STOP SLAVE');
   $ENV{MKDEBUG} && _d($sth->{Statement});
   $sth->execute();
}

sub start_slave {
   my ( $self, $dbh, $pos ) = @_;
   if ( $pos ) {
      my $sql = "START SLAVE UNTIL MASTER_LOG_FILE='$pos->{file}', "
              . "MASTER_LOG_POS=$pos->{position}";
      $ENV{MKDEBUG} && _d($sql);
      $dbh->do($sql);
   }
   else {
      my $sth = $self->{sths}->{$dbh}->{START_SLAVE}
            ||= $dbh->prepare('START SLAVE');
      $ENV{MKDEBUG} && _d($sth->{Statement});
      $sth->execute();
   }
}

sub catchup_to_master {
   my ( $self, $slave, $master, $time ) = @_;
   my $slave_status  = $self->get_slave_status($slave);
   my $slave_pos     = $self->repl_posn($slave_status);
   my $master_status = $self->get_master_status($master);
   my $master_pos    = $self->repl_posn($master_status);
   $ENV{MKDEBUG} && _d("Master position: ", $self->pos_to_string($master_pos),
      " Slave position: ", $self->pos_to_string($slave_pos));
   if ( $self->pos_cmp($slave_pos, $master_pos) < 0 ) {
      $ENV{MKDEBUG} && _d('Waiting for slave to catch up to master');
      $self->start_slave($slave, $master_pos);
      $self->wait_for_master($master, $slave, $time, 0, $master_status);
   }
}

sub catchup_to_same_pos {
   my ( $self, $s1_dbh, $s2_dbh ) = @_;
   $self->stop_slave($s1_dbh);
   $self->stop_slave($s2_dbh);
   my $s1_status = $self->get_slave_status($s1_dbh);
   my $s2_status = $self->get_slave_status($s2_dbh);
   my $s1_pos    = $self->repl_posn($s1_status);
   my $s2_pos    = $self->repl_posn($s2_status);
   if ( $self->pos_cmp($s1_pos, $s2_pos) < 0 ) {
      $self->start_slave($s1_dbh, $s2_pos);
   }
   elsif ( $self->pos_cmp($s2_pos, $s1_pos) < 0 ) {
      $self->start_slave($s2_dbh, $s1_pos);
   }

   $s1_status = $self->get_slave_status($s1_dbh);
   $s2_status = $self->get_slave_status($s2_dbh);
   $s1_pos    = $self->repl_posn($s1_status);
   $s2_pos    = $self->repl_posn($s2_status);

   if ( $self->slave_is_running($s1_status)
     || $self->slave_is_running($s2_status)
     || $self->pos_cmp($s1_pos, $s2_pos) != 0)
   {
      die "The servers aren't both stopped at the same position";
   }

}

sub change_master_to {
   my ( $self, $dbh, $master_dsn, $master_pos ) = @_;
   my $sql = "CHANGE MASTER TO MASTER_HOST='$master_dsn->{h}', "
      . "MASTER_PORT= $master_dsn->{P}, MASTER_LOG_FILE='$master_pos->{file}', "
      . "MASTER_LOG_POS=$master_pos->{position}";
   $ENV{MKDEBUG} && _d($sql);
   $dbh->do($sql);
}

sub make_sibling_of_master {
   my ( $self, $slave_dbh, $slave_dsn, $dsn_parser, $timeout) = @_;

   my $master_dsn  = $self->get_master_dsn($slave_dbh, $slave_dsn, $dsn_parser)
      or die "This server is not a slave";
   my $master_dbh  = $dsn_parser->get_dbh(
      $dsn_parser->get_cxn_params($master_dsn), { AutoCommit => 1 });
   my $gmaster_dsn
      = $self->get_master_dsn($master_dbh, $master_dsn, $dsn_parser)
      or die "This server's master is not a slave";
   my $gmaster_dbh = $dsn_parser->get_dbh(
      $dsn_parser->get_cxn_params($gmaster_dsn), { AutoCommit => 1 });
   if ( $self->short_host($slave_dsn) eq $self->short_host($gmaster_dsn) ) {
      die "The slave's master's master is the slave: master-master replication";
   }

   $self->stop_slave($master_dbh);
   $self->catchup_to_master($slave_dbh, $master_dbh, $timeout);
   $self->stop_slave($slave_dbh);

   my $master_status = $self->get_master_status($master_dbh);
   my $mslave_status = $self->get_slave_status($master_dbh);
   my $slave_status  = $self->get_slave_status($slave_dbh);
   my $master_pos    = $self->repl_posn($master_status);
   my $slave_pos     = $self->repl_posn($slave_status);

   if ( !$self->slave_is_running($mslave_status)
     && !$self->slave_is_running($slave_status)
     && $self->pos_cmp($master_pos, $slave_pos) == 0)
   {
      $self->change_master_to($slave_dbh, $gmaster_dsn,
         $self->repl_posn($mslave_status)); # Note it's not $master_pos!
   }
   else {
      die "The servers aren't both stopped at the same position";
   }

   $mslave_status = $self->get_slave_status($master_dbh);
   $slave_status  = $self->get_slave_status($slave_dbh);
   my $mslave_pos = $self->repl_posn($mslave_status);
   $slave_pos     = $self->repl_posn($slave_status);
   if ( $self->short_host($mslave_status) ne $self->short_host($slave_status)
     || $self->pos_cmp($mslave_pos, $slave_pos) != 0)
   {
      die "The servers don't have the same master/position after the change";
   }
}

sub make_slave_of_sibling {
   my ( $self, $slave_dbh, $slave_dsn, $sib_dbh, $sib_dsn,
        $dsn_parser, $timeout) = @_;

   if ( $self->short_host($slave_dsn) eq $self->short_host($sib_dsn) ) {
      die "You are trying to make the slave a slave of itself";
   }

   my $master_dsn1 = $self->get_master_dsn($slave_dbh, $slave_dsn, $dsn_parser)
      or die "This server is not a slave";
   my $master_dbh1 = $dsn_parser->get_dbh(
      $dsn_parser->get_cxn_params($master_dsn1), { AutoCommit => 1 });
   my $master_dsn2 = $self->get_master_dsn($slave_dbh, $slave_dsn, $dsn_parser)
      or die "The sibling is not a slave";
   if ( $self->short_host($master_dsn1) ne $self->short_host($master_dsn2) ) {
      die "This server isn't a sibling of the slave";
   }
   my $sib_master_stat = $self->get_master_status($sib_dbh)
      or die "Binary logging is not enabled on the sibling";
   die "The log_slave_updates option is not enabled on the sibling"
      unless $self->has_slave_updates($sib_dbh);

   $self->catchup_to_same_pos($slave_dbh, $sib_dbh);

   $sib_master_stat = $self->get_master_status($sib_dbh);
   $self->change_master_to($slave_dbh, $sib_dsn,
         $self->repl_posn($sib_master_stat));

   my $slave_status = $self->get_slave_status($slave_dbh);
   my $slave_pos    = $self->repl_posn($slave_status);
   $sib_master_stat = $self->get_master_status($sib_dbh);
   if ( $self->short_host($slave_status) ne $self->short_host($sib_dsn)
     || $self->pos_cmp($self->repl_posn($sib_master_stat), $slave_pos) != 0)
   {
      die "After changing the slave's master, it isn't a slave of the sibling, "
         . "or it has a different replication position than the sibling";
   }
}

sub make_slave_of_uncle {
   my ( $self, $slave_dbh, $slave_dsn, $unc_dbh, $unc_dsn,
        $dsn_parser, $timeout) = @_;

   if ( $self->short_host($slave_dsn) eq $self->short_host($unc_dsn) ) {
      die "You are trying to make the slave a slave of itself";
   }

   my $master_dsn = $self->get_master_dsn($slave_dbh, $slave_dsn, $dsn_parser)
      or die "This server is not a slave";
   my $master_dbh = $dsn_parser->get_dbh(
      $dsn_parser->get_cxn_params($master_dsn), { AutoCommit => 1 });
   my $gmaster_dsn
      = $self->get_master_dsn($master_dbh, $master_dsn, $dsn_parser)
      or die "The master is not a slave";
   my $unc_master_dsn
      = $self->get_master_dsn($unc_dbh, $unc_dsn, $dsn_parser)
      or die "The uncle is not a slave";
   if ($self->short_host($gmaster_dsn) ne $self->short_host($unc_master_dsn)) {
      die "The uncle isn't really the slave's uncle";
   }

   my $unc_master_stat = $self->get_master_status($unc_dbh)
      or die "Binary logging is not enabled on the uncle";
   die "The log_slave_updates option is not enabled on the uncle"
      unless $self->has_slave_updates($unc_dbh);

   $self->catchup_to_same_pos($master_dbh, $unc_dbh);
   $self->catchup_to_master($slave_dbh, $master_dbh, $timeout);

   my $slave_status  = $self->get_slave_status($slave_dbh);
   my $master_status = $self->get_master_status($master_dbh);
   if ( $self->pos_cmp(
         $self->repl_posn($slave_status),
         $self->repl_posn($master_status)) != 0 )
   {
      die "The slave is not caught up to its master";
   }

   $unc_master_stat = $self->get_master_status($unc_dbh);
   $self->change_master_to($slave_dbh, $unc_dsn,
      $self->repl_posn($unc_master_stat));


   $slave_status    = $self->get_slave_status($slave_dbh);
   my $slave_pos    = $self->repl_posn($slave_status);
   if ( $self->short_host($slave_status) ne $self->short_host($unc_dsn)
     || $self->pos_cmp($self->repl_posn($unc_master_stat), $slave_pos) != 0)
   {
      die "After changing the slave's master, it isn't a slave of the uncle, "
         . "or it has a different replication position than the uncle";
   }
}

sub detach_slave {
   my ( $self, $dbh ) = @_;
   $self->stop_slave($dbh);
   my $stat = $self->get_slave_status($dbh)
      or die "This server is not a slave";
   $dbh->do('CHANGE MASTER TO MASTER_HOST=""');
   $dbh->do('RESET SLAVE'); # Wipes out master.info, etc etc
   return $stat;
}

sub slave_is_running {
   my ( $self, $slave_status ) = @_;
   return ($slave_status->{slave_sql_running} || 'No') eq 'Yes';
}

sub has_slave_updates {
   my ( $self, $dbh ) = @_;
   my $sql = q{SHOW VARIABLES LIKE 'log_slave_updates'};
   $ENV{MKDEBUG} && _d($sql);
   my ($name, $value) = $dbh->selectrow_array($sql);
   return $value && $value =~ m/^(1|ON)$/;
}

sub repl_posn {
   my ( $self, $status ) = @_;
   if ( exists $status->{file} && exists $status->{position} ) {
      return {
         file     => $status->{file},
         position => $status->{position},
      };
   }
   else {
      return {
         file     => $status->{relay_master_log_file},
         position => $status->{exec_master_log_pos},
      };
   }
}

sub pos_cmp {
   my ( $self, $a, $b ) = @_;
   return $self->pos_to_string($a) cmp $self->pos_to_string($b);
}

sub short_host {
   my ( $self, $dsn ) = @_;
   my ($host, $port);
   if ( $dsn->{master_host} ) {
      $host = $dsn->{master_host};
      $port = $dsn->{master_port};
   }
   else {
      $host = $dsn->{h};
      $port = $dsn->{P};
   }
   return ($host || '[default]') . ( ($port || 3306) == 3306 ? '' : ":$port" );
}

sub pos_to_string {
   my ( $self, $pos ) = @_;
   my $fmt  = '%s/%020d';
   return sprintf($fmt, @{$pos}{qw(file position)});
}

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# MasterSlave:$line $PID ", @_, "\n";
}

1;

# ###########################################################################
# End MasterSlave package
# ###########################################################################

package main;

use English qw(-no_match_vars);
use List::Util qw(sum max min);
use POSIX qw(ceil);

$OUTPUT_AUTOFLUSH = 1;

# ############################################################################
# Get configuration information.
# ############################################################################

my @opt_spec = (
   { s => 'algorithm|a=s',
     d => 'Algorithm to use'},
   { s => 'askpass',
     d => 'Prompt for password for connections' },
   { s => 'bufferresults',
     d => 'Fetch all rows from MySQL before comparing' },
   { s => 'chunksize=s',
     d => 'Number of rows or data size per chunk (default 1000)'},
   { s => 'columns|c=a',
     d => 'Compare this comma-separated list of columns' },
   { s => 'databases|d=h',
     d => 'Do only this comma-separated list of databases' },
   { s => 'engine|e=h',
     d => 'Do only this comma-separated list of storage engines' },
   { s => 'execute|x',
     d => 'Execute queries to sync tables' },
   { s => 'function|f=s',
     d => 'Hash function to use for checksums' },
   { s => 'ignoredb|g=H',
     d => 'Ignore this comma-separated list of databases' },
   { s => 'ignoreengine|E=H',
     d => 'Ignore this comma-separated list of storage engines '
        . '(default FEDERATED,MRG_MyISAM)' },
   { s => 'ignoretbl|n=H',
     d => 'Ignore this comma-separated list of tables' },
   { s => 'lock|k=i',
     d => 'Lock tables: 0=none, 1=per sync cycle, 2=per table, or 3=globally' },
   { s => 'print|p',
     d => 'Print queries that will resolve differences' },
   { s => 'replace|r',
     d => 'Use REPLACE instead of INSERT and UPDATE' },
   { s => 'replicate|R=s',
     d => 'Sync tables listed as different in this table' },
   { s => 'setvars=s',
     d => 'Set these MySQL variables (default wait_timeout=10000)' },
   { s => 'skipbinlog',
     d => 'Do not log to the binary log' },
   { s => 'skipforeignkey|K',
     d => 'Disable foreign key checks' },
   { s => 'skipuniquekey',
     d => 'Disable unique key checks' },
   { s => 'synctomaster|s',
     d => 'Treat the DSN as a slave and sync it to its master'},
   { s => 'tables|t=h',
     d => 'Do only this comma-separated list of tables' },
   { s => 'test',
     d => 'Analyze, decide the sync algorithm to use, print and exit' },
   { s => 'timeoutok',
     d => 'Keep going if --wait fails' },
   { s => 'transaction!',
     d => 'Use transactions INSTEAD OF locking tables' },
   { s => 'utf8!',
     d => 'Enable UTF-8 options in Perl and MySQL (default)' },
   { s => 'verbose|v',
     d => 'Print results of sync operations' },
   { s => 'wait|w=m',
     d => 'How long to wait for slaves to catch up to their master' },
   { s => 'where|W=s',
     d => 'WHERE clause to restrict syncing to part of the table' },
);

my $du         = new MySQLDump();
my $tp         = new TableParser();
my $q          = new Quoter();
my $vp         = new VersionParser();
my $chunker    = new TableChunker( quoter => $q );
my $nibbler    = new TableNibbler();
my $checksum   = new TableChecksum();
my $ms         = new MasterSlave();
my $syncer     = new TableSyncer();
my $dsn_parser = new DSNParser(
   {
      key  => 'D',
      desc => 'Database containing the table to be synced',
      dsn  => 'database',
      copy => 1,
   },
   {
      key  => 't',
      desc => 'Table to be synced',
      dsn  => undef,
      copy => 1,
   },
);
$dsn_parser->prop('autokey', 'h');
my $opt_parser = new OptionParser(@opt_spec);
$opt_parser->{dsn}    = $dsn_parser;
$opt_parser->{strict} = 0;
$opt_parser->{prompt} = '[OPTION].. DSN [DSN...]';
$opt_parser->{descr} = 'finds and fixes data differences between MySQL tables.';
my %opts = $opt_parser->parse();
$dsn_parser->prop('setvars', $opts{setvars});

if ( $opts{R} || $opts{s} ) {
   $opts{w} = 60 unless defined $opts{w};
}
if ( $opts{w} ) {
   $opts{k} = 1 unless defined $opts{k};
}
if ( $opts{test} ) {
   $opts{v} = 1;
}

my @dsns;
while ( my $arg = shift(@ARGV) ) {
   if ( $opts{utf8} ) {
      $arg .= ',A=utf8';
   }
   my $dsn = $dsn_parser->parse($arg, $dsns[0]);
   die "You specified a t part, but not a D part in $arg"
      if ($dsn->{t} && !$dsn->{D});
   push @dsns, $dsn;
}

if ( !@dsns || (@dsns ==1 && !$opts{R} && !$opts{s}) ) {
   $opt_parser->error('At least one DSN is required, and at least two are '
      . 'required unless --synctomaster or --replicate is specified');
}

$opt_parser->usage_or_errors(%opts);

# ############################################################################
# Do the work.
# ############################################################################
my $exit_status = 0; # 1: internal error, 2: tables differed, 3: both
my $hdr         = "# %6s %7s %6s %6s %-9s %s.%s\n";

if ( $dsns[0]->{t} ) {
   $ENV{MKDEBUG} && _d('DSN has t part; syncing ONE TABLE between servers');
   if ( @dsns == 1 ) {
      die "You specified one DSN with a t part, but not --synctomaster.\n"
         unless $opts{s};
      $dsns[0]->{dbh} = get_cxn($dsns[0]);
      unshift @dsns,
         $ms->get_master_dsn($dsns[0]->{dbh}, $dsns[0], $dsn_parser);
   }

   if ( $dsns[0]->{t} ne $dsns[1]->{t}
         && -t STDIN && -t STDOUT
         && (  (($dsns[0]->{h} || '') ne ($dsns[1]->{h} || ''))
            || (($dsns[0]->{P} || '') ne ($dsns[1]->{P} || '')) ))
   {
      print "Continue syncing tables with different names?  y/n: ";
      my $answer = <STDIN>;
      if ( $answer !~ m/y/i ) {
         exit(1);
      }
   }

   foreach my $dsn ( @dsns[1 .. $#dsns] ) {

      if ( $opts{v} ) {
         print "# Syncing ", $dsn_parser->as_string($dsn), "\n";
         printf($hdr, @ChangeHandler::ACTIONS, qw(ALGORITHM DATABASE TABLE));
      }

      my %args = (
         src     => $dsns[0],
         dst     => $dsn,
      );

      lock_server(%args);
      sync_a_table(
         %args,
         src_db  => $dsns[0]->{D},
         src_tbl => $dsns[0]->{t},
         dst_db  => $dsn->{D},
         dst_tbl => $dsn->{t},
         where   => $opts{W},
      );
      unlock_server(%args);
   }
}
elsif ( $opts{R} ) {

   if ( @dsns > 1 ) {
      die "You should specify only one DSN with --replicate\n";
   }

   # Connect to the master and treat it as the source, then find differences on
   # the slave and sync them.
   if ( $opts{s} ) {

      $dsns[0]->{dbh} = get_cxn($dsns[0]);
      unshift @dsns,
         $ms->get_master_dsn($dsns[0]->{dbh}, $dsns[0], $dsn_parser);
      $dsns[0]->{dbh} = get_cxn($dsns[0]);
      my %args = (
         src     => $dsns[0],
         dst     => $dsns[1],
      );

      # First, check that the master (source) has no discrepancies itself, and
      # ignore tables that do.
      my %skip_table;
      foreach my $diff (
         $checksum->find_replication_differences($dsns[0]->{dbh}, $opts{R}) )
      {
         $skip_table{$diff->{db}}->{$diff->{tbl}}++;
      }

      # Now check the slave for differences and sync them if necessary.
      my @diffs = $checksum
         ->find_replication_differences($dsns[1]->{dbh}, $opts{R});
      @diffs = grep { !$skip_table{$_->{db}}->{$_->{tbl}} } @diffs;
      if ( $opts{v} ) {
         print "# Syncing ", $dsn_parser->as_string($dsns[1]), "\n";
         printf($hdr, @ChangeHandler::ACTIONS, qw(ALGORITHM DATABASE TABLE));
      }
      if ( @diffs ) {
         lock_server(%args);
         foreach my $diff ( @diffs ) {
            sync_a_table(
               %args,
               src_db  => $diff->{db},
               src_tbl => $diff->{tbl},
               dst_db  => $diff->{db},
               dst_tbl => $diff->{tbl},
               where   => $diff->{boundaries},
            );
         }
         unlock_server(%args);
      }
   }

   # The DSN is the master.  Connect to each slave and find differences, then
   # sync them.
   else {
      my %skip_table;
      $dsns[0]->{dbh} = get_cxn($dsns[0]);
      $ms->recurse_to_slaves(
         {  dbh        => $dsns[0]->{dbh},
            dsn        => $dsns[0],
            dsn_parser => $dsn_parser,
            recurse    => 1,
            callback   => sub {
               my ( $dsn, $dbh, $level, $parent ) = @_;
               my @diffs = $checksum
                  ->find_replication_differences($dbh, $opts{R});
               if ( !$level ) {
                  # This is the master; don't sync any tables that are wrong
                  # here, for obvious reasons.
                  map { $skip_table{$_->{db}}->{$_->{tbl}}++ } @diffs;
               }
               else {
                  # Must save a reference to the DBH to close gracefully later
                  $dsn->{dbh} ||= $dbh;
                  push @dsns, $dsn;
                  @diffs = grep { !$skip_table{$_->{db}}->{$_->{tbl}} } @diffs;
                  return unless @diffs;
                  if ( $opts{v} ) {
                     print "# Syncing ", $dsn_parser->as_string($dsn), "\n";
                     printf($hdr, @ChangeHandler::ACTIONS,
                        qw(ALGORITHM DATABASE TABLE));
                  }
                  my %args = (
                     src     => $dsns[0],
                     dst     => $dsn,
                  );
                  lock_server(%args);
                  foreach my $diff ( @diffs ) {
                     sync_a_table(
                        %args,
                        src_db  => $diff->{db},
                        src_tbl => $diff->{tbl},
                        dst_db  => $diff->{db},
                        dst_tbl => $diff->{tbl},
                        where => $diff->{boundaries},
                     );
                  }
                  unlock_server(%args);
               }
            },
         }
      );
   }

}
else {
   $dsns[0]->{dbh} = get_cxn($dsns[0]);

   if ( @dsns == 1 ) {
      if ( $opts{s} ) {
         unshift @dsns,
            $ms->get_master_dsn($dsns[0]->{dbh}, $dsns[0], $dsn_parser);
         $dsns[0]->{dbh} = get_cxn($dsns[0]);
      }
      else {
         die "You specified only one DSN, "
            . "but not --synctomaster or --replicate.\n";
      }
   }

   my $finder = new MySQLFind(
      quoter    => $q,
      useddl    => 1,
      parser    => $tp,
      dumper    => $du,
      databases => {
         permit => $opts{d},
         reject => $opts{g},
      },
      tables => {
         permit => $opts{t},
         reject => $opts{n},
      },
      engines => {
         views  => 0,
         permit => $opts{e},
         reject => $opts{E},
      },
   );

   foreach my $dsn ( @dsns[ 1 .. $#dsns ] ) {

      if ( $opts{v} ) {
         print "# Syncing ", $dsn_parser->as_string($dsn), "\n";
         printf($hdr, @ChangeHandler::ACTIONS, qw(ALGORITHM DATABASE TABLE));
      }

      my %args = (
         src     => $dsns[0],
         dst     => $dsn,
      );

      lock_server(%args);
      foreach my $database ( $finder->find_databases($dsns[0]->{dbh}) ) {
         foreach my $table (
            $finder->find_tables($dsns[0]->{dbh}, database => $database)
         ) {
            sync_a_table(
               %args,
               src_db  => $database,
               src_tbl => $table,
               dst_db  => $database,
               dst_tbl => $table,
               where   => $opts{W},
            );
         }
      }
      unlock_server(%args);
   }
}

sub lock_server {
   my (%args) = @_;
   eval {
      # Open connections as needed.
      $args{src}->{dbh}      ||= get_cxn($args{src});
      $args{dst}->{dbh}      ||= get_cxn($args{dst});
      $args{src}->{misc_dbh} ||= get_cxn($args{src});
      $syncer->lock_and_wait(%args, lock_level => 3);
   }
}

sub unlock_server {
   my (%args) = @_;
   eval {
      # Open connections as needed.
      $args{src}->{dbh}      ||= get_cxn($args{src});
      $args{dst}->{dbh}      ||= get_cxn($args{dst});
      $args{src}->{misc_dbh} ||= get_cxn($args{src});
      $syncer->unlock(%args, lock_level => 3);
   }
}

sub sync_a_table {
   my (%args) = @_;

   eval {
      # Open connections as needed.
      $args{src}->{dbh}      ||= get_cxn($args{src});
      $args{dst}->{dbh}      ||= get_cxn($args{dst});
      $args{src}->{misc_dbh} ||= get_cxn($args{src});

      my $tbl_struct = $tp->parse(
         $du->get_create_table($args{src}->{dbh}, $q, @args{qw(src_db src_tbl)}));
      my @possible_keys = $tp->find_possible_keys(
         $args{src}->{dbh}, @args{qw(src_db src_tbl)}, $q, $opts{W});

      # If the table is InnoDB, prefer to sync it with transactions, unless
      # the user explicitly said not to.
      my $use_txn = defined $opts{transaction}        ? $opts{transaction}
                  : $tbl_struct->{engine} eq 'InnoDB' ? 1
                  :                                     0;

      # If we're using transactions, turn on AutoCommit on the handles.
      foreach my $dbh (
         $args{src}->{dbh},
         $args{dst}->{dbh},
         $args{src}->{misc_dbh},
      ) {
         $dbh->{AutoCommit} = !$use_txn;
      }

      my %status = $syncer->sync_table(
         algorithm     => $opts{a} || '',
         buffer        => $opts{bufferresults} || 0,
         checksum      => $checksum,
         chunker       => $chunker,
         chunksize     => $opts{chunksize},
         cols          => $opts{c} || $tbl_struct->{cols},
         dst_db        => $args{dst_db},
         dst_dbh       => $args{dst}->{dbh},
         dst_tbl       => $args{dst_tbl},
         dumper        => $du,
         execute       => $opts{x} || 0,
         lock          => $opts{k} || 0,
         misc_dbh      => $args{src}->{misc_dbh},
         nibbler       => $nibbler,
         possible_keys => \@possible_keys,
         parser        => $tp,
         print         => $opts{p} || 0,
         quoter        => $q,
         replace       => $opts{r} || 0,
         replicate     => ($opts{R} || $opts{s}) || 0,
         transaction   => $use_txn,
         src_db        => $args{src_db},
         src_dbh       => $args{src}->{dbh},
         src_tbl       => $args{src_tbl},
         test          => $opts{test} || 0,
         tbl_struct    => $tbl_struct,
         timeoutok     => $opts{timeoutok} || 0,
         versionparser => $vp,
         wait          => $opts{w} || 0,
         where         => $args{where} || '',
         master_slave  => $ms,
         func          => $opts{f} || '',
      );

      if ( $opts{v} ) {
         printf($hdr,
            @status{@ChangeHandler::ACTIONS, 'ALGORITHM'},
            @args{qw(src_db src_tbl)});
      }

      if ( sum(@status{@ChangeHandler::ACTIONS}) ) {
         $exit_status |= 2;
      }
   };

   if ( $EVAL_ERROR ) {
      print_err($EVAL_ERROR, @args{qw(dst_db dst_tbl)}, $args{dst_dsn}->{h} );
      $exit_status |= 1;
   }
}

foreach my $dsn (@dsns) {
   foreach my $thing ( qw(dbh misc_dbh) ) {
      my $dbh = $dsn->{$thing};
      next unless $dbh;
      $dbh->commit unless $dbh->{AutoCommit};
      $dsn_parser->disconnect($dbh);
   }
}

exit $exit_status;

# ############################################################################
# Subroutines
# ############################################################################

sub _d {
   my ( $line ) = (caller(0))[2];
   print "# main:$line $PID ", @_, "\n";
}

# Tries to extract the MySQL error message and print it
sub print_err {
   my ( $msg, $database, $table, $host ) = @_;
   return if !defined $msg;
   $msg =~ s/^.*?failed: (.*?) at \S+ line (\d+).*$/$1 at line $2/s;
   $msg =~ s/\s+/ /g;
   if ( $database && $table ) {
      $msg .= " while doing $database.$table";
   }
   if ( $host ) {
      $msg .= " on $host";
   }
   print STDERR $msg, "\n";
}

sub get_cxn {
   my ( $info ) = @_;
   my $db_options = {};

   if ( !$info->{p} && $opts{askpass} ) {
      $info->{p}
         = OptionParser::prompt_noecho("Enter password for $info->{h}: ");
   }
   my $dbh = $dsn_parser->get_dbh(
      $dsn_parser->get_cxn_params($info), $db_options);
   my $sql;
   if ( $opts{skipbinlog} ) {
      $sql = "/*!32316 SET SQL_LOG_BIN=0 */";
      $ENV{MKDEBUG} && _d("$dbh: $sql");
      $dbh->do($sql);
   }
   if ( $opts{skipuniquekey} ) {
      $sql = "/*!40014 SET UNIQUE_CHECKS=0 */";
      $ENV{MKDEBUG} && _d("$dbh: $sql");
      $dbh->do($sql);
   }
   if ( $opts{K} ) {
      $sql = "/*!40014 SET FOREIGN_KEY_CHECKS=0 */";
      $ENV{MKDEBUG} && _d("$dbh: $sql");
      $dbh->do($sql);
   }
   return $dbh;
}

# ############################################################################
# Documentation
# ############################################################################
=pod

=head1 NAME

mk-table-sync - Synchronize MySQL tables efficiently.

=head1 SYNOPSIS

It is a good idea to back up your data and run C<mk-table-sync> with L<"--test">
first, to see what will happen.  

To sync db.tbl1 from host1 to host2:

   mk-table-sync -x u=user,p=pass,h=host1,D=db,t=tbl host2

Sync all tables in host1 to host2 and host3:

   mk-table-sync -x host1 host2 host3

Resolve differences L<mk-table-checksum> found on this master's slaves:

   mk-table-sync -x --replicate test.checksum master1

Sync this slave to its replication master:

   mk-table-sync -x --synctomaster slave1

Sync this slave to its replication master, resolving differences
L<mk-table-checksum> found:

   mk-table-sync -x --synctomaster --replicate test.checksum slave1

Sync server2 in a master-master replication configuration, where server2's copy
of db1.tbl2 is known or suspected to be incorrect:

   mk-table-sync -x --synctomaster h=server2,D=db1,t=tbl1

Note that in the master-master configuration, the following will NOT do what you
want, because it will make changes directly on server2, which will then flow
through replication and change server1's data:

   # Don't do this in a master-master setup!
   mk-table-sync -x h=server1,D=db1,t=tbl1 h=server2

=head1 DESCRIPTION

B<WARNING> this tool is unfinished and could perform slowly.  The Chunk
algorithm is great when it can be used, and so is Nibble, but otherwise Stream
is the only choice and it does not perform very well.  Please run with
L<"--test"> before subjecting your servers to this tool, and make backups of
your data!

This tool is designed to do one-way synchronization of data (two-way sync is
planned for the future).  It finds differences efficiently with one of several
algorithms (see L<"ALGORITHMS">).  It makes changes on the destination table(s)
so it matches the source.

It does B<not> synchronize table structures, indexes, or any other schema
changes.  It synchronizes only data.

It can operate through replication by comparing a slave with its master and
making changes on the master.  These changes will flow through replication and
correct any differences found on the slave.

It accepts a list of DSNs (see the L<"--help"> output) to tell it where and how
to connect.

There are many ways to invoke it.  The following is the abbreviated logic:

   if DSN has a t part, sync only that table:
      if 1 DSN:
         if --synctomaster:
            The DSN is a slave.  Connect to its master and sync.
      if more than 1 DSN:
         The first DSN is the source.  Sync each DSN in turn.
   else if --replicate:
      if --synctomaster:
         The DSN is a slave.  Connect to its master, find records
         of differences, and fix.
      else:
         The DSN is the master.  Find slaves and connect to each,
         find records of differences, and fix.
   else:
      if only 1 DSN and --synctomaster:
         The DSN is a slave.  Connect to its master, find tables and
         filter with --databases etc, and sync each table to the master.
      else:
         find tables, filtering with --databases etc, and sync each
         DSN to the first.

=head1 ALGORITHMS

This tool has a generic data-syncing framework, within which it is possible to
use any number of different algorithms to actually find differences.  It chooses
the best algorithm automatically.  While I plan to add more algorithms in the
future, the following are implemented now:

=over

=item Chunk

Finds an index whose first column is numeric (including date and time types),
and divides the column's range of values into chunks of approximately
L<"--chunksize"> rows.  Syncs a chunk at a time by checksumming the entire
chunk.  If the chunk differs on the source and destination, checksums each
chunk's rows individually to find the rows that differ.

It is efficient when the column has sufficient cardinality to make the chunks
end up about the right size.

The initial per-chunk checksum is quite small and results in minimal network
traffic and memory consumption.  If a chunk's rows must be examined, only the
primary key columns and a checksum are sent over the network, not the entire
row.  If a row is found to be different, the entire row will be fetched, but not
before.

=item Nibble

Finds an index and ascends the index in fixed-size nibbles of L<"--chunksize">
rows, using a non-backtracking algorithm (see L<mk-archiver> for more on this
algorithm).  It is very similar to L<"Chunk">, but instead of pre-calculating
the boundaries of each piece of the table based on index cardinality, it uses
C<LIMIT> to define each nibble's upper limit, and the previous nibble's upper
limit to define the lower limit.

It works in steps: one query finds the row that will define the next nibble's
upper boundary, and the next query checksums the entire nibble.  If the nibble
differs between the source and destination, it examines the nibble row-by-row,
just as L<"Chunk"> does.

=item Stream

Selects the entire table in one big stream and compares all columns.  Selects
all columns.  Much less efficient than the other algorithms, but works when
there is no suitable index for them to use.

=item Future Plans

Possibilities for future algorithms are TempTable (what I originally called
bottom-up in earlier versions of this tool), DrillDown (what I originallly
called top-down), and GroupByPrefix (similar to how SqlYOG Job Agent works).
Each algorithm has strengths and weaknesses.  If you'd like to implement your
favorite technique for finding differences between two sources of data on
possibly different servers, I'm willing to help.  The algorithms adhere to a
simple interface that makes it pretty easy to write your own.

=back

=head1 OPTIONS

=over

=item --algorithm

The algorithm to use when comparing the tables.  This is a suggestion.  The tool
will auto-detect the best algorithm, and if your chosen algorithm can't be used,
will use the best available one instead.  See L<"ALGORITHMS">.

=item --askpass

Prompts the user for a password when connecting to MySQL.

=item --bufferresults

Fetch all rows from MySQL before comparing.  This is disabled by default.  If
enabled, all rows will be fetched into memory for comparing.  This may result in
the results "cursor" being held open for a shorter time on the server, but if
the tables are large, it could take a long time anyway, and eat all your memory.
For most non-trivial data sizes, you want to leave this disabled.

=item --chunksize

The size of each chunk of rows for the L<"Chunk"> and L<"Nibble"> algorithms.
The size can be either a number of rows, or a data size.  Data sizes are
specified with a suffix of k=kibibytes, M=mebibytes, G=gibibytes.  Data sizes
are converted to a number of rows by dividing by the average row length.

=item --columns

Comma-separated list of columns to compare when looking for differences.

=item --databases

Sync only this comma-separated list of databases.

=item --engine

Sync only tables whose storage engine is in this comma-separated list.

=item --execute

After finding differences, execute the queries required to sync the tables.

=item --function

Which hash function you'd like to use for row checksums.  Good choices include
C<MD5> and C<SHA1>.  If you have installed the C<FNV_64> user-defined function,
C<mk-table-sync> will detect it and prefer to use it, because it is much faster
than the others.  See L<mk-table-checksum> for more information and benchmarks.

=item --help

Show a brief help message.

=item --ignoredb

Use this option to skip a comma-separated list of databases.

=item --ignoreengine

Use this option to skip a comma-separated list of storage engines (table types).

=item --ignoretbl

Use this option to skip a comma-separated list of tables.  Table names may be
qualified with the database name.

=item --lock

Lock tables when beginning work.  This uses C<LOCK TABLES>.  This can help
prevent tables being changed while you're examining them.

The argument is an integer, default value 0.  The possible values are as
follows:

  VALUE  MEANING
  =====  =======================================================
  0      Never lock tables.
  1      Lock and unlock one time per sync cycle (as implemented
         by the syncing algorithm).  This is the most granular
         level of locking available.  For example, the Chunk
         algorithm will lock each chunk of C<N> rows, and then
         unlock them if they are the same on the source and the
         destination, before moving on to the next chunk.
  2      Lock and unlock before and after each table.
  3      Lock and unlock once for every server (DSN) synced, with
         C<FLUSH TABLES WITH READ LOCK>.

A replication slave is never locked if L<"--replicate"> or L<"--synctomaster">
is specified, since in theory locking the table on the master should prevent any
changes from taking place.  (You are not changing data on your slave, right?)
If L<"--wait"> is given, the master (source) is locked and then the tool waits
for the slave to catch up to the master before continuing.

If L<"--transaction"> is specified, C<LOCK TABLES> is not used.  Instead, lock
and unlock are implemented by beginning and committing transactions.  The exception
is if L<"--lock"> is 3.

=item --print

Print queries that will resolve the differences between the tables.  If you
don't trust C<mk-table-sync>, or just want to see what it will do, this is a
good way to be safe.  These queries are valid SQL and you can run them yourself
if you want to sync the tables manually.

=item --replace

Write all C<INSERT> and C<UPDATE> statements as C<REPLACE>.  This is
automatically switched on as needed when there are unique index violations.

=item --replicate

Specifies that C<mk-table-sync> should examine the specified table to find data
that differs.  The table is exactly the same as the argument of the same name to
L<mk-table-checksum>.  That is, it contains records of which tables (and ranges
of values) differ between the master and slave.

For each table and range of values that shows differences between the master and
slave, C<mk-table-checksum> will sync that table, with the appropriate C<WHERE>
clause, to its master.

This automatically sets L<"--wait"> to 60 and causes changes to be made on the
master instead of the slave.

If L<"--synctomaster"> is specified, the tool will assume the server you
specified is the slave, and connect to the master as usual to sync.

Otherwise, it will try to use C<SHOW PROCESSLIST> to find slaves of the server
you specified.  If it is unable to find any slaves via C<SHOW PROCESSLIST>, it
will inspect C<SHOW SLAVE HOSTS> instead.  You must configure each slave's
C<report-host>, C<report-port> and other options for this to work right.  After
finding slaves, it will inspect the specified table on each slave to find data
that needs to be synced, and sync it. 

The tool examines the master's copy of the table first, assuming that the master
is potentially a slave as well.  Any table that shows differences there will
B<NOT> be synced on the slave(s).  For example, suppose your replication is set
up as A->B, B->C, B->D.  Suppose you use this argument and specify server B.
The tool will examine server B's copy of the table.  If it looks like server B's
data in table C<test.tbl1> is different from server A's copy, the tool will not
sync that table on servers C and D.

=item --setvars

Specify any variables you want to be set immediately after connecting to MySQL.
These will be included in a C<SET> command.

=item --skipbinlog

Do not log statements to the binary log (C<SET SQL_LOG_BIN=0>).

=item --skipforeignkey

Disable foreign key checks (C<SET FOREIGN_KEY_CHECKS=0>).

=item --skipuniquekey

Disable unique index checks (C<SET UNIQUE_CHECKS=0>).

=item --synctomaster

Treat the server you specified as a slave.  Inspect C<SHOW SLAVE STATUS>,
connect to the server's master, and treat the master as the source and the slave
as the destination.  Causes changes to be made on the master.  Sets L<"--wait">
to 60 by default, sets L<"--lock"> to 1 by default, and disables
L<"--transaction"> by default.  See also L<"--replicate">, which changes this
option's behavior.

=item --tables

Restrict syncing to this comma-separated list of tables.  Table names may be
qualified with the database name.

=item --test

Analyze tables and decide which sync algorithm to use, then bail out before
doing any work.  Implies L<"--verbose"> so you can see the results.

=item --timeoutok

Keep going if L<"--wait"> fails.  If you specify L<"--wait"> and the slave
doesn't catch up to the master's position before the wait times out, the default
behavior is to abort.  This option makes the tool keep going anyway.
B<Warning>: if you are trying to get a consistent comparision between the two
servers, you probably don't want to keep going after a timeout.

=item --transaction

Use transactions instead of C<LOCK TABLES>.  The granularity of beginning and
committing transactions is controlled by L<"--lock">.  This is enabled by
default, but since L<"--lock"> is disabled by default, it has no effect.

Most options that enable locking also disable transactions by default, so if
you want to use transactional locking (via C<LOCK IN SHARE MODE> and C<FOR
UPDATE>, you must specify --transaction explicitly.

If you don't specify --transaction explicitly, C<mk-table-sync> will decide on
a per-table basis whether to use transactions or table locks.  It currently
uses transactions on InnoDB tables, and table locks on all others.

=item --utf8

This option is deprecated.  Pass the C<A> option in a DSN instead.  For
backwards compatibility, this option adds C<A=utf8> to all DSNs.

Enables character set settings in Perl and MySQL.  If the value is C<utf8>, sets
Perl's binmode on STDOUT to utf8, passes the C<mysql_enable_utf8> option to
DBD::mysql, and runs C<SET NAMES UTF8> after connecting to MySQL.  Any other
value sets binmode on STDOUT without the utf8 layer, and runs C<SET NAMES> after
connecting to MySQL.

=item --verbose

Explain the differences found while comparing the tables.  See L<"OUTPUT"> for
more details about the output.

=item --version

Output version information and exit.

=item --wait

Make the master wait for the slave to catch up in replication before comparing
the tables.  The value is the number of seconds to wait before timing out (see
also L<"--timeoutok">).  Sets L<"--lock"> to 1 and L<"--transaction"> to 0 by
default.

=item --where

A C<WHERE> clause to restrict the portion of the table being synchronized.

=back

=head1 EXIT STATUS

Exit status is as follows:

   STATUS  MEANING
   ======  =======================================================
   0       Success.
   1       Internal error.
   2       At least one table differed on the destination.
   3       Combination of 1 and 2.

=head1 OUTPUT

If you specify the L<"--verbose"> option, you'll see information about the 
differences between the tables.  There is one row per table.  Each server is
printed separately.  For example,

   # Syncing D=test,t=test2
   # DELETE REPLACE INSERT UPDATE ALGORITHM DATABASE.TABLE
   #      0       0      2      0 Chunk     test.test1

This table required 2 C<UPDATE> statements to synchronize.

There are cases where no combination of C<INSERT>, C<UPDATE> or C<DELETE>
statements can resolve differences without violating some unique key.  For
example, suppose there's a primary key on column a and a unique key on column b.
Then there is no way to sync these two tables with straightforward UPDATE
statements:

 +---+---+  +---+---+
 | a | b |  | a | b |
 +---+---+  +---+---+
 | 1 | 2 |  | 1 | 1 |
 | 2 | 1 |  | 2 | 2 |
 +---+---+  +---+---+

The tool rewrites queries to C<DELETE> and C<REPLACE> in this case.  This is
automatically handled after the first index violation, so you don't have to
worry about it.

=head1 ENVIRONMENT

The environment variable MKDEBUG enables verbose debugging output in all of the
Maatkit tools:

   MKDEBUG=1 mk-....

=head1 HISTORY AND ACKNOWLEDGEMENTS

My work is based in part on Giuseppe Maxia's work on distributed databases,
L<http://www.sysadminmag.com/articles/2004/0408/> and code derived from that
article.  There is more explanation, and a link to the code, at
L<http://www.perlmonks.org/?node_id=381053>.

Another programmer extended Maxia's work even further.  Fabien Coelho changed
and generalized Maxia's technique, introducing symmetry and avoiding some
problems that might have caused too-frequent checksum collisions.  This work
grew into pg_comparator, L<http://www.coelho.net/pg_comparator/>.  Coelho also
explained the technique further in a paper titled "Remote Comparison of Database
Tables" (L<http://cri.ensmp.fr/classement/doc/A-375.pdf>).

This existing literature mostly addressed how to find the differences between
the tables, not how to resolve them once found.  I needed a tool that would not
only find them efficiently, but would then resolve them.  I first began thinking
about how to improve the technique further with my article
L<http://www.xaprb.com/blog/2007/03/05/an-algorithm-to-find-and-resolve-data-differences-between-mysql-tables/>,
where I discussed a number of problems with the Maxia/Coelho "bottom-up"
algorithm.  After writing that article, I began to write this tool.  I wanted to
actually implement their algorithm with some improvements so I was sure I
understood it completely.  I discovered it is not what I thought it was, and is
considerably more complex than it appeared to me at first.  Fabien Coelho was
kind enough to address some questions over email.

The first versions of this tool implemented a version of the Coelho/Maxia
algorithm, which I called "bottom-up", and my own, which I called "top-down."
Those algorithms are considerably more complex than the current algorithms and
I have removed them from this tool, and may add them back later.  The
improvements to the bottom-up algorithm are my original work, as is the
top-down algorithm.  The techniques to actually resolve the differences are
also my own work.

Another tool that can synchronize tables is the SQLyog Job Agent from webyog.
Thanks to Rohit Nadhani, SJA's author, for the conversations about the general
techniques.  There is a comparison of mk-table-sync and SJA at
L<http://www.xaprb.com/blog/2007/04/05/mysql-table-sync-vs-sqlyog-job-agent/>

Thanks to the following people and organizations for helping in many ways:

The Rimm-Kaufman Group L<http://www.rimmkaufman.com/>,
MySQL AB L<http://www.mysql.com/>,
Blue Ridge InternetWorks L<http://www.briworks.com/>,
Percona L<http://www.percona.com/>,
Fabien Coelho,
Giuseppe Maxia and others at MySQL AB,
Kristian Koehntopp (MySQL AB),
Rohit Nadhani (WebYog),
The helpful monks at Perlmonks,
And others too numerous to mention.

=head1 SYSTEM REQUIREMENTS

You need Perl, DBI, DBD::mysql, and some core packages that ought to be
installed in any reasonably new version of Perl.

=head1 BUGS

Please use the Sourceforge bug tracker, forums, and mailing lists to request
support or report bugs: L<http://sourceforge.net/projects/maatkit/>.

=head1 COPYRIGHT, LICENSE AND WARRANTY

This program is copyright (c) 2007 Baron Schwartz.
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.

=head1 AUTHOR

Baron Schwartz.

=head1 VERSION

This manual page documents Ver 1.0.6 Distrib 1877 $Revision: 1870 $.

=cut