Merge lp:~percona-toolkit-dev/percona-toolkit/pt-table-checksum-2.0 into lp:percona-toolkit/2.0

[Daniel Nichter]

[Daniel Nichter (daniel-nichter)]

review: Approve

=== modified file '.bzrignore'
=== modified file 'Changelog'
--- Changelog	2011-10-11 14:31:42 +0000
+++ Changelog	2011-12-30 00:34:26 +0000
@@ -1,7 +1,13 @@
 Changelog for Percona Toolkit
 
-  * Fixed bug 821717: pt-tcp-model --type=requests crashes
-
+<<<<<<< TREE
+  * Fixed bug 821717: pt-tcp-model --type=requests crashes
+
+=======
+  * Completely redesigned pt-table-checksum.
+  * Fixed bug 821717: pt-tcp-model --type=requests crashes
+
+>>>>>>> MERGE-SOURCE
 v1.0.1 released 2011-09-01
 
   * Fixed bug 819421: MasterSlave::is_replication_thread() doesn't match all

=== modified file 'bin/pt-table-checksum'
--- bin/pt-table-checksum	2011-09-01 16:00:38 +0000
+++ bin/pt-table-checksum	2011-12-30 00:34:26 +0000
@@ -9,15 +9,15 @@
 use constant MKDEBUG => $ENV{MKDEBUG} || 0;
 
 # ###########################################################################
-# TableParser package
+# DSNParser 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/TableParser.pm
-#   t/lib/TableParser.t
+#   lib/DSNParser.pm
+#   t/lib/DSNParser.t
 # See https://launchpad.net/percona-toolkit for more information.
 # ###########################################################################
 {
-package TableParser;
+package DSNParser;
 
 use strict;
 use warnings FATAL => 'all';
@@ -25,757 +25,331 @@
 use constant MKDEBUG => $ENV{MKDEBUG} || 0;
 
 use Data::Dumper;
-$Data::Dumper::Indent    = 1;
-$Data::Dumper::Sortkeys  = 1;
+$Data::Dumper::Indent    = 0;
 $Data::Dumper::Quotekeys = 0;
 
+eval {
+   require DBI;
+};
+my $have_dbi = $EVAL_ERROR ? 0 : 1;
+
 sub new {
    my ( $class, %args ) = @_;
-   my @required_args = qw(Quoter);
-   foreach my $arg ( @required_args ) {
+   foreach my $arg ( qw(opts) ) {
       die "I need a $arg argument" unless $args{$arg};
    }
-   my $self = { %args };
+   my $self = {
+      opts => {}  # h, P, u, etc.  Should come from DSN OPTIONS section in POD.
+   };
+   foreach my $opt ( @{$args{opts}} ) {
+      if ( !$opt->{key} || !$opt->{desc} ) {
+         die "Invalid DSN option: ", Dumper($opt);
+      }
+      MKDEBUG && _d('DSN option:',
+         join(', ',
+            map { "$_=" . (defined $opt->{$_} ? ($opt->{$_} || '') : 'undef') }
+               keys %$opt
+         )
+      );
+      $self->{opts}->{$opt->{key}} = {
+         dsn  => $opt->{dsn},
+         desc => $opt->{desc},
+         copy => $opt->{copy} || 0,
+      };
+   }
    return bless $self, $class;
 }
 
+sub prop {
+   my ( $self, $prop, $value ) = @_;
+   if ( @_ > 2 ) {
+      MKDEBUG && _d('Setting', $prop, 'property');
+      $self->{$prop} = $value;
+   }
+   return $self->{$prop};
+}
+
 sub parse {
-   my ( $self, $ddl, $opts ) = @_;
-   return unless $ddl;
-   if ( ref $ddl eq 'ARRAY' ) {
-      if ( lc $ddl->[0] eq 'table' ) {
-         $ddl = $ddl->[1];
-      }
-      else {
-         return {
-            engine => 'VIEW',
-         };
-      }
-   }
-
-   if ( $ddl !~ m/CREATE (?:TEMPORARY )?TABLE `/ ) {
-      die "Cannot parse table definition; is ANSI quoting "
-         . "enabled or SQL_QUOTE_SHOW_CREATE disabled?";
-   }
-
-   my ($name)     = $ddl =~ m/CREATE (?:TEMPORARY )?TABLE\s+(`.+?`)/;
-   (undef, $name) = $self->{Quoter}->split_unquote($name) if $name;
-
-   $ddl =~ s/(`[^`]+`)/\L$1/g;
-
-   my $engine = $self->get_engine($ddl);
-
-   my @defs   = $ddl =~ m/^(\s+`.*?),?$/gm;
-   my @cols   = map { $_ =~ m/`([^`]+)`/ } @defs;
-   MKDEBUG && _d('Table cols:', join(', ', map { "`$_`" } @cols));
-
-   my %def_for;
-   @def_for{@cols} = @defs;
-
-   my (@nums, @null);
-   my (%type_for, %is_nullable, %is_numeric, %is_autoinc);
-   foreach my $col ( @cols ) {
-      my $def = $def_for{$col};
-      my ( $type ) = $def =~ m/`[^`]+`\s([a-z]+)/;
-      die "Can't determine column type for $def" unless $type;
-      $type_for{$col} = $type;
-      if ( $type =~ m/(?:(?:tiny|big|medium|small)?int|float|double|decimal|year)/ ) {
-         push @nums, $col;
-         $is_numeric{$col} = 1;
-      }
-      if ( $def !~ m/NOT NULL/ ) {
-         push @null, $col;
-         $is_nullable{$col} = 1;
-      }
-      $is_autoinc{$col} = $def =~ m/AUTO_INCREMENT/i ? 1 : 0;
-   }
-
-   my ($keys, $clustered_key) = $self->get_keys($ddl, $opts, \%is_nullable);
-
-   my ($charset) = $ddl =~ m/DEFAULT CHARSET=(\w+)/;
-
-   return {
-      name           => $name,
-      cols           => \@cols,
-      col_posn       => { map { $cols[$_] => $_ } 0..$#cols },
-      is_col         => { map { $_ => 1 } @cols },
-      null_cols      => \@null,
-      is_nullable    => \%is_nullable,
-      is_autoinc     => \%is_autoinc,
-      clustered_key  => $clustered_key,
-      keys           => $keys,
-      defs           => \%def_for,
-      numeric_cols   => \@nums,
-      is_numeric     => \%is_numeric,
-      engine         => $engine,
-      type_for       => \%type_for,
-      charset        => $charset,
-   };
-}
-
-sub sort_indexes {
-   my ( $self, $tbl ) = @_;
-
-   my @indexes
-      = sort {
-         (($a ne 'PRIMARY') <=> ($b ne 'PRIMARY'))
-         || ( !$tbl->{keys}->{$a}->{is_unique} <=> !$tbl->{keys}->{$b}->{is_unique} )
-         || ( $tbl->{keys}->{$a}->{is_nullable} <=> $tbl->{keys}->{$b}->{is_nullable} )
-         || ( scalar(@{$tbl->{keys}->{$a}->{cols}}) <=> scalar(@{$tbl->{keys}->{$b}->{cols}}) )
-      }
-      grep {
-         $tbl->{keys}->{$_}->{type} eq 'BTREE'
-      }
-      sort keys %{$tbl->{keys}};
-
-   MKDEBUG && _d('Indexes sorted best-first:', join(', ', @indexes));
-   return @indexes;
-}
-
-sub find_best_index {
-   my ( $self, $tbl, $index ) = @_;
-   my $best;
-   if ( $index ) {
-      ($best) = grep { uc $_ eq uc $index } keys %{$tbl->{keys}};
-   }
-   if ( !$best ) {
-      if ( $index ) {
-         die "Index '$index' does not exist in table";
-      }
-      else {
-         ($best) = $self->sort_indexes($tbl);
-      }
-   }
-   MKDEBUG && _d('Best index found is', $best);
-   return $best;
-}
-
-sub find_possible_keys {
-   my ( $self, $dbh, $database, $table, $quoter, $where ) = @_;
-   return () unless $where;
-   my $sql = 'EXPLAIN SELECT * FROM ' . $quoter->quote($database, $table)
-      . ' WHERE ' . $where;
-   MKDEBUG && _d($sql);
-   my $expl = $dbh->selectrow_hashref($sql);
-   $expl = { map { lc($_) => $expl->{$_} } keys %$expl };
-   if ( $expl->{possible_keys} ) {
-      MKDEBUG && _d('possible_keys =', $expl->{possible_keys});
-      my @candidates = split(',', $expl->{possible_keys});
-      my %possible   = map { $_ => 1 } @candidates;
-      if ( $expl->{key} ) {
-         MKDEBUG && _d('MySQL chose', $expl->{key});
-         unshift @candidates, grep { $possible{$_} } split(',', $expl->{key});
-         MKDEBUG && _d('Before deduping:', join(', ', @candidates));
-         my %seen;
-         @candidates = grep { !$seen{$_}++ } @candidates;
-      }
-      MKDEBUG && _d('Final list:', join(', ', @candidates));
-      return @candidates;
-   }
-   else {
-      MKDEBUG && _d('No keys in possible_keys');
-      return ();
-   }
-}
-
-sub check_table {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db tbl);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $db, $tbl) = @args{@required_args};
-   my $q      = $self->{Quoter};
-   my $db_tbl = $q->quote($db, $tbl);
-   MKDEBUG && _d('Checking', $db_tbl);
-
-   my $sql = "SHOW TABLES FROM " . $q->quote($db)
-           . ' LIKE ' . $q->literal_like($tbl);
-   MKDEBUG && _d($sql);
-   my $row;
-   eval {
-      $row = $dbh->selectrow_arrayref($sql);
-   };
-   if ( $EVAL_ERROR ) {
-      MKDEBUG && _d($EVAL_ERROR);
-      return 0;
-   }
-   if ( !$row->[0] || $row->[0] ne $tbl ) {
-      MKDEBUG && _d('Table does not exist');
-      return 0;
-   }
-
-   MKDEBUG && _d('Table exists; no privs to check');
-   return 1 unless $args{all_privs};
-
-   $sql = "SHOW FULL COLUMNS FROM $db_tbl";
-   MKDEBUG && _d($sql);
-   eval {
-      $row = $dbh->selectrow_hashref($sql);
-   };
-   if ( $EVAL_ERROR ) {
-      MKDEBUG && _d($EVAL_ERROR);
-      return 0;
-   }
-   if ( !scalar keys %$row ) {
-      MKDEBUG && _d('Table has no columns:', Dumper($row));
-      return 0;
-   }
-   my $privs = $row->{privileges} || $row->{Privileges};
-
-   $sql = "DELETE FROM $db_tbl LIMIT 0";
-   MKDEBUG && _d($sql);
-   eval {
-      $dbh->do($sql);
-   };
-   my $can_delete = $EVAL_ERROR ? 0 : 1;
-
-   MKDEBUG && _d('User privs on', $db_tbl, ':', $privs,
-      ($can_delete ? 'delete' : ''));
-
-   if ( !($privs =~ m/select/ && $privs =~ m/insert/ && $privs =~ m/update/
-          && $can_delete) ) {
-      MKDEBUG && _d('User does not have all privs');
-      return 0;
-   }
-
-   MKDEBUG && _d('User has all privs');
-   return 1;
-}
-
-sub get_engine {
-   my ( $self, $ddl, $opts ) = @_;
-   my ( $engine ) = $ddl =~ m/\).*?(?:ENGINE|TYPE)=(\w+)/;
-   MKDEBUG && _d('Storage engine:', $engine);
-   return $engine || undef;
-}
-
-sub get_keys {
-   my ( $self, $ddl, $opts, $is_nullable ) = @_;
-   my $engine        = $self->get_engine($ddl);
-   my $keys          = {};
-   my $clustered_key = undef;
-
-   KEY:
-   foreach my $key ( $ddl =~ m/^  ((?:[A-Z]+ )?KEY .*)$/gm ) {
-
-      next KEY if $key =~ m/FOREIGN/;
-
-      my $key_ddl = $key;
-      MKDEBUG && _d('Parsed key:', $key_ddl);
-
-      if ( $engine !~ m/MEMORY|HEAP/ ) {
-         $key =~ s/USING HASH/USING BTREE/;
-      }
-
-      my ( $type, $cols ) = $key =~ m/(?:USING (\w+))? \((.+)\)/;
-      my ( $special ) = $key =~ m/(FULLTEXT|SPATIAL)/;
-      $type = $type || $special || 'BTREE';
-      if ( $opts->{mysql_version} && $opts->{mysql_version} lt '004001000'
-         && $engine =~ m/HEAP|MEMORY/i )
+   my ( $self, $dsn, $prev, $defaults ) = @_;
+   if ( !$dsn ) {
+      MKDEBUG && _d('No DSN to parse');
+      return;
+   }
+   MKDEBUG && _d('Parsing', $dsn);
+   $prev     ||= {};
+   $defaults ||= {};
+   my %given_props;
+   my %final_props;
+   my $opts = $self->{opts};
+
+   foreach my $dsn_part ( split(/,/, $dsn) ) {
+      if ( my ($prop_key, $prop_val) = $dsn_part =~  m/^(.)=(.*)$/ ) {
+         $given_props{$prop_key} = $prop_val;
+      }
+      else {
+         MKDEBUG && _d('Interpreting', $dsn_part, 'as h=', $dsn_part);
+         $given_props{h} = $dsn_part;
+      }
+   }
+
+   foreach my $key ( keys %$opts ) {
+      MKDEBUG && _d('Finding value for', $key);
+      $final_props{$key} = $given_props{$key};
+      if (   !defined $final_props{$key}
+           && defined $prev->{$key} && $opts->{$key}->{copy} )
       {
-         $type = 'HASH'; # MySQL pre-4.1 supports only HASH indexes on HEAP
-      }
-
-      my ($name) = $key =~ m/(PRIMARY|`[^`]*`)/;
-      my $unique = $key =~ m/PRIMARY|UNIQUE/ ? 1 : 0;
-      my @cols;
-      my @col_prefixes;
-      foreach my $col_def ( $cols =~ m/`[^`]+`(?:\(\d+\))?/g ) {
-         my ($name, $prefix) = $col_def =~ m/`([^`]+)`(?:\((\d+)\))?/;
-         push @cols, $name;
-         push @col_prefixes, $prefix;
-      }
-      $name =~ s/`//g;
-
-      MKDEBUG && _d( $name, 'key cols:', join(', ', map { "`$_`" } @cols));
-
-      $keys->{$name} = {
-         name         => $name,
-         type         => $type,
-         colnames     => $cols,
-         cols         => \@cols,
-         col_prefixes => \@col_prefixes,
-         is_unique    => $unique,
-         is_nullable  => scalar(grep { $is_nullable->{$_} } @cols),
-         is_col       => { map { $_ => 1 } @cols },
-         ddl          => $key_ddl,
-      };
-
-      if ( $engine =~ m/InnoDB/i && !$clustered_key ) {
-         my $this_key = $keys->{$name};
-         if ( $this_key->{name} eq 'PRIMARY' ) {
-            $clustered_key = 'PRIMARY';
-         }
-         elsif ( $this_key->{is_unique} && !$this_key->{is_nullable} ) {
-            $clustered_key = $this_key->{name};
-         }
-         MKDEBUG && $clustered_key && _d('This key is the clustered key');
-      }
-   }
-
-   return $keys, $clustered_key;
-}
-
-sub get_fks {
-   my ( $self, $ddl, $opts ) = @_;
-   my $q   = $self->{Quoter};
-   my $fks = {};
-
-   foreach my $fk (
-      $ddl =~ m/CONSTRAINT .* FOREIGN KEY .* REFERENCES [^\)]*\)/mg )
-   {
-      my ( $name ) = $fk =~ m/CONSTRAINT `(.*?)`/;
-      my ( $cols ) = $fk =~ m/FOREIGN KEY \(([^\)]+)\)/;
-      my ( $parent, $parent_cols ) = $fk =~ m/REFERENCES (\S+) \(([^\)]+)\)/;
-
-      my ($db, $tbl) = $q->split_unquote($parent, $opts->{database});
-      my %parent_tbl = (tbl => $tbl);
-      $parent_tbl{db} = $db if $db;
-
-      if ( $parent !~ m/\./ && $opts->{database} ) {
-         $parent = $q->quote($opts->{database}) . ".$parent";
-      }
-
-      $fks->{$name} = {
-         name           => $name,
-         colnames       => $cols,
-         cols           => [ map { s/[ `]+//g; $_; } split(',', $cols) ],
-         parent_tbl     => \%parent_tbl,
-         parent_tblname => $parent,
-         parent_cols    => [ map { s/[ `]+//g; $_; } split(',', $parent_cols) ],
-         parent_colnames=> $parent_cols,
-         ddl            => $fk,
-      };
-   }
-
-   return $fks;
-}
-
-sub remove_auto_increment {
-   my ( $self, $ddl ) = @_;
-   $ddl =~ s/(^\).*?) AUTO_INCREMENT=\d+\b/$1/m;
-   return $ddl;
-}
-
-sub remove_secondary_indexes {
-   my ( $self, $ddl ) = @_;
-   my $sec_indexes_ddl;
-   my $tbl_struct = $self->parse($ddl);
-
-   if ( ($tbl_struct->{engine} || '') =~ m/InnoDB/i ) {
-      my $clustered_key = $tbl_struct->{clustered_key};
-      $clustered_key  ||= '';
-
-      my @sec_indexes   = map {
-         my $key_def = $_->{ddl};
-         $key_def =~ s/([\(\)])/\\$1/g;
-         $ddl =~ s/\s+$key_def//i;
-
-         my $key_ddl = "ADD $_->{ddl}";
-         $key_ddl   .= ',' unless $key_ddl =~ m/,$/;
-         $key_ddl;
-      }
-      grep { $_->{name} ne $clustered_key }
-      values %{$tbl_struct->{keys}};
-      MKDEBUG && _d('Secondary indexes:', Dumper(\@sec_indexes));
-
-      if ( @sec_indexes ) {
-         $sec_indexes_ddl = join(' ', @sec_indexes);
-         $sec_indexes_ddl =~ s/,$//;
-      }
-
-      $ddl =~ s/,(\n\) )/$1/s;
+         $final_props{$key} = $prev->{$key};
+         MKDEBUG && _d('Copying value for', $key, 'from previous DSN');
+      }
+      if ( !defined $final_props{$key} ) {
+         $final_props{$key} = $defaults->{$key};
+         MKDEBUG && _d('Copying value for', $key, 'from defaults');
+      }
+   }
+
+   foreach my $key ( keys %given_props ) {
+      die "Unknown DSN option '$key' in '$dsn'.  For more details, "
+            . "please use the --help option, or try 'perldoc $PROGRAM_NAME' "
+            . "for complete documentation."
+         unless exists $opts->{$key};
+   }
+   if ( (my $required = $self->prop('required')) ) {
+      foreach my $key ( keys %$required ) {
+         die "Missing required DSN option '$key' in '$dsn'.  For more details, "
+               . "please use the --help option, or try 'perldoc $PROGRAM_NAME' "
+               . "for complete documentation."
+            unless $final_props{$key};
+      }
+   }
+
+   return \%final_props;
+}
+
+sub parse_options {
+   my ( $self, $o ) = @_;
+   die 'I need an OptionParser object' unless ref $o eq 'OptionParser';
+   my $dsn_string
+      = join(',',
+          map  { "$_=".$o->get($_); }
+          grep { $o->has($_) && $o->get($_) }
+          keys %{$self->{opts}}
+        );
+   MKDEBUG && _d('DSN string made from options:', $dsn_string);
+   return $self->parse($dsn_string);
+}
+
+sub as_string {
+   my ( $self, $dsn, $props ) = @_;
+   return $dsn unless ref $dsn;
+   my @keys = $props ? @$props : sort keys %$dsn;
+   return join(',',
+      map  { "$_=" . ($_ eq 'p' ? '...' : $dsn->{$_}) }
+      grep {
+         exists $self->{opts}->{$_}
+         && exists $dsn->{$_}
+         && defined $dsn->{$_}
+      } @keys);
+}
+
+sub usage {
+   my ( $self ) = @_;
+   my $usage
+      = "DSN syntax is key=value[,key=value...]  Allowable DSN keys:\n\n"
+      . "  KEY  COPY  MEANING\n"
+      . "  ===  ====  =============================================\n";
+   my %opts = %{$self->{opts}};
+   foreach my $key ( sort keys %opts ) {
+      $usage .= "  $key    "
+             .  ($opts{$key}->{copy} ? 'yes   ' : 'no    ')
+             .  ($opts{$key}->{desc} || '[No description]')
+             . "\n";
+   }
+   $usage .= "\n  If the DSN is a bareword, the word is treated as the 'h' key.\n";
+   return $usage;
+}
+
+sub get_cxn_params {
+   my ( $self, $info ) = @_;
+   my $dsn;
+   my %opts = %{$self->{opts}};
+   my $driver = $self->prop('dbidriver') || '';
+   if ( $driver eq 'Pg' ) {
+      $dsn = 'DBI:Pg:dbname=' . ( $info->{D} || '' ) . ';'
+         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
+                     grep { defined $info->{$_} }
+                     qw(h P));
    }
    else {
-      MKDEBUG && _d('Not removing secondary indexes from',
-         $tbl_struct->{engine}, 'table');
-   }
-
-   return $ddl, $sec_indexes_ddl, $tbl_struct;
-}
-
-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 TableParser package
-# ###########################################################################
-
-# ###########################################################################
-# TableChecksum 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/TableChecksum.pm
-#   t/lib/TableChecksum.t
-# See https://launchpad.net/percona-toolkit for more information.
-# ###########################################################################
-{
-package TableChecksum;
-
-use strict;
-use warnings FATAL => 'all';
-use English qw(-no_match_vars);
-use constant MKDEBUG => $ENV{MKDEBUG} || 0;
-
-use List::Util qw(max);
-
-our %ALGOS = (
-   CHECKSUM => { pref => 0, hash => 0 },
-   BIT_XOR  => { pref => 2, hash => 1 },
-   ACCUM    => { pref => 3, hash => 1 },
-);
-
-sub new {
-   my ( $class, %args ) = @_;
-   foreach my $arg ( qw(Quoter VersionParser) ) {
-      die "I need a $arg argument" unless defined $args{$arg};
-   }
-   my $self = { %args };
-   return bless $self, $class;
-}
-
-sub crc32 {
-   my ( $self, $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 get_crc_wid {
-   my ( $self, $dbh, $func ) = @_;
-   my $crc_wid = 16;
-   if ( uc $func ne 'FNV_64' && uc $func ne 'FNV1A_64' ) {
-      eval {
-         my ($val) = $dbh->selectrow_array("SELECT $func('a')");
-         $crc_wid = max(16, length($val));
-      };
-   }
-   return $crc_wid;
-}
-
-sub get_crc_type {
-   my ( $self, $dbh, $func ) = @_;
-   my $type   = '';
-   my $length = 0;
-   my $sql    = "SELECT $func('a')";
-   my $sth    = $dbh->prepare($sql);
-   eval {
-      $sth->execute();
-      $type   = $sth->{mysql_type_name}->[0];
-      $length = $sth->{mysql_length}->[0];
-      MKDEBUG && _d($sql, $type, $length);
-      if ( $type eq 'bigint' && $length < 20 ) {
-         $type = 'int';
-      }
+      $dsn = 'DBI:mysql:' . ( $info->{D} || '' ) . ';'
+         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
+                     grep { defined $info->{$_} }
+                     qw(F h P S A))
+         . ';mysql_read_default_group=client';
+   }
+   MKDEBUG && _d($dsn);
+   return ($dsn, $info->{u}, $info->{p});
+}
+
+sub fill_in_dsn {
+   my ( $self, $dbh, $dsn ) = @_;
+   my $vars = $dbh->selectall_hashref('SHOW VARIABLES', 'Variable_name');
+   my ($user, $db) = $dbh->selectrow_array('SELECT USER(), DATABASE()');
+   $user =~ s/@.*//;
+   $dsn->{h} ||= $vars->{hostname}->{Value};
+   $dsn->{S} ||= $vars->{'socket'}->{Value};
+   $dsn->{P} ||= $vars->{port}->{Value};
+   $dsn->{u} ||= $user;
+   $dsn->{D} ||= $db;
+}
+
+sub get_dbh {
+   my ( $self, $cxn_string, $user, $pass, $opts ) = @_;
+   $opts ||= {};
+   my $defaults = {
+      AutoCommit         => 0,
+      RaiseError         => 1,
+      PrintError         => 0,
+      ShowErrorStatement => 1,
+      mysql_enable_utf8 => ($cxn_string =~ m/charset=utf8/i ? 1 : 0),
    };
-   $sth->finish;
-   MKDEBUG && _d('crc_type:', $type, 'length:', $length);
-   return ($type, $length);
-}
-
-sub best_algorithm {
-   my ( $self, %args ) = @_;
-   my ( $alg, $dbh ) = @args{ qw(algorithm dbh) };
-   my $vp = $self->{VersionParser};
-   my @choices = sort { $ALGOS{$a}->{pref} <=> $ALGOS{$b}->{pref} } keys %ALGOS;
-   die "Invalid checksum algorithm $alg"
-      if $alg && !$ALGOS{$alg};
-
-   if (
-      $args{where} || $args{chunk}        # CHECKSUM does whole table
-      || $args{replicate}                 # CHECKSUM can't do INSERT.. SELECT
-      || !$vp->version_ge($dbh, '4.1.1')) # CHECKSUM doesn't exist
-   {
-      MKDEBUG && _d('Cannot use CHECKSUM algorithm');
-      @choices = grep { $_ ne 'CHECKSUM' } @choices;
-   }
-
-   if ( !$vp->version_ge($dbh, '4.1.1') ) {
-      MKDEBUG && _d('Cannot use BIT_XOR algorithm because MySQL < 4.1.1');
-      @choices = grep { $_ ne 'BIT_XOR' } @choices;
-   }
-
-   if ( $alg && grep { $_ eq $alg } @choices ) {
-      MKDEBUG && _d('User requested', $alg, 'algorithm');
-      return $alg;
-   }
-
-   if ( $args{count} && grep { $_ ne 'CHECKSUM' } @choices ) {
-      MKDEBUG && _d('Not using CHECKSUM algorithm because COUNT desired');
-      @choices = grep { $_ ne 'CHECKSUM' } @choices;
-   }
-
-   MKDEBUG && _d('Algorithms, in order:', @choices);
-   return $choices[0];
-}
-
-sub is_hash_algorithm {
-   my ( $self, $algorithm ) = @_;
-   return $ALGOS{$algorithm} && $ALGOS{$algorithm}->{hash};
-}
-
-sub choose_hash_func {
-   my ( $self, %args ) = @_;
-   my @funcs = qw(CRC32 FNV1A_64 FNV_64 MD5 SHA1);
-   if ( $args{function} ) {
-      unshift @funcs, $args{function};
-   }
-   my ($result, $error);
-   do {
-      my $func;
+   @{$defaults}{ keys %$opts } = values %$opts;
+
+   if ( $opts->{mysql_use_result} ) {
+      $defaults->{mysql_use_result} = 1;
+   }
+
+   if ( !$have_dbi ) {
+      die "Cannot connect to MySQL because the Perl DBI module is not "
+         . "installed or not found.  Run 'perl -MDBI' to see the directories "
+         . "that Perl searches for DBI.  If DBI is not installed, try:\n"
+         . "  Debian/Ubuntu  apt-get install libdbi-perl\n"
+         . "  RHEL/CentOS    yum install perl-DBI\n"
+         . "  OpenSolaris    pgk install pkg:/SUNWpmdbi\n";
+
+   }
+
+   my $dbh;
+   my $tries = 2;
+   while ( !$dbh && $tries-- ) {
+      MKDEBUG && _d($cxn_string, ' ', $user, ' ', $pass, 
+         join(', ', map { "$_=>$defaults->{$_}" } keys %$defaults ));
+
       eval {
-         $func = shift(@funcs);
-         my $sql = "SELECT $func('test-string')";
-         MKDEBUG && _d($sql);
-         $args{dbh}->do($sql);
-         $result = $func;
+         $dbh = DBI->connect($cxn_string, $user, $pass, $defaults);
+
+         if ( $cxn_string =~ m/mysql/i ) {
+            my $sql;
+
+            $sql = 'SELECT @@SQL_MODE';
+            MKDEBUG && _d($dbh, $sql);
+            my ($sql_mode) = $dbh->selectrow_array($sql);
+
+            $sql = 'SET @@SQL_QUOTE_SHOW_CREATE = 1'
+                 . '/*!40101, @@SQL_MODE=\'NO_AUTO_VALUE_ON_ZERO'
+                 . ($sql_mode ? ",$sql_mode" : '')
+                 . '\'*/';
+            MKDEBUG && _d($dbh, $sql);
+            $dbh->do($sql);
+
+            if ( my ($charset) = $cxn_string =~ m/charset=(\w+)/ ) {
+               $sql = "/*!40101 SET NAMES $charset*/";
+               MKDEBUG && _d($dbh, ':', $sql);
+               $dbh->do($sql);
+               MKDEBUG && _d('Enabling charset for STDOUT');
+               if ( $charset eq 'utf8' ) {
+                  binmode(STDOUT, ':utf8')
+                     or die "Can't binmode(STDOUT, ':utf8'): $OS_ERROR";
+               }
+               else {
+                  binmode(STDOUT) or die "Can't binmode(STDOUT): $OS_ERROR";
+               }
+            }
+
+            if ( $self->prop('set-vars') ) {
+               $sql = "SET " . $self->prop('set-vars');
+               MKDEBUG && _d($dbh, ':', $sql);
+               $dbh->do($sql);
+            }
+         }
       };
-      if ( $EVAL_ERROR && $EVAL_ERROR =~ m/failed: (.*?) at \S+ line/ ) {
-         $error .= qq{$func cannot be used because "$1"\n};
-         MKDEBUG && _d($func, 'cannot be used because', $1);
-      }
-   } while ( @funcs && !$result );
-
-   die $error unless $result;
-   MKDEBUG && _d('Chosen hash func:', $result);
-   return $result;
-}
-
-sub optimize_xor {
-   my ( $self, %args ) = @_;
-   my ($dbh, $func) = @args{qw(dbh function)};
-
-   die "$func never needs the BIT_XOR optimization"
-      if $func =~ m/^(?:FNV1A_64|FNV_64|CRC32)$/i;
-
-   my $opt_slice = 0;
-   my $unsliced  = uc $dbh->selectall_arrayref("SELECT $func('a')")->[0]->[0];
-   my $sliced    = '';
-   my $start     = 1;
-   my $crc_wid   = length($unsliced) < 16 ? 16 : length($unsliced);
-
-   do { # Try different positions till sliced result equals non-sliced.
-      MKDEBUG && _d('Trying slice', $opt_slice);
-      $dbh->do('SET @crc := "", @cnt := 0');
-      my $slices = $self->make_xor_slices(
-         query     => "\@crc := $func('a')",
-         crc_wid   => $crc_wid,
-         opt_slice => $opt_slice,
-      );
-
-      my $sql = "SELECT CONCAT($slices) AS TEST FROM (SELECT NULL) AS x";
-      $sliced = ($dbh->selectrow_array($sql))[0];
-      if ( $sliced ne $unsliced ) {
-         MKDEBUG && _d('Slice', $opt_slice, 'does not work');
-         $start += 16;
-         ++$opt_slice;
-      }
-   } while ( $start < $crc_wid && $sliced ne $unsliced );
-
-   if ( $sliced eq $unsliced ) {
-      MKDEBUG && _d('Slice', $opt_slice, 'works');
-      return $opt_slice;
-   }
-   else {
-      MKDEBUG && _d('No slice works');
-      return undef;
-   }
-}
-
-sub make_xor_slices {
-   my ( $self, %args ) = @_;
-   foreach my $arg ( qw(query crc_wid) ) {
-      die "I need a $arg argument" unless defined $args{$arg};
-   }
-   my ( $query, $crc_wid, $opt_slice ) = @args{qw(query crc_wid opt_slice)};
-
-   my @slices;
-   for ( my $start = 1; $start <= $crc_wid; $start += 16 ) {
-      my $len = $crc_wid - $start + 1;
-      if ( $len > 16 ) {
-         $len = 16;
-      }
-      push @slices,
-         "LPAD(CONV(BIT_XOR("
-         . "CAST(CONV(SUBSTRING(\@crc, $start, $len), 16, 10) AS UNSIGNED))"
-         . ", 10, 16), $len, '0')";
-   }
-
-   if ( defined $opt_slice && $opt_slice < @slices ) {
-      $slices[$opt_slice] =~ s/\@crc/\@crc := $query/;
-   }
-   else {
-      map { s/\@crc/$query/ } @slices;
-   }
-
-   return join(', ', @slices);
-}
-
-sub make_row_checksum {
-   my ( $self, %args ) = @_;
-   my ( $tbl_struct, $func ) = @args{ qw(tbl_struct function) };
-   my $q = $self->{Quoter};
-
-   my $sep = $args{sep} || '#';
-   $sep =~ s/'//g;
-   $sep ||= '#';
-
-   my $ignorecols = $args{ignorecols} || {};
-
-   my %cols = map { lc($_) => 1 }
-              grep { !exists $ignorecols->{$_} }
-              ($args{cols} ? @{$args{cols}} : @{$tbl_struct->{cols}});
-   my %seen;
-   my @cols =
-      map {
-         my $type = $tbl_struct->{type_for}->{$_};
-         my $result = $q->quote($_);
-         if ( $type eq 'timestamp' ) {
-            $result .= ' + 0';
-         }
-         elsif ( $args{float_precision} && $type =~ m/float|double/ ) {
-            $result = "ROUND($result, $args{float_precision})";
-         }
-         elsif ( $args{trim} && $type =~ m/varchar/ ) {
-            $result = "TRIM($result)";
-         }
-         $result;
-      }
-      grep {
-         $cols{$_} && !$seen{$_}++
-      }
-      @{$tbl_struct->{cols}};
-
-   my $query;
-   if ( !$args{no_cols} ) {
-      $query = join(', ',
-                  map { 
-                     my $col = $_;
-                     if ( $col =~ m/\+ 0/ ) {
-                        my ($real_col) = /^(\S+)/;
-                        $col .= " AS $real_col";
-                     }
-                     elsif ( $col =~ m/TRIM/ ) {
-                        my ($real_col) = m/TRIM\(([^\)]+)\)/;
-                        $col .= " AS $real_col";
-                     }
-                     $col;
-                  } @cols)
-             . ', ';
-   }
-
-   if ( uc $func ne 'FNV_64' && uc $func ne 'FNV1A_64' ) {
-      my @nulls = grep { $cols{$_} } @{$tbl_struct->{null_cols}};
-      if ( @nulls ) {
-         my $bitmap = "CONCAT("
-            . join(', ', map { 'ISNULL(' . $q->quote($_) . ')' } @nulls)
-            . ")";
-         push @cols, $bitmap;
-      }
-
-      $query .= @cols > 1
-              ? "$func(CONCAT_WS('$sep', " . join(', ', @cols) . '))'
-              : "$func($cols[0])";
-   }
-   else {
-      my $fnv_func = uc $func;
-      $query .= "$fnv_func(" . join(', ', @cols) . ')';
-   }
-
-   return $query;
-}
-
-sub make_checksum_query {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(db tbl tbl_struct algorithm crc_wid crc_type);
-   foreach my $arg( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ( $db, $tbl, $tbl_struct, $algorithm,
-        $crc_wid, $crc_type) = @args{@required_args};
-   my $func = $args{function};
-   my $q = $self->{Quoter};
-   my $result;
-
-   die "Invalid or missing checksum algorithm"
-      unless $algorithm && $ALGOS{$algorithm};
-
-   if ( $algorithm eq 'CHECKSUM' ) {
-      return "CHECKSUM TABLE " . $q->quote($db, $tbl);
-   }
-
-   my $expr = $self->make_row_checksum(%args, no_cols=>1);
-
-   if ( $algorithm eq 'BIT_XOR' ) {
-      if ( $crc_type =~ m/int$/ ) {
-         $result = "COALESCE(LOWER(CONV(BIT_XOR(CAST($expr AS UNSIGNED)), 10, 16)), 0) AS crc ";
-      }
-      else {
-         my $slices = $self->make_xor_slices( query => $expr, %args );
-         $result = "COALESCE(LOWER(CONCAT($slices)), 0) AS crc ";
-      }
-   }
-   else {
-      if ( $crc_type =~ m/int$/ ) {
-         $result = "COALESCE(RIGHT(MAX("
-            . "\@crc := CONCAT(LPAD(\@cnt := \@cnt + 1, 16, '0'), "
-            . "CONV(CAST($func(CONCAT(\@crc, $expr)) AS UNSIGNED), 10, 16))"
-            . "), $crc_wid), 0) AS crc ";
-      }
-      else {
-         $result = "COALESCE(RIGHT(MAX("
-            . "\@crc := CONCAT(LPAD(\@cnt := \@cnt + 1, 16, '0'), "
-            . "$func(CONCAT(\@crc, $expr)))"
-            . "), $crc_wid), 0) AS crc ";
-      }
-   }
-   if ( $args{replicate} ) {
-      $result = "REPLACE /*PROGRESS_COMMENT*/ INTO $args{replicate} "
-         . "(db, tbl, chunk, boundaries, this_cnt, this_crc) "
-         . "SELECT ?, ?, /*CHUNK_NUM*/ ?, COUNT(*) AS cnt, $result";
-   }
-   else {
-      $result = "SELECT "
-         . ($args{buffer} ? 'SQL_BUFFER_RESULT ' : '')
-         . "/*PROGRESS_COMMENT*//*CHUNK_NUM*/ COUNT(*) AS cnt, $result";
-   }
-   return $result . "FROM /*DB_TBL*//*INDEX_HINT*//*WHERE*/";
-}
-
-sub find_replication_differences {
-   my ( $self, $dbh, $table ) = @_;
-
-   (my $sql = <<"   EOF") =~ s/\s+/ /gm;
-      SELECT db, tbl, chunk, boundaries,
-         COALESCE(this_cnt-master_cnt, 0) AS cnt_diff,
-         COALESCE(
-            this_crc <> master_crc OR ISNULL(master_crc) <> ISNULL(this_crc),
-            0
-         ) AS crc_diff,
-         this_cnt, master_cnt, this_crc, master_crc
-      FROM $table
-      WHERE master_cnt <> this_cnt OR master_crc <> this_crc
-      OR ISNULL(master_crc) <> ISNULL(this_crc)
-   EOF
-
-   MKDEBUG && _d($sql);
-   my $diffs = $dbh->selectall_arrayref($sql, { Slice => {} });
-   return @$diffs;
+      if ( !$dbh && $EVAL_ERROR ) {
+         MKDEBUG && _d($EVAL_ERROR);
+         if ( $EVAL_ERROR =~ m/not a compiled character set|character set utf8/ ) {
+            MKDEBUG && _d('Going to try again without utf8 support');
+            delete $defaults->{mysql_enable_utf8};
+         }
+         elsif ( $EVAL_ERROR =~ m/locate DBD\/mysql/i ) {
+            die "Cannot connect to MySQL because the Perl DBD::mysql module is "
+               . "not installed or not found.  Run 'perl -MDBD::mysql' to see "
+               . "the directories that Perl searches for DBD::mysql.  If "
+               . "DBD::mysql is not installed, try:\n"
+               . "  Debian/Ubuntu  apt-get install libdbd-mysql-perl\n"
+               . "  RHEL/CentOS    yum install perl-DBD-MySQL\n"
+               . "  OpenSolaris    pgk install pkg:/SUNWapu13dbd-mysql\n";
+         }
+         if ( !$tries ) {
+            die $EVAL_ERROR;
+         }
+      }
+   }
+
+   MKDEBUG && _d('DBH info: ',
+      $dbh,
+      Dumper($dbh->selectrow_hashref(
+         'SELECT DATABASE(), CONNECTION_ID(), VERSION()/*!50038 , @@hostname*/')),
+      'Connection info:',      $dbh->{mysql_hostinfo},
+      'Character set info:',   Dumper($dbh->selectall_arrayref(
+                     'SHOW VARIABLES LIKE "character_set%"', { Slice => {}})),
+      '$DBD::mysql::VERSION:', $DBD::mysql::VERSION,
+      '$DBI::VERSION:',        $DBI::VERSION,
+   );
+
+   return $dbh;
+}
+
+sub get_hostname {
+   my ( $self, $dbh ) = @_;
+   if ( my ($host) = ($dbh->{mysql_hostinfo} || '') =~ m/^(\w+) via/ ) {
+      return $host;
+   }
+   my ( $hostname, $one ) = $dbh->selectrow_array(
+      'SELECT /*!50038 @@hostname, */ 1');
+   return $hostname;
+}
+
+sub disconnect {
+   my ( $self, $dbh ) = @_;
+   MKDEBUG && $self->print_active_handles($dbh);
+   $dbh->disconnect;
+}
+
+sub print_active_handles {
+   my ( $self, $thing, $level ) = @_;
+   $level ||= 0;
+   printf("# Active %sh: %s %s %s\n", ($thing->{Type} || 'undef'), "\t" x $level,
+      $thing, (($thing->{Type} || '') eq 'st' ? $thing->{Statement} || '' : ''))
+      or die "Cannot print: $OS_ERROR";
+   foreach my $handle ( grep {defined} @{ $thing->{ChildHandles} } ) {
+      $self->print_active_handles( $handle, $level + 1 );
+   }
+}
+
+sub copy {
+   my ( $self, $dsn_1, $dsn_2, %args ) = @_;
+   die 'I need a dsn_1 argument' unless $dsn_1;
+   die 'I need a dsn_2 argument' unless $dsn_2;
+   my %new_dsn = map {
+      my $key = $_;
+      my $val;
+      if ( $args{overwrite} ) {
+         $val = defined $dsn_1->{$key} ? $dsn_1->{$key} : $dsn_2->{$key};
+      }
+      else {
+         $val = defined $dsn_2->{$key} ? $dsn_2->{$key} : $dsn_1->{$key};
+      }
+      $key => $val;
+   } keys %{$self->{opts}};
+   return \%new_dsn;
 }
 
 sub _d {
@@ -789,7 +363,7 @@
 1;
 }
 # ###########################################################################
-# End TableChecksum package
+# End DSNParser package
 # ###########################################################################
 
 # ###########################################################################
@@ -1743,7 +1317,7 @@
       $opt->{value} = ($pre || '') . $num;
    }
    else {
-      $self->save_error("Invalid size for --$opt->{long}");
+      $self->save_error("Invalid size for --$opt->{long}: $val");
    }
    return;
 }
@@ -1818,345 +1392,129 @@
 # ###########################################################################
 
 # ###########################################################################
-# DSNParser package
+# 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/DSNParser.pm
-#   t/lib/DSNParser.t
+#   lib/Cxn.pm
+#   t/lib/Cxn.t
 # See https://launchpad.net/percona-toolkit for more information.
 # ###########################################################################
 {
-package DSNParser;
+package Cxn;
 
 use strict;
 use warnings FATAL => 'all';
 use English qw(-no_match_vars);
 use constant MKDEBUG => $ENV{MKDEBUG} || 0;
 
-use Data::Dumper;
-$Data::Dumper::Indent    = 0;
-$Data::Dumper::Quotekeys = 0;
-
-eval {
-   require DBI;
-};
-my $have_dbi = $EVAL_ERROR ? 0 : 1;
+use constant PERCONA_TOOLKIT_TEST_USE_DSN_NAMES => $ENV{PERCONA_TOOLKIT_TEST_USE_DSN_NAMES} || 0;
 
 sub new {
    my ( $class, %args ) = @_;
-   foreach my $arg ( qw(opts) ) {
+   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 = {
-      opts => {}  # h, P, u, etc.  Should come from DSN OPTIONS section in POD.
+      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,
    };
-   foreach my $opt ( @{$args{opts}} ) {
-      if ( !$opt->{key} || !$opt->{desc} ) {
-         die "Invalid DSN option: ", Dumper($opt);
-      }
-      MKDEBUG && _d('DSN option:',
-         join(', ',
-            map { "$_=" . (defined $opt->{$_} ? ($opt->{$_} || '') : 'undef') }
-               keys %$opt
-         )
-      );
-      $self->{opts}->{$opt->{key}} = {
-         dsn  => $opt->{dsn},
-         desc => $opt->{desc},
-         copy => $opt->{copy} || 0,
-      };
-   }
+
    return bless $self, $class;
 }
 
-sub prop {
-   my ( $self, $prop, $value ) = @_;
-   if ( @_ > 2 ) {
-      MKDEBUG && _d('Setting', $prop, 'property');
-      $self->{$prop} = $value;
-   }
-   return $self->{$prop};
-}
-
-sub parse {
-   my ( $self, $dsn, $prev, $defaults ) = @_;
-   if ( !$dsn ) {
-      MKDEBUG && _d('No DSN to parse');
-      return;
-   }
-   MKDEBUG && _d('Parsing', $dsn);
-   $prev     ||= {};
-   $defaults ||= {};
-   my %given_props;
-   my %final_props;
-   my $opts = $self->{opts};
-
-   foreach my $dsn_part ( split(/,/, $dsn) ) {
-      if ( my ($prop_key, $prop_val) = $dsn_part =~  m/^(.)=(.*)$/ ) {
-         $given_props{$prop_key} = $prop_val;
-      }
-      else {
-         MKDEBUG && _d('Interpreting', $dsn_part, 'as h=', $dsn_part);
-         $given_props{h} = $dsn_part;
-      }
-   }
-
-   foreach my $key ( keys %$opts ) {
-      MKDEBUG && _d('Finding value for', $key);
-      $final_props{$key} = $given_props{$key};
-      if (   !defined $final_props{$key}
-           && defined $prev->{$key} && $opts->{$key}->{copy} )
-      {
-         $final_props{$key} = $prev->{$key};
-         MKDEBUG && _d('Copying value for', $key, 'from previous DSN');
-      }
-      if ( !defined $final_props{$key} ) {
-         $final_props{$key} = $defaults->{$key};
-         MKDEBUG && _d('Copying value for', $key, 'from defaults');
-      }
-   }
-
-   foreach my $key ( keys %given_props ) {
-      die "Unknown DSN option '$key' in '$dsn'.  For more details, "
-            . "please use the --help option, or try 'perldoc $PROGRAM_NAME' "
-            . "for complete documentation."
-         unless exists $opts->{$key};
-   }
-   if ( (my $required = $self->prop('required')) ) {
-      foreach my $key ( keys %$required ) {
-         die "Missing required DSN option '$key' in '$dsn'.  For more details, "
-               . "please use the --help option, or try 'perldoc $PROGRAM_NAME' "
-               . "for complete documentation."
-            unless $final_props{$key};
-      }
-   }
-
-   return \%final_props;
-}
-
-sub parse_options {
-   my ( $self, $o ) = @_;
-   die 'I need an OptionParser object' unless ref $o eq 'OptionParser';
-   my $dsn_string
-      = join(',',
-          map  { "$_=".$o->get($_); }
-          grep { $o->has($_) && $o->get($_) }
-          keys %{$self->{opts}}
-        );
-   MKDEBUG && _d('DSN string made from options:', $dsn_string);
-   return $self->parse($dsn_string);
-}
-
-sub as_string {
-   my ( $self, $dsn, $props ) = @_;
-   return $dsn unless ref $dsn;
-   my %allowed = $props ? map { $_=>1 } @$props : ();
-   return join(',',
-      map  { "$_=" . ($_ eq 'p' ? '...' : $dsn->{$_})  }
-      grep { defined $dsn->{$_} && $self->{opts}->{$_} }
-      grep { !$props || $allowed{$_}                   }
-      sort keys %$dsn );
-}
-
-sub usage {
+sub connect {
    my ( $self ) = @_;
-   my $usage
-      = "DSN syntax is key=value[,key=value...]  Allowable DSN keys:\n\n"
-      . "  KEY  COPY  MEANING\n"
-      . "  ===  ====  =============================================\n";
-   my %opts = %{$self->{opts}};
-   foreach my $key ( sort keys %opts ) {
-      $usage .= "  $key    "
-             .  ($opts{$key}->{copy} ? 'yes   ' : 'no    ')
-             .  ($opts{$key}->{desc} || '[No description]')
-             . "\n";
-   }
-   $usage .= "\n  If the DSN is a bareword, the word is treated as the 'h' key.\n";
-   return $usage;
-}
-
-sub get_cxn_params {
-   my ( $self, $info ) = @_;
-   my $dsn;
-   my %opts = %{$self->{opts}};
-   my $driver = $self->prop('dbidriver') || '';
-   if ( $driver eq 'Pg' ) {
-      $dsn = 'DBI:Pg:dbname=' . ( $info->{D} || '' ) . ';'
-         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
-                     grep { defined $info->{$_} }
-                     qw(h P));
-   }
-   else {
-      $dsn = 'DBI:mysql:' . ( $info->{D} || '' ) . ';'
-         . join(';', map  { "$opts{$_}->{dsn}=$info->{$_}" }
-                     grep { defined $info->{$_} }
-                     qw(F h P S A))
-         . ';mysql_read_default_group=client';
-   }
-   MKDEBUG && _d($dsn);
-   return ($dsn, $info->{u}, $info->{p});
-}
-
-sub fill_in_dsn {
-   my ( $self, $dbh, $dsn ) = @_;
-   my $vars = $dbh->selectall_hashref('SHOW VARIABLES', 'Variable_name');
-   my ($user, $db) = $dbh->selectrow_array('SELECT USER(), DATABASE()');
-   $user =~ s/@.*//;
-   $dsn->{h} ||= $vars->{hostname}->{Value};
-   $dsn->{S} ||= $vars->{'socket'}->{Value};
-   $dsn->{P} ||= $vars->{port}->{Value};
-   $dsn->{u} ||= $user;
-   $dsn->{D} ||= $db;
-}
-
-sub get_dbh {
-   my ( $self, $cxn_string, $user, $pass, $opts ) = @_;
-   $opts ||= {};
-   my $defaults = {
-      AutoCommit         => 0,
-      RaiseError         => 1,
-      PrintError         => 0,
-      ShowErrorStatement => 1,
-      mysql_enable_utf8 => ($cxn_string =~ m/charset=utf8/i ? 1 : 0),
-   };
-   @{$defaults}{ keys %$opts } = values %$opts;
-
-   if ( $opts->{mysql_use_result} ) {
-      $defaults->{mysql_use_result} = 1;
-   }
-
-   if ( !$have_dbi ) {
-      die "Cannot connect to MySQL because the Perl DBI module is not "
-         . "installed or not found.  Run 'perl -MDBI' to see the directories "
-         . "that Perl searches for DBI.  If DBI is not installed, try:\n"
-         . "  Debian/Ubuntu  apt-get install libdbi-perl\n"
-         . "  RHEL/CentOS    yum install perl-DBI\n"
-         . "  OpenSolaris    pgk install pkg:/SUNWpmdbi\n";
-
-   }
-
-   my $dbh;
-   my $tries = 2;
-   while ( !$dbh && $tries-- ) {
-      MKDEBUG && _d($cxn_string, ' ', $user, ' ', $pass, 
-         join(', ', map { "$_=>$defaults->{$_}" } keys %$defaults ));
-
-      eval {
-         $dbh = DBI->connect($cxn_string, $user, $pass, $defaults);
-
-         if ( $cxn_string =~ m/mysql/i ) {
-            my $sql;
-
-            $sql = 'SELECT @@SQL_MODE';
-            MKDEBUG && _d($dbh, $sql);
-            my ($sql_mode) = $dbh->selectrow_array($sql);
-
-            $sql = 'SET @@SQL_QUOTE_SHOW_CREATE = 1'
-                 . '/*!40101, @@SQL_MODE=\'NO_AUTO_VALUE_ON_ZERO'
-                 . ($sql_mode ? ",$sql_mode" : '')
-                 . '\'*/';
-            MKDEBUG && _d($dbh, $sql);
-            $dbh->do($sql);
-
-            if ( my ($charset) = $cxn_string =~ m/charset=(\w+)/ ) {
-               $sql = "/*!40101 SET NAMES $charset*/";
-               MKDEBUG && _d($dbh, ':', $sql);
-               $dbh->do($sql);
-               MKDEBUG && _d('Enabling charset for STDOUT');
-               if ( $charset eq 'utf8' ) {
-                  binmode(STDOUT, ':utf8')
-                     or die "Can't binmode(STDOUT, ':utf8'): $OS_ERROR";
-               }
-               else {
-                  binmode(STDOUT) or die "Can't binmode(STDOUT): $OS_ERROR";
-               }
-            }
-
-            if ( $self->prop('set-vars') ) {
-               $sql = "SET " . $self->prop('set-vars');
-               MKDEBUG && _d($dbh, ':', $sql);
-               $dbh->do($sql);
-            }
-         }
-      };
-      if ( !$dbh && $EVAL_ERROR ) {
-         MKDEBUG && _d($EVAL_ERROR);
-         if ( $EVAL_ERROR =~ m/not a compiled character set|character set utf8/ ) {
-            MKDEBUG && _d('Going to try again without utf8 support');
-            delete $defaults->{mysql_enable_utf8};
-         }
-         elsif ( $EVAL_ERROR =~ m/locate DBD\/mysql/i ) {
-            die "Cannot connect to MySQL because the Perl DBD::mysql module is "
-               . "not installed or not found.  Run 'perl -MDBD::mysql' to see "
-               . "the directories that Perl searches for DBD::mysql.  If "
-               . "DBD::mysql is not installed, try:\n"
-               . "  Debian/Ubuntu  apt-get install libdbd-mysql-perl\n"
-               . "  RHEL/CentOS    yum install perl-DBD-MySQL\n"
-               . "  OpenSolaris    pgk install pkg:/SUNWapu13dbd-mysql\n";
-         }
-         if ( !$tries ) {
-            die $EVAL_ERROR;
-         }
+   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;
       }
-   }
-
-   MKDEBUG && _d('DBH info: ',
-      $dbh,
-      Dumper($dbh->selectrow_hashref(
-         'SELECT DATABASE(), CONNECTION_ID(), VERSION()/*!50038 , @@hostname*/')),
-      'Connection info:',      $dbh->{mysql_hostinfo},
-      'Character set info:',   Dumper($dbh->selectall_arrayref(
-                     'SHOW VARIABLES LIKE "character_set%"', { Slice => {}})),
-      '$DBD::mysql::VERSION:', $DBD::mysql::VERSION,
-      '$DBI::VERSION:',        $DBI::VERSION,
-   );
-
+      $dbh = $dp->get_dbh($dp->get_cxn_params($dsn),  { AutoCommit => 1 });
+   }
+   MKDEBUG && _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} ) {
+      MKDEBUG && _d($dbh, 'Already set dbh');
+      return $dbh;
+   }
+
+   MKDEBUG && _d($dbh, 'Setting dbh');
+
+   $dbh->{FetchHashKeyName} = 'NAME_lc';
+   
+   my $sql = 'SELECT @@hostname, @@server_id';
+   MKDEBUG && _d($dbh, $sql);
+   my ($hostname, $server_id) = $dbh->selectrow_array($sql);
+   MKDEBUG && _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 get_hostname {
-   my ( $self, $dbh ) = @_;
-   if ( my ($host) = ($dbh->{mysql_hostinfo} || '') =~ m/^(\w+) via/ ) {
-      return $host;
-   }
-   my ( $hostname, $one ) = $dbh->selectrow_array(
-      'SELECT /*!50038 @@hostname, */ 1');
-   return $hostname;
-}
-
-sub disconnect {
-   my ( $self, $dbh ) = @_;
-   MKDEBUG && $self->print_active_handles($dbh);
-   $dbh->disconnect;
-}
-
-sub print_active_handles {
-   my ( $self, $thing, $level ) = @_;
-   $level ||= 0;
-   printf("# Active %sh: %s %s %s\n", ($thing->{Type} || 'undef'), "\t" x $level,
-      $thing, (($thing->{Type} || '') eq 'st' ? $thing->{Statement} || '' : ''))
-      or die "Cannot print: $OS_ERROR";
-   foreach my $handle ( grep {defined} @{ $thing->{ChildHandles} } ) {
-      $self->print_active_handles( $handle, $level + 1 );
-   }
-}
-
-sub copy {
-   my ( $self, $dsn_1, $dsn_2, %args ) = @_;
-   die 'I need a dsn_1 argument' unless $dsn_1;
-   die 'I need a dsn_2 argument' unless $dsn_2;
-   my %new_dsn = map {
-      my $key = $_;
-      my $val;
-      if ( $args{overwrite} ) {
-         $val = defined $dsn_1->{$key} ? $dsn_1->{$key} : $dsn_2->{$key};
-      }
-      else {
-         $val = defined $dsn_2->{$key} ? $dsn_2->{$key} : $dsn_1->{$key};
-      }
-      $key => $val;
-   } keys %{$self->{opts}};
-   return \%new_dsn;
+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} ) {
+      MKDEBUG && _d('Disconnecting dbh', $self->{dbh}, $self->{name});
+      $self->{dbh}->disconnect();
+   }
+   return;
 }
 
 sub _d {
@@ -2170,7 +1528,126 @@
 1;
 }
 # ###########################################################################
-# End DSNParser package
+# End Cxn package
+# ###########################################################################
+
+# ###########################################################################
+# Quoter 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/Quoter.pm
+#   t/lib/Quoter.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package Quoter;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant MKDEBUG => $ENV{MKDEBUG} || 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   return bless {}, $class;
+}
+
+sub quote {
+   my ( $self, @vals ) = @_;
+   foreach my $val ( @vals ) {
+      $val =~ s/`/``/g;
+   }
+   return join('.', map { '`' . $_ . '`' } @vals);
+}
+
+sub quote_val {
+   my ( $self, $val ) = @_;
+
+   return 'NULL' unless defined $val;          # undef = NULL
+   return "''" if $val eq '';                  # blank string = ''
+   return $val if $val =~ m/^0x[0-9a-fA-F]+$/;  # hex data
+
+   $val =~ s/(['\\])/\\$1/g;
+   return "'$val'";
+}
+
+sub split_unquote {
+   my ( $self, $db_tbl, $default_db ) = @_;
+   $db_tbl =~ s/`//g;
+   my ( $db, $tbl ) = split(/[.]/, $db_tbl);
+   if ( !$tbl ) {
+      $tbl = $db;
+      $db  = $default_db;
+   }
+   return ($db, $tbl);
+}
+
+sub literal_like {
+   my ( $self, $like ) = @_;
+   return unless $like;
+   $like =~ s/([%_])/\\$1/g;
+   return "'$like'";
+}
+
+sub join_quote {
+   my ( $self, $default_db, $db_tbl ) = @_;
+   return unless $db_tbl;
+   my ($db, $tbl) = split(/[.]/, $db_tbl);
+   if ( !$tbl ) {
+      $tbl = $db;
+      $db  = $default_db;
+   }
+   $db  = "`$db`"  if $db  && $db  !~ m/^`/;
+   $tbl = "`$tbl`" if $tbl && $tbl !~ m/^`/;
+   return $db ? "$db.$tbl" : $tbl;
+}
+
+sub serialize_list {
+   my ( $self, @args ) = @_;
+   return unless @args;
+
+   return $args[0] if @args == 1 && !defined $args[0];
+
+   die "Cannot serialize multiple values with undef/NULL"
+      if grep { !defined $_ } @args;
+
+   return join ',', map { quotemeta } @args;
+}
+
+sub deserialize_list {
+   my ( $self, $string ) = @_;
+   return $string unless defined $string;
+   my @escaped_parts = $string =~ /
+         \G             # Start of string, or end of previous match.
+         (              # Each of these is an element in the original list.
+            [^\\,]*     # Anything not a backslash or a comma
+            (?:         # When we get here, we found one of the above.
+               \\.      # A backslash followed by something so we can continue
+               [^\\,]*  # Same as above.
+            )*          # Repeat zero of more times.
+         )
+         ,              # Comma dividing elements
+      /sxgc;
+
+   push @escaped_parts, pos($string) ? substr( $string, pos($string) ) : $string;
+
+   my @unescaped_parts = map {
+      my $part = $_;
+
+      my $char_class = utf8::is_utf8($part)  # If it's a UTF-8 string,
+                     ? qr/(?=\p{ASCII})\W/   # We only care about non-word
+                     : qr/(?=\p{ASCII})\W|[\x{80}-\x{FF}]/; # Otherwise,
+      $part =~ s/\\($char_class)/$1/g;
+      $part;
+   } @escaped_parts;
+
+   return @unescaped_parts;
+}
+
+1;
+}
+# ###########################################################################
+# End Quoter package
 # ###########################################################################
 
 # ###########################################################################
@@ -2257,294 +1734,417 @@
 # ###########################################################################
 
 # ###########################################################################
-# MySQLDump package
+# TableParser 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/MySQLDump.pm
-#   t/lib/MySQLDump.t
+#   lib/TableParser.pm
+#   t/lib/TableParser.t
 # See https://launchpad.net/percona-toolkit for more information.
 # ###########################################################################
 {
-package MySQLDump;
+package TableParser;
 
 use strict;
 use warnings FATAL => 'all';
 use English qw(-no_match_vars);
 use constant MKDEBUG => $ENV{MKDEBUG} || 0;
 
-( our $before = <<'EOF') =~ s/^   //gm;
-   /*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
-   /*!40101 SET @OLD_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS */;
-   /*!40101 SET @OLD_COLLATION_CONNECTION=@@COLLATION_CONNECTION */;
-   /*!40101 SET NAMES utf8 */;
-   /*!40103 SET @OLD_TIME_ZONE=@@TIME_ZONE */;
-   /*!40103 SET TIME_ZONE='+00:00' */;
-   /*!40014 SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0 */;
-   /*!40014 SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0 */;
-   /*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;
-   /*!40111 SET @OLD_SQL_NOTES=@@SQL_NOTES, SQL_NOTES=0 */;
-EOF
-
-( our $after = <<'EOF') =~ s/^   //gm;
-   /*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;
-   /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
-   /*!40014 SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS */;
-   /*!40014 SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS */;
-   /*!40101 SET CHARACTER_SET_CLIENT=@OLD_CHARACTER_SET_CLIENT */;
-   /*!40101 SET CHARACTER_SET_RESULTS=@OLD_CHARACTER_SET_RESULTS */;
-   /*!40101 SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION */;
-   /*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;
-EOF
+use Data::Dumper;
+$Data::Dumper::Indent    = 1;
+$Data::Dumper::Sortkeys  = 1;
+$Data::Dumper::Quotekeys = 0;
 
 sub new {
    my ( $class, %args ) = @_;
-   my $self = {
-      cache => 0,  # Afaik no script uses this cache any longer because
-   };
+   my @required_args = qw(Quoter);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my $self = { %args };
    return bless $self, $class;
 }
 
-sub dump {
-   my ( $self, $dbh, $quoter, $db, $tbl, $what ) = @_;
-
-   if ( $what eq 'table' ) {
-      my $ddl = $self->get_create_table($dbh, $quoter, $db, $tbl);
-      return unless $ddl;
-      if ( $ddl->[0] eq 'table' ) {
-         return $before
-            . 'DROP TABLE IF EXISTS ' . $quoter->quote($tbl) . ";\n"
-            . $ddl->[1] . ";\n";
-      }
-      else {
-         return 'DROP TABLE IF EXISTS ' . $quoter->quote($tbl) . ";\n"
-            . '/*!50001 DROP VIEW IF EXISTS '
-            . $quoter->quote($tbl) . "*/;\n/*!50001 "
-            . $self->get_tmp_table($dbh, $quoter, $db, $tbl) . "*/;\n";
-      }
-   }
-   elsif ( $what eq 'triggers' ) {
-      my $trgs = $self->get_triggers($dbh, $quoter, $db, $tbl);
-      if ( $trgs && @$trgs ) {
-         my $result = $before . "\nDELIMITER ;;\n";
-         foreach my $trg ( @$trgs ) {
-            if ( $trg->{sql_mode} ) {
-               $result .= qq{/*!50003 SET SESSION SQL_MODE='$trg->{sql_mode}' */;;\n};
-            }
-            $result .= "/*!50003 CREATE */ ";
-            if ( $trg->{definer} ) {
-               my ( $user, $host )
-                  = map { s/'/''/g; "'$_'"; }
-                    split('@', $trg->{definer}, 2);
-               $result .= "/*!50017 DEFINER=$user\@$host */ ";
-            }
-            $result .= sprintf("/*!50003 TRIGGER %s %s %s ON %s\nFOR EACH ROW %s */;;\n\n",
-               $quoter->quote($trg->{trigger}),
-               @{$trg}{qw(timing event)},
-               $quoter->quote($trg->{table}),
-               $trg->{statement});
-         }
-         $result .= "DELIMITER ;\n\n/*!50003 SET SESSION SQL_MODE=\@OLD_SQL_MODE */;\n\n";
-         return $result;
-      }
-      else {
-         return undef;
-      }
-   }
-   elsif ( $what eq 'view' ) {
-      my $ddl = $self->get_create_table($dbh, $quoter, $db, $tbl);
-      return '/*!50001 DROP TABLE IF EXISTS ' . $quoter->quote($tbl) . "*/;\n"
-         . '/*!50001 DROP VIEW IF EXISTS ' . $quoter->quote($tbl) . "*/;\n"
-         . '/*!50001 ' . $ddl->[1] . "*/;\n";
-   }
-   else {
-      die "You didn't say what to dump.";
-   }
-}
-
-sub _use_db {
-   my ( $self, $dbh, $quoter, $new ) = @_;
-   if ( !$new ) {
-      MKDEBUG && _d('No new DB to use');
-      return;
-   }
-   my $sql = 'USE ' . $quoter->quote($new);
-   MKDEBUG && _d($dbh, $sql);
-   $dbh->do($sql);
-   return;
-}
-
 sub get_create_table {
-   my ( $self, $dbh, $quoter, $db, $tbl ) = @_;
-   if ( !$self->{cache} || !$self->{tables}->{$db}->{$tbl} ) {
-      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 */';
-      MKDEBUG && _d($sql);
-      eval { $dbh->do($sql); };
-      MKDEBUG && $EVAL_ERROR && _d($EVAL_ERROR);
-      $self->_use_db($dbh, $quoter, $db);
-      $sql = "SHOW CREATE TABLE " . $quoter->quote($db, $tbl);
-      MKDEBUG && _d($sql);
-      my $href;
-      eval { $href = $dbh->selectrow_hashref($sql); };
-      if ( $EVAL_ERROR ) {
-         warn "Failed to $sql.  The table may be damaged.\nError: $EVAL_ERROR";
-         return;
-      }
-
-      $sql = '/*!40101 SET @@SQL_MODE := @OLD_SQL_MODE, '
-         . '@@SQL_QUOTE_SHOW_CREATE := @OLD_QUOTE */';
-      MKDEBUG && _d($sql);
-      $dbh->do($sql);
-      my ($key) = grep { m/create table/i } keys %$href;
-      if ( $key ) {
-         MKDEBUG && _d('This table is a base table');
-         $self->{tables}->{$db}->{$tbl} = [ 'table', $href->{$key} ];
+   my ( $self, $dbh, $db, $tbl ) = @_;
+   die "I need a dbh parameter" unless $dbh;
+   die "I need a db parameter"  unless $db;
+   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 */';
+   MKDEBUG && _d($sql);
+   eval { $dbh->do($sql); };
+   MKDEBUG && $EVAL_ERROR && _d($EVAL_ERROR);
+
+   $sql = 'USE ' . $q->quote($db);
+   MKDEBUG && _d($dbh, $sql);
+   $dbh->do($sql);
+
+   $sql = "SHOW CREATE TABLE " . $q->quote($db, $tbl);
+   MKDEBUG && _d($sql);
+   my $href;
+   eval { $href = $dbh->selectrow_hashref($sql); };
+   if ( $EVAL_ERROR ) {
+      MKDEBUG && _d($EVAL_ERROR);
+      return;
+   }
+
+   $sql = '/*!40101 SET @@SQL_MODE := @OLD_SQL_MODE, '
+        . '@@SQL_QUOTE_SHOW_CREATE := @OLD_QUOTE */';
+   MKDEBUG && _d($sql);
+   $dbh->do($sql);
+
+   my ($key) = grep { m/create table/i } keys %$href;
+   if ( $key ) {
+      MKDEBUG && _d('This table is a base table');
+      $href->{$key}  =~ s/\b[ ]{2,}/ /g;
+      $href->{$key} .= "\n";
+   }
+   else {
+      MKDEBUG && _d('This table is a view');
+      ($key) = grep { m/create view/i } keys %$href;
+   }
+
+   return $href->{$key};
+}
+
+sub parse {
+   my ( $self, $ddl, $opts ) = @_;
+   return unless $ddl;
+
+   if ( $ddl !~ m/CREATE (?:TEMPORARY )?TABLE `/ ) {
+      die "Cannot parse table definition; is ANSI quoting "
+         . "enabled or SQL_QUOTE_SHOW_CREATE disabled?";
+   }
+
+   my ($name)     = $ddl =~ m/CREATE (?:TEMPORARY )?TABLE\s+(`.+?`)/;
+   (undef, $name) = $self->{Quoter}->split_unquote($name) if $name;
+
+   $ddl =~ s/(`[^`]+`)/\L$1/g;
+
+   my $engine = $self->get_engine($ddl);
+
+   my @defs   = $ddl =~ m/^(\s+`.*?),?$/gm;
+   my @cols   = map { $_ =~ m/`([^`]+)`/ } @defs;
+   MKDEBUG && _d('Table cols:', join(', ', map { "`$_`" } @cols));
+
+   my %def_for;
+   @def_for{@cols} = @defs;
+
+   my (@nums, @null);
+   my (%type_for, %is_nullable, %is_numeric, %is_autoinc);
+   foreach my $col ( @cols ) {
+      my $def = $def_for{$col};
+      my ( $type ) = $def =~ m/`[^`]+`\s([a-z]+)/;
+      die "Can't determine column type for $def" unless $type;
+      $type_for{$col} = $type;
+      if ( $type =~ m/(?:(?:tiny|big|medium|small)?int|float|double|decimal|year)/ ) {
+         push @nums, $col;
+         $is_numeric{$col} = 1;
+      }
+      if ( $def !~ m/NOT NULL/ ) {
+         push @null, $col;
+         $is_nullable{$col} = 1;
+      }
+      $is_autoinc{$col} = $def =~ m/AUTO_INCREMENT/i ? 1 : 0;
+   }
+
+   my ($keys, $clustered_key) = $self->get_keys($ddl, $opts, \%is_nullable);
+
+   my ($charset) = $ddl =~ m/DEFAULT CHARSET=(\w+)/;
+
+   return {
+      name           => $name,
+      cols           => \@cols,
+      col_posn       => { map { $cols[$_] => $_ } 0..$#cols },
+      is_col         => { map { $_ => 1 } @cols },
+      null_cols      => \@null,
+      is_nullable    => \%is_nullable,
+      is_autoinc     => \%is_autoinc,
+      clustered_key  => $clustered_key,
+      keys           => $keys,
+      defs           => \%def_for,
+      numeric_cols   => \@nums,
+      is_numeric     => \%is_numeric,
+      engine         => $engine,
+      type_for       => \%type_for,
+      charset        => $charset,
+   };
+}
+
+sub sort_indexes {
+   my ( $self, $tbl ) = @_;
+
+   my @indexes
+      = sort {
+         (($a ne 'PRIMARY') <=> ($b ne 'PRIMARY'))
+         || ( !$tbl->{keys}->{$a}->{is_unique} <=> !$tbl->{keys}->{$b}->{is_unique} )
+         || ( $tbl->{keys}->{$a}->{is_nullable} <=> $tbl->{keys}->{$b}->{is_nullable} )
+         || ( scalar(@{$tbl->{keys}->{$a}->{cols}}) <=> scalar(@{$tbl->{keys}->{$b}->{cols}}) )
+      }
+      grep {
+         $tbl->{keys}->{$_}->{type} eq 'BTREE'
+      }
+      sort keys %{$tbl->{keys}};
+
+   MKDEBUG && _d('Indexes sorted best-first:', join(', ', @indexes));
+   return @indexes;
+}
+
+sub find_best_index {
+   my ( $self, $tbl, $index ) = @_;
+   my $best;
+   if ( $index ) {
+      ($best) = grep { uc $_ eq uc $index } keys %{$tbl->{keys}};
+   }
+   if ( !$best ) {
+      if ( $index ) {
+         die "Index '$index' does not exist in table";
       }
       else {
-         MKDEBUG && _d('This table is a view');
-         ($key) = grep { m/create view/i } keys %$href;
-         $self->{tables}->{$db}->{$tbl} = [ 'view', $href->{$key} ];
-      }
-   }
-   return $self->{tables}->{$db}->{$tbl};
-}
-
-sub get_columns {
-   my ( $self, $dbh, $quoter, $db, $tbl ) = @_;
-   MKDEBUG && _d('Get columns for', $db, $tbl);
-   if ( !$self->{cache} || !$self->{columns}->{$db}->{$tbl} ) {
-      $self->_use_db($dbh, $quoter, $db);
-      my $sql = "SHOW COLUMNS FROM " . $quoter->quote($db, $tbl);
-      MKDEBUG && _d($sql);
-      my $cols = $dbh->selectall_arrayref($sql, { Slice => {} });
-
-      $self->{columns}->{$db}->{$tbl} = [
-         map {
-            my %row;
-            @row{ map { lc $_ } keys %$_ } = values %$_;
-            \%row;
-         } @$cols
-      ];
-   }
-   return $self->{columns}->{$db}->{$tbl};
-}
-
-sub get_tmp_table {
-   my ( $self, $dbh, $quoter, $db, $tbl ) = @_;
-   my $result = 'CREATE TABLE ' . $quoter->quote($tbl) . " (\n";
-   $result .= join(",\n",
-      map { '  ' . $quoter->quote($_->{field}) . ' ' . $_->{type} }
-      @{$self->get_columns($dbh, $quoter, $db, $tbl)});
-   $result .= "\n)";
-   MKDEBUG && _d($result);
-   return $result;
-}
-
-sub get_triggers {
-   my ( $self, $dbh, $quoter, $db, $tbl ) = @_;
-   if ( !$self->{cache} || !$self->{triggers}->{$db} ) {
-      $self->{triggers}->{$db} = {};
-      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 */';
-      MKDEBUG && _d($sql);
-      eval { $dbh->do($sql); };
-      MKDEBUG && $EVAL_ERROR && _d($EVAL_ERROR);
-      $sql = "SHOW TRIGGERS FROM " . $quoter->quote($db);
-      MKDEBUG && _d($sql);
-      my $sth = $dbh->prepare($sql);
-      $sth->execute();
-      if ( $sth->rows ) {
-         my $trgs = $sth->fetchall_arrayref({});
-         foreach my $trg (@$trgs) {
-            my %trg;
-            @trg{ map { lc $_ } keys %$trg } = values %$trg;
-            push @{ $self->{triggers}->{$db}->{ $trg{table} } }, \%trg;
-         }
-      }
-      $sql = '/*!40101 SET @@SQL_MODE := @OLD_SQL_MODE, '
-         . '@@SQL_QUOTE_SHOW_CREATE := @OLD_QUOTE */';
-      MKDEBUG && _d($sql);
+         ($best) = $self->sort_indexes($tbl);
+      }
+   }
+   MKDEBUG && _d('Best index found is', $best);
+   return $best;
+}
+
+sub find_possible_keys {
+   my ( $self, $dbh, $database, $table, $quoter, $where ) = @_;
+   return () unless $where;
+   my $sql = 'EXPLAIN SELECT * FROM ' . $quoter->quote($database, $table)
+      . ' WHERE ' . $where;
+   MKDEBUG && _d($sql);
+   my $expl = $dbh->selectrow_hashref($sql);
+   $expl = { map { lc($_) => $expl->{$_} } keys %$expl };
+   if ( $expl->{possible_keys} ) {
+      MKDEBUG && _d('possible_keys =', $expl->{possible_keys});
+      my @candidates = split(',', $expl->{possible_keys});
+      my %possible   = map { $_ => 1 } @candidates;
+      if ( $expl->{key} ) {
+         MKDEBUG && _d('MySQL chose', $expl->{key});
+         unshift @candidates, grep { $possible{$_} } split(',', $expl->{key});
+         MKDEBUG && _d('Before deduping:', join(', ', @candidates));
+         my %seen;
+         @candidates = grep { !$seen{$_}++ } @candidates;
+      }
+      MKDEBUG && _d('Final list:', join(', ', @candidates));
+      return @candidates;
+   }
+   else {
+      MKDEBUG && _d('No keys in possible_keys');
+      return ();
+   }
+}
+
+sub check_table {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(dbh db tbl);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($dbh, $db, $tbl) = @args{@required_args};
+   my $q      = $self->{Quoter};
+   my $db_tbl = $q->quote($db, $tbl);
+   MKDEBUG && _d('Checking', $db_tbl);
+
+   my $sql = "SHOW TABLES FROM " . $q->quote($db)
+           . ' LIKE ' . $q->literal_like($tbl);
+   MKDEBUG && _d($sql);
+   my $row;
+   eval {
+      $row = $dbh->selectrow_arrayref($sql);
+   };
+   if ( $EVAL_ERROR ) {
+      MKDEBUG && _d($EVAL_ERROR);
+      return 0;
+   }
+   if ( !$row->[0] || $row->[0] ne $tbl ) {
+      MKDEBUG && _d('Table does not exist');
+      return 0;
+   }
+
+   MKDEBUG && _d('Table exists; no privs to check');
+   return 1 unless $args{all_privs};
+
+   $sql = "SHOW FULL COLUMNS FROM $db_tbl";
+   MKDEBUG && _d($sql);
+   eval {
+      $row = $dbh->selectrow_hashref($sql);
+   };
+   if ( $EVAL_ERROR ) {
+      MKDEBUG && _d($EVAL_ERROR);
+      return 0;
+   }
+   if ( !scalar keys %$row ) {
+      MKDEBUG && _d('Table has no columns:', Dumper($row));
+      return 0;
+   }
+   my $privs = $row->{privileges} || $row->{Privileges};
+
+   $sql = "DELETE FROM $db_tbl LIMIT 0";
+   MKDEBUG && _d($sql);
+   eval {
       $dbh->do($sql);
-   }
-   if ( $tbl ) {
-      return $self->{triggers}->{$db}->{$tbl};
-   }
-   return values %{$self->{triggers}->{$db}};
-}
-
-sub get_databases {
-   my ( $self, $dbh, $quoter, $like ) = @_;
-   if ( !$self->{cache} || !$self->{databases} || $like ) {
-      my $sql = 'SHOW DATABASES';
-      my @params;
-      if ( $like ) {
-         $sql .= ' LIKE ?';
-         push @params, $like;
-      }
-      my $sth = $dbh->prepare($sql);
-      MKDEBUG && _d($sql, @params);
-      $sth->execute( @params );
-      my @dbs = map { $_->[0] } @{$sth->fetchall_arrayref()};
-      $self->{databases} = \@dbs unless $like;
-      return @dbs;
-   }
-   return @{$self->{databases}};
+   };
+   my $can_delete = $EVAL_ERROR ? 0 : 1;
+
+   MKDEBUG && _d('User privs on', $db_tbl, ':', $privs,
+      ($can_delete ? 'delete' : ''));
+
+   if ( !($privs =~ m/select/ && $privs =~ m/insert/ && $privs =~ m/update/
+          && $can_delete) ) {
+      MKDEBUG && _d('User does not have all privs');
+      return 0;
+   }
+
+   MKDEBUG && _d('User has all privs');
+   return 1;
+}
+
+sub get_engine {
+   my ( $self, $ddl, $opts ) = @_;
+   my ( $engine ) = $ddl =~ m/\).*?(?:ENGINE|TYPE)=(\w+)/;
+   MKDEBUG && _d('Storage engine:', $engine);
+   return $engine || undef;
+}
+
+sub get_keys {
+   my ( $self, $ddl, $opts, $is_nullable ) = @_;
+   my $engine        = $self->get_engine($ddl);
+   my $keys          = {};
+   my $clustered_key = undef;
+
+   KEY:
+   foreach my $key ( $ddl =~ m/^  ((?:[A-Z]+ )?KEY .*)$/gm ) {
+
+      next KEY if $key =~ m/FOREIGN/;
+
+      my $key_ddl = $key;
+      MKDEBUG && _d('Parsed key:', $key_ddl);
+
+      if ( $engine !~ m/MEMORY|HEAP/ ) {
+         $key =~ s/USING HASH/USING BTREE/;
+      }
+
+      my ( $type, $cols ) = $key =~ m/(?:USING (\w+))? \((.+)\)/;
+      my ( $special ) = $key =~ m/(FULLTEXT|SPATIAL)/;
+      $type = $type || $special || 'BTREE';
+      if ( $opts->{mysql_version} && $opts->{mysql_version} lt '004001000'
+         && $engine =~ m/HEAP|MEMORY/i )
+      {
+         $type = 'HASH'; # MySQL pre-4.1 supports only HASH indexes on HEAP
+      }
+
+      my ($name) = $key =~ m/(PRIMARY|`[^`]*`)/;
+      my $unique = $key =~ m/PRIMARY|UNIQUE/ ? 1 : 0;
+      my @cols;
+      my @col_prefixes;
+      foreach my $col_def ( $cols =~ m/`[^`]+`(?:\(\d+\))?/g ) {
+         my ($name, $prefix) = $col_def =~ m/`([^`]+)`(?:\((\d+)\))?/;
+         push @cols, $name;
+         push @col_prefixes, $prefix;
+      }
+      $name =~ s/`//g;
+
+      MKDEBUG && _d( $name, 'key cols:', join(', ', map { "`$_`" } @cols));
+
+      $keys->{$name} = {
+         name         => $name,
+         type         => $type,
+         colnames     => $cols,
+         cols         => \@cols,
+         col_prefixes => \@col_prefixes,
+         is_unique    => $unique,
+         is_nullable  => scalar(grep { $is_nullable->{$_} } @cols),
+         is_col       => { map { $_ => 1 } @cols },
+         ddl          => $key_ddl,
+      };
+
+      if ( $engine =~ m/InnoDB/i && !$clustered_key ) {
+         my $this_key = $keys->{$name};
+         if ( $this_key->{name} eq 'PRIMARY' ) {
+            $clustered_key = 'PRIMARY';
+         }
+         elsif ( $this_key->{is_unique} && !$this_key->{is_nullable} ) {
+            $clustered_key = $this_key->{name};
+         }
+         MKDEBUG && $clustered_key && _d('This key is the clustered key');
+      }
+   }
+
+   return $keys, $clustered_key;
+}
+
+sub get_fks {
+   my ( $self, $ddl, $opts ) = @_;
+   my $q   = $self->{Quoter};
+   my $fks = {};
+
+   foreach my $fk (
+      $ddl =~ m/CONSTRAINT .* FOREIGN KEY .* REFERENCES [^\)]*\)/mg )
+   {
+      my ( $name ) = $fk =~ m/CONSTRAINT `(.*?)`/;
+      my ( $cols ) = $fk =~ m/FOREIGN KEY \(([^\)]+)\)/;
+      my ( $parent, $parent_cols ) = $fk =~ m/REFERENCES (\S+) \(([^\)]+)\)/;
+
+      my ($db, $tbl) = $q->split_unquote($parent, $opts->{database});
+      my %parent_tbl = (tbl => $tbl);
+      $parent_tbl{db} = $db if $db;
+
+      if ( $parent !~ m/\./ && $opts->{database} ) {
+         $parent = $q->quote($opts->{database}) . ".$parent";
+      }
+
+      $fks->{$name} = {
+         name           => $name,
+         colnames       => $cols,
+         cols           => [ map { s/[ `]+//g; $_; } split(',', $cols) ],
+         parent_tbl     => \%parent_tbl,
+         parent_tblname => $parent,
+         parent_cols    => [ map { s/[ `]+//g; $_; } split(',', $parent_cols) ],
+         parent_colnames=> $parent_cols,
+         ddl            => $fk,
+      };
+   }
+
+   return $fks;
+}
+
+sub remove_auto_increment {
+   my ( $self, $ddl ) = @_;
+   $ddl =~ s/(^\).*?) AUTO_INCREMENT=\d+\b/$1/m;
+   return $ddl;
 }
 
 sub get_table_status {
-   my ( $self, $dbh, $quoter, $db, $like ) = @_;
-   if ( !$self->{cache} || !$self->{table_status}->{$db} || $like ) {
-      my $sql = "SHOW TABLE STATUS FROM " . $quoter->quote($db);
-      my @params;
-      if ( $like ) {
-         $sql .= ' LIKE ?';
-         push @params, $like;
-      }
-      MKDEBUG && _d($sql, @params);
-      my $sth = $dbh->prepare($sql);
-      $sth->execute(@params);
-      my @tables = @{$sth->fetchall_arrayref({})};
-      @tables = map {
-         my %tbl; # Make a copy with lowercased keys
-         @tbl{ map { lc $_ } keys %$_ } = values %$_;
-         $tbl{engine} ||= $tbl{type} || $tbl{comment};
-         delete $tbl{type};
-         \%tbl;
-      } @tables;
-      $self->{table_status}->{$db} = \@tables unless $like;
-      return @tables;
-   }
-   return @{$self->{table_status}->{$db}};
-}
-
-sub get_table_list {
-   my ( $self, $dbh, $quoter, $db, $like ) = @_;
-   if ( !$self->{cache} || !$self->{table_list}->{$db} || $like ) {
-      my $sql = "SHOW /*!50002 FULL*/ TABLES FROM " . $quoter->quote($db);
-      my @params;
-      if ( $like ) {
-         $sql .= ' LIKE ?';
-         push @params, $like;
-      }
-      MKDEBUG && _d($sql, @params);
-      my $sth = $dbh->prepare($sql);
-      $sth->execute(@params);
-      my @tables = @{$sth->fetchall_arrayref()};
-      @tables = map {
-         my %tbl = (
-            name   => $_->[0],
-            engine => ($_->[1] || '') eq 'VIEW' ? 'VIEW' : '',
-         );
-         \%tbl;
-      } @tables;
-      $self->{table_list}->{$db} = \@tables unless $like;
-      return @tables;
-   }
-   return @{$self->{table_list}->{$db}};
+   my ( $self, $dbh, $db, $like ) = @_;
+   my $q = $self->{Quoter};
+   my $sql = "SHOW TABLE STATUS FROM " . $q->quote($db);
+   my @params;
+   if ( $like ) {
+      $sql .= ' LIKE ?';
+      push @params, $like;
+   }
+   MKDEBUG && _d($sql, @params);
+   my $sth = $dbh->prepare($sql);
+   eval { $sth->execute(@params); };
+   if ($EVAL_ERROR) {
+      MKDEBUG && _d($EVAL_ERROR);
+      return;
+   }
+   my @tables = @{$sth->fetchall_arrayref({})};
+   @tables = map {
+      my %tbl; # Make a copy with lowercased keys
+      @tbl{ map { lc $_ } keys %$_ } = values %$_;
+      $tbl{engine} ||= $tbl{type} || $tbl{comment};
+      delete $tbl{type};
+      \%tbl;
+   } @tables;
+   return @tables;
 }
 
 sub _d {
@@ -2558,923 +2158,250 @@
 1;
 }
 # ###########################################################################
-# End MySQLDump package
+# End TableParser package
 # ###########################################################################
 
 # ###########################################################################
-# TableChunker 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/TableChunker.pm
-#   t/lib/TableChunker.t
+#   lib/TableNibbler.pm
+#   t/lib/TableNibbler.t
 # See https://launchpad.net/percona-toolkit for more information.
 # ###########################################################################
 {
-package TableChunker;
+package TableNibbler;
 
 use strict;
 use warnings FATAL => 'all';
 use English qw(-no_match_vars);
 use constant MKDEBUG => $ENV{MKDEBUG} || 0;
 
-use POSIX qw(floor ceil);
-use List::Util qw(min max);
-use Data::Dumper;
-$Data::Dumper::Indent    = 1;
-$Data::Dumper::Sortkeys  = 1;
-$Data::Dumper::Quotekeys = 0;
-
 sub new {
    my ( $class, %args ) = @_;
-   foreach my $arg ( qw(Quoter MySQLDump) ) {
+   my @required_args = qw(TableParser Quoter);
+   foreach my $arg ( @required_args ) {
       die "I need a $arg argument" unless $args{$arg};
    }
-
-   my %int_types  = map { $_ => 1 } qw(bigint date datetime int mediumint smallint time timestamp tinyint year);
-   my %real_types = map { $_ => 1 } qw(decimal double float);
-
-   my $self = {
-      %args,
-      int_types  => \%int_types,
-      real_types => \%real_types,
-      EPOCH      => '1970-01-01',
-   };
-
+   my $self = { %args };
    return bless $self, $class;
 }
 
-sub find_chunk_columns {
-   my ( $self, %args ) = @_;
-   foreach my $arg ( qw(tbl_struct) ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my $tbl_struct = $args{tbl_struct};
-
-   my @possible_indexes;
-   foreach my $index ( values %{ $tbl_struct->{keys} } ) {
-
-      next unless $index->{type} eq 'BTREE';
-
-      next if grep { defined } @{$index->{col_prefixes}};
-
-      if ( $args{exact} ) {
-         next unless $index->{is_unique} && @{$index->{cols}} == 1;
-      }
-
-      push @possible_indexes, $index;
-   }
-   MKDEBUG && _d('Possible chunk indexes in order:',
-      join(', ', map { $_->{name} } @possible_indexes));
-
-   my $can_chunk_exact = 0;
-   my @candidate_cols;
-   foreach my $index ( @possible_indexes ) { 
-      my $col = $index->{cols}->[0];
-
-      my $col_type = $tbl_struct->{type_for}->{$col};
-      next unless $self->{int_types}->{$col_type}
-               || $self->{real_types}->{$col_type}
-               || $col_type =~ m/char/;
-
-      push @candidate_cols, { column => $col, index => $index->{name} };
-   }
-
-   $can_chunk_exact = 1 if $args{exact} && scalar @candidate_cols;
-
-   if ( MKDEBUG ) {
-      my $chunk_type = $args{exact} ? 'Exact' : 'Inexact';
-      _d($chunk_type, 'chunkable:',
-         join(', ', map { "$_->{column} on $_->{index}" } @candidate_cols));
-   }
-
-   my @result;
-   MKDEBUG && _d('Ordering columns by order in tbl, PK first');
-   if ( $tbl_struct->{keys}->{PRIMARY} ) {
-      my $pk_first_col = $tbl_struct->{keys}->{PRIMARY}->{cols}->[0];
-      @result          = grep { $_->{column} eq $pk_first_col } @candidate_cols;
-      @candidate_cols  = grep { $_->{column} ne $pk_first_col } @candidate_cols;
-   }
-   my $i = 0;
-   my %col_pos = map { $_ => $i++ } @{$tbl_struct->{cols}};
-   push @result, sort { $col_pos{$a->{column}} <=> $col_pos{$b->{column}} }
-                    @candidate_cols;
-
-   if ( MKDEBUG ) {
-      _d('Chunkable columns:',
-         join(', ', map { "$_->{column} on $_->{index}" } @result));
-      _d('Can chunk exactly:', $can_chunk_exact);
-   }
-
-   return ($can_chunk_exact, @result);
-}
-
-sub calculate_chunks {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db tbl tbl_struct chunk_col rows_in_range chunk_size);
+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};
    }
-   MKDEBUG && _d('Calculate chunks for',
-      join(", ", map {"$_=".(defined $args{$_} ? $args{$_} : "undef")}
-         qw(db tbl chunk_col min max rows_in_range chunk_size zero_chunk exact)
-      ));
-
-   if ( !$args{rows_in_range} ) {
-      MKDEBUG && _d("Empty table");
-      return '1=1';
-   }
-
-   if ( $args{rows_in_range} < $args{chunk_size} ) {
-      MKDEBUG && _d("Chunk size larger than rows in range");
-      return '1=1';
-   }
-
-   my $q          = $self->{Quoter};
-   my $dbh        = $args{dbh};
-   my $chunk_col  = $args{chunk_col};
-   my $tbl_struct = $args{tbl_struct};
-   my $col_type   = $tbl_struct->{type_for}->{$chunk_col};
-   MKDEBUG && _d('chunk col type:', $col_type);
-
-   my %chunker;
-   if ( $tbl_struct->{is_numeric}->{$chunk_col} || $col_type =~ /date|time/ ) {
-      %chunker = $self->_chunk_numeric(%args);
-   }
-   elsif ( $col_type =~ m/char/ ) {
-      %chunker = $self->_chunk_char(%args);
-   }
-   else {
-      die "Cannot chunk $col_type columns";
-   }
-   MKDEBUG && _d("Chunker:", Dumper(\%chunker));
-   my ($col, $start_point, $end_point, $interval, $range_func)
-      = @chunker{qw(col start_point end_point interval range_func)};
-
-   my @chunks;
-   if ( $start_point < $end_point ) {
-
-      push @chunks, "$col = 0" if $chunker{have_zero_chunk};
-
-      my ($beg, $end);
-      my $iter = 0;
-      for ( my $i = $start_point; $i < $end_point; $i += $interval ) {
-         ($beg, $end) = $self->$range_func($dbh, $i, $interval, $end_point);
-
-         if ( $iter++ == 0 ) {
-            push @chunks,
-               ($chunker{have_zero_chunk} ? "$col > 0 AND " : "")
-               ."$col < " . $q->quote_val($end);
+   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};
+   MKDEBUG && _d('Will ascend index', $index);  
+
+   my @asc_cols = @{$tbl_struct->{keys}->{$index}->{cols}};
+   if ( $args{asc_first} ) {
+      @asc_cols = $asc_cols[0];
+      MKDEBUG && _d('Ascending only first column');
+   }
+   MKDEBUG && _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};
+   }
+   MKDEBUG && _d('Will ascend, in ordinal position:', join(', ', @asc_slice));
+
+   my $asc_stmt = {
+      cols  => \@cols,
+      index => $index,
+      where => '',
+      slice => [],
+      scols => [],
+   };
+
+   if ( @asc_slice ) {
+      my $cmp_where;
+      foreach my $cmp ( qw(< <= >= >) ) {
+         $cmp_where = $self->generate_cmp_where(
+            type        => $cmp,
+            slice       => \@asc_slice,
+            cols        => \@cols,
+            quoter      => $q,
+            is_nullable => $tbl_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 @chunks, "$col >= " . $q->quote_val($beg) . " AND $col < " . $q->quote_val($end);
-         }
-      }
-
-      my $chunk_range = lc $args{chunk_range} || 'open';
-      my $nullable    = $args{tbl_struct}->{is_nullable}->{$args{chunk_col}};
-      pop @chunks;
-      if ( @chunks ) {
-         push @chunks, "$col >= " . $q->quote_val($beg)
-            . ($chunk_range eq 'openclosed'
-               ? " AND $col <= " . $q->quote_val($args{max}) : "");
-      }
-      else {
-         push @chunks, $nullable ? "$col IS NOT NULL" : '1=1';
-      }
-      if ( $nullable ) {
-         push @chunks, "$col IS NULL";
-      }
-   }
-   else {
-      MKDEBUG && _d('No chunks; using single chunk 1=1');
-      push @chunks, '1=1';
-   }
-
-   return @chunks;
-}
-
-sub _chunk_numeric {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db tbl tbl_struct chunk_col rows_in_range chunk_size);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless defined $args{$arg};
-   }
-   my $q        = $self->{Quoter};
-   my $db_tbl   = $q->quote($args{db}, $args{tbl});
-   my $col_type = $args{tbl_struct}->{type_for}->{$args{chunk_col}};
-
-   my $range_func;
-   if ( $col_type =~ m/(?:int|year|float|double|decimal)$/ ) {
-      $range_func  = 'range_num';
-   }
-   elsif ( $col_type =~ m/^(?:timestamp|date|time)$/ ) {
-      $range_func  = "range_$col_type";
-   }
-   elsif ( $col_type eq 'datetime' ) {
-      $range_func  = 'range_datetime';
-   }
-
-   my ($start_point, $end_point);
-   eval {
-      $start_point = $self->value_to_number(
-         value       => $args{min},
-         column_type => $col_type,
-         dbh         => $args{dbh},
-      );
-      $end_point  = $self->value_to_number(
-         value       => $args{max},
-         column_type => $col_type,
-         dbh         => $args{dbh},
-      );
-   };
-   if ( $EVAL_ERROR ) {
-      if ( $EVAL_ERROR =~ m/don't know how to chunk/ ) {
-         die $EVAL_ERROR;
-      }
-      else {
-         die "Error calculating chunk start and end points for table "
-            . "`$args{tbl_struct}->{name}` on column `$args{chunk_col}` "
-            . "with min/max values "
-            . join('/',
-                  map { defined $args{$_} ? $args{$_} : 'undef' } qw(min max))
-            . ":\n\n"
-            . $EVAL_ERROR
-            . "\nVerify that the min and max values are valid for the column.  "
-            . "If they are valid, this error could be caused by a bug in the "
-            . "tool.";
-      }
-   }
-
-   if ( !defined $start_point ) {
-      MKDEBUG && _d('Start point is undefined');
-      $start_point = 0;
-   }
-   if ( !defined $end_point || $end_point < $start_point ) {
-      MKDEBUG && _d('End point is undefined or before start point');
-      $end_point = 0;
-   }
-   MKDEBUG && _d("Actual chunk range:", $start_point, "to", $end_point);
-
-   my $have_zero_chunk = 0;
-   if ( $args{zero_chunk} ) {
-      if ( $start_point != $end_point && $start_point >= 0 ) {
-         MKDEBUG && _d('Zero chunking');
-         my $nonzero_val = $self->get_nonzero_value(
-            %args,
-            db_tbl   => $db_tbl,
-            col      => $args{chunk_col},
-            col_type => $col_type,
-            val      => $args{min}
-         );
-         $start_point = $self->value_to_number(
-            value       => $nonzero_val,
-            column_type => $col_type,
-            dbh         => $args{dbh},
-         );
-         $have_zero_chunk = 1;
-      }
-      else {
-         MKDEBUG && _d("Cannot zero chunk");
-      }
-   }
-   MKDEBUG && _d("Using chunk range:", $start_point, "to", $end_point);
-
-   my $interval = $args{chunk_size}
-                * ($end_point - $start_point)
-                / $args{rows_in_range};
-   if ( $self->{int_types}->{$col_type} ) {
-      $interval = ceil($interval);
-   }
-   $interval ||= $args{chunk_size};
-   if ( $args{exact} ) {
-      $interval = $args{chunk_size};
-   }
-   MKDEBUG && _d('Chunk interval:', $interval, 'units');
-
-   return (
-      col             => $q->quote($args{chunk_col}),
-      start_point     => $start_point,
-      end_point       => $end_point,
-      interval        => $interval,
-      range_func      => $range_func,
-      have_zero_chunk => $have_zero_chunk,
-   );
-}
-
-sub _chunk_char {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db tbl tbl_struct chunk_col min max rows_in_range chunk_size);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless defined $args{$arg};
-   }
-   my $q         = $self->{Quoter};
-   my $db_tbl    = $q->quote($args{db}, $args{tbl});
-   my $dbh       = $args{dbh};
-   my $chunk_col = $args{chunk_col};
-   my $row;
-   my $sql;
-
-   my ($min_col, $max_col) = @{args}{qw(min max)};
-   $sql = "SELECT ORD(?) AS min_col_ord, ORD(?) AS max_col_ord";
-   MKDEBUG && _d($dbh, $sql);
-   my $ord_sth = $dbh->prepare($sql);  # avoid quoting issues
-   $ord_sth->execute($min_col, $max_col);
-   $row = $ord_sth->fetchrow_arrayref();
-   my ($min_col_ord, $max_col_ord) = ($row->[0], $row->[1]);
-   MKDEBUG && _d("Min/max col char code:", $min_col_ord, $max_col_ord);
-
-   my $base;
-   my @chars;
-   MKDEBUG && _d("Table charset:", $args{tbl_struct}->{charset});
-   if ( ($args{tbl_struct}->{charset} || "") eq "latin1" ) {
-      my @sorted_latin1_chars = (
-          32,  33,  34,  35,  36,  37,  38,  39,  40,  41,  42,  43,  44,  45,
-          46,  47,  48,  49,  50,  51,  52,  53,  54,  55,  56,  57,  58,  59,
-          60,  61,  62,  63,  64,  65,  66,  67,  68,  69,  70,  71,  72,  73,
-          74,  75,  76,  77,  78,  79,  80,  81,  82,  83,  84,  85,  86,  87,
-          88,  89,  90,  91,  92,  93,  94,  95,  96, 123, 124, 125, 126, 161,
-         162, 163, 164, 165, 166, 167, 168, 169, 170, 171, 172, 173, 174, 175,
-         176, 177, 178, 179, 180, 181, 182, 183, 184, 185, 186, 187, 188, 189,
-         190, 191, 215, 216, 222, 223, 247, 255);
-
-      my ($first_char, $last_char);
-      for my $i ( 0..$#sorted_latin1_chars ) {
-         $first_char = $i and last if $sorted_latin1_chars[$i] >= $min_col_ord;
-      }
-      for my $i ( $first_char..$#sorted_latin1_chars ) {
-         $last_char = $i and last if $sorted_latin1_chars[$i] >= $max_col_ord;
-      };
-
-      @chars = map { chr $_; } @sorted_latin1_chars[$first_char..$last_char];
-      $base  = scalar @chars;
-   }
-   else {
-
-      my $tmp_tbl    = '__maatkit_char_chunking_map';
-      my $tmp_db_tbl = $q->quote($args{db}, $tmp_tbl);
-      $sql = "DROP TABLE IF EXISTS $tmp_db_tbl";
-      MKDEBUG && _d($dbh, $sql);
-      $dbh->do($sql);
-      my $col_def = $args{tbl_struct}->{defs}->{$chunk_col};
-      $sql        = "CREATE TEMPORARY TABLE $tmp_db_tbl ($col_def) "
-                  . "ENGINE=MEMORY";
-      MKDEBUG && _d($dbh, $sql);
-      $dbh->do($sql);
-
-      $sql = "INSERT INTO $tmp_db_tbl VALUE (CHAR(?))";
-      MKDEBUG && _d($dbh, $sql);
-      my $ins_char_sth = $dbh->prepare($sql);  # avoid quoting issues
-      for my $char_code ( $min_col_ord..$max_col_ord ) {
-         $ins_char_sth->execute($char_code);
-      }
-
-      $sql = "SELECT `$chunk_col` FROM $tmp_db_tbl "
-           . "WHERE `$chunk_col` BETWEEN ? AND ? "
-           . "ORDER BY `$chunk_col`";
-      MKDEBUG && _d($dbh, $sql);
-      my $sel_char_sth = $dbh->prepare($sql);
-      $sel_char_sth->execute($min_col, $max_col);
-
-      @chars = map { $_->[0] } @{ $sel_char_sth->fetchall_arrayref() };
-      $base  = scalar @chars;
-
-      $sql = "DROP TABLE $tmp_db_tbl";
-      MKDEBUG && _d($dbh, $sql);
-      $dbh->do($sql);
-   }
-   MKDEBUG && _d("Base", $base, "chars:", @chars);
-
-
-   $sql = "SELECT MAX(LENGTH($chunk_col)) FROM $db_tbl "
-        . ($args{where} ? "WHERE $args{where} " : "") 
-        . "ORDER BY `$chunk_col`";
-   MKDEBUG && _d($dbh, $sql);
-   $row = $dbh->selectrow_arrayref($sql);
-   my $max_col_len = $row->[0];
-   MKDEBUG && _d("Max column value:", $max_col, $max_col_len);
-   my $n_values;
-   for my $n_chars ( 1..$max_col_len ) {
-      $n_values = $base**$n_chars;
-      if ( $n_values >= $args{chunk_size} ) {
-         MKDEBUG && _d($n_chars, "chars in base", $base, "expresses",
-            $n_values, "values");
-         last;
-      }
-   }
-
-   my $n_chunks = $args{rows_in_range} / $args{chunk_size};
-   my $interval = floor($n_values / $n_chunks) || 1;
-
-   my $range_func = sub {
-      my ( $self, $dbh, $start, $interval, $max ) = @_;
-      my $start_char = $self->base_count(
-         count_to => $start,
-         base     => $base,
-         symbols  => \@chars,
-      );
-      my $end_char = $self->base_count(
-         count_to => min($max, $start + $interval),
-         base     => $base,
-         symbols  => \@chars,
-      );
-      return $start_char, $end_char;
-   };
-
-   return (
-      col         => $q->quote($chunk_col),
-      start_point => 0,
-      end_point   => $n_values,
-      interval    => $interval,
-      range_func  => $range_func,
-   );
-}
-
-sub get_first_chunkable_column {
-   my ( $self, %args ) = @_;
-   foreach my $arg ( qw(tbl_struct) ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-
-   my ($exact, @cols) = $self->find_chunk_columns(%args);
-   my $col = $cols[0]->{column};
-   my $idx = $cols[0]->{index};
-
-   my $wanted_col = $args{chunk_column};
-   my $wanted_idx = $args{chunk_index};
-   MKDEBUG && _d("Preferred chunk col/idx:", $wanted_col, $wanted_idx);
-
-   if ( $wanted_col && $wanted_idx ) {
-      foreach my $chunkable_col ( @cols ) {
-         if (    $wanted_col eq $chunkable_col->{column}
-              && $wanted_idx eq $chunkable_col->{index} ) {
-            $col = $wanted_col;
-            $idx = $wanted_idx;
-            last;
-         }
-      }
-   }
-   elsif ( $wanted_col ) {
-      foreach my $chunkable_col ( @cols ) {
-         if ( $wanted_col eq $chunkable_col->{column} ) {
-            $col = $wanted_col;
-            $idx = $chunkable_col->{index};
-            last;
-         }
-      }
-   }
-   elsif ( $wanted_idx ) {
-      foreach my $chunkable_col ( @cols ) {
-         if ( $wanted_idx eq $chunkable_col->{index} ) {
-            $col = $chunkable_col->{column};
-            $idx = $wanted_idx;
-            last;
-         }
-      }
-   }
-
-   MKDEBUG && _d('First chunkable col/index:', $col, $idx);
-   return $col, $idx;
-}
-
-sub size_to_rows {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db tbl chunk_size);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $db, $tbl, $chunk_size) = @args{@required_args};
-   my $q  = $self->{Quoter};
-   my $du = $self->{MySQLDump};
-
-   my ($n_rows, $avg_row_length);
-
-   my ( $num, $suffix ) = $chunk_size =~ m/^(\d+)([MGk])?$/;
-   if ( $suffix ) { # Convert to bytes.
-      $chunk_size = $suffix eq 'k' ? $num * 1_024
-                  : $suffix eq 'M' ? $num * 1_024 * 1_024
-                  :                  $num * 1_024 * 1_024 * 1_024;
-   }
-   elsif ( $num ) {
-      $n_rows = $num;
-   }
-   else {
-      die "Invalid chunk size $chunk_size; must be an integer "
-         . "with optional suffix kMG";
-   }
-
-   if ( $suffix || $args{avg_row_length} ) {
-      my ($status) = $du->get_table_status($dbh, $q, $db, $tbl);
-      $avg_row_length = $status->{avg_row_length};
-      if ( !defined $n_rows ) {
-         $n_rows = $avg_row_length ? ceil($chunk_size / $avg_row_length) : undef;
-      }
-   }
-
-   return $n_rows, $avg_row_length;
-}
-
-sub get_range_statistics {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db tbl chunk_col tbl_struct);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $db, $tbl, $col) = @args{@required_args};
-   my $where = $args{where};
-   my $q     = $self->{Quoter};
-
-   my $col_type       = $args{tbl_struct}->{type_for}->{$col};
-   my $col_is_numeric = $args{tbl_struct}->{is_numeric}->{$col};
-
-   my $db_tbl = $q->quote($db, $tbl);
-   $col       = $q->quote($col);
-
-   my ($min, $max);
-   eval {
-      my $sql = "SELECT MIN($col), MAX($col) FROM $db_tbl"
-              . ($args{index_hint} ? " $args{index_hint}" : "")
-              . ($where ? " WHERE ($where)" : '');
-      MKDEBUG && _d($dbh, $sql);
-      ($min, $max) = $dbh->selectrow_array($sql);
-      MKDEBUG && _d("Actual end points:", $min, $max);
-
-      ($min, $max) = $self->get_valid_end_points(
-         %args,
-         dbh      => $dbh,
-         db_tbl   => $db_tbl,
-         col      => $col,
-         col_type => $col_type,
-         min      => $min,
-         max      => $max,
-      );
-      MKDEBUG && _d("Valid end points:", $min, $max);
-   };
-   if ( $EVAL_ERROR ) {
-      die "Error getting min and max values for table $db_tbl "
-         . "on column $col: $EVAL_ERROR";
-   }
-
-   my $sql = "EXPLAIN SELECT * FROM $db_tbl"
-           . ($args{index_hint} ? " $args{index_hint}" : "")
-           . ($where ? " WHERE $where" : '');
-   MKDEBUG && _d($sql);
-   my $expl = $dbh->selectrow_hashref($sql);
-
-   return (
-      min           => $min,
-      max           => $max,
-      rows_in_range => $expl->{rows},
-   );
-}
-
-sub inject_chunks {
-   my ( $self, %args ) = @_;
-   foreach my $arg ( qw(database table chunks chunk_num query) ) {
-      die "I need a $arg argument" unless defined $args{$arg};
-   }
-   MKDEBUG && _d('Injecting chunk', $args{chunk_num});
-   my $query   = $args{query};
-   my $comment = sprintf("/*%s.%s:%d/%d*/",
-      $args{database}, $args{table},
-      $args{chunk_num} + 1, scalar @{$args{chunks}});
-   $query =~ s!/\*PROGRESS_COMMENT\*/!$comment!;
-   my $where = "WHERE (" . $args{chunks}->[$args{chunk_num}] . ')';
-   if ( $args{where} && grep { $_ } @{$args{where}} ) {
-      $where .= " AND ("
-         . join(" AND ", map { "($_)" } grep { $_ } @{$args{where}} )
-         . ")";
-   }
-   my $db_tbl     = $self->{Quoter}->quote(@args{qw(database table)});
-   my $index_hint = $args{index_hint} || '';
-
-   MKDEBUG && _d('Parameters:',
-      Dumper({WHERE => $where, DB_TBL => $db_tbl, INDEX_HINT => $index_hint}));
-   $query =~ s!/\*WHERE\*/! $where!;
-   $query =~ s!/\*DB_TBL\*/!$db_tbl!;
-   $query =~ s!/\*INDEX_HINT\*/! $index_hint!;
-   $query =~ s!/\*CHUNK_NUM\*/! $args{chunk_num} AS chunk_num,!;
-
-   return $query;
-}
-
-
-sub value_to_number {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(column_type dbh);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless defined $args{$arg};
-   }
-   my $val = $args{value};
-   my ($col_type, $dbh) = @args{@required_args};
-   MKDEBUG && _d('Converting MySQL', $col_type, $val);
-
-   return unless defined $val;  # value is NULL
-
-   my %mysql_conv_func_for = (
-      timestamp => 'UNIX_TIMESTAMP',
-      date      => 'TO_DAYS',
-      time      => 'TIME_TO_SEC',
-      datetime  => 'TO_DAYS',
-   );
-
-   my $num;
-   if ( $col_type =~ m/(?:int|year|float|double|decimal)$/ ) {
-      $num = $val;
-   }
-   elsif ( $col_type =~ m/^(?:timestamp|date|time)$/ ) {
-      my $func = $mysql_conv_func_for{$col_type};
-      my $sql = "SELECT $func(?)";
-      MKDEBUG && _d($dbh, $sql, $val);
-      my $sth = $dbh->prepare($sql);
-      $sth->execute($val);
-      ($num) = $sth->fetchrow_array();
-   }
-   elsif ( $col_type eq 'datetime' ) {
-      $num = $self->timestampdiff($dbh, $val);
-   }
-   else {
-      die "I don't know how to chunk $col_type\n";
-   }
-   MKDEBUG && _d('Converts to', $num);
-   return $num;
-}
-
-sub range_num {
-   my ( $self, $dbh, $start, $interval, $max ) = @_;
-   my $end = min($max, $start + $interval);
-
-
-   $start = sprintf('%.17f', $start) if $start =~ /e/;
-   $end   = sprintf('%.17f', $end)   if $end   =~ /e/;
-
-   $start =~ s/\.(\d{5}).*$/.$1/;
-   $end   =~ s/\.(\d{5}).*$/.$1/;
-
-   if ( $end > $start ) {
-      return ( $start, $end );
-   }
-   else {
-      die "Chunk size is too small: $end !> $start\n";
-   }
-}
-
-sub range_time {
-   my ( $self, $dbh, $start, $interval, $max ) = @_;
-   my $sql = "SELECT SEC_TO_TIME($start), SEC_TO_TIME(LEAST($max, $start + $interval))";
-   MKDEBUG && _d($sql);
-   return $dbh->selectrow_array($sql);
-}
-
-sub range_date {
-   my ( $self, $dbh, $start, $interval, $max ) = @_;
-   my $sql = "SELECT FROM_DAYS($start), FROM_DAYS(LEAST($max, $start + $interval))";
-   MKDEBUG && _d($sql);
-   return $dbh->selectrow_array($sql);
-}
-
-sub range_datetime {
-   my ( $self, $dbh, $start, $interval, $max ) = @_;
-   my $sql = "SELECT DATE_ADD('$self->{EPOCH}', INTERVAL $start SECOND), "
-       . "DATE_ADD('$self->{EPOCH}', INTERVAL LEAST($max, $start + $interval) SECOND)";
-   MKDEBUG && _d($sql);
-   return $dbh->selectrow_array($sql);
-}
-
-sub range_timestamp {
-   my ( $self, $dbh, $start, $interval, $max ) = @_;
-   my $sql = "SELECT FROM_UNIXTIME($start), FROM_UNIXTIME(LEAST($max, $start + $interval))";
-   MKDEBUG && _d($sql);
-   return $dbh->selectrow_array($sql);
-}
-
-sub timestampdiff {
-   my ( $self, $dbh, $time ) = @_;
-   my $sql = "SELECT (COALESCE(TO_DAYS('$time'), 0) * 86400 + TIME_TO_SEC('$time')) "
-      . "- TO_DAYS('$self->{EPOCH} 00:00:00') * 86400";
-   MKDEBUG && _d($sql);
-   my ( $diff ) = $dbh->selectrow_array($sql);
-   $sql = "SELECT DATE_ADD('$self->{EPOCH}', INTERVAL $diff SECOND)";
-   MKDEBUG && _d($sql);
-   my ( $check ) = $dbh->selectrow_array($sql);
-   die <<"   EOF"
-   Incorrect datetime math: given $time, calculated $diff but checked to $check.
-   This could be due to a version of MySQL that overflows on large interval
-   values to DATE_ADD(), or the given datetime is not a valid date.  If not,
-   please report this as a bug.
-   EOF
-      unless $check eq $time;
-   return $diff;
-}
-
-
-
-
-sub get_valid_end_points {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db_tbl col col_type);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $db_tbl, $col, $col_type) = @args{@required_args};
-   my ($real_min, $real_max)           = @args{qw(min max)};
-
-   my $err_fmt = "Error finding a valid %s value for table $db_tbl on "
-               . "column $col. The real %s value %s is invalid and "
-               . "no other valid values were found.  Verify that the table "
-               . "has at least one valid value for this column"
-               . ($args{where} ? " where $args{where}." : ".");
-
-   my $valid_min = $real_min;
-   if ( defined $valid_min ) {
-      MKDEBUG && _d("Validating min end point:", $real_min);
-      $valid_min = $self->_get_valid_end_point(
-         %args,
-         val      => $real_min,
-         endpoint => 'min',
-      );
-      die sprintf($err_fmt, 'minimum', 'minimum',
-         (defined $real_min ? $real_min : "NULL"))
-         unless defined $valid_min;
-   }
-
-   my $valid_max = $real_max;
-   if ( defined $valid_max ) {
-      MKDEBUG && _d("Validating max end point:", $real_min);
-      $valid_max = $self->_get_valid_end_point(
-         %args,
-         val      => $real_max,
-         endpoint => 'max',
-      );
-      die sprintf($err_fmt, 'maximum', 'maximum',
-         (defined $real_max ? $real_max : "NULL"))
-         unless defined $valid_max;
-   }
-
-   return $valid_min, $valid_max;
-}
-
-sub _get_valid_end_point {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db_tbl col col_type);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $db_tbl, $col, $col_type) = @args{@required_args};
-   my $val = $args{val};
-
-   return $val unless defined $val;
-
-   my $validate = $col_type =~ m/time|date/ ? \&_validate_temporal_value
-                :                             undef;
-
-   if ( !$validate ) {
-      MKDEBUG && _d("No validator for", $col_type, "values");
-      return $val;
-   }
-
-   return $val if defined $validate->($dbh, $val);
-
-   MKDEBUG && _d("Value is invalid, getting first valid value");
-   $val = $self->get_first_valid_value(
-      %args,
-      val      => $val,
-      validate => $validate,
-   );
-
-   return $val;
-}
-
-sub get_first_valid_value {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db_tbl col validate endpoint);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $db_tbl, $col, $validate, $endpoint) = @args{@required_args};
-   my $tries = defined $args{tries} ? $args{tries} : 5;
-   my $val   = $args{val};
-
-   return unless defined $val;
-
-   my $cmp = $endpoint =~ m/min/i ? '>'
-           : $endpoint =~ m/max/i ? '<'
-           :                        die "Invalid endpoint arg: $endpoint";
-   my $sql = "SELECT $col FROM $db_tbl "
-           . ($args{index_hint} ? "$args{index_hint} " : "")
-           . "WHERE $col $cmp ? AND $col IS NOT NULL "
-           . ($args{where} ? "AND ($args{where}) " : "")
-           . "ORDER BY $col LIMIT 1";
-   MKDEBUG && _d($dbh, $sql);
-   my $sth = $dbh->prepare($sql);
-
-   my $last_val = $val;
-   while ( $tries-- ) {
-      $sth->execute($last_val);
-      my ($next_val) = $sth->fetchrow_array();
-      MKDEBUG && _d('Next value:', $next_val, '; tries left:', $tries);
-      if ( !defined $next_val ) {
-         MKDEBUG && _d('No more rows in table');
-         last;
-      }
-      if ( defined $validate->($dbh, $next_val) ) {
-         MKDEBUG && _d('First valid value:', $next_val);
-         $sth->finish();
-         return $next_val;
-      }
-      $last_val = $next_val;
-   }
-   $sth->finish();
-   $val = undef;  # no valid value found
-
-   return $val;
-}
-
-sub _validate_temporal_value {
-   my ( $dbh, $val ) = @_;
-   my $sql = "SELECT IF(TIME_FORMAT(?,'%H:%i:%s')=?, TIME_TO_SEC(?), TO_DAYS(?))";
-   my $res;
-   eval {
-      MKDEBUG && _d($dbh, $sql, $val);
-      my $sth = $dbh->prepare($sql);
-      $sth->execute($val, $val, $val, $val);
-      ($res) = $sth->fetchrow_array();
-      $sth->finish();
-   };
-   if ( $EVAL_ERROR ) {
-      MKDEBUG && _d($EVAL_ERROR);
-   }
-   return $res;
-}
-
-sub get_nonzero_value {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(dbh db_tbl col col_type);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my ($dbh, $db_tbl, $col, $col_type) = @args{@required_args};
-   my $tries = defined $args{tries} ? $args{tries} : 5;
-   my $val   = $args{val};
-
-   my $is_nonzero = $col_type =~ m/time|date/ ? \&_validate_temporal_value
-                  :                             sub { return $_[1]; };
-
-   if ( !$is_nonzero->($dbh, $val) ) {  # quasi-double-negative, sorry
-      MKDEBUG && _d('Discarding zero value:', $val);
-      my $sql = "SELECT $col FROM $db_tbl "
-              . ($args{index_hint} ? "$args{index_hint} " : "")
-              . "WHERE $col > ? AND $col IS NOT NULL "
-              . ($args{where} ? "AND ($args{where}) " : '')
-              . "ORDER BY $col LIMIT 1";
-      MKDEBUG && _d($sql);
-      my $sth = $dbh->prepare($sql);
-
-      my $last_val = $val;
-      while ( $tries-- ) {
-         $sth->execute($last_val);
-         my ($next_val) = $sth->fetchrow_array();
-         if ( $is_nonzero->($dbh, $next_val) ) {
-            MKDEBUG && _d('First non-zero value:', $next_val);
-            $sth->finish();
-            return $next_val;
-         }
-         $last_val = $next_val;
-      }
-      $sth->finish();
-      $val = undef;  # no non-zero value found
-   }
-
-   return $val;
-}
-
-sub base_count {
-   my ( $self, %args ) = @_;
-   my @required_args = qw(count_to base symbols);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless defined $args{$arg};
-   }
-   my ($n, $base, $symbols) = @args{@required_args};
-
-   return $symbols->[0] if $n == 0;
-
-   my $highest_power = floor(log($n)/log($base));
-   if ( $highest_power == 0 ){
-      return $symbols->[$n];
-   }
-
-   my @base_powers;
-   for my $power ( 0..$highest_power ) {
-      push @base_powers, ($base**$power) || 1;  
-   }
-
-   my @base_multiples;
-   foreach my $base_power ( reverse @base_powers ) {
-      my $multiples = floor($n / $base_power);
-      push @base_multiples, $multiples;
-      $n -= $multiples * $base_power;
-   }
-
-   return join('', map { $symbols->[$_] } @base_multiples);
+            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}};
+   }
+   MKDEBUG && _d('Columns needed for DELETE:', join(', ', @del_cols));
+
+   my %col_posn = do { my $i = 0; map { $_ => $i++ } @cols };
+   foreach my $col ( @del_cols ) {
+      if ( !exists $col_posn{$col} ) {
+         push @cols, $col;
+         $col_posn{$col} = $#cols;
+      }
+      push @del_slice, $col_posn{$col};
+   }
+   MKDEBUG && _d('Ordinals needed for DELETE:', join(', ', @del_slice));
+
+   my $del_stmt = {
+      cols  => \@cols,
+      index => $index,
+      where => '',
+      slice => [],
+      scols => [],
+   };
+
+   my @clauses;
+   foreach my $i ( 0 .. $#del_slice ) {
+      my $ord = $del_slice[$i];
+      my $col = $cols[$ord];
+      my $quo = $q->quote($col);
+      if ( $tbl->{is_nullable}->{$col} ) {
+         push @clauses, "((? IS NULL AND $quo IS NULL) OR ($quo = ?))";
+         push @{$del_stmt->{slice}}, $ord, $ord;
+         push @{$del_stmt->{scols}}, $col, $col;
+      }
+      else {
+         push @clauses, "$quo = ?";
+         push @{$del_stmt->{slice}}, $ord;
+         push @{$del_stmt->{scols}}, $col;
+      }
+   }
+
+   $del_stmt->{where} = '(' . join(' AND ', @clauses) . ')';
+
+   return $del_stmt;
+}
+
+sub generate_ins_stmt {
+   my ( $self, %args ) = @_;
+   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 {
@@ -3488,84 +2415,7 @@
 1;
 }
 # ###########################################################################
-# End TableChunker package
-# ###########################################################################
-
-# ###########################################################################
-# Quoter 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/Quoter.pm
-#   t/lib/Quoter.t
-# See https://launchpad.net/percona-toolkit for more information.
-# ###########################################################################
-{
-package Quoter;
-
-use strict;
-use warnings FATAL => 'all';
-use English qw(-no_match_vars);
-use constant MKDEBUG => $ENV{MKDEBUG} || 0;
-
-sub new {
-   my ( $class, %args ) = @_;
-   return bless {}, $class;
-}
-
-sub quote {
-   my ( $self, @vals ) = @_;
-   foreach my $val ( @vals ) {
-      $val =~ s/`/``/g;
-   }
-   return join('.', map { '`' . $_ . '`' } @vals);
-}
-
-sub quote_val {
-   my ( $self, $val ) = @_;
-
-   return 'NULL' unless defined $val;          # undef = NULL
-   return "''" if $val eq '';                  # blank string = ''
-   return $val if $val =~ m/^0x[0-9a-fA-F]+$/;  # hex data
-
-   $val =~ s/(['\\])/\\$1/g;
-   return "'$val'";
-}
-
-sub split_unquote {
-   my ( $self, $db_tbl, $default_db ) = @_;
-   $db_tbl =~ s/`//g;
-   my ( $db, $tbl ) = split(/[.]/, $db_tbl);
-   if ( !$tbl ) {
-      $tbl = $db;
-      $db  = $default_db;
-   }
-   return ($db, $tbl);
-}
-
-sub literal_like {
-   my ( $self, $like ) = @_;
-   return unless $like;
-   $like =~ s/([%_])/\\$1/g;
-   return "'$like'";
-}
-
-sub join_quote {
-   my ( $self, $default_db, $db_tbl ) = @_;
-   return unless $db_tbl;
-   my ($db, $tbl) = split(/[.]/, $db_tbl);
-   if ( !$tbl ) {
-      $tbl = $db;
-      $db  = $default_db;
-   }
-   $db  = "`$db`"  if $db  && $db  !~ m/^`/;
-   $tbl = "`$tbl`" if $tbl && $tbl !~ m/^`/;
-   return $db ? "$db.$tbl" : $tbl;
-}
-
-1;
-}
-# ###########################################################################
-# End Quoter package
+# End TableNibbler package
 # ###########################################################################
 
 # ###########################################################################
@@ -3593,6 +2443,54 @@
    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');
+   MKDEBUG && _d('Slave recursion method:', $method);
+   if ( !$method || $method =~ m/proocesslist|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;
+               MKDEBUG && _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 --recusion-method: $method.  Valid values are: "
+        . "dsn=DSN, hosts, or processlist.\n";
+   }
+
+   return $slaves;
+}
+
 sub recurse_to_slaves {
    my ( $self, $args, $level ) = @_;
    $level ||= 0;
@@ -4169,6 +3067,43 @@
    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};
+   MKDEBUG && _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";
+   MKDEBUG && _d($sql);
+   my $dsn_strings = $dbh->selectcol_arrayref($sql);
+   my @cxn;
+   if ( $dsn_strings ) {
+      foreach my $dsn_string ( @$dsn_strings ) {
+         MKDEBUG && _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; }
@@ -4184,6 +3119,1206 @@
 # ###########################################################################
 
 # ###########################################################################
+# RowChecksum 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/RowChecksum.pm
+#   t/lib/RowChecksum.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package RowChecksum;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant MKDEBUG => $ENV{MKDEBUG} || 0;
+
+use List::Util qw(max);
+use Data::Dumper;
+$Data::Dumper::Indent    = 1;
+$Data::Dumper::Sortkeys  = 1;
+$Data::Dumper::Quotekeys = 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   foreach my $arg ( qw(OptionParser Quoter) ) {
+      die "I need a $arg argument" unless defined $args{$arg};
+   }
+   my $self = { %args };
+   return bless $self, $class;
+}
+
+sub make_row_checksum {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(tbl);
+   foreach my $arg( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($tbl) = @args{@required_args};
+
+   my $o          = $self->{OptionParser};
+   my $q          = $self->{Quoter};
+   my $tbl_struct = $tbl->{tbl_struct};
+   my $func       = $args{func} || uc($o->get('function'));
+   my $cols       = $self->get_checksum_columns(%args);
+
+   my $query;
+   if ( !$args{no_cols} ) {
+      $query = join(', ',
+                  map { 
+                     my $col = $_;
+                     if ( $col =~ m/\+ 0/ ) {
+                        my ($real_col) = /^(\S+)/;
+                        $col .= " AS $real_col";
+                     }
+                     elsif ( $col =~ m/TRIM/ ) {
+                        my ($real_col) = m/TRIM\(([^\)]+)\)/;
+                        $col .= " AS $real_col";
+                     }
+                     $col;
+                  } @{$cols->{select}})
+             . ', ';
+   }
+
+   if ( uc $func ne 'FNV_64' && uc $func ne 'FNV1A_64' ) {
+      my $sep = $o->get('separator') || '#';
+      $sep    =~ s/'//g;
+      $sep  ||= '#';
+
+      my @nulls = grep { $cols->{allowed}->{$_} } @{$tbl_struct->{null_cols}};
+      if ( @nulls ) {
+         my $bitmap = "CONCAT("
+            . join(', ', map { 'ISNULL(' . $q->quote($_) . ')' } @nulls)
+            . ")";
+         push @{$cols->{select}}, $bitmap;
+      }
+
+      $query .= @{$cols->{select}} > 1
+              ? "$func(CONCAT_WS('$sep', " . join(', ', @{$cols->{select}}) . '))'
+              : "$func($cols->{select}->[0])";
+   }
+   else {
+      my $fnv_func = uc $func;
+      $query .= "$fnv_func(" . join(', ', @{$cols->{select}}) . ')';
+   }
+
+   MKDEBUG && _d('Row checksum:', $query);
+   return $query;
+}
+
+sub make_chunk_checksum {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(tbl);
+   foreach my $arg( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   if ( !$args{dbh} && !($args{func} && $args{crc_width} && $args{crc_type}) ) {
+      die "I need a dbh argument"
+   }
+   my ($tbl) = @args{@required_args};
+   my $o     = $self->{OptionParser};
+   my $q     = $self->{Quoter};
+
+   my %crc_args = $self->get_crc_args(%args);
+   MKDEBUG && _d("Checksum strat:", Dumper(\%crc_args));
+
+   my $row_checksum = $self->make_row_checksum(
+      %args,
+      %crc_args,
+      no_cols => 1
+   );
+   my $crc;
+   if ( $crc_args{crc_type} =~ m/int$/ ) {
+      $crc = "COALESCE(LOWER(CONV(BIT_XOR(CAST($row_checksum AS UNSIGNED)), "
+           . "10, 16)), 0)";
+   }
+   else {
+      my $slices = $self->_make_xor_slices(
+         row_checksum => $row_checksum,
+         %crc_args,
+      );
+      $crc = "COALESCE(LOWER(CONCAT($slices)), 0)";
+   }
+
+   my $select = "COUNT(*) AS cnt, $crc AS crc";
+   MKDEBUG && _d('Chunk checksum:', $select);
+   return $select;
+}
+
+sub get_checksum_columns {
+   my ($self, %args) = @_;
+   my @required_args = qw(tbl);
+   foreach my $arg( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($tbl) = @args{@required_args};
+   my $o     = $self->{OptionParser};
+   my $q     = $self->{Quoter};
+
+   my $trim            = $o->get('trim');
+   my $float_precision = $o->get('float-precision');
+
+   my $tbl_struct = $tbl->{tbl_struct};
+   my $ignore_col = $o->get('ignore-columns') || {};
+   my $all_cols   = $o->get('columns') || $tbl_struct->{cols};
+   my %cols       = map { lc($_) => 1 } grep { !$ignore_col->{$_} } @$all_cols;
+   my %seen;
+   my @cols =
+      map {
+         my $type   = $tbl_struct->{type_for}->{$_};
+         my $result = $q->quote($_);
+         if ( $type eq 'timestamp' ) {
+            $result .= ' + 0';
+         }
+         elsif ( $float_precision && $type =~ m/float|double/ ) {
+            $result = "ROUND($result, $float_precision)";
+         }
+         elsif ( $trim && $type =~ m/varchar/ ) {
+            $result = "TRIM($result)";
+         }
+         $result;
+      }
+      grep {
+         $cols{$_} && !$seen{$_}++
+      }
+      @{$tbl_struct->{cols}};
+
+   return {
+      select  => \@cols,
+      allowed => \%cols,
+   };
+}
+
+sub get_crc_args {
+   my ($self, %args) = @_;
+   my $func      = $args{func}     || $self->_get_hash_func(%args);
+   my $crc_width = $args{crc_width}|| $self->_get_crc_width(%args, func=>$func);
+   my $crc_type  = $args{crc_type} || $self->_get_crc_type(%args, func=>$func);
+   my $opt_slice; 
+   if ( $args{dbh} && $crc_type !~ m/int$/ ) {
+      $opt_slice = $self->_optimize_xor(%args, func=>$func);
+   }
+
+   return (
+      func      => $func,
+      crc_width => $crc_width,
+      crc_type  => $crc_type,
+      opt_slice => $opt_slice,
+   );
+}
+
+sub _get_hash_func {
+   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 $o     = $self->{OptionParser};
+   my @funcs = qw(CRC32 FNV1A_64 FNV_64 MD5 SHA1);
+
+   if ( my $func = $o->get('function') ) {
+      unshift @funcs, $func;
+   }
+
+   my ($result, $error);
+   foreach my $func ( @funcs ) {
+      eval {
+         my $sql = "SELECT $func('test-string')";
+         MKDEBUG && _d($sql);
+         $args{dbh}->do($sql);
+      };
+      if ( $EVAL_ERROR && $EVAL_ERROR =~ m/failed: (.*?) at \S+ line/ ) {
+         $error .= qq{$func cannot be used because "$1"\n};
+         MKDEBUG && _d($func, 'cannot be used because', $1);
+      }
+      MKDEBUG && _d('Chosen hash func:', $result);
+      return $func;
+   }
+   die $error || 'No hash functions (CRC32, MD5, etc.) are available';
+}
+
+sub _get_crc_width {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(dbh func);
+   foreach my $arg( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($dbh, $func) = @args{@required_args};
+
+   my $crc_width = 16;
+   if ( uc $func ne 'FNV_64' && uc $func ne 'FNV1A_64' ) {
+      eval {
+         my ($val) = $dbh->selectrow_array("SELECT $func('a')");
+         $crc_width = max(16, length($val));
+      };
+   }
+   return $crc_width;
+}
+
+sub _get_crc_type {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(dbh func);
+   foreach my $arg( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($dbh, $func) = @args{@required_args};
+
+   my $type   = '';
+   my $length = 0;
+   my $sql    = "SELECT $func('a')";
+   my $sth    = $dbh->prepare($sql);
+   eval {
+      $sth->execute();
+      $type   = $sth->{mysql_type_name}->[0];
+      $length = $sth->{mysql_length}->[0];
+      MKDEBUG && _d($sql, $type, $length);
+      if ( $type eq 'bigint' && $length < 20 ) {
+         $type = 'int';
+      }
+   };
+   $sth->finish;
+   MKDEBUG && _d('crc_type:', $type, 'length:', $length);
+   return $type;
+}
+
+sub _optimize_xor {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(dbh func);
+   foreach my $arg( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($dbh, $func) = @args{@required_args};
+
+   die "$func never needs BIT_XOR optimization"
+      if $func =~ m/^(?:FNV1A_64|FNV_64|CRC32)$/i;
+
+   my $opt_slice = 0;
+   my $unsliced  = uc $dbh->selectall_arrayref("SELECT $func('a')")->[0]->[0];
+   my $sliced    = '';
+   my $start     = 1;
+   my $crc_width = length($unsliced) < 16 ? 16 : length($unsliced);
+
+   do { # Try different positions till sliced result equals non-sliced.
+      MKDEBUG && _d('Trying slice', $opt_slice);
+      $dbh->do('SET @crc := "", @cnt := 0');
+      my $slices = $self->_make_xor_slices(
+         row_checksum => "\@crc := $func('a')",
+         crc_width    => $crc_width,
+         opt_slice    => $opt_slice,
+      );
+
+      my $sql = "SELECT CONCAT($slices) AS TEST FROM (SELECT NULL) AS x";
+      $sliced = ($dbh->selectrow_array($sql))[0];
+      if ( $sliced ne $unsliced ) {
+         MKDEBUG && _d('Slice', $opt_slice, 'does not work');
+         $start += 16;
+         ++$opt_slice;
+      }
+   } while ( $start < $crc_width && $sliced ne $unsliced );
+
+   if ( $sliced eq $unsliced ) {
+      MKDEBUG && _d('Slice', $opt_slice, 'works');
+      return $opt_slice;
+   }
+   else {
+      MKDEBUG && _d('No slice works');
+      return undef;
+   }
+}
+
+sub _make_xor_slices {
+   my ( $self, %args ) = @_;
+   my @required_args = qw(row_checksum crc_width);
+   foreach my $arg( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($row_checksum, $crc_width) = @args{@required_args};
+   my ($opt_slice) = $args{opt_slice};
+
+   my @slices;
+   for ( my $start = 1; $start <= $crc_width; $start += 16 ) {
+      my $len = $crc_width - $start + 1;
+      if ( $len > 16 ) {
+         $len = 16;
+      }
+      push @slices,
+         "LPAD(CONV(BIT_XOR("
+         . "CAST(CONV(SUBSTRING(\@crc, $start, $len), 16, 10) AS UNSIGNED))"
+         . ", 10, 16), $len, '0')";
+   }
+
+   if ( defined $opt_slice && $opt_slice < @slices ) {
+      $slices[$opt_slice] =~ s/\@crc/\@crc := $row_checksum/;
+   }
+   else {
+      map { s/\@crc/$row_checksum/ } @slices;
+   }
+
+   return join(', ', @slices);
+}
+
+sub find_replication_differences {
+   my ($self, %args) = @_;
+   my @required_args = qw(dbh repl_table);
+   foreach my $arg( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my ($dbh, $repl_table) = @args{@required_args};
+
+   my $sql
+      = "SELECT CONCAT(db, '.', tbl) AS `table`, "
+      . "chunk, chunk_index, lower_boundary, upper_boundary, "
+      . "COALESCE(this_cnt-master_cnt, 0) AS cnt_diff, "
+      . "COALESCE("
+      .   "this_crc <> master_crc OR ISNULL(master_crc) <> ISNULL(this_crc), 0"
+      . ") AS crc_diff, this_cnt, master_cnt, this_crc, master_crc "
+      . "FROM $repl_table "
+      . "WHERE (master_cnt <> this_cnt OR master_crc <> this_crc "
+      .        "OR ISNULL(master_crc) <> ISNULL(this_crc))"
+      . ($args{where} ? " AND ($args{where})" : "");
+   MKDEBUG && _d($sql);
+   my $diffs = $dbh->selectall_arrayref($sql, { Slice => {} });
+   return $diffs;
+}
+
+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 RowChecksum 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 MKDEBUG => $ENV{MKDEBUG} || 0;
+
+use Data::Dumper;
+$Data::Dumper::Indent    = 1;
+$Data::Dumper::Sortkeys  = 1;
+$Data::Dumper::Quotekeys = 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   my @required_args = qw(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 $where = $o->get('where');
+   my ($row_est, $mysql_index) = get_row_estimate(%args, where => $where);
+   my $one_nibble = !defined $args{one_nibble} || $args{one_nibble}
+                  ? $row_est <= $chunk_size * $o->get('chunk-size-limit')
+                  : 0;
+   MKDEBUG && _d('One nibble:', $one_nibble ? 'yes' : 'no');
+
+   if ( $args{resume}
+        && !defined $args{resume}->{lower_boundary}
+        && !defined $args{resume}->{upper_boundary} ) {
+      MKDEBUG && _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.";
+   }
+
+   my $tbl_struct = $tbl->{tbl_struct};
+   my $ignore_col = $o->get('ignore-columns') || {};
+   my $all_cols   = $o->get('columns') || $tbl_struct->{cols};
+   my @cols       = grep { !$ignore_col->{$_} } @$all_cols;
+   my $self;
+   if ( $one_nibble ) {
+      my $nibble_sql
+         = ($args{dml} ? "$args{dml} " : "SELECT ")
+         . ($args{select} ? $args{select}
+                          : join(', ', map { $q->quote($_) } @cols))
+         . " FROM " . $q->quote(@{$tbl}{qw(db tbl)})
+         . ($where ? " WHERE $where" : '')
+         . " /*checksum table*/";
+      MKDEBUG && _d('One nibble statement:', $nibble_sql);
+
+      my $explain_nibble_sql
+         = "EXPLAIN SELECT "
+         . ($args{select} ? $args{select}
+                          : join(', ', map { $q->quote($_) } @cols))
+         . " FROM " . $q->quote(@{$tbl}{qw(db tbl)})
+         . ($where ? " WHERE $where" : '')
+         . " /*explain checksum table*/";
+      MKDEBUG && _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_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,
+      );
+      MKDEBUG && _d('Ascend params:', Dumper($asc));
+
+      my $from     = $q->quote(@{$tbl}{qw(db tbl)}) . " 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*/";
+      MKDEBUG && _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*/";
+         MKDEBUG && _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*/";
+      MKDEBUG && _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*/";
+      MKDEBUG && _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" : "")
+         . " /*checksum chunk*/";
+      MKDEBUG && _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 checksum chunk*/";
+      MKDEBUG && _d('Explain nibble statement:', $explain_nibble_sql);
+
+      my $limit = $chunk_size - 1;
+      MKDEBUG && _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}    = $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} ) {
+      MKDEBUG && _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);
+         MKDEBUG && _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}++;
+         MKDEBUG && _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();
+         }
+         MKDEBUG && _d($self->{have_rows}, 'rows in nibble', $self->{nibbleno});
+      }
+
+      if ( $self->{have_rows} ) {
+         my $row = $self->{nibble_sth}->fetchrow_arrayref();
+         if ( $row ) {
+            $self->{rowno}++;
+            MKDEBUG && _d('Row', $self->{rowno}, 'in nibble',$self->{nibbleno});
+            return [ @$row ];
+         }
+      }
+
+      MKDEBUG && _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;
+   }
+
+   MKDEBUG && _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;
+   MKDEBUG && _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;
+   MKDEBUG && _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;
+   MKDEBUG && _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 _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 ) {
+      MKDEBUG && _d('User wants to use index', $want_index);
+      if ( !exists $indexes->{$want_index} ) {
+         MKDEBUG && _d('Cannot use user index because it does not exist');
+         $want_index = undef;
+      }
+   }
+
+   if ( !$want_index && $args{mysql_index} ) {
+      MKDEBUG && _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} ) {
+         MKDEBUG && _d('Will use wanted index');
+         $best_index = $want_index;
+      }
+      else {
+         MKDEBUG && _d('Wanted index is a possible index');
+         push @possible_indexes, $want_index;
+      }
+   }
+   else {
+      MKDEBUG && _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 ) {
+      MKDEBUG && _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];
+   }
+
+   MKDEBUG && _d('Best index:', $best_index);
+   return $best_index;
+}
+
+sub _get_index_cardinality {
+   my (%args) = @_;
+   my @required_args = qw(Cxn tbl index Quoter);
+   my ($cxn, $tbl, $index, $q) = @args{@required_args};
+
+   my $sql = "SHOW INDEXES FROM " . $q->quote(@{$tbl}{qw(db tbl)})
+           . " WHERE Key_name = '$index'";
+   MKDEBUG && _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};
+   }
+   MKDEBUG && _d('Index', $index, 'cardinality:', $cardinality);
+   return $cardinality;
+}
+
+sub get_row_estimate {
+   my (%args) = @_;
+   my @required_args = qw(Cxn tbl OptionParser TableParser Quoter);
+   my ($cxn, $tbl, $o, $tp, $q) = @args{@required_args};
+
+   if ( $args{where} ) {
+      MKDEBUG && _d('WHERE clause, using explain plan for row estimate');
+      my $table = $q->quote(@{$tbl}{qw(db tbl)});
+      my $sql   = "EXPLAIN SELECT * FROM $table WHERE $args{where}";
+      MKDEBUG && _d($sql);
+      my $expl = $cxn->dbh()->selectrow_hashref($sql);
+      MKDEBUG && _d(Dumper($expl));
+      return ($expl->{rows} || 0), $expl->{key};
+   }
+   else {
+      MKDEBUG && _d('No WHERE clause, using table status for row estimate');
+      return $tbl->{tbl_status}->{rows} || 0;
+   }
+}
+
+sub _prepare_sths {
+   my ($self) = @_;
+   MKDEBUG && _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});
+   MKDEBUG && _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};
+         MKDEBUG && _d($sth->{Statement}, 'params:', @ub);
+         $sth->execute(@ub);
+         $self->{next_lower} = $sth->fetchrow_arrayref();
+         $sth->finish();
+      }
+   }
+   else {
+      $self->{next_lower}  = $self->{first_lower};   
+   }
+   MKDEBUG && _d('Next lower boundary:', Dumper($self->{next_lower}));  
+
+   if ( !$self->{next_lower} ) {
+      MKDEBUG && _d('At end of table, or no more boundaries to resume');
+      $self->{no_more_boundaries} = 1;
+   }
+
+   $self->{last_upper} = $dbh->selectrow_arrayref($self->{last_ub_sql});
+   MKDEBUG && _d('Last upper boundary:', Dumper($self->{last_upper}));
+
+   return;
+}
+
+sub _next_boundaries {
+   my ($self) = @_;
+
+   if ( $self->{no_more_boundaries} ) {
+      MKDEBUG && _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}) ) {
+      MKDEBUG && _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,
+      );
+      MKDEBUG && _d('next_boundaries callback returned', $oktonibble);
+      if ( !$oktonibble ) {
+         $self->{no_more_boundaries} = 1;
+         return; # stop nibbling
+      }
+   }
+
+   MKDEBUG && _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();
+   MKDEBUG && _d('Next boundary:', Dumper($boundary));
+   if ( $boundary && @$boundary ) {
+      $self->{upper} = $boundary->[0]; # this nibble
+      if ( $boundary->[1] ) {
+         $self->{next_lower} = $boundary->[1]; # next nibble
+      }
+      else {
+         $self->{no_more_boundaries} = 1;  # for next call
+         MKDEBUG && _d('Last upper boundary:', Dumper($boundary->[0]));
+      }
+   }
+   else {
+      $self->{no_more_boundaries} = 1;  # for next call
+      $self->{upper} = $self->{last_upper};
+      MKDEBUG && _d('Last upper boundary:', Dumper($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$/ ) {
+         MKDEBUG && _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
+# ###########################################################################
+
+# ###########################################################################
+# OobNibbleIterator 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/OobNibbleIterator.pm
+#   t/lib/OobNibbleIterator.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package OobNibbleIterator;
+use base 'NibbleIterator';
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant MKDEBUG => $ENV{MKDEBUG} || 0;
+
+use Data::Dumper;
+$Data::Dumper::Indent    = 1;
+$Data::Dumper::Sortkeys  = 1;
+$Data::Dumper::Quotekeys = 0;
+
+sub new {
+   my ( $class, %args ) = @_;
+   my @required_args = qw();
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+
+   my $self = $class->SUPER::new(%args);
+
+   my $q     = $self->{Quoter};
+   my $o     = $self->{OptionParser};
+   my $where = $o->get('where');
+
+   if ( !$self->one_nibble() ) {
+      my $head_sql
+         = ($args{past_dml} || "SELECT ")
+         . ($args{past_select}
+            || join(', ', map { $q->quote($_) } @{$self->{sql}->{columns}}))
+         . " FROM "  . $self->{sql}->{from};
+
+      my $tail_sql
+         = ($where ? " AND ($where)" : '')
+         . " ORDER BY " . $self->{sql}->{order_by};
+
+      my $past_lower_sql
+         = $head_sql
+         . " WHERE " . $self->{sql}->{boundaries}->{'<'}
+         . $tail_sql
+         . " /*past lower chunk*/";
+      MKDEBUG && _d('Past lower statement:', $past_lower_sql);
+
+      my $explain_past_lower_sql
+         = "EXPLAIN SELECT "
+         . ($args{past_select}
+            || join(', ', map { $q->quote($_) } @{$self->{sql}->{columns}}))
+         . " FROM "  . $self->{sql}->{from}
+         . " WHERE " . $self->{sql}->{boundaries}->{'<'}
+         . $tail_sql
+         . " /*explain past lower chunk*/";
+      MKDEBUG && _d('Past lower statement:', $explain_past_lower_sql);
+
+      my $past_upper_sql
+         = $head_sql
+         . " WHERE " . $self->{sql}->{boundaries}->{'>'}
+         . $tail_sql
+         . " /*past upper chunk*/";
+      MKDEBUG && _d('Past upper statement:', $past_upper_sql);
+      
+      my $explain_past_upper_sql
+         = "EXPLAIN SELECT "
+         . ($args{past_select}
+            || join(', ', map { $q->quote($_) } @{$self->{sql}->{columns}}))
+         . " FROM "  . $self->{sql}->{from}
+         . " WHERE " . $self->{sql}->{boundaries}->{'>'}
+         . $tail_sql
+         . " /*explain past upper chunk*/";
+      MKDEBUG && _d('Past upper statement:', $explain_past_upper_sql);
+
+      $self->{past_lower_sql}         = $past_lower_sql;
+      $self->{past_upper_sql}         = $past_upper_sql;
+      $self->{explain_past_lower_sql} = $explain_past_lower_sql;
+      $self->{explain_past_upper_sql} = $explain_past_upper_sql;
+
+      $self->{past_nibbles} = [qw(lower upper)];
+      if ( my $nibble = $args{resume} ) {
+         if (    !defined $nibble->{lower_boundary}
+              || !defined $nibble->{upper_boundary} ) {
+            $self->{past_nibbles} = !defined $nibble->{lower_boundary}
+                                  ? ['upper']
+                                  : [];
+         }
+      }
+      MKDEBUG && _d('Nibble past', @{$self->{past_nibbles}});
+
+   } # not one nibble
+
+   return bless $self, $class;
+}
+
+sub more_boundaries {
+   my ($self) = @_;
+   return $self->SUPER::more_boundaries() if $self->{one_nibble};
+   return scalar @{$self->{past_nibbles}} ? 1 : 0;
+}
+
+sub statements {
+   my ($self) = @_;
+
+   my $sths = $self->SUPER::statements();
+
+   $sths->{past_lower_boundary} = $self->{past_lower_sth};
+   $sths->{past_upper_boundary} = $self->{past_upper_sth};
+
+   return $sths;
+}
+
+sub _prepare_sths {
+   my ($self) = @_;
+   MKDEBUG && _d('Preparing out-of-bound statement handles');
+
+   if ( !$self->{one_nibble} ) {
+      my $dbh = $self->{Cxn}->dbh();
+      $self->{past_lower_sth}         = $dbh->prepare($self->{past_lower_sql});
+      $self->{past_upper_sth}         = $dbh->prepare($self->{past_upper_sql});
+      $self->{explain_past_lower_sth} = $dbh->prepare($self->{explain_past_lower_sql});
+      $self->{explain_past_upper_sth} = $dbh->prepare($self->{explain_past_upper_sql});
+   }
+
+   return $self->SUPER::_prepare_sths();
+}
+
+sub _next_boundaries {
+   my ($self) = @_;
+
+   return $self->SUPER::_next_boundaries() unless $self->{no_more_boundaries};
+
+   if ( my $past = shift @{$self->{past_nibbles}} ) {
+      if ( $past eq 'lower' ) {
+         MKDEBUG && _d('Nibbling values below lower boundary');
+         $self->{nibble_sth}         = $self->{past_lower_sth};
+         $self->{explain_nibble_sth} = $self->{explain_past_lower_sth};
+         $self->{lower}              = [];
+         $self->{upper}              = $self->boundaries()->{first_lower};
+         $self->{next_lower}         = undef;
+      }
+      elsif ( $past eq 'upper' ) {
+         MKDEBUG && _d('Nibbling values above upper boundary');
+         $self->{nibble_sth}         = $self->{past_upper_sth};
+         $self->{explain_nibble_sth} = $self->{explain_past_upper_sth};
+         $self->{lower}              = $self->boundaries()->{last_upper};
+         $self->{upper}              = [];
+         $self->{next_lower}         = undef;
+      }
+      else {
+         die "Invalid past nibble: $past";
+      }
+      return 1; # continue nibbling
+   }
+
+   MKDEBUG && _d('Done nibbling past boundaries');
+   return; # stop nibbling
+}
+
+sub DESTROY {
+   my ( $self ) = @_;
+   foreach my $key ( keys %$self ) {
+      if ( $key =~ m/_sth$/ ) {
+         MKDEBUG && _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 OobNibbleIterator package
+# ###########################################################################
+
+# ###########################################################################
 # Daemon 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,
@@ -4370,12 +4505,16 @@
 # ###########################################################################
 
 # ###########################################################################
-# SchemaIterator r7512
-# Don't update this package!
+# SchemaIterator 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/SchemaIterator.pm
+#   t/lib/SchemaIterator.t
+# See https://launchpad.net/percona-toolkit for more information.
 # ###########################################################################
+{
 package SchemaIterator;
 
-{ # package scope
 use strict;
 use warnings FATAL => 'all';
 use English qw(-no_match_vars);
@@ -4407,8 +4546,19 @@
    die "I need either a dbh or file_itr argument"
       if (!$dbh && !$file_itr) || ($dbh && $file_itr);
 
+   my %resume;
+   if ( my $table = $args{resume} ) {
+      MKDEBUG && _d('Will resume from or after', $table);
+      my ($db, $tbl) = $args{Quoter}->split_unquote($table);
+      die "Resume table must be database-qualified: $table"
+         unless $db && $tbl;
+      $resume{db}  = $db;
+      $resume{tbl} = $tbl;
+   }
+
    my $self = {
       %args,
+      resume  => \%resume,
       filters => _make_filters(%args),
    };
 
@@ -4469,19 +4619,43 @@
    return \%filters;
 }
 
-sub next_schema_object {
+sub next {
    my ( $self ) = @_;
 
-   my %schema_object;
+   if ( !$self->{initialized} ) {
+      $self->{initialized} = 1;
+      if ( $self->{resume}->{tbl}
+           && !$self->table_is_allowed(@{$self->{resume}}{qw(db tbl)}) ) {
+         MKDEBUG && _d('Will resume after',
+            join('.', @{$self->{resume}}{qw(db tbl)}));
+         $self->{resume}->{after} = 1;
+      }
+   }
+
+   my $schema_obj;
    if ( $self->{file_itr} ) {
-      %schema_object = $self->_iterate_files();
+      $schema_obj= $self->_iterate_files();
    }
    else { # dbh
-      %schema_object = $self->_iterate_dbh();
-   }
-
-   MKDEBUG && _d('Next schema object:', Dumper(\%schema_object));
-   return %schema_object;
+      $schema_obj= $self->_iterate_dbh();
+   }
+
+   if ( $schema_obj ) {
+      if ( $schema_obj->{ddl} && $self->{TableParser} ) {
+         $schema_obj->{tbl_struct}
+            = $self->{TableParser}->parse($schema_obj->{ddl});
+      }
+
+      delete $schema_obj->{ddl} unless $self->{keep_ddl};
+      delete $schema_obj->{tbl_status} unless $self->{keep_tbl_status};
+
+      if ( my $schema = $self->{Schema} ) {