Merge lp:~percona-toolkit-dev/percona-toolkit/pt-osc-2.1.1 into lp:percona-toolkit/2.1

[Daniel Nichter]

[Daniel Nichter (daniel-nichter)]

review: Approve

[Baron Schwartz (baron-xaprb)]

review: Approve

[lp:~percona-toolkit-dev/percona-toolkit/pt-osc-2.1.1]

251. By Daniel Nichter on 2012-04-03

Add comment about code that will fail with new Cxn due to NAME_lc.

252. By Daniel Nichter on 2012-04-03

Catch lost cleanup task (should rarely happen).

253. By Daniel Nichter on 2012-04-03

Handle Key_name or key_name in NibbleIterator::_get_index_cardinality().

254. By Daniel Nichter on 2012-04-03

Add XXX comment about improperly copying .

255. By Daniel Nichter on 2012-04-03

Uncomment cleanup lines in alter_active_table.t.

256. By Daniel Nichter on 2012-04-03

Update modules in pt-online-schema-change.

257. By Daniel Nichter on 2012-04-03

Fix typo.

=== modified file 'bin/pt-online-schema-change'
--- bin/pt-online-schema-change	2012-03-30 16:37:16 +0000
+++ bin/pt-online-schema-change	2012-04-03 02:13:24 +0000
@@ -1784,245 +1784,246 @@
 # ###########################################################################
 
 # ###########################################################################
-# Transformers package
+# TableNibbler package
 # This package is a copy without comments from the original.  The original
 # with comments and its test file can be found in the Bazaar repository at,
-#   lib/Transformers.pm
-#   t/lib/Transformers.t
+#   lib/TableNibbler.pm
+#   t/lib/TableNibbler.t
 # See https://launchpad.net/percona-toolkit for more information.
 # ###########################################################################
 {
-package Transformers;
+package TableNibbler;
 
 use strict;
 use warnings FATAL => 'all';
 use English qw(-no_match_vars);
 use constant PTDEBUG => $ENV{PTDEBUG} || 0;
 
-use Time::Local qw(timegm timelocal);
-use Digest::MD5 qw(md5_hex);
-
-require Exporter;
-our @ISA         = qw(Exporter);
-our %EXPORT_TAGS = ();
-our @EXPORT      = ();
-our @EXPORT_OK   = qw(
-   micro_t
-   percentage_of
-   secs_to_time
-   time_to_secs
-   shorten
-   ts
-   parse_timestamp
-   unix_timestamp
-   any_unix_timestamp
-   make_checksum
-   crc32
-);
-
-our $mysql_ts  = qr/(\d\d)(\d\d)(\d\d) +(\d+):(\d+):(\d+)(\.\d+)?/;
-our $proper_ts = qr/(\d\d\d\d)-(\d\d)-(\d\d)[T ](\d\d):(\d\d):(\d\d)(\.\d+)?/;
-our $n_ts      = qr/(\d{1,5})([shmd]?)/; # Limit \d{1,5} because \d{6} looks
-
-sub micro_t {
-   my ( $t, %args ) = @_;
-   my $p_ms = defined $args{p_ms} ? $args{p_ms} : 0;  # precision for ms vals
-   my $p_s  = defined $args{p_s}  ? $args{p_s}  : 0;  # precision for s vals
-   my $f;
-
-   $t = 0 if $t < 0;
-
-   $t = sprintf('%.17f', $t) if $t =~ /e/;
-
-   $t =~ s/\.(\d{1,6})\d*/\.$1/;
-
-   if ($t > 0 && $t <= 0.000999) {
-      $f = ($t * 1000000) . 'us';
-   }
-   elsif ($t >= 0.001000 && $t <= 0.999999) {
-      $f = sprintf("%.${p_ms}f", $t * 1000);
-      $f = ($f * 1) . 'ms'; # * 1 to remove insignificant zeros
-   }
-   elsif ($t >= 1) {
-      $f = sprintf("%.${p_s}f", $t);
-      $f = ($f * 1) . 's'; # * 1 to remove insignificant zeros
-   }
-   else {
-      $f = 0;  # $t should = 0 at this point
-   }
-
-   return $f;
-}
-
-sub percentage_of {
-   my ( $is, $of, %args ) = @_;
-   my $p   = $args{p} || 0; # float precision
-   my $fmt = $p ? "%.${p}f" : "%d";
-   return sprintf $fmt, ($is * 100) / ($of ||= 1);
-}
-
-sub secs_to_time {
-   my ( $secs, $fmt ) = @_;
-   $secs ||= 0;
-   return '00:00' unless $secs;
-
-   $fmt ||= $secs >= 86_400 ? 'd'
-          : $secs >= 3_600  ? 'h'
-          :                   'm';
-
-   return
-      $fmt eq 'd' ? sprintf(
-         "%d+%02d:%02d:%02d",
-         int($secs / 86_400),
-         int(($secs % 86_400) / 3_600),
-         int(($secs % 3_600) / 60),
-         $secs % 60)
-      : $fmt eq 'h' ? sprintf(
-         "%02d:%02d:%02d",
-         int(($secs % 86_400) / 3_600),
-         int(($secs % 3_600) / 60),
-         $secs % 60)
-      : sprintf(
-         "%02d:%02d",
-         int(($secs % 3_600) / 60),
-         $secs % 60);
-}
-
-sub time_to_secs {
-   my ( $val, $default_suffix ) = @_;
-   die "I need a val argument" unless defined $val;
-   my $t = 0;
-   my ( $prefix, $num, $suffix ) = $val =~ m/([+-]?)(\d+)([a-z])?$/;
-   $suffix = $suffix || $default_suffix || 's';
-   if ( $suffix =~ m/[smhd]/ ) {
-      $t = $suffix eq 's' ? $num * 1        # Seconds
-         : $suffix eq 'm' ? $num * 60       # Minutes
-         : $suffix eq 'h' ? $num * 3600     # Hours
-         :                  $num * 86400;   # Days
-
-      $t *= -1 if $prefix && $prefix eq '-';
-   }
-   else {
-      die "Invalid suffix for $val: $suffix";
-   }
-   return $t;
-}
-
-sub shorten {
-   my ( $num, %args ) = @_;
-   my $p = defined $args{p} ? $args{p} : 2;     # float precision
-   my $d = defined $args{d} ? $args{d} : 1_024; # divisor
-   my $n = 0;
-   my @units = ('', qw(k M G T P E Z Y));
-   while ( $num >= $d && $n < @units - 1 ) {
-      $num /= $d;
-      ++$n;
-   }
-   return sprintf(
-      $num =~ m/\./ || $n
-         ? "%.${p}f%s"
-         : '%d',
-      $num, $units[$n]);
-}
-
-sub ts {
-   my ( $time, $gmt ) = @_;
-   my ( $sec, $min, $hour, $mday, $mon, $year )
-      = $gmt ? gmtime($time) : localtime($time);
-   $mon  += 1;
-   $year += 1900;
-   my $val = sprintf("%d-%02d-%02dT%02d:%02d:%02d",
-      $year, $mon, $mday, $hour, $min, $sec);
-   if ( my ($us) = $time =~ m/(\.\d+)$/ ) {
-      $us = sprintf("%.6f", $us);
-      $us =~ s/^0\././;
-      $val .= $us;
-   }
-   return $val;
-}
-
-sub parse_timestamp {
-   my ( $val ) = @_;
-   if ( my($y, $m, $d, $h, $i, $s, $f)
-         = $val =~ m/^$mysql_ts$/ )
-   {
-      return sprintf "%d-%02d-%02d %02d:%02d:"
-                     . (defined $f ? '%09.6f' : '%02d'),
-                     $y + 2000, $m, $d, $h, $i, (defined $f ? $s + $f : $s);
-   }
-   return $val;
-}
-
-sub unix_timestamp {
-   my ( $val, $gmt ) = @_;
-   if ( my($y, $m, $d, $h, $i, $s, $us) = $val =~ m/^$proper_ts$/ ) {
-      $val = $gmt
-         ? timegm($s, $i, $h, $d, $m - 1, $y)
-         : timelocal($s, $i, $h, $d, $m - 1, $y);
-      if ( defined $us ) {
-         $us = sprintf('%.6f', $us);
-         $us =~ s/^0\././;
-         $val .= $us;
-      }
-   }
-   return $val;
-}
-
-sub any_unix_timestamp {
-   my ( $val, $callback ) = @_;
-
-   if ( my ($n, $suffix) = $val =~ m/^$n_ts$/ ) {
-      $n = $suffix eq 's' ? $n            # Seconds
-         : $suffix eq 'm' ? $n * 60       # Minutes
-         : $suffix eq 'h' ? $n * 3600     # Hours
-         : $suffix eq 'd' ? $n * 86400    # Days
-         :                  $n;           # default: Seconds
-      PTDEBUG && _d('ts is now - N[shmd]:', $n);
-      return time - $n;
-   }
-   elsif ( $val =~ m/^\d{9,}/ ) {
-      PTDEBUG && _d('ts is already a unix timestamp');
-      return $val;
-   }
-   elsif ( my ($ymd, $hms) = $val =~ m/^(\d{6})(?:\s+(\d+:\d+:\d+))?/ ) {
-      PTDEBUG && _d('ts is MySQL slow log timestamp');
-      $val .= ' 00:00:00' unless $hms;
-      return unix_timestamp(parse_timestamp($val));
-   }
-   elsif ( ($ymd, $hms) = $val =~ m/^(\d{4}-\d\d-\d\d)(?:[T ](\d+:\d+:\d+))?/) {
-      PTDEBUG && _d('ts is properly formatted timestamp');
-      $val .= ' 00:00:00' unless $hms;
-      return unix_timestamp($val);
-   }
-   else {
-      PTDEBUG && _d('ts is MySQL expression');
-      return $callback->($val) if $callback && ref $callback eq 'CODE';
-   }
-
-   PTDEBUG && _d('Unknown ts type:', $val);
-   return;
-}
-
-sub make_checksum {
-   my ( $val ) = @_;
-   my $checksum = uc substr(md5_hex($val), -16);
-   PTDEBUG && _d($checksum, 'checksum for', $val);
-   return $checksum;
-}
-
-sub crc32 {
-   my ( $string ) = @_;
-   return unless $string;
-   my $poly = 0xEDB88320;
-   my $crc  = 0xFFFFFFFF;
-   foreach my $char ( split(//, $string) ) {
-      my $comp = ($crc ^ ord($char)) & 0xFF;
-      for ( 1 .. 8 ) {
-         $comp = $comp & 1 ? $poly ^ ($comp >> 1) : $comp >> 1;
-      }
-      $crc = (($crc >> 8) & 0x00FFFFFF) ^ $comp;
-   }
-   return $crc ^ 0xFFFFFFFF;
+sub new {
+   my ( $class, %args ) = @_;
+   my @required_args = qw(TableParser Quoter);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my $self = { %args };
+   return bless $self, $class;
+}
+
+sub generate_asc_stmt {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(tbl_struct index);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless defined $args{$arg};
+   }
+   my ($tbl_struct, $index) = @args{@required_args};
+   my @cols = $args{cols} ? @{$args{cols}} : @{$tbl_struct->{cols}};
+   my $q    = $self->{Quoter};
+
+   die "Index '$index' does not exist in table"
+      unless exists $tbl_struct->{keys}->{$index};
+   PTDEBUG && _d('Will ascend index', $index);  
+
+   my @asc_cols = @{$tbl_struct->{keys}->{$index}->{cols}};
+   if ( $args{asc_first} ) {
+      @asc_cols = $asc_cols[0];
+      PTDEBUG && _d('Ascending only first column');
+   }
+   PTDEBUG && _d('Will ascend columns', join(', ', @asc_cols));
+
+   my @asc_slice;
+   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};
+   }
+   PTDEBUG && _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_struct->{is_nullable},
+         );
+         $asc_stmt->{boundaries}->{$cmp} = $cmp_where->{where};
+      }
+      my $cmp = $args{asc_only} ? '>' : '>=';
+      $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 is_nullable) ) {
+      die "I need a $arg arg" unless defined $args{$arg};
+   }
+   my @slice       = @{$args{slice}};
+   my @cols        = @{$args{cols}};
+   my $is_nullable = $args{is_nullable};
+   my $type        = $args{type};
+   my $q           = $self->{Quoter};
+
+   (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) . ')';
+   my $where = {
+      slice => \@r_slice,
+      scols => \@r_scols,
+      where => $result,
+   };
+   return $where;
+}
+
+sub generate_del_stmt {
+   my ( $self, %args ) = @_;
+
+   my $tbl  = $args{tbl_struct};
+   my @cols = $args{cols} ? @{$args{cols}} : ();
+   my $tp   = $self->{TableParser};
+   my $q    = $self->{Quoter};
+
+   my @del_cols;
+   my @del_slice;
+
+   my $index = $tp->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}};
+   }
+   PTDEBUG && _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};
+   }
+   PTDEBUG && _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 ) = @_;
+   foreach my $arg ( qw(ins_tbl sel_cols) ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my $ins_tbl  = $args{ins_tbl};
+   my @sel_cols = @{$args{sel_cols}};
+
+   die "You didn't specify any SELECT columns" unless @sel_cols;
+
+   my @ins_cols;
+   my @ins_slice;
+   for my $i ( 0..$#sel_cols ) {
+      next unless $ins_tbl->{is_col}->{$sel_cols[$i]};
+      push @ins_cols, $sel_cols[$i];
+      push @ins_slice, $i;
+   }
+
+   return {
+      cols  => \@ins_cols,
+      slice => \@ins_slice,
+   };
 }
 
 sub _d {
@@ -2036,7 +2037,7 @@
 1;
 }
 # ###########################################################################
-# End Transformers package
+# End TableNibbler package
 # ###########################################################################
 
 # ###########################################################################
@@ -2077,41 +2078,43 @@
    die "I need a tbl parameter" unless $tbl;
    my $q = $self->{Quoter};
 
-   my $sql = '/*!40101 SET @OLD_SQL_MODE := @@SQL_MODE, '
-           . q{@@SQL_MODE := REPLACE(REPLACE(@@SQL_MODE, 'ANSI_QUOTES', ''), ',,', ','), }
-           . '@OLD_QUOTE := @@SQL_QUOTE_SHOW_CREATE, '
-           . '@@SQL_QUOTE_SHOW_CREATE := 1 */';
-   PTDEBUG && _d($sql);
-   eval { $dbh->do($sql); };
+   my $new_sql_mode
+      = '/*!40101 SET @OLD_SQL_MODE := @@SQL_MODE, '
+      . q{@@SQL_MODE := REPLACE(REPLACE(@@SQL_MODE, 'ANSI_QUOTES', ''), ',,', ','), }
+      . '@OLD_QUOTE := @@SQL_QUOTE_SHOW_CREATE, '
+      . '@@SQL_QUOTE_SHOW_CREATE := 1 */';
+
+   my $old_sql_mode = '/*!40101 SET @@SQL_MODE := @OLD_SQL_MODE, '
+                     . '@@SQL_QUOTE_SHOW_CREATE := @OLD_QUOTE */';
+
+   PTDEBUG && _d($new_sql_mode);
+   eval { $dbh->do($new_sql_mode); };
    PTDEBUG && $EVAL_ERROR && _d($EVAL_ERROR);
 
-   $sql = 'USE ' . $q->quote($db);
-   PTDEBUG && _d($dbh, $sql);
-   $dbh->do($sql);
+   my $use_sql = 'USE ' . $q->quote($db);
+   PTDEBUG && _d($dbh, $use_sql);
+   $dbh->do($use_sql);
 
-   $sql = "SHOW CREATE TABLE " . $q->quote($db, $tbl);
-   PTDEBUG && _d($sql);
+   my $show_sql = "SHOW CREATE TABLE " . $q->quote($db, $tbl);
+   PTDEBUG && _d($show_sql);
    my $href;
-   eval { $href = $dbh->selectrow_hashref($sql); };
+   eval { $href = $dbh->selectrow_hashref($show_sql); };
    if ( $EVAL_ERROR ) {
       PTDEBUG && _d($EVAL_ERROR);
+
+      PTDEBUG && _d($old_sql_mode);
+      $dbh->do($old_sql_mode);
+
       return;
    }
 
-   $sql = '/*!40101 SET @@SQL_MODE := @OLD_SQL_MODE, '
-        . '@@SQL_QUOTE_SHOW_CREATE := @OLD_QUOTE */';
-   PTDEBUG && _d($sql);
-   $dbh->do($sql);
+   PTDEBUG && _d($old_sql_mode);
+   $dbh->do($old_sql_mode);
 
-   my ($key) = grep { m/create table/i } keys %$href;
-   if ( $key ) {
-      PTDEBUG && _d('This table is a base table');
-      $href->{$key}  =~ s/\b[ ]{2,}/ /g;
-      $href->{$key} .= "\n";
-   }
-   else {
-      PTDEBUG && _d('This table is a view');
-      ($key) = grep { m/create view/i } keys %$href;
+   my ($key) = grep { m/create (?:table|view)/i } keys %$href;
+   if ( !$key ) {
+      die "Error: no 'Create Table' or 'Create View' in result set from "
+         . "$show_sql: " . Dumper($href);
    }
 
    return $href->{$key};
@@ -2468,6 +2471,7 @@
 # ###########################################################################
 
 # ###########################################################################
+<<<<<<< TREE
 # TableChunker package
 # This package is a copy without comments from the original.  The original
 # with comments and its test file can be found in the Bazaar repository at,
@@ -3399,6 +3403,8 @@
 # ###########################################################################
 
 # ###########################################################################
+=======
+>>>>>>> MERGE-SOURCE
 # Progress package
 # This package is a copy without comments from the original.  The original
 # with comments and its test file can be found in the Bazaar repository at,
@@ -3547,241 +3553,6 @@
 # ###########################################################################
 
 # ###########################################################################
-# OSCCaptureSync package
-# This package is a copy without comments from the original.  The original
-# with comments and its test file can be found in the Bazaar repository at,
-#   lib/OSCCaptureSync.pm
-#   t/lib/OSCCaptureSync.t
-# See https://launchpad.net/percona-toolkit for more information.
-# ###########################################################################
-{
-package OSCCaptureSync;
-
-use strict;
-use warnings FATAL => 'all';
-use English qw(-no_match_vars);
-use constant PTDEBUG => $ENV{PTDEBUG} || 0;
-
-sub new {
-   my ( $class, %args ) = @_;
-   my @required_args = qw(Quoter);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-
-   my $self = {
-      Quoter => $args{Quoter},
-   };
-
-   return bless $self, $class;
-}
-
-sub capture {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(msg dbh db tbl tmp_tbl columns chunk_column);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($msg, $dbh) = @args{@required_args};
-
-   my @triggers = $self->_make_triggers(%args);
-   foreach my $sql ( @triggers ) {
-      $msg->($sql);
-      $dbh->do($sql) unless $args{print};
-   }
-
-   return;
-}
-
-sub _make_triggers {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(db tbl tmp_tbl chunk_column columns);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($db, $tbl, $tmp_tbl, $chunk_column) = @args{@required_args};
-   my $q = $self->{Quoter};
-
-   $chunk_column = $q->quote($chunk_column);
-
-   my $old_table  = $q->quote($db, $tbl);
-   my $new_table  = $q->quote($db, $tmp_tbl);
-   my $new_values = join(', ', map { "NEW.".$q->quote($_) } @{$args{columns}});
-   my $columns    = join(', ', map { $q->quote($_) }        @{$args{columns}});
-
-   my $delete_trigger = "CREATE TRIGGER mk_osc_del AFTER DELETE ON $old_table "
-                      . "FOR EACH ROW "
-                      . "DELETE IGNORE FROM $new_table "
-                      . "WHERE $new_table.$chunk_column = OLD.$chunk_column";
-
-   my $insert_trigger = "CREATE TRIGGER mk_osc_ins AFTER INSERT ON $old_table "
-                      . "FOR EACH ROW "
-                      . "REPLACE INTO $new_table ($columns) "
-                      . "VALUES($new_values)";
-
-   my $update_trigger = "CREATE TRIGGER mk_osc_upd AFTER UPDATE ON $old_table "
-                      . "FOR EACH ROW "
-                      . "REPLACE INTO $new_table ($columns) "
-                      . "VALUES ($new_values)";
-
-   return $delete_trigger, $update_trigger, $insert_trigger;
-}
-
-sub sync {
-   my ( $self, %args ) = @_;
-   my @required_args = qw();
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   return;
-}
-
-sub cleanup {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db msg);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $db, $msg) = @args{@required_args};
-   my $q = $self->{Quoter};
-
-   foreach my $trigger ( qw(del ins upd) ) {
-      my $sql = "DROP TRIGGER IF EXISTS " . $q->quote($db, "mk_osc_$trigger");
-      $msg->($sql);
-      $dbh->do($sql) unless $args{print};
-   }
-
-   return;
-}
-
-sub _d {
-   my ($package, undef, $line) = caller 0;
-   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
-        map { defined $_ ? $_ : 'undef' }
-        @_;
-   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
-}
-
-1;
-}
-# ###########################################################################
-# End OSCCaptureSync package
-# ###########################################################################
-
-# ###########################################################################
-# CopyRowsInsertSelect package
-# This package is a copy without comments from the original.  The original
-# with comments and its test file can be found in the Bazaar repository at,
-#   lib/CopyRowsInsertSelect.pm
-#   t/lib/CopyRowsInsertSelect.t
-# See https://launchpad.net/percona-toolkit for more information.
-# ###########################################################################
-{
-package CopyRowsInsertSelect;
-
-use strict;
-use warnings FATAL => 'all';
-use English qw(-no_match_vars);
-use constant PTDEBUG => $ENV{PTDEBUG} || 0;
-
-sub new {
-   my ( $class, %args ) = @_;
-   my @required_args = qw(Retry Quoter);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-
-   my $self = {
-      Retry  => $args{Retry},
-      Quoter => $args{Quoter},
-   };
-
-   return bless $self, $class;
-}
-
-sub copy {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh msg from_table to_table chunks columns);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $msg, $from_table, $to_table, $chunks) = @args{@required_args};
-   my $q        = $self->{Quoter};
-   my $pr       = $args{Progress};
-   my $sleep    = $args{sleep};
-   my $columns  = join(', ', map { $q->quote($_) } @{$args{columns}});
-   my $n_chunks = @$chunks - 1;
-
-   for my $chunkno ( 0..$n_chunks ) {
-      if ( !$chunks->[$chunkno] ) {
-         warn "Chunk number ", ($chunkno + 1), "is undefined";
-         next;
-      }
-
-      my $sql = "INSERT IGNORE INTO $to_table ($columns) "
-              . "SELECT $columns FROM $from_table "
-              . "WHERE ($chunks->[$chunkno])"
-              . ($args{where}        ? " AND ($args{where})"  : "")
-              . ($args{engine_flags} ? " $args{engine_flags}" : "");
-
-      if ( $args{print} ) {
-         $msg->($sql);
-      }
-      else {
-         PTDEBUG && _d($dbh, $sql);
-         my $error;
-         $self->{Retry}->retry(
-            wait  => sub { sleep 1; },
-            tries => 3,
-            try   => sub {
-               $dbh->do($sql);
-               return;
-            },
-            fail => sub {
-               my (%args) = @_;
-               my $error = $args{error};
-               PTDEBUG && _d($error);
-               if ( $error =~ m/Lock wait timeout exceeded/ ) {
-                  $msg->("Lock wait timeout exceeded; retrying $sql");
-                  return 1; # call wait, call try
-               }
-               return 0; # call final_fail
-            },
-            final_fail => sub {
-               my (%args) = @_;
-               die $args{error};
-            },
-         );
-      }
-
-      $pr->update(sub { return $chunkno + 1; }) if $pr;
-
-      $sleep->($chunkno + 1) if $sleep && $chunkno < $n_chunks;
-   }
-
-   return;
-}
-
-sub cleanup {
-   my ( $self, %args ) = @_;
-   return;
-}
-
-sub _d {
-   my ($package, undef, $line) = caller 0;
-   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
-        map { defined $_ ? $_ : 'undef' }
-        @_;
-   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
-}
-
-1;
-}
-# ###########################################################################
-# End CopyRowsInsertSelect package
-# ###########################################################################
-
-# ###########################################################################
 # Retry package
 # This package is a copy without comments from the original.  The original
 # with comments and its test file can be found in the Bazaar repository at,
@@ -3860,6 +3631,2187 @@
 # ###########################################################################
 
 # ###########################################################################
+# Cxn package
+# This package is a copy without comments from the original.  The original
+# with comments and its test file can be found in the Bazaar repository at,
+#   lib/Cxn.pm
+#   t/lib/Cxn.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package Cxn;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+use constant PERCONA_TOOLKIT_TEST_USE_DSN_NAMES => $ENV{PERCONA_TOOLKIT_TEST_USE_DSN_NAMES} || 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   my @required_args = qw(DSNParser OptionParser);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   };
+   my ($dp, $o) = @args{@required_args};
+
+   my $dsn_defaults = $dp->parse_options($o);
+   my $prev_dsn     = $args{prev_dsn};
+   my $dsn          = $args{dsn};
+   if ( !$dsn ) {
+      $args{dsn_string} ||= 'h=' . ($dsn_defaults->{h} || 'localhost');
+
+      $dsn = $dp->parse(
+         $args{dsn_string}, $prev_dsn, $dsn_defaults);
+   }
+   elsif ( $prev_dsn ) {
+      $dsn = $dp->copy($prev_dsn, $dsn);
+   }
+
+   my $self = {
+      dsn          => $dsn,
+      dbh          => $args{dbh},
+      dsn_name     => $dp->as_string($dsn, [qw(h P S)]),
+      hostname     => '',
+      set          => $args{set},
+      dbh_set      => 0,
+      OptionParser => $o,
+      DSNParser    => $dp,
+   };
+
+   return bless $self, $class;
+}
+
+sub connect {
+   my ( $self ) = @_;
+   my $dsn = $self->{dsn};
+   my $dp  = $self->{DSNParser};
+   my $o   = $self->{OptionParser};
+
+   my $dbh = $self->{dbh};
+   if ( !$dbh || !$dbh->ping() ) {
+      if ( $o->get('ask-pass') && !$self->{asked_for_pass} ) {
+         $dsn->{p} = OptionParser::prompt_noecho("Enter MySQL password: ");
+         $self->{asked_for_pass} = 1;
+      }
+      $dbh = $dp->get_dbh($dp->get_cxn_params($dsn),  { AutoCommit => 1 });
+   }
+   PTDEBUG && _d($dbh, 'Connected dbh to', $self->{name});
+
+   return $self->set_dbh($dbh);
+}
+
+sub set_dbh {
+   my ($self, $dbh) = @_;
+
+   if ( $self->{dbh} && $self->{dbh} == $dbh && $self->{dbh_set} ) {
+      PTDEBUG && _d($dbh, 'Already set dbh');
+      return $dbh;
+   }
+
+   PTDEBUG && _d($dbh, 'Setting dbh');
+
+   $dbh->{FetchHashKeyName} = 'NAME_lc';
+   
+   my $sql = 'SELECT @@hostname, @@server_id';
+   PTDEBUG && _d($dbh, $sql);
+   my ($hostname, $server_id) = $dbh->selectrow_array($sql);
+   PTDEBUG && _d($dbh, 'hostname:', $hostname, $server_id);
+   if ( $hostname ) {
+      $self->{hostname} = $hostname;
+   }
+
+   if ( my $set = $self->{set}) {
+      $set->($dbh);
+   }
+
+   $self->{dbh}     = $dbh;
+   $self->{dbh_set} = 1;
+   return $dbh;
+}
+
+sub dbh {
+   my ($self) = @_;
+   return $self->{dbh};
+}
+
+sub dsn {
+   my ($self) = @_;
+   return $self->{dsn};
+}
+
+sub name {
+   my ($self) = @_;
+   return $self->{dsn_name} if PERCONA_TOOLKIT_TEST_USE_DSN_NAMES;
+   return $self->{hostname} || $self->{dsn_name} || 'unknown host';
+}
+
+sub DESTROY {
+   my ($self) = @_;
+   if ( $self->{dbh} ) {
+      PTDEBUG && _d('Disconnecting dbh', $self->{dbh}, $self->{name});
+      $self->{dbh}->disconnect();
+   }
+   return;
+}
+
+sub _d {
+   my ($package, undef, $line) = caller 0;
+   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
+        map { defined $_ ? $_ : 'undef' }
+        @_;
+   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
+}
+
+1;
+}
+# ###########################################################################
+# End Cxn package
+# ###########################################################################
+
+# ###########################################################################
+# MasterSlave package
+# This package is a copy without comments from the original.  The original
+# with comments and its test file can be found in the Bazaar repository at,
+#   lib/MasterSlave.pm
+#   t/lib/MasterSlave.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package MasterSlave;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   my $self = {
+      %args,
+      replication_thread => {},
+   };
+   return bless $self, $class;
+}
+
+sub get_slaves {
+   my ($self, %args) = @_;
+   my @required_args = qw(make_cxn OptionParser DSNParser Quoter);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($make_cxn, $o, $dp) = @args{@required_args};
+
+   my $slaves = [];
+   my $method = $o->get('recursion-method');
+   PTDEBUG && _d('Slave recursion method:', $method);
+   if ( !$method || $method =~ m/processlist|hosts/i ) {
+      my @required_args = qw(dbh dsn);
+      foreach my $arg ( @required_args ) {
+         die "I need a $arg argument" unless $args{$arg};
+      }
+      my ($dbh, $dsn) = @args{@required_args};
+      $self->recurse_to_slaves(
+         {  dbh        => $dbh,
+            dsn        => $dsn,
+            dsn_parser => $dp,
+            recurse    => $o->get('recurse'),
+            method     => $o->get('recursion-method'),
+            callback   => sub {
+               my ( $dsn, $dbh, $level, $parent ) = @_;
+               return unless $level;
+               PTDEBUG && _d('Found slave:', $dp->as_string($dsn));
+               push @$slaves, $make_cxn->(dsn => $dsn, dbh => $dbh);
+               return;
+            },
+         }
+      );
+   }
+   elsif ( $method =~ m/^dsn=/i ) {
+      my ($dsn_table_dsn) = $method =~ m/^dsn=(.+)/i;
+      $slaves = $self->get_cxn_from_dsn_table(
+         %args,
+         dsn_table_dsn => $dsn_table_dsn,
+      );
+   }
+   else {
+      die "Invalid --recursion-method: $method.  Valid values are: "
+        . "dsn=DSN, hosts, or processlist.\n";
+   }
+
+   return $slaves;
+}
+
+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 });
+      PTDEBUG && _d('Connected to', $dp->as_string($dsn));
+   };
+   if ( $EVAL_ERROR ) {
+      print STDERR "Cannot connect to ", $dp->as_string($dsn), "\n"
+         or die "Cannot print: $OS_ERROR";
+      return;
+   }
+
+   my $sql  = 'SELECT @@SERVER_ID';
+   PTDEBUG && _d($sql);
+   my ($id) = $dbh->selectrow_array($sql);
+   PTDEBUG && _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}++
+   ) {
+      PTDEBUG && _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 ) {
+         PTDEBUG && _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 ) = @_;
+
+   my @methods = qw(processlist hosts);
+   if ( $method ) {
+      @methods = grep { $_ ne $method } @methods;
+      unshift @methods, $method;
+   }
+   else {
+      if ( ($dsn->{P} || 3306) != 3306 ) {
+         PTDEBUG && _d('Port number is non-standard; using only hosts method');
+         @methods = qw(hosts);
+      }
+   }
+   PTDEBUG && _d('Looking for slaves on', $dsn_parser->as_string($dsn),
+      'using methods', @methods);
+
+   my @slaves;
+   METHOD:
+   foreach my $method ( @methods ) {
+      my $find_slaves = "_find_slaves_by_$method";
+      PTDEBUG && _d('Finding slaves with', $find_slaves);
+      @slaves = $self->$find_slaves($dsn_parser, $dbh, $dsn);
+      last METHOD if @slaves;
+   }
+
+   PTDEBUG && _d('Found', scalar(@slaves), 'slaves');
+   return @slaves;
+}
+
+sub _find_slaves_by_processlist {
+   my ( $self, $dsn_parser, $dbh, $dsn ) = @_;
+
+   my @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;
+   } $self->get_connected_slaves($dbh);
+
+   return @slaves;
+}
+
+sub _find_slaves_by_hosts {
+   my ( $self, $dsn_parser, $dbh, $dsn ) = @_;
+
+   my @slaves;
+   my $sql = 'SHOW SLAVE HOSTS';
+   PTDEBUG && _d($dbh, $sql);
+   @slaves = @{$dbh->selectall_arrayref($sql, { Slice => {} })};
+
+   if ( @slaves ) {
+      PTDEBUG && _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;
+   }
+
+   return @slaves;
+}
+
+sub get_connected_slaves {
+   my ( $self, $dbh ) = @_;
+
+   my $show = "SHOW GRANTS FOR ";
+   my $user = 'CURRENT_USER()';
+   my $vp   = $self->{VersionParser};
+   if ( $vp && !$vp->version_ge($dbh, '4.1.2') ) {
+      $user = $dbh->selectrow_arrayref('SELECT USER()')->[0];
+      $user =~ s/([^@]+)@(.+)/'$1'\@'$2'/;
+   }
+   my $sql = $show . $user;
+   PTDEBUG && _d($dbh, $sql);
+
+   my $proc;
+   eval {
+      $proc = grep {
+         m/ALL PRIVILEGES.*?\*\.\*|PROCESS/
+      } @{$dbh->selectcol_arrayref($sql)};
+   };
+   if ( $EVAL_ERROR ) {
+
+      if ( $EVAL_ERROR =~ m/no such grant defined for user/ ) {
+         PTDEBUG && _d('Retrying SHOW GRANTS without host; error:',
+            $EVAL_ERROR);
+         ($user) = split('@', $user);
+         $sql    = $show . $user;
+         PTDEBUG && _d($sql);
+         eval {
+            $proc = grep {
+               m/ALL PRIVILEGES.*?\*\.\*|PROCESS/
+            } @{$dbh->selectcol_arrayref($sql)};
+         };
+      }
+
+      die "Failed to $sql: $EVAL_ERROR" if $EVAL_ERROR;
+   }
+   if ( !$proc ) {
+      die "You do not have the PROCESS privilege";
+   }
+
+   $sql = 'SHOW PROCESSLIST';
+   PTDEBUG && _d($dbh, $sql);
+   grep { $_->{command} =~ m/Binlog Dump/i }
+   map  { # Lowercase the column names
+      my %hash;
+      @hash{ map { lc $_ } keys %$_ } = values %$_;
+      \%hash;
+   }
+   @{$dbh->selectall_arrayref($sql, { Slice => {} })};
+}
+
+sub is_master_of {
+   my ( $self, $master, $slave ) = @_;
+   my $master_status = $self->get_master_status($master)
+      or die "The server specified as a master is not a master";
+   my $slave_status  = $self->get_slave_status($slave)
+      or die "The server specified as a slave is not a slave";
+   my @connected     = $self->get_connected_slaves($master)
+      or die "The server specified as a master has no connected slaves";
+   my (undef, $port) = $master->selectrow_array('SHOW VARIABLES LIKE "port"');
+
+   if ( $port != $slave_status->{master_port} ) {
+      die "The slave is connected to $slave_status->{master_port} "
+         . "but the master's port is $port";
+   }
+
+   if ( !grep { $slave_status->{master_user} eq $_->{user} } @connected ) {
+      die "I don't see any slave I/O thread connected with user "
+         . $slave_status->{master_user};
+   }
+
+   if ( ($slave_status->{slave_io_state} || '')
+      eq 'Waiting for master to send event' )
+   {
+      my ( $master_log_name, $master_log_num )
+         = $master_status->{file} =~ m/^(.*?)\.0*([1-9][0-9]*)$/;
+      my ( $slave_log_name, $slave_log_num )
+         = $slave_status->{master_log_file} =~ m/^(.*?)\.0*([1-9][0-9]*)$/;
+      if ( $master_log_name ne $slave_log_name
+         || abs($master_log_num - $slave_log_num) > 1 )
+      {
+         die "The slave thinks it is reading from "
+            . "$slave_status->{master_log_file},  but the "
+            . "master is writing to $master_status->{file}";
+      }
+   }
+   return 1;
+}
+
+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');
+      PTDEBUG && _d($dbh, 'SHOW SLAVE STATUS');
+      $sth->execute();
+      my ($ss) = @{$sth->fetchall_arrayref({})};
+
+      if ( $ss && %$ss ) {
+         $ss = { map { lc($_) => $ss->{$_} } keys %$ss }; # lowercase the keys
+         return $ss;
+      }
+
+      PTDEBUG && _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} ) {
+      PTDEBUG && _d('Server on dbh', $dbh, 'is not a master');
+      return;
+   }
+
+   my $sth = $self->{sths}->{$dbh}->{MASTER_STATUS}
+         ||= $dbh->prepare('SHOW MASTER STATUS');
+   PTDEBUG && _d($dbh, 'SHOW MASTER STATUS');
+   $sth->execute();
+   my ($ms) = @{$sth->fetchall_arrayref({})};
+   PTDEBUG && _d(
+      $ms ? map { "$_=" . (defined $ms->{$_} ? $ms->{$_} : '') } keys %$ms
+          : '');
+
+   if ( !$ms || scalar keys %$ms < 2 ) {
+      PTDEBUG && _d('Server on dbh', $dbh, 'does not seem to be a master');
+      $self->{not_a_master}->{$dbh}++;
+   }
+
+  return { map { lc($_) => $ms->{$_} } keys %$ms }; # lowercase the keys
+}
+
+sub wait_for_master {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(master_status slave_dbh);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($master_status, $slave_dbh) = @args{@required_args};
+   my $timeout       = $args{timeout} || 60;
+
+   my $result;
+   my $waited;
+   if ( $master_status ) {
+      my $sql = "SELECT MASTER_POS_WAIT('$master_status->{file}', "
+              . "$master_status->{position}, $timeout)";
+      PTDEBUG && _d($slave_dbh, $sql);
+      my $start = time;
+      ($result) = $slave_dbh->selectrow_array($sql);
+
+      $waited = time - $start;
+
+      PTDEBUG && _d('Result of waiting:', $result);
+      PTDEBUG && _d("Waited", $waited, "seconds");
+   }
+   else {
+      PTDEBUG && _d('Not waiting: this server is not a master');
+   }
+
+   return {
+      result => $result,
+      waited => $waited,
+   };
+}
+
+sub stop_slave {
+   my ( $self, $dbh ) = @_;
+   my $sth = $self->{sths}->{$dbh}->{STOP_SLAVE}
+         ||= $dbh->prepare('STOP SLAVE');
+   PTDEBUG && _d($dbh, $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}";
+      PTDEBUG && _d($dbh, $sql);
+      $dbh->do($sql);
+   }
+   else {
+      my $sth = $self->{sths}->{$dbh}->{START_SLAVE}
+            ||= $dbh->prepare('START SLAVE');
+      PTDEBUG && _d($dbh, $sth->{Statement});
+      $sth->execute();
+   }
+}
+
+sub catchup_to_master {
+   my ( $self, $slave, $master, $timeout ) = @_;
+   $self->stop_slave($master);
+   $self->stop_slave($slave);
+   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);
+   PTDEBUG && _d('Master position:', $self->pos_to_string($master_pos),
+      'Slave position:', $self->pos_to_string($slave_pos));
+
+   my $result;
+   if ( $self->pos_cmp($slave_pos, $master_pos) < 0 ) {
+      PTDEBUG && _d('Waiting for slave to catch up to master');
+      $self->start_slave($slave, $master_pos);
+
+      $result = $self->wait_for_master(
+            master_status => $master_status,
+            slave_dbh     => $slave,
+            timeout       => $timeout,
+            master_status => $master_status
+      );
+      if ( !defined $result->{result} ) {
+         $slave_status = $self->get_slave_status($slave);
+         if ( !$self->slave_is_running($slave_status) ) {
+            PTDEBUG && _d('Master position:',
+               $self->pos_to_string($master_pos),
+               'Slave position:', $self->pos_to_string($slave_pos));
+            $slave_pos = $self->repl_posn($slave_status);
+            if ( $self->pos_cmp($slave_pos, $master_pos) != 0 ) {
+               die "MASTER_POS_WAIT() returned NULL but slave has not "
+                  . "caught up to master";
+            }
+            PTDEBUG && _d('Slave is caught up to master and stopped');
+         }
+         else {
+            die "Slave has not caught up to master and it is still running";
+         }
+      }
+   }
+   else {
+      PTDEBUG && _d("Slave is already caught up to master");
+   }
+
+   return $result;
+}
+
+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 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'};
+   PTDEBUG && _d($dbh, $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 get_slave_lag {
+   my ( $self, $dbh ) = @_;
+   my $stat = $self->get_slave_status($dbh);
+   return unless $stat;  # server is not a slave
+   return $stat->{seconds_behind_master};
+}
+
+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 is_replication_thread {
+   my ( $self, $query, %args ) = @_; 
+   return unless $query;
+
+   my $type = lc($args{type} || 'all');
+   die "Invalid type: $type"
+      unless $type =~ m/^binlog_dump|slave_io|slave_sql|all$/i;
+
+   my $match = 0;
+   if ( $type =~ m/binlog_dump|all/i ) {
+      $match = 1
+         if ($query->{Command} || $query->{command} || '') eq "Binlog Dump";
+   }
+   if ( !$match ) {
+      if ( ($query->{User} || $query->{user} || '') eq "system user" ) {
+         PTDEBUG && _d("Slave replication thread");
+         if ( $type ne 'all' ) { 
+            my $state = $query->{State} || $query->{state} || '';
+
+            if ( $state =~ m/^init|end$/ ) {
+               PTDEBUG && _d("Special state:", $state);
+               $match = 1;
+            }
+            else {
+               my ($slave_sql) = $state =~ m/
+                  ^(Waiting\sfor\sthe\snext\sevent
+                   |Reading\sevent\sfrom\sthe\srelay\slog
+                   |Has\sread\sall\srelay\slog;\swaiting
+                   |Making\stemp\sfile
+                   |Waiting\sfor\sslave\smutex\son\sexit)/xi; 
+
+               $match = $type eq 'slave_sql' &&  $slave_sql ? 1
+                      : $type eq 'slave_io'  && !$slave_sql ? 1
+                      :                                       0;
+            }
+         }
+         else {
+            $match = 1;
+         }
+      }
+      else {
+         PTDEBUG && _d('Not system user');
+      }
+
+      if ( !defined $args{check_known_ids} || $args{check_known_ids} ) {
+         my $id = $query->{Id} || $query->{id};
+         if ( $match ) {
+            $self->{replication_thread}->{$id} = 1;
+         }
+         else {
+            if ( $self->{replication_thread}->{$id} ) {
+               PTDEBUG && _d("Thread ID is a known replication thread ID");
+               $match = 1;
+            }
+         }
+      }
+   }
+
+   PTDEBUG && _d('Matches', $type, 'replication thread:',
+      ($match ? 'yes' : 'no'), '; match:', $match);
+
+   return $match;
+}
+
+
+sub get_replication_filters {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(dbh);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($dbh) = @args{@required_args};
+
+   my %filters = ();
+
+   my $status = $self->get_master_status($dbh);
+   if ( $status ) {
+      map { $filters{$_} = $status->{$_} }
+      grep { defined $status->{$_} && $status->{$_} ne '' }
+      qw(
+         binlog_do_db
+         binlog_ignore_db
+      );
+   }
+
+   $status = $self->get_slave_status($dbh);
+   if ( $status ) {
+      map { $filters{$_} = $status->{$_} }
+      grep { defined $status->{$_} && $status->{$_} ne '' }
+      qw(
+         replicate_do_db
+         replicate_ignore_db
+         replicate_do_table
+         replicate_ignore_table 
+         replicate_wild_do_table
+         replicate_wild_ignore_table
+      );
+
+      my $sql = "SHOW VARIABLES LIKE 'slave_skip_errors'";
+      PTDEBUG && _d($dbh, $sql);
+      my $row = $dbh->selectrow_arrayref($sql);
+      $filters{slave_skip_errors} = $row->[1] if $row->[1] && $row->[1] ne 'OFF';
+   }
+
+   return \%filters; 
+}
+
+
+sub pos_to_string {
+   my ( $self, $pos ) = @_;
+   my $fmt  = '%s/%020d';
+   return sprintf($fmt, @{$pos}{qw(file position)});
+}
+
+sub reset_known_replication_threads {
+   my ( $self ) = @_;
+   $self->{replication_thread} = {};
+   return;
+}
+
+sub get_cxn_from_dsn_table {
+   my ($self, %args) = @_;
+   my @required_args = qw(dsn_table_dsn make_cxn DSNParser Quoter);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($dsn_table_dsn, $make_cxn, $dp, $q) = @args{@required_args};
+   PTDEBUG && _d('DSN table DSN:', $dsn_table_dsn);
+
+   my $dsn = $dp->parse($dsn_table_dsn);
+   my $dsn_table;
+   if ( $dsn->{D} && $dsn->{t} ) {
+      $dsn_table = $q->quote($dsn->{D}, $dsn->{t});
+   }
+   elsif ( $dsn->{t} && $dsn->{t} =~ m/\./ ) {
+      $dsn_table = $q->quote($q->split_unquote($dsn->{t}));
+   }
+   else {
+      die "DSN table DSN does not specify a database (D) "
+        . "or a database-qualified table (t)";
+   }
+
+   my $dsn_tbl_cxn = $make_cxn->(dsn => $dsn);
+   my $dbh         = $dsn_tbl_cxn->connect();
+   my $sql         = "SELECT dsn FROM $dsn_table ORDER BY id";
+   PTDEBUG && _d($sql);
+   my $dsn_strings = $dbh->selectcol_arrayref($sql);
+   my @cxn;
+   if ( $dsn_strings ) {
+      foreach my $dsn_string ( @$dsn_strings ) {
+         PTDEBUG && _d('DSN from DSN table:', $dsn_string);
+         push @cxn, $make_cxn->(dsn_string => $dsn_string);
+      }
+   }
+   return \@cxn;
+}
+
+sub _d {
+   my ($package, undef, $line) = caller 0;
+   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
+        map { defined $_ ? $_ : 'undef' }
+        @_;
+   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
+}
+
+1;
+}
+# ###########################################################################
+# End MasterSlave package
+# ###########################################################################
+
+# ###########################################################################
+# ReplicaLagWaiter package
+# This package is a copy without comments from the original.  The original
+# with comments and its test file can be found in the Bazaar repository at,
+#   lib/ReplicaLagWaiter.pm
+#   t/lib/ReplicaLagWaiter.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package ReplicaLagWaiter;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+use Time::HiRes qw(sleep time);
+use Data::Dumper;
+
+sub new {
+   my ( $class, %args ) = @_;
+   my @required_args = qw(oktorun get_lag sleep max_lag slaves);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless defined $args{$arg};
+   }
+
+   my $self = {
+      %args,
+   };
+
+   return bless $self, $class;
+}
+
+sub wait {
+   my ( $self, %args ) = @_;
+   my @required_args = qw();
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my $pr = $args{Progress};
+
+   my $oktorun = $self->{oktorun};
+   my $get_lag = $self->{get_lag};
+   my $sleep   = $self->{sleep};
+   my $slaves  = $self->{slaves};
+   my $max_lag = $self->{max_lag};
+
+   my $worst;  # most lagging slave
+   my $pr_callback;
+   my $pr_first_report;
+   if ( $pr ) {
+      $pr_callback = sub {
+         my ($fraction, $elapsed, $remaining, $eta, $completed) = @_;
+         my $dsn_name = $worst->{cxn}->name();
+         if ( defined $worst->{lag} ) {
+            print STDERR "Replica lag is " . ($worst->{lag} || '?')
+               . " seconds on $dsn_name.  Waiting.\n";
+         }
+         else {
+            print STDERR "Replica $dsn_name is stopped.  Waiting.\n";
+         }
+         return;
+      };
+      $pr->set_callback($pr_callback);
+
+      $pr_first_report = sub {
+         my $dsn_name = $worst->{cxn}->name();
+         if ( !defined $worst->{lag} ) {
+            print STDERR "Replica $dsn_name is stopped.  Waiting.\n";
+         }
+         return;
+      };
+   }
+
+   my @lagged_slaves = map { {cxn=>$_, lag=>undef} } @$slaves;  
+   while ( $oktorun->() && @lagged_slaves ) {
+      PTDEBUG && _d('Checking slave lag');
+      for my $i ( 0..$#lagged_slaves ) {
+         my $lag = $get_lag->($lagged_slaves[$i]->{cxn});
+         PTDEBUG && _d($lagged_slaves[$i]->{cxn}->name(),
+            'slave lag:', $lag);
+         if ( !defined $lag || $lag > $max_lag ) {
+            $lagged_slaves[$i]->{lag} = $lag;
+         }
+         else {
+            delete $lagged_slaves[$i];
+         }
+      }
+
+      @lagged_slaves = grep { defined $_ } @lagged_slaves;
+      if ( @lagged_slaves ) {
+         @lagged_slaves = reverse sort {
+              defined $a->{lag} && defined $b->{lag} ? $a->{lag} <=> $b->{lag}
+            : defined $a->{lag}                      ? -1
+            :                                           1;
+         } @lagged_slaves;
+         $worst = $lagged_slaves[0];
+         PTDEBUG && _d(scalar @lagged_slaves, 'slaves are lagging, worst:',
+            $worst->{lag}, 'on', Dumper($worst->{cxn}->dsn()));
+
+         if ( $pr ) {
+            $pr->update(
+               sub { return 0; },
+               first_report => $pr_first_report,
+            );
+         }
+
+         PTDEBUG && _d('Calling sleep callback');
+         $sleep->($worst->{cxn}, $worst->{lag});
+      }
+   }
+
+   PTDEBUG && _d('All slaves caught up');
+   return;
+}
+
+sub _d {
+   my ($package, undef, $line) = caller 0;
+   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
+        map { defined $_ ? $_ : 'undef' }
+        @_;
+   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
+}
+
+1;
+}
+# ###########################################################################
+# End ReplicaLagWaiter package
+# ###########################################################################
+
+# ###########################################################################
+# MySQLStatusWaiter package
+# This package is a copy without comments from the original.  The original
+# with comments and its test file can be found in the Bazaar repository at,
+#   lib/MySQLStatusWaiter.pm
+#   t/lib/MySQLStatusWaiter.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package MySQLStatusWaiter;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   my @required_args = qw(max_spec get_status sleep oktorun);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless defined $args{$arg};
+   }
+
+   PTDEBUG && _d('Parsing spec for max thresholds');
+   my $max_val_for = _parse_spec(
+      spec             => $args{max_spec},
+      get_status       => $args{get_status},
+      threshold_factor => 0.2, # +20%
+   );
+
+   PTDEBUG && _d('Parsing spec for critical thresholds');
+   my $critical_val_for = _parse_spec(
+      spec             => $args{critical_spec} || [],
+      get_status       => $args{get_status},
+      threshold_factor => 1.0, # double (x2; +100%)
+   );
+
+   my $self = {
+      get_status       => $args{get_status},
+      sleep            => $args{sleep},
+      oktorun          => $args{oktorun},
+      max_val_for      => $max_val_for,
+      critical_val_for => $critical_val_for,
+   };
+
+   return bless $self, $class;
+}
+
+sub _parse_spec {
+   my ( %args ) = @_;
+   my @required_args = qw(spec get_status);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless defined $args{$arg};
+   }
+   my ($spec, $get_status) = @args{@required_args};
+
+   return unless $spec && scalar @$spec;
+   my $threshold_factor = $args{threshold_factor} || 0.20;
+
+   my %max_val_for;
+   foreach my $var_val ( @$spec ) {
+      my ($var, $val) = split /[:=]/, $var_val;
+      die "Invalid spec: $var_val" unless $var;
+      if ( !$val ) {
+         my $init_val = $get_status->($var);
+         PTDEBUG && _d('Initial', $var, 'value:', $init_val);
+         $val = int(($init_val * $threshold_factor) + $init_val);
+      }
+      PTDEBUG && _d('Wait if', $var, '>=', $val);
+      $max_val_for{$var} = $val;
+   }
+
+   return \%max_val_for; 
+}
+
+sub max_values {
+   my ($self) = @_;
+   return $self->{max_val_for};
+}
+
+sub critical_values {
+   my ($self) = @_;
+   return $self->{critical_val_for};
+}
+
+sub wait {
+   my ( $self, %args ) = @_;
+
+   return unless $self->{max_val_for};
+
+   my $pr = $args{Progress}; # optional
+
+   my $oktorun    = $self->{oktorun};
+   my $get_status = $self->{get_status};
+   my $sleep      = $self->{sleep};
+
+   my %vals_too_high = %{$self->{max_val_for}};
+   my $pr_callback;
+   if ( $pr ) {
+      $pr_callback = sub {
+         print STDERR "Pausing because "
+            . join(', ',
+                 map {
+                    "$_="
+                    . (defined $vals_too_high{$_} ? $vals_too_high{$_}
+                                                  : 'unknown')
+                 } sort keys %vals_too_high
+              )
+            . ".\n";
+         return;
+      };
+      $pr->set_callback($pr_callback);
+   }
+
+   while ( $oktorun->() ) {
+      PTDEBUG && _d('Checking status variables');
+      foreach my $var ( sort keys %vals_too_high ) {
+         my $val = $get_status->($var);
+         PTDEBUG && _d($var, '=', $val);
+         if ( $val
+              && exists $self->{critical_val_for}->{$var}
+              && $val >= $self->{critical_val_for}->{$var} ) {
+            die "$var=$val exceeds its critical threshold "
+               . "$self->{critical_val_for}->{$var}\n";
+         }
+         if ( !$val || $val >= $self->{max_val_for}->{$var} ) {
+            $vals_too_high{$var} = $val;
+         }
+         else {
+            delete $vals_too_high{$var};
+         }
+      }
+
+      last unless scalar keys %vals_too_high;
+
+      PTDEBUG && _d(scalar keys %vals_too_high, 'values are too high:',
+         %vals_too_high);
+      if ( $pr ) {
+         $pr->update(sub { return 0; });
+      }
+      PTDEBUG && _d('Calling sleep callback');
+      $sleep->();
+      %vals_too_high = %{$self->{max_val_for}}; # recheck all vars
+   }
+
+   PTDEBUG && _d('All var vals are low enough');
+   return;
+}
+
+sub _d {
+   my ($package, undef, $line) = caller 0;
+   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
+        map { defined $_ ? $_ : 'undef' }
+        @_;
+   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
+}
+
+1;
+}
+# ###########################################################################
+# End MySQLStatusWaiter package
+# ###########################################################################
+
+# ###########################################################################
+# WeightedAvgRate package
+# This package is a copy without comments from the original.  The original
+# with comments and its test file can be found in the Bazaar repository at,
+#   lib/WeightedAvgRate.pm
+#   t/lib/WeightedAvgRate.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package WeightedAvgRate;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   my @required_args = qw(target_t);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless defined $args{$arg};
+   }
+
+   my $self = {
+      %args,
+      avg_n  => 0,
+      avg_t  => 0,
+      weight => $args{weight} || 0.75,
+   };
+
+   return bless $self, $class;
+}
+
+sub update {
+   my ($self, $n, $t) = @_;
+   PTDEBUG && _d('Master op time:', $n, 'n /', $t, 's');
+
+   if ( $self->{avg_n} && $self->{avg_t} ) {
+      $self->{avg_n}    = ($self->{avg_n} * $self->{weight}) + $n;
+      $self->{avg_t}    = ($self->{avg_t} * $self->{weight}) + $t;
+      $self->{avg_rate} = $self->{avg_n}  / $self->{avg_t};
+      PTDEBUG && _d('Weighted avg rate:', $self->{avg_rate}, 'n/s');
+   }
+   else {
+      $self->{avg_n}    = $n;
+      $self->{avg_t}    = $t;
+      $self->{avg_rate} = $self->{avg_n}  / $self->{avg_t};
+      PTDEBUG && _d('Initial avg rate:', $self->{avg_rate}, 'n/s');
+   }
+
+   my $new_n = int($self->{avg_rate} * $self->{target_t});
+   PTDEBUG && _d('Adjust n to', $new_n);
+   return $new_n;
+}
+
+sub _d {
+   my ($package, undef, $line) = caller 0;
+   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
+        map { defined $_ ? $_ : 'undef' }
+        @_;
+   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
+}
+
+1;
+}
+# ###########################################################################
+# End WeightedAvgRate package
+# ###########################################################################
+
+# ###########################################################################
+# NibbleIterator package
+# This package is a copy without comments from the original.  The original
+# with comments and its test file can be found in the Bazaar repository at,
+#   lib/NibbleIterator.pm
+#   t/lib/NibbleIterator.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package NibbleIterator;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+use Data::Dumper;
+$Data::Dumper::Indent    = 1;
+$Data::Dumper::Sortkeys  = 1;
+$Data::Dumper::Quotekeys = 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   my @required_args = qw(Cxn tbl chunk_size OptionParser Quoter TableNibbler TableParser);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($cxn, $tbl, $chunk_size, $o, $q) = @args{@required_args};
+
+   my $nibble_params = can_nibble(%args);
+
+   my %comments = (
+      bite   => "bite table",
+      nibble => "nibble table",
+   );
+   if ( $args{comments} ) {
+      map  { $comments{$_} = $args{comments}->{$_} }
+      grep { defined $args{comments}->{$_}         }
+      keys %{$args{comments}};
+   }
+
+   my $where      = $o->has('where') ? $o->get('where') : '';
+   my $tbl_struct = $tbl->{tbl_struct};
+   my $ignore_col = $o->has('ignore-columns')
+                  ? ($o->get('ignore-columns') || {})
+                  : {};
+   my $all_cols   = $o->has('columns')
+                  ? ($o->get('columns') || $tbl_struct->{cols})
+                  : $tbl_struct->{cols};
+   my @cols       = grep { !$ignore_col->{$_} } @$all_cols;
+   my $self;
+   if ( $nibble_params->{one_nibble} ) {
+      my $nibble_sql
+         = ($args{dml} ? "$args{dml} " : "SELECT ")
+         . ($args{select} ? $args{select}
+                          : join(', ', map { $q->quote($_) } @cols))
+         . " FROM $tbl->{name}"
+         . ($where ? " WHERE $where" : '')
+         . " /*$comments{bite}*/";
+      PTDEBUG && _d('One nibble statement:', $nibble_sql);
+
+      my $explain_nibble_sql
+         = "EXPLAIN SELECT "
+         . ($args{select} ? $args{select}
+                          : join(', ', map { $q->quote($_) } @cols))
+         . " FROM $tbl->{name}"
+         . ($where ? " WHERE $where" : '')
+         . " /*explain $comments{bite}*/";
+      PTDEBUG && _d('Explain one nibble statement:', $explain_nibble_sql);
+
+      $self = {
+         %args,
+         one_nibble         => 1,
+         limit              => 0,
+         nibble_sql         => $nibble_sql,
+         explain_nibble_sql => $explain_nibble_sql,
+      };
+   }
+   else {
+      my $index      = $nibble_params->{index}; # brevity
+      my $index_cols = $tbl->{tbl_struct}->{keys}->{$index}->{cols};
+
+      my $asc = $args{TableNibbler}->generate_asc_stmt(
+         %args,
+         tbl_struct => $tbl->{tbl_struct},
+         index      => $index,
+         cols       => \@cols,
+         asc_only   => 1,
+      );
+      PTDEBUG && _d('Ascend params:', Dumper($asc));
+
+      my $from     = "$tbl->{name} FORCE INDEX(`$index`)";
+      my $order_by = join(', ', map {$q->quote($_)} @{$index_cols});
+
+      my $first_lb_sql
+         = "SELECT /*!40001 SQL_NO_CACHE */ "
+         . join(', ', map { $q->quote($_) } @{$asc->{scols}})
+         . " FROM $from"
+         . ($where ? " WHERE $where" : '')
+         . " ORDER BY $order_by"
+         . " LIMIT 1"
+         . " /*first lower boundary*/";
+      PTDEBUG && _d('First lower boundary statement:', $first_lb_sql);
+
+      my $resume_lb_sql;
+      if ( $args{resume} ) {
+         $resume_lb_sql
+            = "SELECT /*!40001 SQL_NO_CACHE */ "
+            . join(', ', map { $q->quote($_) } @{$asc->{scols}})
+            . " FROM $from"
+            . " WHERE " . $asc->{boundaries}->{'>'}
+            . ($where ? " AND ($where)" : '')
+            . " ORDER BY $order_by"
+            . " LIMIT 1"
+            . " /*resume lower boundary*/";
+         PTDEBUG && _d('Resume lower boundary statement:', $resume_lb_sql);
+      }
+
+      my $last_ub_sql
+         = "SELECT /*!40001 SQL_NO_CACHE */ "
+         . join(', ', map { $q->quote($_) } @{$asc->{scols}})
+         . " FROM $from"
+         . ($where ? " WHERE $where" : '')
+         . " ORDER BY "
+         . join(' DESC, ', map {$q->quote($_)} @{$index_cols}) . ' DESC'
+         . " LIMIT 1"
+         . " /*last upper boundary*/";
+      PTDEBUG && _d('Last upper boundary statement:', $last_ub_sql);
+
+      my $ub_sql
+         = "SELECT /*!40001 SQL_NO_CACHE */ "
+         . join(', ', map { $q->quote($_) } @{$asc->{scols}})
+         . " FROM $from"
+         . " WHERE " . $asc->{boundaries}->{'>='}
+                     . ($where ? " AND ($where)" : '')
+         . " ORDER BY $order_by"
+         . " LIMIT ?, 2"
+         . " /*next chunk boundary*/";
+      PTDEBUG && _d('Upper boundary statement:', $ub_sql);
+
+      my $nibble_sql
+         = ($args{dml} ? "$args{dml} " : "SELECT ")
+         . ($args{select} ? $args{select}
+                          : join(', ', map { $q->quote($_) } @{$asc->{cols}}))
+         . " FROM $from"
+         . " WHERE " . $asc->{boundaries}->{'>='}  # lower boundary
+         . " AND "   . $asc->{boundaries}->{'<='}  # upper boundary
+         . ($where ? " AND ($where)" : '')
+         . ($args{order_by} ? " ORDER BY $order_by" : "")
+         . " /*$comments{nibble}*/";
+      PTDEBUG && _d('Nibble statement:', $nibble_sql);
+
+      my $explain_nibble_sql 
+         = "EXPLAIN SELECT "
+         . ($args{select} ? $args{select}
+                          : join(', ', map { $q->quote($_) } @{$asc->{cols}}))
+         . " FROM $from"
+         . " WHERE " . $asc->{boundaries}->{'>='}  # lower boundary
+         . " AND "   . $asc->{boundaries}->{'<='}  # upper boundary
+         . ($where ? " AND ($where)" : '')
+         . ($args{order_by} ? " ORDER BY $order_by" : "")
+         . " /*explain $comments{nibble}*/";
+      PTDEBUG && _d('Explain nibble statement:', $explain_nibble_sql);
+
+      my $limit = $chunk_size - 1;
+      PTDEBUG && _d('Initial chunk size (LIMIT):', $limit);
+
+      $self = {
+         %args,
+         index              => $index,
+         limit              => $limit,
+         first_lb_sql       => $first_lb_sql,
+         last_ub_sql        => $last_ub_sql,
+         ub_sql             => $ub_sql,
+         nibble_sql         => $nibble_sql,
+         explain_ub_sql     => "EXPLAIN $ub_sql",
+         explain_nibble_sql => $explain_nibble_sql,
+         resume_lb_sql      => $resume_lb_sql,
+         sql                => {
+            columns    => $asc->{scols},
+            from       => $from,
+            where      => $where,
+            boundaries => $asc->{boundaries},
+            order_by   => $order_by,
+         },
+      };
+   }
+
+   $self->{row_est}    = $nibble_params->{row_est},
+   $self->{nibbleno}   = 0;
+   $self->{have_rows}  = 0;
+   $self->{rowno}      = 0;
+   $self->{oktonibble} = 1;
+
+   return bless $self, $class;
+}
+
+sub next {
+   my ($self) = @_;
+
+   if ( !$self->{oktonibble} ) {
+      PTDEBUG && _d('Not ok to nibble');
+      return;
+   }
+
+   my %callback_args = (
+      Cxn            => $self->{Cxn},
+      tbl            => $self->{tbl},
+      NibbleIterator => $self,
+   );
+
+   if ($self->{nibbleno} == 0) {
+      $self->_prepare_sths();
+      $self->_get_bounds();
+      if ( my $callback = $self->{callbacks}->{init} ) {
+         $self->{oktonibble} = $callback->(%callback_args);
+         PTDEBUG && _d('init callback returned', $self->{oktonibble});
+         if ( !$self->{oktonibble} ) {
+            $self->{no_more_boundaries} = 1;
+            return;
+         }
+      }
+   }
+
+   NIBBLE:
+   while ( $self->{have_rows} || $self->_next_boundaries() ) {
+      if ( !$self->{have_rows} ) {
+         $self->{nibbleno}++;
+         PTDEBUG && _d($self->{nibble_sth}->{Statement}, 'params:',
+            join(', ', (@{$self->{lower}}, @{$self->{upper}})));
+         if ( my $callback = $self->{callbacks}->{exec_nibble} ) {
+            $self->{have_rows} = $callback->(%callback_args);
+         }
+         else {
+            $self->{nibble_sth}->execute(@{$self->{lower}}, @{$self->{upper}});
+            $self->{have_rows} = $self->{nibble_sth}->rows();
+         }
+         PTDEBUG && _d($self->{have_rows}, 'rows in nibble', $self->{nibbleno});
+      }
+
+      if ( $self->{have_rows} ) {
+         my $row = $self->{nibble_sth}->fetchrow_arrayref();
+         if ( $row ) {
+            $self->{rowno}++;
+            PTDEBUG && _d('Row', $self->{rowno}, 'in nibble',$self->{nibbleno});
+            return [ @$row ];
+         }
+      }
+
+      PTDEBUG && _d('No rows in nibble or nibble skipped');
+      if ( my $callback = $self->{callbacks}->{after_nibble} ) {
+         $callback->(%callback_args);
+      }
+      $self->{rowno}     = 0;
+      $self->{have_rows} = 0;
+   }
+
+   PTDEBUG && _d('Done nibbling');
+   if ( my $callback = $self->{callbacks}->{done} ) {
+      $callback->(%callback_args);
+   }
+
+   return;
+}
+
+sub nibble_number {
+   my ($self) = @_;
+   return $self->{nibbleno};
+}
+
+sub set_nibble_number {
+   my ($self, $n) = @_;
+   die "I need a number" unless $n;
+   $self->{nibbleno} = $n;
+   PTDEBUG && _d('Set new nibble number:', $n);
+   return;
+}
+
+sub nibble_index {
+   my ($self) = @_;
+   return $self->{index};
+}
+
+sub statements {
+   my ($self) = @_;
+   return {
+      nibble                 => $self->{nibble_sth},
+      explain_nibble         => $self->{explain_nibble_sth},
+      upper_boundary         => $self->{ub_sth},
+      explain_upper_boundary => $self->{explain_ub_sth},
+   }
+}
+
+sub boundaries {
+   my ($self) = @_;
+   return {
+      first_lower => $self->{first_lower},
+      lower       => $self->{lower},
+      upper       => $self->{upper},
+      next_lower  => $self->{next_lower},
+      last_upper  => $self->{last_upper},
+   };
+}
+
+sub set_boundary {
+   my ($self, $boundary, $values) = @_;
+   die "I need a boundary parameter"
+      unless $boundary;
+   die "Invalid boundary: $boundary"
+      unless $boundary =~ m/^(?:lower|upper|next_lower|last_upper)$/;
+   die "I need a values arrayref parameter"
+      unless $values && ref $values eq 'ARRAY';
+   $self->{$boundary} = $values;
+   PTDEBUG && _d('Set new', $boundary, 'boundary:', Dumper($values));
+   return;
+}
+
+sub one_nibble {
+   my ($self) = @_;
+   return $self->{one_nibble};
+}
+
+sub chunk_size {
+   my ($self) = @_;
+   return $self->{limit} + 1;
+}
+
+sub set_chunk_size {
+   my ($self, $limit) = @_;
+   return if $self->{one_nibble};
+   die "Chunk size must be > 0" unless $limit;
+   $self->{limit} = $limit - 1;
+   PTDEBUG && _d('Set new chunk size (LIMIT):', $limit);
+   return;
+}
+
+sub sql {
+   my ($self) = @_;
+   return $self->{sql};
+}
+
+sub more_boundaries {
+   my ($self) = @_;
+   return !$self->{no_more_boundaries};
+}
+
+sub row_estimate {
+   my ($self) = @_;
+   return $self->{row_est};
+}
+
+sub can_nibble {
+   my (%args) = @_;
+   my @required_args = qw(Cxn tbl chunk_size OptionParser TableParser);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($cxn, $tbl, $chunk_size, $o) = @args{@required_args};
+
+   my ($row_est, $mysql_index) = get_row_estimate(
+      Cxn   => $cxn,
+      tbl   => $tbl,
+      where => $o->has('where') ? $o->get('where') : '',
+   );
+
+   my $one_nibble = !defined $args{one_nibble} || $args{one_nibble}
+                  ? $row_est <= $chunk_size * $o->get('chunk-size-limit')
+                  : 0;
+   PTDEBUG && _d('One nibble:', $one_nibble ? 'yes' : 'no');
+
+   if ( $args{resume}
+        && !defined $args{resume}->{lower_boundary}
+        && !defined $args{resume}->{upper_boundary} ) {
+      PTDEBUG && _d('Resuming from one nibble table');
+      $one_nibble = 1;
+   }
+
+   my $index = _find_best_index(%args, mysql_index => $mysql_index);
+   if ( !$index && !$one_nibble ) {
+      die "There is no good index and the table is oversized.";
+   }
+
+   return {
+      row_est     => $row_est,      # nibble about this many rows
+      index       => $index,        # using this index
+      one_nibble  => $one_nibble,   # if the table fits in one nibble/chunk
+   };
+}
+
+sub _find_best_index {
+   my (%args) = @_;
+   my @required_args = qw(Cxn tbl TableParser);
+   my ($cxn, $tbl, $tp) = @args{@required_args};
+   my $tbl_struct = $tbl->{tbl_struct};
+   my $indexes    = $tbl_struct->{keys};
+
+   my $want_index = $args{chunk_index};
+   if ( $want_index ) {
+      PTDEBUG && _d('User wants to use index', $want_index);
+      if ( !exists $indexes->{$want_index} ) {
+         PTDEBUG && _d('Cannot use user index because it does not exist');
+         $want_index = undef;
+      }
+   }
+
+   if ( !$want_index && $args{mysql_index} ) {
+      PTDEBUG && _d('MySQL wants to use index', $args{mysql_index});
+      $want_index = $args{mysql_index};
+   }
+
+   my $best_index;
+   my @possible_indexes;
+   if ( $want_index ) {
+      if ( $indexes->{$want_index}->{is_unique} ) {
+         PTDEBUG && _d('Will use wanted index');
+         $best_index = $want_index;
+      }
+      else {
+         PTDEBUG && _d('Wanted index is a possible index');
+         push @possible_indexes, $want_index;
+      }
+   }
+   else {
+      PTDEBUG && _d('Auto-selecting best index');
+      foreach my $index ( $tp->sort_indexes($tbl_struct) ) {
+         if ( $index eq 'PRIMARY' || $indexes->{$index}->{is_unique} ) {
+            $best_index = $index;
+            last;
+         }
+         else {
+            push @possible_indexes, $index;
+         }
+      }
+   }
+
+   if ( !$best_index && @possible_indexes ) {
+      PTDEBUG && _d('No PRIMARY or unique indexes;',
+         'will use index with highest cardinality');
+      foreach my $index ( @possible_indexes ) {
+         $indexes->{$index}->{cardinality} = _get_index_cardinality(
+            %args,
+            index => $index,
+         );
+      }
+      @possible_indexes = sort {
+         my $cmp
+            = $indexes->{$b}->{cardinality} <=> $indexes->{$b}->{cardinality};
+         if ( $cmp == 0 ) {
+            $cmp = scalar @{$indexes->{$b}->{cols}}
+               <=> scalar @{$indexes->{$a}->{cols}};
+         }
+         $cmp;
+      } @possible_indexes;
+      $best_index = $possible_indexes[0];
+   }
+
+   PTDEBUG && _d('Best index:', $best_index);
+   return $best_index;
+}
+
+sub _get_index_cardinality {
+   my (%args) = @_;
+   my @required_args = qw(Cxn tbl index);
+   my ($cxn, $tbl, $index) = @args{@required_args};
+
+   my $sql = "SHOW INDEXES FROM $tbl->{name} "
+           . "WHERE Key_name = '$index'";
+   PTDEBUG && _d($sql);
+   my $cardinality = 1;
+   my $rows = $cxn->dbh()->selectall_hashref($sql, 'key_name');
+   foreach my $row ( values %$rows ) {
+      $cardinality *= $row->{cardinality} if $row->{cardinality};
+   }
+   PTDEBUG && _d('Index', $index, 'cardinality:', $cardinality);
+   return $cardinality;
+}
+
+sub get_row_estimate {
+   my (%args) = @_;
+   my @required_args = qw(Cxn tbl);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($cxn, $tbl) = @args{@required_args};
+
+   if ( !$args{where} && exists $tbl->{tbl_status} ) {
+      PTDEBUG && _d('Using table status for row estimate');
+      return $tbl->{tbl_status}->{rows} || 0;
+   }
+   else {
+      PTDEBUG && _d('Use EXPLAIN for row estimate');
+      my $sql   = "EXPLAIN SELECT * FROM $tbl->{name} "
+                . "WHERE " . ($args{where} || '1=1');
+      PTDEBUG && _d($sql);
+      my $expl = $cxn->dbh()->selectrow_hashref($sql);
+      PTDEBUG && _d(Dumper($expl));
+      return ($expl->{rows} || 0), $expl->{key};
+   }
+}
+
+sub _prepare_sths {
+   my ($self) = @_;
+   PTDEBUG && _d('Preparing statement handles');
+
+   my $dbh = $self->{Cxn}->dbh();
+
+   $self->{nibble_sth}         = $dbh->prepare($self->{nibble_sql});
+   $self->{explain_nibble_sth} = $dbh->prepare($self->{explain_nibble_sql});
+
+   if ( !$self->{one_nibble} ) {
+      $self->{ub_sth} = $dbh->prepare($self->{ub_sql});
+      $self->{explain_ub_sth} = $dbh->prepare($self->{explain_ub_sql});
+   }
+
+   return;
+}
+
+sub _get_bounds { 
+   my ($self) = @_;
+
+   if ( $self->{one_nibble} ) {
+      if ( $self->{resume} ) {
+         $self->{no_more_boundaries} = 1;
+      }
+      return;
+   }
+
+   my $dbh = $self->{Cxn}->dbh();
+
+   $self->{first_lower} = $dbh->selectrow_arrayref($self->{first_lb_sql});
+   PTDEBUG && _d('First lower boundary:', Dumper($self->{first_lower}));  
+
+   if ( my $nibble = $self->{resume} ) {
+      if (    defined $nibble->{lower_boundary}
+           && defined $nibble->{upper_boundary} ) {
+         my $sth = $dbh->prepare($self->{resume_lb_sql});
+         my @ub  = split ',', $nibble->{upper_boundary};
+         PTDEBUG && _d($sth->{Statement}, 'params:', @ub);
+         $sth->execute(@ub);
+         $self->{next_lower} = $sth->fetchrow_arrayref();
+         $sth->finish();
+      }
+   }
+   else {
+      $self->{next_lower}  = $self->{first_lower};   
+   }
+   PTDEBUG && _d('Next lower boundary:', Dumper($self->{next_lower}));  
+
+   if ( !$self->{next_lower} ) {
+      PTDEBUG && _d('At end of table, or no more boundaries to resume');
+      $self->{no_more_boundaries} = 1;
+   }
+
+   return;
+}
+
+sub _next_boundaries {
+   my ($self) = @_;
+
+   if ( $self->{no_more_boundaries} ) {
+      PTDEBUG && _d('No more boundaries');
+      return; # stop nibbling
+   }
+
+   if ( $self->{one_nibble} ) {
+      $self->{lower} = $self->{upper} = [];
+      $self->{no_more_boundaries} = 1;  # for next call
+      return 1; # continue nibbling
+   }
+
+   if ( $self->identical_boundaries($self->{lower}, $self->{next_lower}) ) {
+      PTDEBUG && _d('Infinite loop detected');
+      my $tbl     = $self->{tbl};
+      my $index   = $tbl->{tbl_struct}->{keys}->{$self->{index}};
+      my $n_cols  = scalar @{$index->{cols}};
+      my $chunkno = $self->{nibbleno};
+      die "Possible infinite loop detected!  "
+         . "The lower boundary for chunk $chunkno is "
+         . "<" . join(', ', @{$self->{lower}}) . "> and the lower "
+         . "boundary for chunk " . ($chunkno + 1) . " is also "
+         . "<" . join(', ', @{$self->{next_lower}}) . ">.  "
+         . "This usually happens when using a non-unique single "
+         . "column index.  The current chunk index for table "
+         . "$tbl->{db}.$tbl->{tbl} is $self->{index} which is"
+         . ($index->{is_unique} ? '' : ' not') . " unique and covers "
+         . ($n_cols > 1 ? "$n_cols columns" : "1 column") . ".\n";
+   }
+   $self->{lower} = $self->{next_lower};
+
+   if ( my $callback = $self->{callbacks}->{next_boundaries} ) {
+      my $oktonibble = $callback->(
+         Cxn            => $self->{Cxn},
+         tbl            => $self->{tbl},
+         NibbleIterator => $self,
+      );
+      PTDEBUG && _d('next_boundaries callback returned', $oktonibble);
+      if ( !$oktonibble ) {
+         $self->{no_more_boundaries} = 1;
+         return; # stop nibbling
+      }
+   }
+
+   PTDEBUG && _d($self->{ub_sth}->{Statement}, 'params:',
+      join(', ', @{$self->{lower}}), $self->{limit});
+   $self->{ub_sth}->execute(@{$self->{lower}}, $self->{limit});
+   my $boundary = $self->{ub_sth}->fetchall_arrayref();
+   PTDEBUG && _d('Next boundary:', Dumper($boundary));
+   if ( $boundary && @$boundary ) {
+      $self->{upper} = $boundary->[0];
+
+      if ( $boundary->[1] ) {
+         $self->{next_lower} = $boundary->[1];
+      }
+      else {
+         PTDEBUG && _d('End of table boundary:', Dumper($boundary->[0]));
+         $self->{no_more_boundaries} = 1;  # for next call
+
+         $self->{last_upper} = $boundary->[0];
+      }
+   }
+   else {
+      my $dbh = $self->{Cxn}->dbh();
+      $self->{upper} = $dbh->selectrow_arrayref($self->{last_ub_sql});
+      PTDEBUG && _d('Last upper boundary:', Dumper($self->{upper}));
+      $self->{no_more_boundaries} = 1;  # for next call
+      
+      $self->{last_upper} = $self->{upper};
+   }
+   $self->{ub_sth}->finish();
+
+   return 1; # continue nibbling
+}
+
+sub identical_boundaries {
+   my ($self, $b1, $b2) = @_;
+
+   return 0 if ($b1 && !$b2) || (!$b1 && $b2);
+
+   return 1 if !$b1 && !$b2;
+
+   die "Boundaries have different numbers of values"
+      if scalar @$b1 != scalar @$b2;  # shouldn't happen
+   my $n_vals = scalar @$b1;
+   for my $i ( 0..($n_vals-1) ) {
+      return 0 if $b1->[$i] ne $b2->[$i]; # diff
+   }
+   return 1;
+}
+
+sub DESTROY {
+   my ( $self ) = @_;
+   foreach my $key ( keys %$self ) {
+      if ( $key =~ m/_sth$/ ) {
+         PTDEBUG && _d('Finish', $key);
+         $self->{$key}->finish();
+      }
+   }
+   return;
+}
+
+sub _d {
+   my ($package, undef, $line) = caller 0;
+   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
+        map { defined $_ ? $_ : 'undef' }
+        @_;
+   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
+}
+
+1;
+}
+# ###########################################################################
+# End NibbleIterator package
+# ###########################################################################
+
+# ###########################################################################
+# Transformers package
+# This package is a copy without comments from the original.  The original
+# with comments and its test file can be found in the Bazaar repository at,
+#   lib/Transformers.pm
+#   t/lib/Transformers.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package Transformers;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+use Time::Local qw(timegm timelocal);
+use Digest::MD5 qw(md5_hex);
+
+require Exporter;
+our @ISA         = qw(Exporter);
+our %EXPORT_TAGS = ();
+our @EXPORT      = ();
+our @EXPORT_OK   = qw(
+   micro_t
+   percentage_of
+   secs_to_time
+   time_to_secs
+   shorten
+   ts
+   parse_timestamp
+   unix_timestamp
+   any_unix_timestamp
+   make_checksum
+   crc32
+);
+
+our $mysql_ts  = qr/(\d\d)(\d\d)(\d\d) +(\d+):(\d+):(\d+)(\.\d+)?/;
+our $proper_ts = qr/(\d\d\d\d)-(\d\d)-(\d\d)[T ](\d\d):(\d\d):(\d\d)(\.\d+)?/;
+our $n_ts      = qr/(\d{1,5})([shmd]?)/; # Limit \d{1,5} because \d{6} looks
+
+sub micro_t {
+   my ( $t, %args ) = @_;
+   my $p_ms = defined $args{p_ms} ? $args{p_ms} : 0;  # precision for ms vals
+   my $p_s  = defined $args{p_s}  ? $args{p_s}  : 0;  # precision for s vals
+   my $f;
+
+   $t = 0 if $t < 0;
+
+   $t = sprintf('%.17f', $t) if $t =~ /e/;
+
+   $t =~ s/\.(\d{1,6})\d*/\.$1/;
+
+   if ($t > 0 && $t <= 0.000999) {
+      $f = ($t * 1000000) . 'us';
+   }
+   elsif ($t >= 0.001000 && $t <= 0.999999) {
+      $f = sprintf("%.${p_ms}f", $t * 1000);
+      $f = ($f * 1) . 'ms'; # * 1 to remove insignificant zeros
+   }
+   elsif ($t >= 1) {
+      $f = sprintf("%.${p_s}f", $t);
+      $f = ($f * 1) . 's'; # * 1 to remove insignificant zeros
+   }
+   else {
+      $f = 0;  # $t should = 0 at this point
+   }
+
+   return $f;
+}
+
+sub percentage_of {
+   my ( $is, $of, %args ) = @_;
+   my $p   = $args{p} || 0; # float precision
+   my $fmt = $p ? "%.${p}f" : "%d";
+   return sprintf $fmt, ($is * 100) / ($of ||= 1);
+}
+
+sub secs_to_time {
+   my ( $secs, $fmt ) = @_;
+   $secs ||= 0;
+   return '00:00' unless $secs;
+
+   $fmt ||= $secs >= 86_400 ? 'd'
+          : $secs >= 3_600  ? 'h'
+          :                   'm';
+
+   return
+      $fmt eq 'd' ? sprintf(
+         "%d+%02d:%02d:%02d",
+         int($secs / 86_400),
+         int(($secs % 86_400) / 3_600),
+         int(($secs % 3_600) / 60),
+         $secs % 60)
+      : $fmt eq 'h' ? sprintf(
+         "%02d:%02d:%02d",
+         int(($secs % 86_400) / 3_600),
+         int(($secs % 3_600) / 60),
+         $secs % 60)
+      : sprintf(
+         "%02d:%02d",
+         int(($secs % 3_600) / 60),
+         $secs % 60);
+}
+
+sub time_to_secs {
+   my ( $val, $default_suffix ) = @_;
+   die "I need a val argument" unless defined $val;
+   my $t = 0;
+   my ( $prefix, $num, $suffix ) = $val =~ m/([+-]?)(\d+)([a-z])?$/;
+   $suffix = $suffix || $default_suffix || 's';
+   if ( $suffix =~ m/[smhd]/ ) {
+      $t = $suffix eq 's' ? $num * 1        # Seconds
+         : $suffix eq 'm' ? $num * 60       # Minutes
+         : $suffix eq 'h' ? $num * 3600     # Hours
+         :                  $num * 86400;   # Days
+
+      $t *= -1 if $prefix && $prefix eq '-';
+   }
+   else {
+      die "Invalid suffix for $val: $suffix";
+   }
+   return $t;
+}
+
+sub shorten {
+   my ( $num, %args ) = @_;
+   my $p = defined $args{p} ? $args{p} : 2;     # float precision
+   my $d = defined $args{d} ? $args{d} : 1_024; # divisor
+   my $n = 0;
+   my @units = ('', qw(k M G T P E Z Y));
+   while ( $num >= $d && $n < @units - 1 ) {
+      $num /= $d;
+      ++$n;
+   }
+   return sprintf(
+      $num =~ m/\./ || $n
+         ? "%.${p}f%s"
+         : '%d',
+      $num, $units[$n]);
+}
+
+sub ts {
+   my ( $time, $gmt ) = @_;
+   my ( $sec, $min, $hour, $mday, $mon, $year )
+      = $gmt ? gmtime($time) : localtime($time);
+   $mon  += 1;
+   $year += 1900;
+   my $val = sprintf("%d-%02d-%02dT%02d:%02d:%02d",
+      $year, $mon, $mday, $hour, $min, $sec);
+   if ( my ($us) = $time =~ m/(\.\d+)$/ ) {
+      $us = sprintf("%.6f", $us);
+      $us =~ s/^0\././;
+      $val .= $us;
+   }
+   return $val;
+}
+
+sub parse_timestamp {
+   my ( $val ) = @_;
+   if ( my($y, $m, $d, $h, $i, $s, $f)
+         = $val =~ m/^$mysql_ts$/ )
+   {
+      return sprintf "%d-%02d-%02d %02d:%02d:"
+                     . (defined $f ? '%09.6f' : '%02d'),
+                     $y + 2000, $m, $d, $h, $i, (defined $f ? $s + $f : $s);
+   }
+   return $val;
+}
+
+sub unix_timestamp {
+   my ( $val, $gmt ) = @_;
+   if ( my($y, $m, $d, $h, $i, $s, $us) = $val =~ m/^$proper_ts$/ ) {
+      $val = $gmt
+         ? timegm($s, $i, $h, $d, $m - 1, $y)
+         : timelocal($s, $i, $h, $d, $m - 1, $y);
+      if ( defined $us ) {
+         $us = sprintf('%.6f', $us);
+         $us =~ s/^0\././;
+         $val .= $us;
+      }
+   }
+   return $val;
+}
+
+sub any_unix_timestamp {
+   my ( $val, $callback ) = @_;
+
+   if ( my ($n, $suffix) = $val =~ m/^$n_ts$/ ) {
+      $n = $suffix eq 's' ? $n            # Seconds
+         : $suffix eq 'm' ? $n * 60       # Minutes
+         : $suffix eq 'h' ? $n * 3600     # Hours
+         : $suffix eq 'd' ? $n * 86400    # Days
+         :                  $n;           # default: Seconds
+      PTDEBUG && _d('ts is now - N[shmd]:', $n);
+      return time - $n;
+   }
+   elsif ( $val =~ m/^\d{9,}/ ) {
+      PTDEBUG && _d('ts is already a unix timestamp');
+      return $val;
+   }
+   elsif ( my ($ymd, $hms) = $val =~ m/^(\d{6})(?:\s+(\d+:\d+:\d+))?/ ) {
+      PTDEBUG && _d('ts is MySQL slow log timestamp');
+      $val .= ' 00:00:00' unless $hms;
+      return unix_timestamp(parse_timestamp($val));
+   }
+   elsif ( ($ymd, $hms) = $val =~ m/^(\d{4}-\d\d-\d\d)(?:[T ](\d+:\d+:\d+))?/) {
+      PTDEBUG && _d('ts is properly formatted timestamp');
+      $val .= ' 00:00:00' unless $hms;
+      return unix_timestamp($val);
+   }
+   else {
+      PTDEBUG && _d('ts is MySQL expression');
+      return $callback->($val) if $callback && ref $callback eq 'CODE';
+   }
+
+   PTDEBUG && _d('Unknown ts type:', $val);
+   return;
+}
+
+sub make_checksum {
+   my ( $val ) = @_;
+   my $checksum = uc substr(md5_hex($val), -16);
+   PTDEBUG && _d($checksum, 'checksum for', $val);
+   return $checksum;
+}
+
+sub crc32 {
+   my ( $string ) = @_;
+   return unless $string;
+   my $poly = 0xEDB88320;
+   my $crc  = 0xFFFFFFFF;
+   foreach my $char ( split(//, $string) ) {
+      my $comp = ($crc ^ ord($char)) & 0xFF;
+      for ( 1 .. 8 ) {
+         $comp = $comp & 1 ? $poly ^ ($comp >> 1) : $comp >> 1;
+      }
+      $crc = (($crc >> 8) & 0x00FFFFFF) ^ $comp;
+   }
+   return $crc ^ 0xFFFFFFFF;
+}
+
+sub _d {
+   my ($package, undef, $line) = caller 0;
+   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
+        map { defined $_ ? $_ : 'undef' }
+        @_;
+   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
+}
+
+1;
+}
+# ###########################################################################
+# End Transformers package
+# ###########################################################################
+
+# ###########################################################################
+# CleanupTask package
+# This package is a copy without comments from the original.  The original
+# with comments and its test file can be found in the Bazaar repository at,
+#   lib/CleanupTask.pm
+#   t/lib/CleanupTask.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package CleanupTask;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+sub new {
+   my ( $class, $task ) = @_;
+   die "I need a task parameter" unless $task;
+   die "The task parameter must be a coderef" unless ref $task eq 'CODE';
+   my $self = {
+      task => $task,
+   };
+   PTDEBUG && _d('Created cleanup task', $task);
+   return bless $self, $class;
+}
+
+sub DESTROY {
+   my ($self) = @_;
+   my $task = $self->{task};
+   PTDEBUG && _d('Calling cleanup task', $task);
+   $task->();
+   return;
+}
+
+sub _d {
+   my ($package, undef, $line) = caller 0;
+   @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
+        map { defined $_ ? $_ : 'undef' }
+        @_;
+   print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
+}
+
+1;
+}
+# ###########################################################################
+# End CleanupTask package
+# ###########################################################################
+
+# ###########################################################################
 # This is a combination of modules and programs in one -- a runnable module.
 # http://www.perl.com/pub/a/2006/07/13/lightning-articles.html?page=last
 # Or, look it up in the Camel book on pages 642 and 643 in the 3rd edition.
@@ -3869,29 +5821,37 @@
 # ###########################################################################
 package pt_online_schema_change;
 
+use strict;
+use warnings FATAL => 'all';
 use English qw(-no_match_vars);
-use Time::HiRes qw(sleep);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+use Time::HiRes qw(time);
 use Data::Dumper;
 $Data::Dumper::Indent    = 1;
 $Data::Dumper::Sortkeys  = 1;
 $Data::Dumper::Quotekeys = 0;
 
-Transformers->import(qw(ts));
-
-use constant PTDEBUG => $ENV{PTDEBUG} || 0;
-
-my $quiet = 0;  # for msg()
+use sigtrap 'handler', \&sig_int, 'normal-signals';
+
+my $exit_status = 0;
+my $oktorun     = 1;
+my @drop_trigger_sqls;
+
+$OUTPUT_AUTOFLUSH = 1;
 
 sub main {
-   @ARGV       = @_;  # set global ARGV for this package
-   my $vp      = new VersionParser();
-   my $q       = new Quoter();
-   my $tp      = new TableParser(Quoter => $q);
-   my $chunker = new TableChunker(Quoter => $q, TableParser => $tp);
+   # Reset global vars else tests will fail.
+   @ARGV              = @_;
+   $oktorun           = 1;
+   @drop_trigger_sqls = ();
+
+   $exit_status = 0;
 
    # ########################################################################
    # Get configuration information.
    # ########################################################################
+   my $q = new Quoter();
    my $o = new OptionParser();
    $o->get_specs();
    $o->get_opts();
@@ -3899,41 +5859,36 @@
    my $dp = $o->DSNParser();
    $dp->prop('set-vars', $o->get('set-vars'));
 
-   $quiet = $o->get('quiet');  # for msg()
-
-   my ($dsn, $db, $tbl);
-   $dsn = shift @ARGV;
+   # The original table, i.e. the one being altered, must be specified
+   # on the command line via the DSN.
+   my ($db, $tbl);
+   my $dsn = shift @ARGV;
    if ( !$dsn ) {
-      $o->save_error('A DSN with a t part must be specified');
+      $o->save_error('A DSN must be specified');
    }
    else {
+      # Parse DSN string and convert it to a DSN data struct.
       $dsn = $dp->parse($dsn, $dp->parse_options($o));
-      if ( !$dsn->{t} ) {
-         $o->save_error('The DSN must specify a t (table) part');
-      }
-      else {
-         ($db, $tbl) = $q->split_unquote($dsn->{t} || "", $dsn->{D} || "");
-      }
-   }
-
-   my $rename_fk_method = lc($o->get('update-foreign-keys-method') || '');
-   if ( ($rename_fk_method || '') eq 'drop_old_table' ) {
-      $o->set('rename-tables',  0);
-      $o->set('drop-old-table', 0),
-   }
+      $db  = $dsn->{D};
+      $tbl = $dsn->{t};  
+   }
+
+   my $alter_fk_method = $o->get('alter-foreign-keys-method') || '';
+   if ( $alter_fk_method eq 'drop_swap' ) {
+      $o->set('swap-tables',    0);
+      $o->set('drop-old-table', 0);
+   }
+
+   # Explicit --chunk-size disable auto chunk sizing.
+   $o->set('chunk-time', 0) if $o->got('chunk-size');
 
    if ( !$o->get('help') ) {
       if ( @ARGV ) {
          $o->save_error('Specify only one DSN on the command line');
       }
 
-      if ( !$db ) {
-         $o->save_error("No database was specified in the DSN or by "
-            . "--database (-D)");
-      }
-
-      if ( $tbl && $tbl eq ($o->get('tmp-table') || "") ) {
-         $o->save_error("--tmp-table cannot be the same as the table");
+      if ( !$db || !$tbl ) {
+         $o->save_error("The DSN must specify a database (D) and a table (t)");
       }
 
       if ( $o->get('progress') ) {
@@ -3944,523 +5899,1755 @@
          }
       }
 
-      if ( $o->get('child-tables') && !$o->get('update-foreign-keys-method') ) {
-         $o->save_error("--child-tables requires --update-foreign-keys-method");
-      }
-
-      if ( $rename_fk_method
-           && $rename_fk_method ne 'rebuild_constraints'
-           && $rename_fk_method ne 'drop_old_table' ) {
-         $o->save_error("Invalid --update-foreign-keys-method value");
-      }
-   }
-
-   $o->usage_or_errors();
-
-   msg("$PROGRAM_NAME started");
-   my $exit_status = 0;
+      # See the "pod-based-option-value-validation" spec for how this may
+      # be automagically validated.
+      if ( $alter_fk_method
+           && $alter_fk_method ne 'auto'
+           && $alter_fk_method ne 'rebuild_constraints'
+           && $alter_fk_method ne 'drop_swap'
+           && $alter_fk_method ne 'none' )
+      {
+         $o->save_error("Invalid --alter-foreign-keys-method value: $alter_fk_method");
+      }
+   }
+
+   $o->usage_or_errors(); 
+
+   if ( $o->get('quiet') ) {
+      # BARON: this will fail on Windows, where there is no /dev/null. I feel
+      # it's a hack, like ignoring a problem instead of fixing it somehow. We
+      # should take a look at the things that get printed in a "normal"
+      # non-quiet run, and "if !quiet" them, and then do some kind of Logger.pm
+      # or Messager.pm module for a future release.
+      close STDOUT;
+      open  STDOUT, '>', '/dev/null'
+         or warn "Cannot reopen STDOUT to /dev/null: $OS_ERROR";
+   }
 
    # ########################################################################
    # Connect to MySQL.
    # ########################################################################
-   my $dbh = get_cxn(
-      dsn          => $dsn,
-      DSNParser    => $dp,
+   my $set_on_connect = sub {
+      my ($dbh) = @_;
+
+      # See the same code in pt-table-checksum.
+      my $lock_wait_timeout = $o->get('lock-wait-timeout');
+      my $set_lwt = "SET SESSION innodb_lock_wait_timeout=$lock_wait_timeout";
+      PTDEBUG && _d($set_lwt);
+      eval {
+         $dbh->do($set_lwt);
+      };
+      if ( $EVAL_ERROR ) {
+         PTDEBUG && _d($EVAL_ERROR);
+         # Get the server's current value.
+         my $sql = "SHOW SESSION VARIABLES LIKE 'innodb_lock_wait_timeout'";
+         PTDEBUG && _d($dbh, $sql);
+         my (undef, $curr_lwt) = $dbh->selectrow_array($sql);
+         PTDEBUG && _d('innodb_lock_wait_timeout on server:', $curr_lwt);
+         if ( $curr_lwt > $lock_wait_timeout ) { 
+            warn "Failed to $set_lwt: $EVAL_ERROR\n"
+               . "The current innodb_lock_wait_timeout value "
+               . "$curr_lwt is greater than the --lock-wait-timeout "
+               . "value $lock_wait_timeout and the variable cannot be "
+               . "changed.  innodb_lock_wait_timeout is only dynamic when "
+               . "using the InnoDB plugin.  To prevent this warning, either "
+               . "specify --lock-wait-time=$curr_lwt, or manually set "
+               . "innodb_lock_wait_timeout to a value less than or equal "
+               . "to $lock_wait_timeout and restart MySQL.\n";
+         }
+      }
+   };
+
+   # Do not call "new Cxn(" directly; use this sub so that set_on_connect
+   # is applied to every cxn.
+   # BARON: why not make this a subroutine instead of a subroutine variable? I
+   # think that can be less confusing. Also, the $set_on_connect variable can be
+   # inlined into this subroutine. Many of our tools have a get_dbh() subroutine
+   # and it might be good to just make a convention of it.
+   my $make_cxn = sub {
+      my (%args) = @_;
+      my $cxn = new Cxn(
+         %args,
+         DSNParser    => $dp,
+         OptionParser => $o,
+         set          => $set_on_connect,
+      );
+      eval { $cxn->connect() };  # connect or die trying
+      if ( $EVAL_ERROR ) {
+         die "Cannot connect to MySQL: $EVAL_ERROR\n";
+      }
+      return $cxn;
+   };
+
+   my $cxn = $make_cxn->(dsn => $dsn);
+
+   # ########################################################################
+   # Check if MySQL is new enough to have the triggers we need.
+   # Although triggers were introduced in 5.0.2, "Prior to MySQL 5.0.10,
+   # triggers cannot contain direct references to tables by name."
+   # ########################################################################   
+   my $vp = new VersionParser();
+   if ( !$vp->version_ge($cxn->dbh(), '5.0.10') ) {
+      die "This tool requires MySQL 5.0.10 or newer.\n";
+   }
+
+   # ########################################################################
+   # Setup lag and load monitors.
+   # ########################################################################   
+   my $slaves;         # all slaves that are found or specified
+   my $slave_lag_cxns; # slaves whose lag we'll check
+   my $replica_lag;    # ReplicaLagWaiter object
+   my $replica_lag_pr; # Progress for ReplicaLagWaiter
+   my $sys_load;       # MySQLStatusWaiter object
+   my $sys_load_pr;    # Progress for MySQLStatusWaiter object
+
+   if ( $o->get('execute') ) {
+      # #####################################################################
+      # Find and connect to slaves.
+      # #####################################################################
+      my $ms = new MasterSlave();
+      $slaves = $ms->get_slaves(
+         dbh          => $cxn->dbh(),
+         dsn          => $cxn->dsn(),
+         OptionParser => $o,
+         DSNParser    => $dp,
+         Quoter       => $q,
+         make_cxn     => sub {
+            return $make_cxn->(@_, prev_dsn => $cxn->dsn());
+         },
+      );
+      PTDEBUG && _d(scalar @$slaves, 'slaves found');
+
+      if ( $o->get('check-slave-lag') ) {
+         PTDEBUG && _d('Will use --check-slave-lag to check for slave lag');
+         my $cxn = $make_cxn->(
+            dsn_string => $o->get('check-slave-lag'),
+            prev_dsn   => $cxn->dsn(),
+         );
+         $slave_lag_cxns = [ $cxn ];
+      }
+      else {
+         PTDEBUG && _d('Will check slave lag on all slaves');
+         $slave_lag_cxns = $slaves;
+      }
+
+      # #####################################################################
+      # Check for replication filters.
+      # #####################################################################
+      if ( $o->get('check-replication-filters') ) {
+         PTDEBUG && _d("Checking slave replication filters");
+         my @all_repl_filters;
+         foreach my $slave ( @$slaves ) {
+            my $repl_filters = $ms->get_replication_filters(
+               dbh => $slave->dbh(),
+            );
+            if ( keys %$repl_filters ) {
+               push @all_repl_filters,
+                  { name    => $slave->name(),
+                    filters => $repl_filters,
+                  };
+            }
+         }
+         if ( @all_repl_filters ) {
+            my $msg = "Replication filters are set on these hosts:\n";
+            foreach my $host ( @all_repl_filters ) {
+               my $filters = $host->{filters};
+               $msg .= "  $host->{name}\n"
+                     . join("\n", map { "    $_ = $host->{filters}->{$_}" }
+                            keys %{$host->{filters}})
+                     . "\n";
+            }
+            $msg .= "Please read the --check-replication-filters documentation "
+                  . "to learn how to solve this problem.";
+            die $msg;
+         }
+      }
+
+      # #####################################################################
+      # Make a ReplicaLagWaiter to help wait for slaves after each chunk.
+      # #####################################################################
+      my $sleep = sub {
+         # Don't let the master dbh die while waiting for slaves because we
+         # may wait a very long time for slaves.
+         my $dbh = $cxn->dbh();
+         if ( !$dbh || !$dbh->ping() ) {
+            eval { $dbh = $cxn->connect() };  # connect or die trying
+            if ( $EVAL_ERROR ) {
+               $oktorun = 0;  # flag for cleanup tasks
+               chomp $EVAL_ERROR;
+               die "Lost connection to " . $cxn->name() . " while waiting for "
+                  . "replica lag ($EVAL_ERROR)\n";
+            }
+         }
+         $dbh->do("SELECT 'pt-online-schema-change keepalive'");
+         sleep $o->get('check-interval');
+         return;
+      };
+
+      my $get_lag = sub {
+         my ($cxn) = @_;
+         my $dbh = $cxn->dbh();
+         if ( !$dbh || !$dbh->ping() ) {
+            eval { $dbh = $cxn->connect() };  # connect or die trying
+            if ( $EVAL_ERROR ) {
+               $oktorun = 0;  # flag for cleanup tasks
+               chomp $EVAL_ERROR;
+               die "Lost connection to replica " . $cxn->name()
+                  . " while attempting to get its lag ($EVAL_ERROR)\n";
+            }
+         }
+         return $ms->get_slave_lag($dbh);
+      };
+
+      $replica_lag = new ReplicaLagWaiter(
+         slaves   => $slave_lag_cxns,
+         max_lag  => $o->get('max-lag'),
+         oktorun  => sub { return $oktorun },
+         get_lag  => $get_lag,
+         sleep    => $sleep,
+      );
+
+      my $get_status;
+      {
+         my $sql = "SHOW GLOBAL STATUS LIKE ?";
+         my $sth = $cxn->dbh()->prepare($sql);
+
+         $get_status = sub {
+            my ($var) = @_;
+            PTDEBUG && _d($sth->{Statement}, $var);
+            $sth->execute($var);
+            my (undef, $val) = $sth->fetchrow_array();
+            return $val;
+         };
+      }
+
+      $sys_load = new MySQLStatusWaiter(
+         max_spec      => $o->get('max-load'),
+         critical_spec => $o->get('critical-load'),
+         get_status    => $get_status,
+         oktorun       => sub { return $oktorun },
+         sleep         => $sleep,
+      );
+      
+      if ( $o->get('progress') ) {
+         $replica_lag_pr = new Progress(
+            jobsize => scalar @$slaves,
+            spec    => $o->get('progress'),
+            name    => "Waiting for replicas to catch up",  # not used
+         );
+
+         $sys_load_pr = new Progress(
+            jobsize => scalar @{$o->get('max-load')},
+            spec    => $o->get('progress'),
+            name    => "Waiting for --max-load", # not used
+         );
+      }
+   }
+
+   # ########################################################################
+   # Setup and check the original table.
+   # ########################################################################
+   my $tp = new TableParser(Quoter => $q);
+
+   # Common table data struct (that modules like NibbleIterator expect).
+   my $orig_tbl = {
+      db   => $db,
+      tbl  => $tbl,
+      name => $q->quote($db, $tbl),
+   };
+
+   check_orig_table(
+      orig_tbl     => $orig_tbl,
+      Cxn          => $cxn,
       OptionParser => $o,
-      AutoCommit   => 1,
-   );
-   msg("USE `$db`");
-   $dbh->do("USE `$db`");
-
-   # ########################################################################
-   # Daemonize only after (potentially) asking for passwords for --ask-pass.
+      TableParser  => $tp,
+      Quoter       => $q,
+   );
+
+   # ########################################################################
+   # Get child tables of the original table, if necessary.
+   # ########################################################################
+   my $child_tables = find_child_tables(
+      tbl    => $orig_tbl,
+      Cxn    => $cxn,
+      Quoter => $q,
+   );
+   if ( !$child_tables ) {
+      if ( $alter_fk_method ) {
+         warn "No foreign keys reference $orig_tbl->{name}; ignoring "
+            . "--alter-foreign-keys-method.\n";
+
+         if ( $alter_fk_method eq 'drop_swap' ) {
+            # These opts are disabled at the start if the user specifies
+            # the drop_swap method, but now that we know there are no
+            # child tables, we must re-enable these to make the alter work.
+            $o->set('swap-tables',    1);
+            $o->set('drop-old-table', 1);
+         }
+
+         $alter_fk_method = '';
+      }
+      # No child tables and --alter-fk-method wasn't specified,
+      # so nothing to do.
+   }
+   else {
+      print "Child tables:\n";
+      foreach my $child_table ( @$child_tables ) {
+         printf "  %s (approx. %s rows)\n",
+            $child_table->{name},
+            $child_table->{row_est} || '?';
+      }
+
+      if ( $alter_fk_method ) {
+         # Let the user know how we're going to update the child table fk refs.
+         my $choice
+            = $alter_fk_method eq 'none' ? "not"
+            : $alter_fk_method eq 'auto' ? "automatically choose the method to"
+            :                              "use the $alter_fk_method method to";
+         print "Will $choice update foreign keys.\n";
+      }
+      else {
+         print "You did not specify --alter-foreign-keys-method, but there "
+             . "are foreign keys that reference the table. "
+             . "Please read the tool's documentation carefully.\n";
+         return 1; 
+      }
+   }
+
+   # ########################################################################
+   # XXX
+   # Ready to begin the alter!  Nothing has been changed on the server at
+   # this point; we've just checked and looked for things.  Past this point,
+   # the code is live if --execute, else it's doing a --dry-run.  Or, if
+   # the user didn't read the docs, we may bail out here.
+   # XXX
+   # ########################################################################
+   if ( $o->get('dry-run') ) {
+      print "Starting a dry run.  $orig_tbl->{name} will not be altered.  "
+          . "Specify --execute instead of --dry-run to alter the table.\n";
+   }
+   elsif ( $o->get('execute') ) {
+      print "Altering $orig_tbl->{name}...\n";
+   }
+   else {
+      print "Exiting without altering $orig_tbl->{name} because neither "
+        . "--dry-run nor --execute was specified.  Please read the tool's "
+        . "documentation carefully before using this tool.\n";
+      return 1;
+   }
+
+   # ########################################################################
+   # Create a cleanup task object to undo changes (i.e. clean up) if the
+   # code dies, or we may call this explicitly at the end if all goes well.
+   # ########################################################################
+   my @cleanup_tasks;
+   my $cleanup = new CleanupTask(
+      sub {
+         # XXX We shouldn't copy $EVAL_ERROR here, but I found that
+         # errors are not re-thrown in tests.  If you comment out this
+         # line and the die below, an error fails:
+         # not ok 5 - Doesn't try forever to find a new table name
+         #   Failed test 'Doesn't try forever to find a new table name'
+         #   at /Users/daniel/p/pt-osc-2.1.1/lib/PerconaTest.pm line 559.
+         #                   ''
+         #   doesn't match '(?-xism:Failed to find a unique new table name)'
+         my $original_error = $EVAL_ERROR;
+         foreach my $task ( reverse @cleanup_tasks ) {
+            eval {
+               $task->();
+            };
+            if ( $EVAL_ERROR ) {
+               warn "Error cleaning up: $EVAL_ERROR\n";
+            }
+         }
+         die $original_error if $original_error;  # rethrow original error
+         return;
+      }
+   );
+
+   # The last cleanup task is to report whether or not the orig table
+   # was altered.
+   push @cleanup_tasks, sub {
+      PTDEBUG && _d('Clean up done, report if orig table was altered');
+      if ( $o->get('dry-run') ) {
+         print "Dry run complete.  $orig_tbl->{name} was not altered.\n";
+      }
+      else {
+         if ( $orig_tbl->{swapped} ) {
+            if ( $orig_tbl->{success} ) {
+               print "Successfully altered $orig_tbl->{name}.\n";
+            }
+            else {
+               print "Altered $orig_tbl->{name} but there were errors "
+                  . "or warnings.\n";
+            }
+         }
+         else {
+            print "$orig_tbl->{name} was not altered.\n";
+         }
+      }
+      return;
+   };
+
+   # ########################################################################
+   # Check and create PID file if user specified --pid.
    # ########################################################################
    my $daemon;
-   if ( $o->get('pid') ) {
+   if ( $o->get('execute') && $o->get('pid') ) {
       # We're not daemoninzing, it just handles PID stuff.
       $daemon = new Daemon(o=>$o);
       $daemon->make_PID_file();
    }
 
-   # ########################################################################
-   # Setup/init some vars.
-   # ########################################################################
-   my $tmp_tbl = $o->get('tmp-table') || "__tmp_$tbl";
-   my $old_tbl = "__old_$tbl"; # what tbl becomes after swapped with tmp tbl
-   my %tables  = (
-      db      => $db,
-      tbl     => $tbl,
-      tmp_tbl => $tmp_tbl,
-      old_tbl => $old_tbl,
-   );
-   msg("Alter table $tbl using temporary table $tmp_tbl");
-
-   my %common_modules = (
-      OptionParser  => $o,
-      DSNParser     => $dp,
-      Quoter        => $q,
-      TableParser   => $tp,
-      TableChunker  => $chunker,
-      VersionParser => $vp,
-   );
-
-   # ########################################################################
-   # Create the capture-sync and copy-rows plugins.  Currently, we just have
-   # one method for each.
-   # ########################################################################
-   my $capture_sync = new OSCCaptureSync(Quoter => $q);
-   my $copy_rows    = new CopyRowsInsertSelect(
-      Retry  => new Retry(),
-      Quoter => $q,
-   );
-
-   # More values are added later.  These are the minimum need to do --cleanup.
-   my %plugin_args = (
-      dbh   => $dbh,
-      msg   => \&msg,  # so plugin can talk back to user
-      print => $o->get('print'),
-      %tables,
-      %common_modules,
-   );
-
-   if ( my $sleep_time = $o->get('sleep') ) {
-      PTDEBUG && _d("Sleep time:", $sleep_time);
-      $plugin_args{sleep} = sub {
-         my ( $chunkno ) = @_;
-         PTDEBUG && _d("Sleeping after chunk", $chunkno);
-         sleep($sleep_time);
+   # #####################################################################
+   # Step 1: Create the new table.
+   # #####################################################################
+   my $new_tbl;
+   eval {
+      $new_tbl = create_new_table(
+         orig_tbl     => $orig_tbl,
+         suffix       => '_new',
+         Cxn          => $cxn,
+         Quoter       => $q,
+         OptionParser => $o,
+         TableParser  => $tp,
+      );
+   };
+   if ( $EVAL_ERROR ) {
+      die "Error creating new table: $EVAL_ERROR\n";
+   }
+
+   # If the new table still exists, drop it unless the tool was interrupted.
+   push @cleanup_tasks, sub {
+      PTDEBUG && _('Clean up new table');
+      my $new_tbl_exists = $tp->check_table(
+         dbh => $cxn->dbh(),
+         db  => $new_tbl->{db},
+         tbl => $new_tbl->{tbl},
+      );
+      PTDEBUG && _d('New table exists:', $new_tbl_exists ? 'yes' : 'no');
+      return unless $new_tbl_exists;
+
+      my $sql = "DROP TABLE IF EXISTS $new_tbl->{name};";
+      if ( !$oktorun ) {
+         # The tool was interrupted, so do not drop the new table
+         # in case the user wants to resume (once resume capability
+         # is implemented).
+         print "Not dropping the new table $new_tbl->{name} because "
+             . "the tool was interrupted.  To drop the new table, "
+             . "execute:\n$sql\n";
+      }
+      elsif ( $orig_tbl->{copied} && !$orig_tbl->{swapped} ) {
+         print "Not dropping the new table $new_tbl->{name} because "
+             . "--swap-tables failed.  To drop the new table, "
+             . "execute:\n$sql\n";
+      }
+      else {
+         print "Dropping new table...\n";
+         print $sql, "\n" if $o->get('print');
+         PTDEBUG && _d($sql);
+         eval {
+            $cxn->dbh()->do($sql);
+         };
+         if ( $EVAL_ERROR ) {
+            warn "Error dropping new table $new_tbl->{name}: $EVAL_ERROR\n"
+               . "To try dropping the new table again, execute:\n$sql\n";
+         }
+         print "Dropped new table OK.\n";
+      }
+   };
+
+   # #####################################################################
+   # Step 2: Alter the new, empty table.  This should be very quick,
+   # or die if the user specified a bad alter statement.
+   # #####################################################################
+   if ( my $alter = $o->get('alter') ) {
+      print "Altering new table...\n";
+      my $sql = "ALTER TABLE $new_tbl->{name} $alter";
+      print $sql, "\n" if $o->get('print');
+      PTDEBUG && _d($sql);
+      eval {
+         $cxn->dbh()->do($sql);
       };
-   }
-
-   # ########################################################################
-   # Just cleanup and exit.
-   # ########################################################################
-   if ( $o->get('cleanup-and-exit') ) {
-      msg("Calling " . (ref $copy_rows). "::cleanup()");
-      $copy_rows->cleanup(%plugin_args);
-
-      msg("Calling " . (ref $capture_sync) . "::cleanup()");
-      $capture_sync->cleanup(%plugin_args);
-
-      msg("$PROGRAM_NAME ending for --cleanup-and-exit");
-      return 0;
-   }
-
-   # ########################################################################
-   # Check that table can be altered.
-   # ########################################################################
-   my %tbl_info;
+      if ( $EVAL_ERROR ) {
+         die "Error altering new table $new_tbl->{name}: $EVAL_ERROR\n"
+      }
+      print "Altered $new_tbl->{name} OK.\n"
+   }
+
+   # Get the new table struct.  This shouldn't die because
+   # we just created the table successfully so we know it's
+   # there.  But the ghost of Ryan is everywhere.
+   my $ddl = $tp->get_create_table(
+      $cxn->dbh(),
+      $new_tbl->{db},
+      $new_tbl->{tbl},
+   );
+   $new_tbl->{tbl_struct} = $tp->parse($ddl);
+
+   # Determine what columns the original and new table share.
+   # If the user drops a col, that's easy: just don't copy it.  If they
+   # add a column, it must have a default value.  Other alterations
+   # may or may not affect the copy process--we'll know when we try!
+   # Note: we don't want to examine the --alter statement to see if the
+   # cols have changed because that's messy and prone to parsing errors.
+   # Col posn (position) is just for looks because user's like
+   # to see columns listed in their original order, not Perl's
+   # random hash key sorting.
+   my $col_posn    = $orig_tbl->{tbl_struct}->{col_posn};
+   my $orig_cols   = $orig_tbl->{tbl_struct}->{is_col};
+   my $new_cols    = $new_tbl->{tbl_struct}->{is_col};
+   my @common_cols = sort { $col_posn->{$a} <=> $col_posn->{$b} }
+                     grep { $new_cols->{$_} }
+                     keys %$orig_cols;
+   PTDEBUG && _d('Common columns', @common_cols);
+
+   # ########################################################################
+   # Step 3: Create the triggers to capture changes on the original table and
+   # apply them to the new table.
+   # ########################################################################
+
+   # Drop the triggers.  We can save this cleanup task before
+   # adding the triggers because if adding them fails, this will be
+   # called which will drop whichever triggers were created.
+   push @cleanup_tasks, sub {
+      PTDEBUG && _d('Clean up triggers');
+      if ( $oktorun ) {
+         drop_triggers(
+            tbl          => $orig_tbl,
+            Cxn          => $cxn,
+            Quoter       => $q,
+            OptionParser => $o,
+         );
+      }
+      else {
+         print "Not dropping triggers because the tool was interrupted.  "
+             . "To drop the triggers, execute:\n"
+             . join("\n", @drop_trigger_sqls) . "\n";
+      }
+   };
+
    eval {
-      %tbl_info = check_tables(%plugin_args);
+      create_triggers(
+         orig_tbl     => $orig_tbl,
+         new_tbl      => $new_tbl,
+         columns      => \@common_cols,
+         Cxn          => $cxn,
+         Quoter       => $q,
+         OptionParser => $o,
+      );
    };
    if ( $EVAL_ERROR ) {
-      chomp $EVAL_ERROR;
-      msg("Table $tbl cannot be altered: $EVAL_ERROR");
-      return 1;
-   }
-
-   @plugin_args{keys %tbl_info} = values %tbl_info;
-   msg("Table $tbl can be altered");
-   msg("Chunk column $plugin_args{chunk_column}, index $plugin_args{chunk_index}");
-
-   if ( $o->get('check-tables-and-exit') ) {
-      msg("$PROGRAM_NAME ending for --check-tables-and-exit");
-      return 0;
-   }
-
-   # #####################################################################
-   # Chunk the table.  If the checks pass, then this shouldn't fail.
-   # #####################################################################
-   my %range_stats = $chunker->get_range_statistics(
-      dbh        => $dbh,
-      db         => $db,
-      tbl        => $tbl,
-      chunk_col  => $plugin_args{chunk_column},
-      tbl_struct => $plugin_args{tbl_struct},
-   );
-   my @chunks = $chunker->calculate_chunks(
-      dbh        => $dbh,
-      db         => $db,
-      tbl        => $tbl,
-      chunk_col  => $plugin_args{chunk_column},
-      tbl_struct => $plugin_args{tbl_struct},
-      chunk_size => $o->get('chunk-size'),
-      %range_stats,
-   );
-   $plugin_args{chunks}   = \@chunks;
-   $plugin_args{Progress} = new Progress(
-      jobsize => scalar @chunks,
-      spec    => $o->get('progress'),
-      name    => "Copying rows",
-   );
-   msg("Chunked table $tbl into " . scalar @chunks . " chunks");
-
-   # #####################################################################
-   # Get child tables if necessary.
-   # #####################################################################
-   my @child_tables;
-   if ( my $child_tables = $o->get('child-tables') ) {
-      if ( lc $child_tables eq 'auto_detect' ) {
-         msg("Auto-detecting child tables of $tbl");
-         @child_tables = get_child_tables(%plugin_args);
-         msg("Child tables of $tables{old_tbl}: "
-            .  (@child_tables ? join(', ', @child_tables) : "(none)"));
-      }
-      else {
-         @child_tables = split(',', $child_tables);
-         msg("User-specified child tables: " . join(', ', @child_tables));
-      }
-   }
-
-   # #####################################################################
-   # Do the online alter.
-   # #####################################################################
-   if ( !$o->get('execute') ) {
-      msg("Exiting without altering $db.$tbl because you did not "
-        . "specify --execute.  Please read the tool's documentation "
-        . "carefully before using this tool.");
-      return $exit_status;
-   }
-
-   msg("Starting online schema change");
-   eval {
-      my $sql = "";
-      
-      # #####################################################################
-      # Create and alter the new table.
-      # #####################################################################
-      if ( $o->get('create-tmp-table') ) {
-         $sql = "CREATE TABLE `$db`.`$tmp_tbl` LIKE `$db`.`$tbl`";
-         msg($sql);
-         $dbh->do($sql) unless $o->get('print');
-      }
-
-      if ( my $alter = $o->get('alter') ) {
-         my @stmts;
-         if ( -f $alter && -r $alter ) {
-            msg("Reading ALTER TABLE statements from file $alter");
-            open my $fh, '<', $alter or die "Cannot open $alter: $OS_ERROR";
-            @stmts = <$fh>;
-            close $fh;
+      die "Error creating triggers: $EVAL_ERROR\n";
+   };
+
+   # #####################################################################
+   # Step 4: Copy rows.
+   # #####################################################################
+
+   # The hashref of callbacks below is what NibbleIterator calls internally
+   # to do all the copy work.  The callbacks do not need to eval their work
+   # because the higher call to $nibble_iter->next() is eval'ed which will
+   # catch any errors in the callbacks.
+   my $total_rows = 0;
+   my $total_time = 0;
+   my $avg_rate   = 0;  # rows/second
+   my $retry      = new Retry();  # for retrying to exec the copy statement
+   my $limit      = $o->get('chunk-size-limit');  # brevity
+   my $chunk_time = $o->get('chunk-time');        # brevity
+
+   my $callbacks = {
+      init => sub {
+         my (%args) = @_;
+         my $tbl         = $args{tbl};
+         my $nibble_iter = $args{NibbleIterator};
+
+         if ( $o->get('dry-run') ) {
+            print "Not copying rows because this is a dry run.\n";
          }
          else {
-            @stmts = split(';', $alter);
-         }
-
-         foreach my $stmt ( @stmts ) {
-            $sql = "ALTER TABLE `$db`.`$tmp_tbl` $stmt";
-            msg($sql);
-            $dbh->do($sql) unless $o->get('print');
+            print "Copying approximately ", $nibble_iter->row_estimate(),
+               " rows...\n";
+         }
+
+         if ( $o->get('print') ) {
+            # Print the checksum and next boundary statements.
+            my $statements = $nibble_iter->statements();
+            foreach my $sth ( sort keys %$statements ) {
+               next if $sth =~ m/^explain/;
+               if ( $statements->{$sth} ) {
+                  print $statements->{$sth}->{Statement}, "\n";
+               }
+            }
+         }
+
+         return unless $o->get('execute');
+
+         # If table is a single chunk on the master, make sure it's also
+         # a single chunk on all slaves.  E.g. if a slave is out of sync
+         # and has a lot more rows than the master, single chunking on the
+         # master could cause the slave to choke.
+         if ( $nibble_iter->one_nibble() ) {
+            PTDEBUG && _d('Getting table row estimate on replicas');
+            my @too_large;
+            foreach my $slave ( @$slaves ) {
+               my ($n_rows) = NibbleIterator::get_row_estimate(
+                  Cxn   => $slave,
+                  tbl   => $tbl,
+               );
+               PTDEBUG && _d('Table on',$slave->name(),'has', $n_rows, 'rows');
+               if ( $n_rows && $n_rows > ($tbl->{chunk_size} * $limit) ) {
+                  PTDEBUG && _d('Table too large on', $slave->name());
+                  push @too_large, [$slave->name(), $n_rows || 0];
+               }
+            }
+            if ( @too_large ) {
+               my $msg
+                  = "Cannot copy table $tbl->{name} because"
+                  . " on the master it would be checksummed in one chunk"
+                  . " but on these replicas it has too many rows:\n";
+               foreach my $info ( @too_large ) {
+                  $msg .= "  $info->[1] rows on $info->[0]\n";
+               }
+               $msg .= "The current chunk size limit is "
+                     . ($tbl->{chunk_size} * $limit)
+                     . " rows (chunk size=$tbl->{chunk_size}"
+                     . " * chunk size limit=$limit).\n";
+               die $msg;
+            }
+         }
+
+         return 1; # continue nibbling table
+      },
+      next_boundaries => sub {
+         my (%args) = @_;
+         my $tbl         = $args{tbl};
+         my $nibble_iter = $args{NibbleIterator};
+         my $sth         = $nibble_iter->statements();
+         my $boundary    = $nibble_iter->boundaries();
+
+         return 0 if $o->get('dry-run');
+         return 1 if $nibble_iter->one_nibble();
+
+         # Check that MySQL will use the nibble index for the next upper
+         # boundary sql.  This check applies to the next nibble.  So if
+         # the current nibble number is 5, then nibble 5 is already done
+         # and we're checking nibble number 6.
+         my $expl = explain_statement(
+            tbl  => $tbl,
+            sth  => $sth->{explain_upper_boundary},
+            vals => [ @{$boundary->{lower}}, $nibble_iter->chunk_size() ],
+         );
+         if (lc($expl->{key} || '') ne lc($nibble_iter->nibble_index() || '')) {
+            my $msg
+               = "Aborting copying table $tbl->{name} at chunk "
+               . ($nibble_iter->nibble_number() + 1)
+               . " because it is not safe to ascend.  Chunking should "
+               . "use the "
+               . ($nibble_iter->nibble_index() || '?')
+               . " index, but MySQL EXPLAIN reports that "
+               . ($expl->{key} ? "the $expl->{key}" : "no")
+               . " index will be used for "
+               . $sth->{upper_boundary}->{Statement}
+               . " with values "
+               . join(", ", map { defined $_ ? $_ : "NULL" }
+                      (@{$boundary->{lower}}, $nibble_iter->chunk_size()))
+               . "\n";
+            die $msg;
          } 
-      }
-
-      # #####################################################################
-      # Determine what columns the two tables have in common.
-      # #####################################################################
-      my @columns;
-      # If --print is in effect, then chances are the new table wasn't
-      # created above, so we can't get it's struct.
-      # TODO: check if the new table exists because user might have created
-      # it manually.
-      if ( !$o->get('print') ) {
-         my $tmp_tbl_struct = $tp->parse(
-            $tp->get_create_table($dbh, $db, $tmp_tbl));
-
-         @columns = intersection([
-            $plugin_args{tbl_struct}->{is_col},
-            $tmp_tbl_struct->{is_col},
-         ]);
-
-         # Order columns according to new table because people like/expect
-         # to see things in a certain order (this has been an issue before).
-         # This just matters to us; does't make a difference to MySQL.
-         my $col_posn = $plugin_args{tbl_struct}->{col_posn};
-         @columns = sort { $col_posn->{$a} <=> $col_posn->{$b} } @columns;
-         msg("Shared columns: " . join(', ', @columns));
-      }
-      $plugin_args{columns} = \@columns;
-
-      # #####################################################################
-      # Start capturing changes to the new table.
-      # #####################################################################
-      msg("Calling " . (ref $capture_sync) . "::capture()");
-      $capture_sync->capture(%plugin_args);
-
-      # #####################################################################
-      # Copy rows from new table to old table.
-      # #####################################################################
-      msg("Calling " . (ref $copy_rows) . "::copy()");
-      $copy_rows->copy(
-         from_table => $q->quote($db, $tbl),
-         to_table   => $q->quote($db, $tmp_tbl),
-         %plugin_args
-      );
-
-      # #####################################################################
-      # Sync tables.
-      # #####################################################################
-      msg("Calling " . (ref $capture_sync) . "::sync()");
-      $capture_sync->sync(%plugin_args);
-
-      # #####################################################################
-      # Rename tables.
-      # #####################################################################
-      if ( $o->get('rename-tables') ) {
-         msg("Renaming tables");
-         $sql = "RENAME TABLE `$db`.`$tbl` TO `$db`.`$old_tbl`,"
-              . " `$db`.`$tmp_tbl` TO `$db`.`$tbl`";
-         msg($sql);
-         $dbh->do($sql) unless $o->get('print');
-         msg("Original table $tbl renamed to $old_tbl");
-      }
-
-      # #####################################################################
-      # Update foreign key constraints if there are child tables.
-      # #####################################################################
-      if ( @child_tables ) {
-         msg("Renaming foreign key constraints in child table");
-         if ( $rename_fk_method eq 'rebuild_constraints' ) { 
-            update_foreign_key_constraints(
-               child_tables => \@child_tables,
-               %plugin_args,
-            );
-         }
-         elsif ( $rename_fk_method eq 'drop_old_table' ) {
-            $sql = "SET foreign_key_checks=0";
-            msg($sql);
-            $dbh->do($sql) unless $o->get('print');
-            
-            $sql = "DROP TABLE IF EXISTS `$db`.`$tbl`";
-            msg($sql);
-            $dbh->do($sql) unless $o->get('print');
-
-            $sql = "RENAME TABLE `$db`.`$tmp_tbl` TO `$db`.`$tbl`";
-            msg($sql);
-            $dbh->do($sql) unless $o->get('print');
+
+         # Once nibbling begins for a table, control does not return to this
+         # tool until nibbling is done because, as noted above, all work is
+         # done in these callbacks.  This callback is the only place where we
+         # can prematurely stop nibbling by returning false.  This allows
+         # Ctrl-C to stop the tool between nibbles instead of between tables.
+         return $oktorun; # continue nibbling table?
+      },
+      exec_nibble => sub {
+         my (%args) = @_;
+         my $tbl         = $args{tbl};
+         my $nibble_iter = $args{NibbleIterator};
+         my $sth         = $nibble_iter->statements();
+         my $boundary    = $nibble_iter->boundaries();
+
+         return if $o->get('dry-run');
+
+         # Count every chunk, even if it's ultimately skipped, etc.
+         $tbl->{results}->{n_chunks}++;
+
+         # If the table is being chunk (i.e., it's not small enough to be
+         # consumed by one nibble), then check index usage and chunk size.
+         if ( !$nibble_iter->one_nibble() ) {
+            my $expl = explain_statement(
+               tbl  => $tbl,
+               sth  => $sth->{explain_nibble},
+               vals => [ @{$boundary->{lower}}, @{$boundary->{upper}} ],
+            );
+
+            # Ensure that MySQL is using the chunk index.
+            if (   lc($expl->{key} || '')
+                ne lc($nibble_iter->nibble_index() || '') ) {
+               my $msg
+                  = "Aborting copying table $tbl->{name} at chunk "
+                  . $nibble_iter->nibble_number()
+                  . " because it is not safe to chunk.  Chunking should "
+                  . "use the "
+                  . ($nibble_iter->nibble_index() || '?')
+                  . " index, but MySQL EXPLAIN reports that "
+                  . ($expl->{key} ? "the $expl->{key}" : "no")
+                  . " index will be used for "
+                  . $sth->{explain_nibble}->{Statement}
+                  . " with values "
+                  . join(", ", map { defined $_ ? $_ : "NULL" }
+                         (@{$boundary->{lower}}, @{$boundary->{upper}}))
+                  . "\n";
+               die $msg;
+            }
+
+            # Check chunk size limit if the upper boundary and next lower
+            # boundary are identical.
+            if ( $limit ) {
+               my $boundary = $nibble_iter->boundaries(); 
+               my $oversize_chunk
+                  = $limit ? ($expl->{rows} || 0) >= $tbl->{chunk_size} * $limit
+                  :          0;
+               if ( $oversize_chunk
+                    && $nibble_iter->identical_boundaries(
+                          $boundary->{upper}, $boundary->{next_lower}) )
+               {
+                  my $msg
+                     = "Aborting copying table $tbl->{name} at chunk "
+                     . $nibble_iter->nibble_number()
+                     . " because the chunk is too large: MySQL estimates "
+                     . ($expl->{rows} || 0) . "rows.  The current chunk "
+                     . "size limit is " . ($tbl->{chunk_size} * $limit)
+                     . " rows (chunk size=$tbl->{chunk_size}"
+                     . " * chunk size limit=$limit).\n";
+                  die $msg;
+               }
+            }
+         }
+
+         # Exec and time the chunk checksum query.
+         $tbl->{nibble_time} = exec_nibble(
+            %args,
+            Retry        => $retry,
+            Quoter       => $q,
+            OptionParser => $o,
+         );
+         PTDEBUG && _d('Nibble time:', $tbl->{nibble_time});
+
+         # We're executing REPLACE queries which don't return rows.
+         # Returning 0 from this callback causes the nibble iter to
+         # get the next boundaries/nibble.
+         return 0;
+      },
+      after_nibble => sub {
+         my (%args) = @_;
+         my $tbl         = $args{tbl};
+         my $nibble_iter = $args{NibbleIterator};
+
+         return unless $o->get('execute');
+
+         # Update rate, chunk size, and progress if the nibble actually
+         # selected some rows.
+         my $cnt = $tbl->{row_cnt};
+         if ( ($cnt || 0) > 0 ) {
+            # Update the rate of rows per second for the entire server.
+            # This is used for the initial chunk size of the next table.
+            $total_rows += $cnt;
+            $total_time += $tbl->{nibble_time};
+            $avg_rate    = int($total_rows / $total_time);
+            PTDEBUG && _d('Average copy rate (rows/s):', $avg_rate);
+
+            # Adjust chunk size.  This affects the next chunk.
+            if ( $chunk_time ) {
+               # Calcuate a new chunk-size based on the rate of rows/s.
+               $tbl->{chunk_size} = $tbl->{rate}->update(
+                  $cnt,                # processed this many rows
+                  $tbl->{nibble_time}, # is this amount of time
+               );
+
+               if ( $tbl->{chunk_size} < 1 ) {
+                  # This shouldn't happen.  WeightedAvgRate::update() may
+                  # return a value < 1, but minimum chunk size is 1.
+                  $tbl->{chunk_size} = 1;
+
+                  # This warning is printed once per table.
+                  if ( !$tbl->{warned_slow} ) {
+                     warn "Rows are copying very slowly.  "
+                        . "--chunk-size has been automatically reduced to 1.  "
+                        . "Check that the server is not being overloaded, "
+                        . "or increase --chunk-time.  The last chunk "
+                        . "selected $cnt rows and took "
+                        . sprintf('%.3f', $tbl->{nibble_time})
+                        . " seconds to execute.\n";
+                     $tbl->{warned_slow} = 1;
+                  }
+               }
+
+               # Update chunk-size based on the rate of rows/s.
+               $nibble_iter->set_chunk_size($tbl->{chunk_size});
+            }
+
+            # Every table should have a Progress obj; update it.
+            if ( my $tbl_pr = $tbl->{progress} ) {
+               $tbl_pr->update( sub { return $total_rows } );
+            }
+         }
+
+         # Wait forever for slaves to catch up.
+         $replica_lag_pr->start() if $replica_lag_pr;
+         $replica_lag->wait(Progress => $replica_lag_pr);
+
+         # Wait forever for system load to abate.  wait() will die if
+         # --critical load is reached.
+         $sys_load_pr->start() if $sys_load_pr;
+         $sys_load->wait(Progress => $sys_load_pr);
+
+         return;
+      },
+      done => sub {
+         if ( $o->get('execute') ) {
+            print "Copied rows OK.\n";
+         }
+      },
+   };
+
+   # NibbleIterator combines these two statements and adds
+   # "FROM $orig_table->{name} WHERE <nibble stuff>".
+   my $dml    = "INSERT LOW_PRIORITY IGNORE INTO $new_tbl->{name} "
+              . "(" . join(', ', map { $q->quote($_) } @common_cols) . ") "
+              . "SELECT";
+   my $select = join(', ', map { $q->quote($_) } @common_cols);
+
+   # The chunk size is auto-adjusted, so use --chunk-size as
+   # the initial value, but then save and update the adjusted
+   # chunk size in the table data struct.
+   $orig_tbl->{chunk_size} = $o->get('chunk-size');
+
+   # This won't (shouldn't) fail because we already verified in
+   # check_orig_table() table we can NibbleIterator::can_nibble().
+   my $nibble_iter = new NibbleIterator(
+      Cxn          => $cxn,
+      tbl          => $orig_tbl,
+      chunk_size   => $orig_tbl->{chunk_size},
+      chunk_index  => $o->get('chunk-index'),
+      dml          => $dml,
+      select       => $select,
+      callbacks    => $callbacks,
+      OptionParser => $o,
+      Quoter       => $q,
+      TableParser  => $tp,
+      TableNibbler => new TableNibbler(TableParser => $tp, Quoter => $q),
+      comments     => {
+         bite   => "pt-online-schema-change $PID copy table",
+         nibble => "pt-online-schema-change $PID copy nibble",
+      },
+   );
+ 
+   # Init a new weighted avg rate calculator for the table.
+   $orig_tbl->{rate} = new WeightedAvgRate(target_t => $chunk_time);
+
+   # Make a Progress obj for this table.  It may not be used;
+   # depends on how many rows, chunk size, how fast the server
+   # is, etc.  But just in case, all tables have a Progress obj.
+   if ( $o->get('progress')
+        && !$nibble_iter->one_nibble()
+        &&  $nibble_iter->row_estimate() )
+   {
+      $orig_tbl->{progress} = new Progress(
+         jobsize => $nibble_iter->row_estimate(),
+         spec    => $o->get('progress'),
+         name    => "Copying $orig_tbl->{name}",
+      );
+   }
+
+   # Start copying rows.  This may take awhile, but --progress is on
+   # by default so there will be progress updates to stderr.
+   eval {
+      1 while $nibble_iter->next();
+   };
+   if ( $EVAL_ERROR ) {
+      die "Error copying rows from $orig_tbl->{name} to "
+        . "$new_tbl->{name}: $EVAL_ERROR\n";
+   }
+   $orig_tbl->{copied} = 1;  # flag for cleanup tasks
+
+
+   # XXX Auto-choose the alter fk method BEFORE swapping/renaming tables 
+   # else everything will break because if drop_swap is chosen, then we
+   # most NOT rename tables or drop the old table.
+   if ( $alter_fk_method eq 'auto' ) {
+      # If chunk time is set, then use the average rate of rows/s
+      # from copying the orig table to determine the max size of
+      # a child table that can be altered within one chunk time.
+      # The limit is a fudge factor.  Chunk time won't be set if
+      # the user specified --chunk-size=N on the cmd line, in which
+      # case the max child table size is their specified chunk size
+      # times the fudge factor.
+      my $max_rows
+         = $o->get('dry-run') ? $o->get('chunk-size') * $limit
+         : $chunk_time        ? $avg_rate * $chunk_time * $limit
+         :                      $o->get('chunk-size') * $limit;
+      PTDEBUG && _d('Max allowed child table size:', $max_rows);
+
+      $alter_fk_method = determine_alter_fk_method(
+         child_tables => $child_tables,
+         max_rows     => $max_rows,
+         Cxn          => $cxn,
+         OptionParser => $o,
+      );
+
+      if ( $alter_fk_method eq 'drop_swap' ) {
+         $o->set('swap-tables',    0);
+         $o->set('drop-old-table', 0);
+      }
+   }
+
+   # #####################################################################
+   # XXX
+   # Step 5: Rename tables: orig -> old, new -> orig
+   # Past this step, the original table has been altered. This shouldn't
+   # fail, but if it does, the failure could be serious depending on what
+   # state the tables are left in.
+   # XXX
+   # #####################################################################
+   my $old_tbl;
+   if ( $o->get('swap-tables') ) {
+      eval {
+         $old_tbl = swap_tables(
+            orig_tbl     => $orig_tbl,
+            new_tbl      => $new_tbl,
+            suffix       => '_old',
+            Cxn          => $cxn,
+            Quoter       => $q,
+            OptionParser => $o,
+         );
+      };
+      if ( $EVAL_ERROR ) {
+         die "Error swapping the tables: $EVAL_ERROR\n"
+           . "Verify that the original table $orig_tbl->{name} has not "
+           . "been modified or renamed to the old table $old_tbl->{name}.  "
+           . "Then drop the new table $new_tbl->{name} if it exists.\n";
+      }
+   }
+   $orig_tbl->{swapped} = 1;  # flag for cleanup tasks
+   PTDEBUG && _d('Old table:', Dumper($old_tbl));
+
+   # #####################################################################
+   # Step 6: Update foreign key constraints if there are child tables.
+   # #####################################################################
+   if ( $child_tables ) {
+      eval {
+         if ( $alter_fk_method eq 'none' ) {
+            print "Not updating foreign keys because "
+               . "--alter-foreign-keys-method=none.  Foreign keys "
+               . "that reference the table will no longer work.\n";
+         }
+         elsif ( $alter_fk_method eq 'rebuild_constraints' ) {
+            rebuild_constraints(
+               orig_tbl     => $orig_tbl,
+               old_tbl      => $old_tbl,
+               child_tables => $child_tables,
+               OptionParser => $o,
+               Quoter       => $q,
+               Cxn          => $cxn,
+               TableParser  => $tp,
+            );
+         }
+         elsif ( $alter_fk_method eq 'drop_swap' ) {
+            drop_swap(
+               orig_tbl     => $orig_tbl,
+               new_tbl      => $new_tbl,
+               Cxn          => $cxn,
+               OptionParser => $o,
+            );
          }
          else {
-            die "Invalid --update-foreign-keys-method value: $rename_fk_method";
-         }
-      }
-
-      # #####################################################################
-      # Cleanup.
-      # #####################################################################
-      msg("Calling " . (ref $copy_rows). "::cleanup()");
-      $copy_rows->cleanup(%plugin_args);
-
-      msg("Calling " . (ref $capture_sync) . "::cleanup()");
-      $capture_sync->cleanup(%plugin_args);
-
-      if ( $o->get('rename-tables') &&  $o->get('drop-old-table') ) {
-         $sql = "DROP TABLE IF EXISTS `$db`.`$old_tbl`";
-         msg($sql);
-         $dbh->do($sql) unless $o->get('print');
-      }
-   };
-   if ( $EVAL_ERROR ) {
-      warn "An error occurred:\n\n$EVAL_ERROR\n"
-         . "Some triggers, temp tables, etc. may not have been removed. "
-         . "Run with --cleanup-and-exit to remove these items.\n";
-      $exit_status = 1;
-   }
-
-   msg("$PROGRAM_NAME ended, exit status $exit_status");
+            # This should "never" happen because we check this var earlier.
+            die "Invalid --alter-foreign-keys-method: $alter_fk_method\n";
+         }
+      };
+      if ( $EVAL_ERROR ) {
+         # TODO: improve error message and handling.
+         die "Error updating foreign key constraints: $EVAL_ERROR\n";
+      }
+   }
+
+   # ########################################################################
+   # Step 7: Drop the old table.
+   # ########################################################################
+   if ( $o->get('drop-old-table') ) {
+      if ( $o->get('dry-run') ) {
+         print "Not dropping old table because this is a dry run.\n";
+      }
+      else {
+         print "Dropping old table...\n";
+
+         if ( $alter_fk_method eq 'none' ) {
+            # Child tables still reference the old table, but the user
+            # has chosen to break fks, so we need to disable fk checks
+            # in order to drop the old table.
+            my $sql = "SET foreign_key_checks=0";
+            PTDEBUG && _d($sql);
+            print $sql, "\n" if $o->get('print');
+            $cxn->dbh()->do($sql);
+         }
+
+         my $sql = "DROP TABLE IF EXISTS $old_tbl->{name}";
+         print $sql, "\n" if $o->get('print');
+         PTDEBUG && _d($sql); 
+         eval {
+            $cxn->dbh()->do($sql);
+         };
+         if ( $EVAL_ERROR ) {
+            die "Error dropping the old table: $EVAL_ERROR\n";
+         }
+         print "Dropped old table $old_tbl->{name} OK.\n";
+      }
+   }
+   
+   # ########################################################################
+   # Done.
+   # ########################################################################
+   $orig_tbl->{success} = 1;  # flag for cleanup tasks
+   $cleanup = undef;          # exec cleanup tasks
+
    return $exit_status;
 }
 
 # ############################################################################
 # Subroutines.
 # ############################################################################
-sub check_tables {
+sub create_new_table{
+   my (%args) = @_;
+   my @required_args = qw(orig_tbl Cxn Quoter OptionParser TableParser);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($orig_tbl, $cxn, $q, $o, $tp) = @args{@required_args};
+
+   # Get the original table struct.
+   my $ddl = $tp->get_create_table(
+      $cxn->dbh(),
+      $orig_tbl->{db},
+      $orig_tbl->{tbl},
+   );
+
+   my $tries      = $args{tries}  || 10;  # don't try forever
+   my $prefix     = $args{prefix} || '_';
+   my $suffix     = $args{suffix} || '_new';
+   my $table_name = $orig_tbl->{tbl} . $suffix;
+
+   print "Creating new table...\n";
+   my $tryno = 1;
+   my @old_tables;
+   while ( $tryno++ < $tries ) {
+      $table_name = $prefix . $table_name;
+      my $quoted  = $q->quote($orig_tbl->{db}, $table_name);
+
+      # Generate SQL to create the new table.  We do not use CREATE TABLE LIKE
+      # because it doesn't preserve foreign key constraints. Here we need to
+      # rename the FK constraints, too. This is because FK constraints are
+      # internally stored as <database>.<constraint> and there cannot be
+      # duplicates.  If we don't rename the constraints, then InnoDB will throw
+      # error 121 (duplicate key violation) when we try to execute the CREATE
+      # TABLE.  TODO: this code isn't perfect. If we rename a constraint from
+      # foo to _foo and there is already a constraint with that name in this
+      # or another table, we can still have a collision. But if there are
+      # multiple FKs on this table, it's hard to know which one is causing the
+      # trouble. Should we generate random/UUID FK names or something instead?
+      my $sql    = $ddl;
+      $sql       =~ s/\ACREATE TABLE .*?\($/CREATE TABLE $quoted (/m;
+      $sql       =~ s/^  CONSTRAINT `/  CONSTRAINT `_/gm;
+      PTDEBUG && _d($sql);
+      eval {
+         $cxn->dbh()->do($sql);
+      };
+      if ( $EVAL_ERROR ) {
+         # Ignore this error because if multiple instances of the tool
+         # are running, or previous runs failed and weren't cleaned up,
+         # then there will be other similarly named tables with fewer
+         # leading prefix chars.  Or, in rarer cases, the db just happens
+         # to have a similarly named table created by the user for other
+         # purposes.
+         if ( $EVAL_ERROR =~ m/table.+?already exists/i ) {
+            push @old_tables, $q->quote($orig_tbl->{db}, $table_name);
+            next;
+         }
+
+         # Some other error happened.  Let the caller catch it.
+         die $EVAL_ERROR;
+      }
+      print $sql, "\n" if $o->get('print');  # the sql that work
+      print "Created new table $orig_tbl->{db}.$table_name OK.\n";
+      return { # success
+         db   => $orig_tbl->{db},
+         tbl  => $table_name,
+         name => $q->quote($orig_tbl->{db}, $table_name),
+      };
+   }
+
+   die "Failed to find a unique new table name after $tries attemps.  "
+     . "The following tables exist which may be left over from previous "
+     . "failed runs of the tool:\n"
+     . join("\n", map { "  $_" } @old_tables)
+     . "\nExamine these tables and drop some or all of them if they are "
+     . "no longer need, then re-run the tool.\n";
+}
+
+sub swap_tables {
+   my (%args) = @_;
+   my @required_args = qw(orig_tbl new_tbl Cxn Quoter OptionParser);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($orig_tbl, $new_tbl, $cxn, $q, $o) = @args{@required_args};
+
+   my $prefix     = '_';
+   my $table_name = $orig_tbl->{tbl} . ($args{suffix} || '');
+   my $tries      = 10;  # don't try forever
+
+   # This sub only works for --execute.  Since the options are
+   # mutually exclusive and we return in the if case, the elsif
+   # is just a paranoid check because swapping the tables is one
+   # of the most sensitive/dangerous operations.
+   if ( $o->get('dry-run') ) {
+      print "Not swapping tables because this is a dry run.\n";
+
+      # A return value really isn't needed, but this trick allows
+      # rebuild_constraints() to parse and show the sql statements
+      # it would used.  Otherwise, this has no effect.      
+      return $orig_tbl;
+   }
+   elsif ( $o->get('execute') ) {
+      print "Swapping tables...\n";
+
+      while ( $tries-- ) {
+         $table_name = $prefix . $table_name;
+         my $sql     = "RENAME TABLE $orig_tbl->{name} "
+                     . "TO " . $q->quote($orig_tbl->{db}, $table_name)
+                     . ", $new_tbl->{name} TO $orig_tbl->{name}"; 
+         PTDEBUG && _d($sql); 
+         eval {
+            $cxn->dbh()->do($sql);
+         };
+         if ( $EVAL_ERROR ) {
+            # Ignore this error because if multiple instances of the tool
+            # are running, or previous runs failed and weren't cleaned up,
+            # then there will be other similarly named tables with fewer
+            # leading prefix chars.  Or, in rarer cases, the db just happens
+            # to have a similarly named table created by the user for other
+            # purposes.
+            next if $EVAL_ERROR =~ m/table.+?already exists/i;
+
+            # Some other error happened.  Let caller catch it.
+            die $EVAL_ERROR;
+         }
+         print $sql, "\n" if $o->get('print');
+         print "Swapped original and new tables OK.\n";
+         return { # success
+            db   => $orig_tbl->{db},
+            tbl  => $table_name,
+            name => $q->quote($orig_tbl->{db}, $table_name),
+         };
+      }
+
+      # This shouldn't happen.
+      # Here and in the attempt to find a new table name we probably ought to
+      # use --retries (and maybe a Retry object?)
+      die "Failed to find a unique old table name after serveral attempts.\n";
+   }
+}
+
+sub check_orig_table {
    my ( %args ) = @_;
-   my @required_args = qw(dbh db tbl tmp_tbl old_tbl VersionParser Quoter TableParser OptionParser TableChunker);
+   my @required_args = qw(orig_tbl Cxn TableParser OptionParser Quoter);
    foreach my $arg ( @required_args ) {
       die "I need a $arg argument" unless $args{$arg};
    }
-   my ($dbh, $db, $tbl, $tmp_tbl, $old_tbl, $o, $tp)
-      = @args{qw(dbh db tbl tmp_tbl old_tbl OptionParser TableParser)};
-
-   msg("Checking if table $tbl can be altered");
-   my %tbl_info;
-   my $sql = "";
-
-   # ########################################################################
-   # Check MySQL.
-   # ########################################################################   
-   # Although triggers were introduced in 5.0.2, "Prior to MySQL 5.0.10,
-   # triggers cannot contain direct references to tables by name."
-   if ( !$args{VersionParser}->version_ge($dbh, '5.0.10') ) {
-      die "This tool requires MySQL 5.0.10 or newer\n";
-   }
-
-   # ########################################################################
-   # Check the (original) table.
-   # ########################################################################
-   # The table must exist of course.
-   if ( !$tp->check_table(dbh=>$dbh, db=>$db, tbl=>$tbl) ) {
-      die "Table $db.$tbl does not exist\n";
-   }
-
-   # There cannot be any triggers on the table.
-   $sql = "SHOW TRIGGERS FROM `$db` LIKE '$tbl'";
-   msg($sql);
+   my ($orig_tbl, $cxn, $tp, $o, $q) = @args{@required_args};
+
+   my $dbh = $cxn->dbh();
+
+   # The original table must exist, of course.
+   if (!$tp->check_table(dbh=>$dbh,db=>$orig_tbl->{db},tbl=>$orig_tbl->{tbl})) {
+      die "The original table $orig_tbl->{name} does not exist.\n";
+   }
+
+   # There cannot be any triggers on the original table.
+   my $sql = 'SHOW TRIGGERS FROM ' . $q->quote($orig_tbl->{db})
+           . ' LIKE ' . $q->literal_like($orig_tbl->{tbl});
+   PTDEBUG && _d($sql);
    my $triggers = $dbh->selectall_arrayref($sql);
    if ( $triggers && @$triggers ) {
-      die "Table $db.$tbl has triggers.  This tool needs to create "
-         . "its own triggers, so the table cannot already have triggers.\n";
+      die "The table $orig_tbl->{name} has triggers.  This tool "
+         . "needs to create its own triggers, so the table cannot "
+         . "already have triggers.\n";
    }
 
-   # For now, we require that the old table has an exact-chunkable
-   # column (i.e. unique single-column).
-   $tbl_info{tbl_struct} = $tp->parse($tp->get_create_table($dbh, $db, $tbl));
-   my ($exact, @chunkable_cols) = $args{TableChunker}->find_chunk_columns(
-      tbl_struct => $tbl_info{tbl_struct},
-      exact      => 1,
+   # Get the table struct.  NibbleIterator needs this, and so do we.
+   my $ddl = $tp->get_create_table(
+      $cxn->dbh(),
+      $orig_tbl->{db},
+      $orig_tbl->{tbl},
    );
-   if ( !$exact || !@chunkable_cols ) {
-      die "Table $db.$tbl cannot be chunked because it does not have "
-         . "a unique, single-column index\n";
-   }
-   $tbl_info{chunk_column} = $chunkable_cols[0]->{column};
-   $tbl_info{chunk_index}  = $chunkable_cols[0]->{index};
-
-   # ########################################################################
-   # Check the tmp table.
-   # ########################################################################   
-   # The tmp table should not exist if we're supposed to create it.
-   # Else, if user specifies --no-create-tmp-table, they should ensure
-   # that it exists.
-   if ( $o->get('create-tmp-table')
-        && $tp->check_table(dbh=>$dbh, db=>$db, tbl=>$tmp_tbl) ) {
-      die "Temporary table $db.$tmp_tbl exists which will prevent "
-         . "--create-tmp-table from creating the temporary table.\n";
-   }
-
-   # ########################################################################
-   # Check the old table.
-   # ########################################################################   
-   # If we're going to rename the tables, which we do by default, then
-   # the old table cannot already exist.
-   if ( $o->get('rename-tables')
-        && $tp->check_table(dbh=>$dbh, db=>$db, tbl=>$old_tbl) ) {
-      die "Table $db.$old_tbl exists which will prevent $db.$tbl "
-         . "from being renamed to it.  Table $db.$old_tbl could be from "
-         . "a previous run that failed.  See --drop-old-table for more "
-         . "information.\n";
-   }
-
-   return %tbl_info;
+   $orig_tbl->{tbl_struct} = $tp->parse($ddl);
+
+   # Must be able to nibble the original table (to copy rows to the new table).
+   eval {
+      NibbleIterator::can_nibble(
+         Cxn          => $cxn,
+         tbl          => $orig_tbl,
+         chunk_size   => $o->get('chunk-size'),
+         chunk_indx   => $o->get('chunk-index'),
+         OptionParser => $o,
+         TableParser  => $tp,
+      );
+   };
+   if ( $EVAL_ERROR ) {
+      die "Cannot chunk the original table $orig_tbl->{name}: $EVAL_ERROR\n";
+   }
+
+   # Find a pk or unique index to use for the delete trigger.  can_nibble()
+   # above returns an index, but NibbleIterator will use non-unique indexes,
+   # so we have to do this again here.
+   my $indexes = $orig_tbl->{tbl_struct}->{indexes}; # brevity
+   foreach my $index ( $tp->sort_indexes($orig_tbl->{tbl_struct}) ) {
+      if ( $index eq 'PRIMARY' || $indexes->{$index}->{is_unique} ) {
+         PTDEBUG && _d('Delete trigger index:', Dumper($index));
+         $orig_tbl->{del_index} = $index;
+         last;
+      }
+   }
+   if ( !$orig_tbl->{del_index} ) {
+      die "The original table $orig_tbl->{name} does not have a PRIMARY KEY "
+        . "or a unique index which is required for the DELETE trigger.\n";
+   }
+
+   return;  # success
 }
 
-sub get_child_tables {
+sub find_child_tables {
    my ( %args ) = @_;
-   my @required_args = qw(dbh db tbl Quoter);
+   my @required_args = qw(tbl Cxn Quoter);
    foreach my $arg ( @required_args ) {
       die "I need a $arg argument" unless $args{$arg};
    }
-   my ($dbh, $db, $tbl, $q) = @args{@required_args};
-
-   my $sql = "SELECT table_name "
+   my ($tbl, $cxn, $q) = @args{@required_args};
+   
+   PTDEBUG && _d('Finding child tables');
+   
+   my $sql = "SELECT table_schema, table_name "
            . "FROM information_schema.key_column_usage "
-           . "WHERE constraint_schema='$db' AND referenced_table_name='$tbl'";
-   PTDEBUG && _d($dbh, $sql);
-   my $child_tables;
-   eval {
-      $child_tables = $dbh->selectall_arrayref($sql);
-   };
-   if ( $EVAL_ERROR ) {
-      die "Error executing query to check $tbl for child tables.\n\n"
-         . "Query: $sql\n\n"
-         . "Error: $EVAL_ERROR"
-   }
-
-   PTDEBUG && _d("Child tables:", join(', ', map { $_->[0] } @$child_tables));
-   return map { $_->[0] } @$child_tables;
-}
-
-sub update_foreign_key_constraints {
-   my ( %args ) = @_;
-   my @required_args = qw(msg dbh db tbl old_tbl child_tables Quoter);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($msg, $dbh, $db, $tbl, $old_tbl, $child_tables, $q)
+           . "WHERE constraint_schema='$tbl->{db}' "
+           . "AND referenced_table_name='$tbl->{tbl}'";
+   PTDEBUG && _d($sql);
+   my $rows = $cxn->dbh()->selectall_arrayref($sql);
+   if ( !$rows || !@$rows ) {
+      PTDEBUG && _d('No child tables found');
+      return;
+   }
+
+   my @child_tables;
+   foreach my $row ( @$rows ) {
+      my $tbl = {
+         db   => $row->[0],
+         tbl  => $row->[1],
+         name => $q->quote(@$row),
+      };
+
+      # Get row estimates for each child table so we can give the user
+      # some input on choosing an --alter-foreign-keys-method if they
+      # don't use "auto".
+      my ($n_rows) = NibbleIterator::get_row_estimate(
+         Cxn => $cxn,
+         tbl => $tbl,
+      );
+      $tbl->{row_est} = $n_rows;
+
+      push @child_tables, $tbl;
+   }
+
+   PTDEBUG && _d('Child tables:', Dumper(\@child_tables));
+   return \@child_tables;
+}
+
+sub determine_alter_fk_method {
+   my ( %args ) = @_;
+   my @required_args = qw(child_tables max_rows Cxn OptionParser);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($child_tables, $max_rows, $cxn, $o) = @args{@required_args};
+
+   if ( $o->get('dry-run') ) {
+      print "Not determining the method to update foreign keys "
+         . "because this is a dry run.\n";
+      return '';  # $alter_fk_method can't be undef
+   }
+
+   # The rebuild_constraints method is the default becuase it's safer
+   # and doesn't cause the orig table to go missing for a moment.
+   my $method = 'rebuild_constraints';
+   
+   print "Max rows for the rebuild_constraints method: $max_rows\n",
+         "Determining the method to update foreign keys...\n";
+   foreach my $child_tbl ( @$child_tables ) {
+      print "  $child_tbl->{name}: ";
+      my ($n_rows) = NibbleIterator::get_row_estimate(
+         Cxn   => $cxn,
+         tbl   => $child_tbl,
+      );
+      if ( $n_rows > $max_rows ) {
+         print "too many rows: $n_rows; must use drop_swap\n";
+         $method = 'drop_swap';
+         last;
+      }
+      else {
+         print "$n_rows rows; can use rebuild_constraints\n";
+      }
+   }
+
+   return $method || '';  # $alter_fk_method can't be undef
+}
+
+sub rebuild_constraints {
+   my ( %args ) = @_;
+   my @required_args = qw(orig_tbl old_tbl child_tables
+                          Cxn Quoter OptionParser TableParser);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($orig_tbl, $old_tbl, $child_tables, $cxn, $q, $o, $tp)
       = @args{@required_args};
 
-   my $constraint = qr/^\s+(CONSTRAINT.+?REFERENCES `$old_tbl`.+)$/mo;
+   # MySQL has a "feature" where if the parent tbl is in the same db,
+   # then the child tbl ref is simply `parent_tbl`, but if the parent tbl
+   # is in another db, then the child tbl ref is `other_db`.`parent_tbl`.
+   # When we recreate the ref below, we use the db-qualified form, and
+   # MySQL will automatically trim the db if the tables are in the same db.
+   my $quoted_old_table = $q->quote($old_tbl->{tbl});
+   my $constraint       = qr/
+      ^\s+
+      (
+         CONSTRAINT.+?
+         REFERENCES\s(?:$quoted_old_table|$old_tbl->{name})
+         .+
+      )$
+   /xm;
+   PTDEBUG && _d('Rebuilding fk constraint matching', $constraint);
+
+   if ( $o->get('dry-run') ) {
+      print "Not rebuilding foreign key constraints because this is a dry run.\n";
+   }
+   else {
+      print "Rebuilding foreign key constraints...\n";
+   }
 
    CHILD_TABLE:
-   foreach my $child_table ( @$child_tables ) {
-      my $sql = "SHOW CREATE TABLE `$db`.`$child_table`";
-      PTDEBUG && _d($dbh, $sql);
-      $msg->($sql);
-      my $table_def;
-      eval {
-         $table_def = $dbh->selectrow_arrayref($sql)->[1];
-      };
-      if ( $EVAL_ERROR ) {
-         warn "Skipping child table $child_table: $EVAL_ERROR";
-         next CHILD_TABLE;
-      }
-
+   foreach my $child_tbl ( @$child_tables ) {
+      my $table_def = $tp->get_create_table(
+         $cxn->dbh(),
+         $child_tbl->{db},
+         $child_tbl->{tbl},
+      );
       my @constraints = $table_def =~ m/$constraint/g;
       if ( !@constraints ) {
-         warn "Child table `$child_table` does not have any foreign key "
-            . "constraints referencing $old_tbl";
+         warn "$child_tbl->{name} has no foreign key "
+            . "constraints referencing $old_tbl->{name}.\n";
          next CHILD_TABLE;
       }
 
+      my @rebuilt_constraints;
       foreach my $constraint ( @constraints ) {
-         my ($fk_symbol) = $constraint =~ m/CONSTRAINT\s+(\S+)/;
-         $sql = "ALTER TABLE `$db`.`$child_table` DROP FOREIGN KEY $fk_symbol";
-         $msg->($sql);
-         $dbh->do($sql) unless $args{print};
-
-         $constraint =~ s/REFERENCES `$old_tbl`/REFERENCES `$tbl`/o;
-         $sql = "ALTER TABLE `$db`.`$child_table` ADD $constraint"; 
-         $msg->($sql);
-         $dbh->do($sql) unless $args{print};
-      }
-   }
-
-   return;
-}
-
-sub intersection {
-   my ( $hashes ) = @_;
-   my %keys     = map { $_ => 1 } keys %{$hashes->[0]};
-   my $n_hashes = (scalar @$hashes) - 1;
-   my @isect    = grep { $keys{$_} } map { keys %{$hashes->[$_]} } 1..$n_hashes;
-   return @isect;
-}
-
-sub get_cxn {
-   my ( %args ) = @_;
-   my ($dsn, $ac, $dp, $o) = @args{qw(dsn AutoCommit DSNParser OptionParser)};
-
-   if ( $o->get('ask-pass') ) {
-      $dsn->{p} = OptionParser::prompt_noecho("Enter password: "); 
-   }
-   my $dbh = $dp->get_dbh($dp->get_cxn_params($dsn), {AutoCommit => $ac});
-
-   $dbh->do('SET SQL_LOG_BIN=0') unless $o->get('bin-log');
-   $dbh->do('SET FOREIGN_KEY_CHECKS=0') unless $o->get('foreign-key-checks');
-
-   return $dbh;
-}
-
-sub msg {
-   my ( $msg ) = @_;
-   chomp $msg;
-   print '# ', ts(time), " $msg\n" unless $quiet;
-   PTDEBUG && _d($msg);
-   return;
-}
-
-# Only for tests which may not call main().
-sub __set_quiet {
-   $quiet = $_[0];
+         PTDEBUG && _d('Rebuilding fk constraint:', $constraint);
+
+         # Remove trailing commas in case there are multiple constraints on the
+         # table.
+         $constraint =~ s/,$//;
+
+         # Find the constraint name. It will be quoted already.
+         my ($fk) = $constraint =~ m/CONSTRAINT\s+`([^`]+)`/;
+
+         # Drop the reference to the old table/renamed orig table, and add a new
+         # reference to the new table.  InnoDB will throw an error if the new
+         # constraint has the same name as the old one, so we must rename it.
+         # Example: after renaming sakila.actor to sakila.actor_old (for
+         # example), the foreign key on film_actor looks like this:
+         # CONSTRAINT `fk_film_actor_actor` FOREIGN KEY (`actor_id`) REFERENCES
+         #   `actor_old` (`actor_id`) ON UPDATE CASCADE
+         # We need it to look like this instead:
+         # CONSTRAINT `_fk_film_actor_actor` FOREIGN KEY (`actor_id`) REFERENCES
+         #   `actor` (`actor_id`) ON UPDATE CASCADE
+         # Reference the correct table name...
+         $constraint =~ s/REFERENCES[^\(]+/REFERENCES $orig_tbl->{name} /;
+         # And rename the constraint to avoid conflict
+         $constraint =~ s/CONSTRAINT `$fk`/CONSTRAINT `_$fk`/;
+
+         my $sql = "DROP FOREIGN KEY `$fk`, "
+                 . "ADD $constraint";
+         push @rebuilt_constraints, $sql;
+      }
+
+      my $sql = "ALTER TABLE $child_tbl->{name} "
+              . join(', ', @rebuilt_constraints);
+      print $sql, "\n" if $o->get('print');
+      if ( $o->get('execute') ) {
+         PTDEBUG && _d($sql);
+         $cxn->dbh()->do($sql);
+      }
+   }
+
+   if ( $o->get('execute') ) {
+      print "Rebuilt foreign key constraints OK.\n";
+   }
+
+   return;
+}
+
+sub drop_swap {
+   my ( %args ) = @_;
+   my @required_args = qw(orig_tbl new_tbl Cxn OptionParser);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($orig_tbl, $new_tbl, $cxn, $o) = @args{@required_args};
+
+   if ( $o->get('dry-run') ) {
+      print "Not drop-swapping tables because this is a dry run.\n";