Largely working, but un-tested, pt-upgrade 2.2 for host-to-host comparison. Add EventExecutor.pm.

bin/pt-upgrade

@@ -33,6 +33,7 @@ BEGIN {
       QueryRewriter
       FileIterator
       QueryIterator
+      EventExecutor
       UpgradeResults
    ));
 }
@@ -5042,12 +5043,6 @@ has 'oktorun' => (
 );
 
 
-has 'default_database' => (
-   is       => 'rw',
-   isa      => 'Maybe[Str]',
-   required => 0,
-);
-
 has 'filter' => (
    is       => 'ro',
    isa      => 'CodeRef',
@@ -5076,12 +5071,6 @@ has 'stats' => (
    default  => sub { return {} },
 );
 
-has 'database' => (
-   is       => 'rw',
-   isa      => 'Maybe[Str]',
-   required => 0,
-);
-
 has '_fh' => (
    is       => 'rw',
    isa      => 'Maybe[FileHandle]',
@@ -5213,15 +5202,6 @@ sub next {
 
       $event->{fingerprint} = $self->fingerprint->($event->{arg});
 
-      my $current_db = $self->database;
-      my $db = $event->{db} || $event->{Schema} || $self->default_database;
-      if ( $db && (!$current_db || $current_db ne $db) ) {
-         $self->database($db);
-      }
-      else {
-         $self->database(undef);
-      }
-
       return $event;
    }
 
@@ -5278,6 +5258,113 @@ no Lmo;
 # End QueryIterator package
 # ###########################################################################
 
+# ###########################################################################
+# EventExecutor 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/EventExecutor.pm
+#   t/lib/EventExecutor.t
+# See https://launchpad.net/percona-toolkit for more information.
+# ###########################################################################
+{
+package EventExecutor;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+use Time::HiRes qw(time);
+use Data::Dumper;
+
+use Lmo;
+
+has 'default_database' => (
+   is       => 'rw',
+   isa      => 'Maybe[Str]',
+   required => 0,
+);
+
+has 'current_database' => (
+   is       => 'rw',
+   isa      => 'Maybe[Str]',
+   required => 0,
+);
+
+
+has 'stats' => (
+   is       => 'ro',
+   isa      => 'HashRef',
+   required => 0,
+   default  => sub { return {} },
+);
+
+sub exec_event {
+   my ($self, %args) = @_;
+   my @required_args = qw(host event);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my $host  = $args{host};
+   my $event = $args{event};
+
+   my $results = {
+      query_time => undef,
+      sth        => undef,
+      warnings   => undef,
+      error      => undef,
+   };
+
+   eval {
+      my $db = $event->{db} || $event->{Schema} || $self->default_database;
+      if ( !$host->{current_db} || $host->{current_db} ne $db ) {
+         PTDEBUG && _d('New current db:', $db);
+         $host->dbh->do("USE `$db`");
+         $host->{current_db} = $db;
+      }
+      my $sth = $host->dbh->prepare($event->{arg});
+      my $t0 = time;
+      $sth->execute();
+      my $t1 = time - $t0;
+      $results->{query_time} = sprintf('%.6f', $t1);
+      $results->{sth}        = $sth;
+      $results->{warnings}   = $self->get_warnings(host => $host);
+   };
+   if ( $EVAL_ERROR ) {
+      PTDEBUG && _d($EVAL_ERROR);
+      chomp($EVAL_ERROR);
+      $results->{error} = $EVAL_ERROR;
+   }
+   PTDEBUG && _d('Result on', $host->name, Dumper($results));
+   return $results;
+}
+
+sub get_warnings {
+   my ($self, %args) = @_;
+   my @required_args = qw(host);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my $host = $args{host};
+   my $warnings = $host->dbh->selectall_hashref('SHOW WARNINGS', 'code');
+   return $warnings;
+}
+
+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";
+}
+
+no Lmo;
+1;
+}
+# ###########################################################################
+# End EventExecutor package
+# ###########################################################################
+
 # ###########################################################################
 # UpgradeResults package
 # This package is a copy without comments from the original.  The original
@@ -5328,24 +5415,23 @@ sub save_diffs {
    my $class = $self->class(event => $event);
 
    if ( my $query = $self->_can_save(event => $event, class => $class) ) {
-
       if ( $query_time_diff
            && scalar @{$class->{query_time_diffs}} < $self->max_examples ) {
          push @{$class->{query_time_diffs}}, [
             $query,
-            $query_time_diff,
+            @$query_time_diff,
          ];
       }
 
-      if ( @$warning_diffs
+      if ( $warning_diffs && @$warning_diffs
            && scalar @{$class->{warning_diffs}} < $self->max_examples ) {
-         push @{$class->{warnings_diffs}}, [
+         push @{$class->{warning_diffs}}, [
             $query,
             $warning_diffs,
          ];
       }
 
-      if ( @$row_diffs
+      if ( $row_diffs && @$row_diffs
            && scalar @{$class->{row_diffs}} < $self->max_examples ) {
          push @{$class->{row_diffs}}, [
             $query,
@@ -5354,6 +5440,8 @@ sub save_diffs {
       }
    }
 
+   $self->report_if_ready(class => $class);
+
    return;
 }
 
@@ -5376,6 +5464,32 @@ sub save_error {
       }
    }
 
+   $self->report_if_ready(class => $class);
+
+   return;
+}
+
+sub save_failed_query {
+   my ($self, %args) = @_;
+
+   my $event  = $args{event};
+   my $error1 = $args{error1};
+   my $error2 = $args{error2};
+
+   my $class = $self->class(event => $event);
+
+   if ( my $query = $self->_can_save(event => $event, class => $class) ) {
+      if ( scalar @{$class->{failures}} < $self->max_examples ) {
+         push @{$class->{failures}}, [
+            $query,
+            $error1,
+            $error2,
+         ];
+      }
+   }
+
+   $self->report_if_ready(class => $class);
+
    return;
 }
 
@@ -5384,6 +5498,11 @@ sub _can_save {
    my $event = $args{event};
    my $class = $args{class};
    my $query = $event->{arg};
+   if ( $class->{reported} ) {
+      PTDEBUG && _d('Class already reported');
+      return;
+   }
+   $class->{total_queries}++;
    if ( exists $class->{unique_queries}->{$query}
         || scalar keys %{$class->{unique_queries}} < $self->max_class_size ) {
       $class->{unique_queries}->{$query}++;
@@ -5402,7 +5521,6 @@ sub class {
    my $classes = $self->classes;
    my $class   = $classes->{$id};
    if ( !$class ) {
-      PTDEBUG && _d('New query class:', $id, $event->{fingerprint});
       $class = $self->_new_class(
          id    => $id,
          event => $event,
@@ -5424,6 +5542,8 @@ sub _new_class {
       unique_queries   => {
          $event->{arg} => 0,
       },
+      failures         => [],  # error on both hosts
+      errors           => [],  # error on one host
       query_time_diffs => [],
       warning_diffs    => [],
       row_diffs        => [],
@@ -5431,6 +5551,313 @@ sub _new_class {
    return $class;
 }
 
+sub report_unreported_classes {
+   my ($self) = @_;
+   my $classes = $self->classes;
+   foreach my $id ( sort keys %$classes ) {
+      my $class = $classes->{$id};
+      my $reason;
+      if ( !scalar @{$class->{failures}} ) {
+         $reason = 'it has diffs';
+      }
+      elsif (    scalar @{$class->{errors}}
+              || scalar @{$class->{query_time_diffs}}
+              || scalar @{$class->{warning_diffs}}
+              || scalar @{$class->{row_diffs}} ) {
+         $reason = 'it has SQL errors and diffs';
+      }
+      else {
+         $reason = 'it has SQL errors'
+      }
+      $self->report_class(
+         class   => $class,
+         reasons => ["$reason, but hasn't been reported yet"],
+      );
+      $class = { reported => 1 };
+   }
+   return;
+}
+
+sub report_if_ready {
+   my ($self, %args) = @_;
+   my $class = $args{class};
+
+   my $max_examples = $self->max_class_size;
+   my @report_reasons;
+
+   if ( scalar keys %{$class->{unique_queries}} >= $self->max_class_size ) {
+      push @report_reasons, "it's full (--max-class-size)";
+   }
+
+   if ( scalar @{$class->{query_time_diffs}} >= $max_examples ) {
+      push @report_reasons, "there are $max_examples query diffs";
+   }
+
+   if ( scalar @{$class->{warning_diffs}} >= $max_examples ) {
+      push @report_reasons, "there are $max_examples warning diffs";
+   }
+
+   if ( scalar @{$class->{row_diffs}} >= $self->max_examples ) {
+      push @report_reasons, "there are $max_examples row diffs";
+   }
+
+   if ( scalar @{$class->{errors}} >= $self->max_examples ) {
+      push @report_reasons, "there are $max_examples query errors";
+   }
+
+   if ( scalar @{$class->{failures}} >= $self->max_examples ) {
+      push @report_reasons, "there are $max_examples failed queries";
+   }
+
+   if ( scalar @report_reasons ) {
+      PTDEBUG && _d('Reporting class because', @report_reasons);
+      $self->report_class(
+         class   => $class,
+         reasons => \@report_reasons,
+      );
+      $class = { reported => 1 };
+   }
+
+   return;
+}
+
+sub report_class {
+   my ($self, %args) = @_;
+   my $class   = $args{class};
+   my $reasons = $args{reasons};
+
+   PTDEBUG && _d('Reporting class', $class->{id}, $class->{fingerprint});
+
+   $self->_print_class_header(
+      class   => $class,
+      reasons => $reasons,
+   );
+
+   if ( scalar @{$class->{failures}} ) {
+      $self->_print_failures(
+         failures => $class->{failures},
+      );
+   }
+
+   if ( scalar @{$class->{errors}} ) {
+      $self->_print_diffs(
+         diffs         => $class->{errors},
+         name          => 'Query error',
+         inline        => 0,
+         default_value => 'No error',
+      );
+   }
+
+   if ( scalar @{$class->{query_time_diffs}} ) {
+      $self->_print_diffs(
+         diffs  => $class->{query_time_diffs},
+         name   => 'Query time',
+         inline => 1,
+      );
+   }
+
+   if ( scalar @{$class->{warning_diffs}} ) {
+      $self->_print_multiple_diffs(
+         diffs     => $class->{warning_diffs},
+         name      => 'Warning',
+         formatter => \&_format_warnings,
+      );
+   }
+
+   if ( scalar @{$class->{row_diffs}} ) {
+      $self->_print_multiple_diffs(
+         diffs     => $class->{row_diffs},
+         name      => 'Row',
+         formatter => \&_format_rows,
+      );
+   }
+
+   return;
+}
+
+my $class_header_format = <<'EOF';
+%s
+%s
+%s
+
+Reporting class because %s.
+
+Total queries      %s
+Unique queries     %s
+Discarded queries  %s
+
+%s
+EOF
+
+sub _print_class_header {
+   my ($self, %args) = @_;
+   my $class   = $args{class};
+   my @reasons = @{ $args{reasons} };
+
+   my $unique_queries = do {
+      my $i = 0;
+      map { $i += $_ } values %{$class->{unique_queries}};
+      $i;
+   };
+   PTDEBUG && _d('Unique queries:', $unique_queries);
+
+   my $reasons;
+   if ( scalar @reasons > 1 ) {
+      $reasons = join(', ', @reasons[0..($#reasons - 1)])
+               . ', and ' . $reasons[-1];
+   }
+   else {
+      $reasons = $reasons[0];
+   }
+   PTDEBUG && _d('Reasons:', $reasons);
+
+   printf $class_header_format,
+      ('#' x 72),
+      ('# Query class ' . ($class->{id} || '?')),
+      ('#' x 72),
+      ($reasons                || '?'),
+      (defined $class->{total_queries} ? $class->{total_queries} : '?'),
+      (defined $unique_queries         ? $unique_queries         : '?'),
+      (defined $class->{discarded}     ? $class->{discarded}     : '?'),
+      ($class->{fingerprint}   || '?');
+
+   return;
+}
+
+sub _print_diff_header {
+   my ($self, %args) = @_;
+   my $name  = $args{name}  || '?';
+   my $count = $args{count} || '?';
+   print "\n##\n## $name diffs: $count\n##\n";
+   return;
+}
+
+sub _print_diffs {
+   my ($self, %args) = @_;
+   my $diffs         = $args{diffs};
+   my $name          = $args{name};
+   my $inline        = $args{inline};
+   my $default_value = $args{default_value} || '?';
+
+   $self->_print_diff_header(
+      name  => $name,
+      count => scalar @$diffs,
+   );
+
+   my $fmt = $inline ? "\n%s vs. %s\n" : "\n%s\n\nvs.\n\n%s\n";
+
+   my $diffno = 1;
+   foreach my $diff ( @$diffs ) {
+      print "\n-- $diffno.\n";
+      printf $fmt,
+         ($diff->[1] || $default_value), 
+         ($diff->[2] || $default_value);
+      print "\n" . ($diff->[0] || '?') . "\n";
+      $diffno++;
+   }
+
+   return;
+}
+
+sub _print_multiple_diffs {
+   my ($self, %args) = @_;
+   my $diffs     = $args{diffs};
+   my $name      = $args{name};
+   my $formatter = $args{formatter};
+
+   $self->_print_diff_header(
+      name  => $name,
+      count => scalar @$diffs,
+   );
+
+   my $diffno = 1;
+   foreach my $diff ( @$diffs ) {
+      print "\n-- $diffno.\n";
+      my $formatted_diff = $formatter->($diff->[1]);
+      print $formatted_diff || '?';
+      print "\n" . ($diff->[0] || '?') . "\n";
+      $diffno++;
+   }
+
+   return;
+}
+
+sub _print_failures {
+   my ($self, %args) = @_;
+   my $failures = $args{failures};
+
+   my $n_failures = scalar @$failures;
+
+   print "\n##\n## SQL errors: $n_failures\n##\n";
+
+   my $failno = 1;
+   foreach my $failure ( @$failures ) {
+      print "\n-- $failno.\n";
+      if ( ($failure->[1] || '') eq ($failure->[2] || '') ) {
+         printf "\nOn both hosts:\n\n" . ($failure->[1] || '') . "\n";
+      }
+      else {
+         printf "\n%s\n\nvs.\n\n%s\n",
+            ($failure->[1] || ''),
+            ($failure->[2] || '');
+      }
+      print "\n" . ($failure->[0] || '?') . "\n";
+      $failno++;
+   }
+
+   return;
+}
+
+my $warning_format = <<'EOL';
+   Code: %s
+  Level: %s
+Message: %s
+EOL
+
+sub _format_warnings {
+   my ($warnings) = @_;
+   return unless $warnings && @$warnings;
+   my @warnings;
+   foreach my $warn ( @$warnings ) {
+      my $code  = $warn->[0];
+      my $warn1 = $warn->[1];
+      my $warn2 = $warn->[2];
+      my $host1_warn
+         = $warn1 ? sprintf $warning_format, 
+                       ($warn1->{Code}    || $warn1->{code}    || '?'),
+                       ($warn1->{Level}   || $warn1->{level}   || '?'),
+                       ($warn1->{Message} || $warn1->{message} || '?')
+         :          "No warning $code\n";
+      my $host2_warn
+         = $warn2 ? sprintf $warning_format, 
+                       ($warn2->{Code}    || $warn2->{code}    || '?'),
+                       ($warn2->{Level}   || $warn2->{level}   || '?'),
+                       ($warn2->{Message} || $warn2->{message} || '?')
+         :          "No warning $code\n";
+
+      my $warning = sprintf "\n%s\nvs.\n\n%s", $host1_warn, $host2_warn;
+      push @warnings, $warning;
+   }
+   return join("\n\n", @warnings);
+}
+
+sub _format_rows {
+   my ($rows) = @_;
+   return unless $rows && @$rows;
+   my @diffs;
+   foreach my $row ( @$rows ) {
+      my $rowno = $row->[0];
+      my $cols1 = $row->[1];
+      my $cols2 = $row->[2];
+      my $diff
+         = "@ row " . ($rowno || '?') . "\n"
+         . '< ' . join(',', map {defined $_ ? $_ : 'NULL'} @$cols1) . "\n"
+         . '> ' . join(',', map {defined $_ ? $_ : 'NULL'} @$cols2) . "\n";
+      push @diffs, $diff;
+   }
+   return "\n" . join("\n", @diffs);
+}
+
 sub _d {
    my ($package, undef, $line) = caller 0;
    @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
@@ -5474,6 +5901,7 @@ use sigtrap 'handler', \&sig_int, 'normal-signals';
 # Global variables.  Only really essential variables should be here.
 my $oktorun     = 1;
 my $exit_status = 0;
+my $stats       = {};
 
 sub main {
    local @ARGV = @_;  # set global ARGV for this package
@@ -5481,6 +5909,14 @@ sub main {
    # Reset global vars, else tests will fail.
    $oktorun     = 1;
    $exit_status = 0;
+   $stats       = {
+      queries_read        => 0,
+      queries_filtered    => 0,
+      queries_with_diffs  => 0,
+      queries_no_diffs    => 0,
+      queries_with_errors => 0,
+      failed_queries      => 0,
+   };
 
    # ##########################################################################
    # Get configuration information.
@@ -5578,14 +6014,14 @@ sub main {
    # ########################################################################
    # Do the version-check
    # ########################################################################
-   if ( $o->get('version-check') && (!$o->has('quiet') || !$o->get('quiet')) ) {
+#   if ( $o->get('version-check') && (!$o->has('quiet') || !$o->get('quiet')) ) {
-      VersionCheck::version_check(
+#      VersionCheck::version_check(
-         instances => [
+#         instances => [
-            ($host1 ? { dbh => $host1->dbh, dsn => $host1->dsn } : ()),
+#            ($host1 ? { dbh => $host1->dbh, dsn => $host1->dsn } : ()),
-            ($host2 ? { dbh => $host2->dbh, dsn => $host2->dsn } : ()),
+#            ($host2 ? { dbh => $host2->dbh, dsn => $host2->dsn } : ()),
-         ],
+#         ],
-      );
+#      );
-   }
+#   }
 
    # ########################################################################
    # Daemonize now that everything is setup and ready to work.
@@ -5613,6 +6049,7 @@ sub main {
          max_class_size  => $o->get('max-class-size'),
          max_examples    => $o->get('max-examples'),
          # Optional
+         database        => $o->get('database'),
          filter          => $o->get('filter'),
          ignore_warnings => $o->get('ignore-warnings'),
          read_only       => $o->get('read-only') ?  1 : 0,
@@ -5634,12 +6071,13 @@ sub main {
    }
    else {
       # Shouldn't get here, unless you're Ryan.
-      die "Error: the combination of command line options are invalid, "
+      die "Invalid combination of command line arguments, and pt-upgrade "
-         . "and pt-upgrade failed to detect this error earlier.  Please "
+         . "failed to detect this error earlier.  Please report this bug "
-         . "report this bug with the exact command line used to run "
+         . "with the exact command line used to run the tool.\n";
-         . "the tool.\n";
    }
 
+   report_stats();
+
    return $exit_status;
 }
 
@@ -5662,20 +6100,12 @@ sub compare_host_to_host {
    PTDEBUG && _d('Compare', $host1->name, 'to', $host2->name);
 
    # Optional args
+   my $database        = $args{database};
    my $filter          = $args{filter};
    my $ignore_warnings = $args{ignore_warnings};
    my $read_only       = $args{read_only};
    my $read_timeout    = $args{read_timeout};
 
-   my $stats = {
-      queries_read        => 0,
-      queries_filtered    => 0,
-      queries_with_diffs  => 0,
-      queries_no_diffs    => 0,
-      queries_with_errors => 0,
-      failed_queries      => 0,
-   };
-
    my $results = UpgradeResults->new(
       max_class_size => $max_class_size,
       max_examples   => $max_examples,
@@ -5692,34 +6122,45 @@ sub compare_host_to_host {
       fingerprint      => sub { return $qr->fingerprint(@_) },
       oktorun          => sub { return $oktorun },
       stats            => $stats,
-      default_database => $host1->dsn->{D},
+      ($database     ? (default_database => $database)     : ()),
-      ($filter       ? (filter       => $filter)       : ()),
+      ($filter       ? (filter           => $filter)       : ()),
-      ($read_only    ? (read_only    => $read_only)    : ()),
+      ($read_only    ? (read_only        => $read_only)    : ()),
-      ($read_timeout ? (read_timeout => $read_timeout) : ()),
+      ($read_timeout ? (read_timeout     => $read_timeout) : ()),
+   );
+
+   my $executor = EventExecutor->new(
+      default_database => $database,
    );
 
    while ( my $event = $query_iter->next() ) {
-      my $results1 = exec_event(
+      my $results1 = $executor->exec_event(
          event => $event,
          host  => $host1,
       );
 
-      my $results2 = exec_event(
+      my $results2 = $executor->exec_event(
-         event => $event,
+         event    => $event,
-         host  => $host2,
+         host     => $host2,
+         database => $executor->current_database,
       );
 
       if ( $results1->{error} && $results2->{error} ) {
-         PTDEBUG && _d('Failed query:', $event->{arg});
+         PTDEBUG && _d('Failed query');
          $stats->{failed_queries}++;
+         $results->save_failed_query(
+            event  => $event,
+            error1 => $results1->{error},
+            error2 => $results2->{error},
+         );
       }
       elsif (   ($results1->{error} && !$results2->{error})
              || ($results2->{error} && !$results1->{error}) ) {
-         $stats->{queries_wit_errors}++;
+         PTDEBUG && _d('Query error');
+         $stats->{queries_with_errors}++;
          $results->save_error(
-            event    => $event,
+            event  => $event,
-            results1 => $results1,
+            error1 => $results1->{error},
-            results2 => $results2,
+            error2 => $results2->{error},
          );
       }
       else {
@@ -5733,16 +6174,29 @@ sub compare_host_to_host {
             warnings2 => $results2->{warnings},
          );
 
-         my $row_diffs = diff_rows(
+         my $row_diffs;
-            sth1 => $results1->{sth},
+         if ( $event->{arg} =~ m/(?:^\s*SELECT|(?:\*\/\s*SELECT))/i ) {
-            sth2 => $results2->{sth},
+            $row_diffs = diff_rows(
-         );
+               sth1 => $results1->{sth},
+               sth2 => $results2->{sth},
+            );
+         }
+
+         eval {
+            foreach my $result ( $results1, $results2 ) {
+               $result->{sth}->finish();
+               delete $result->{sth};
+            }
+         };
+         if ( $EVAL_ERROR ) {
+            PTDEBUG && _d($EVAL_ERROR);
+         }
 
          if ( $query_time_diff
-              || scalar @$warning_diffs
+              || ($warning_diffs && scalar @$warning_diffs)
-              || scalar @$row_diffs )
+              || ($row_diffs     && scalar @$row_diffs) )
          {
-            PTDEBUG && _d('Query has diffs');
+            PTDEBUG && _d('Query diffs');
             $stats->{queries_with_diffs}++;
             $results->save_diffs(
                event           => $event,
@@ -5752,13 +6206,13 @@ sub compare_host_to_host {
             );
          }
          else {
+            PTDEBUG && _d('Query OK, no diffs');
             $stats->{queries_no_diffs}++;
          }
       }
    }
 
-   warn Dumper($results);
+   $results->report_unreported_classes();
-   warn Dumper($stats);
 
    return;
 }
@@ -5801,50 +6255,6 @@ sub disable_query_cache {
    return;
 }
 
-sub exec_event {
-   my (%args) = @_;
-   my @required_args = qw(host event);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my $host  = $args{host};
-   my $event = $args{event};
-
-   my $results = {
-      query_time => undef,
-      sth        => undef,
-      warnings   => undef,
-      error      => undef,
-   };
-
-   eval {
-      my $sth = $host->dbh->prepare($event->{arg});
-      my $t0 = time;
-      $sth->execute();
-      my $t1 = time - $t0;
-      $results->{query_time} = sprintf('%.6f', $t1);
-      $results->{sth}        = $sth;
-      $results->{warnings}   = get_warnings(dbh => $host->dbh);
-   };
-   if ( $EVAL_ERROR ) {
-      PTDEBUG && _d($EVAL_ERROR);
-      $results->{error} = $EVAL_ERROR;
-   }
-   PTDEBUG && _d('Result on', $host->name, Dumper($results));
-   return $results;
-}
-
-sub get_warnings {
-   my (%args) = @_;
-   my @required_args = qw(dbh);
-   foreach my $arg ( @required_args ) {
-      die "I need a $arg argument" unless $args{$arg};
-   }
-   my $dbh = $args{dbh};
-   my $warnings = $dbh->selectall_hashref('SHOW WARNINGS', 'code');
-   return $warnings;
-}
-
 sub diff_query_times {
    my (%args) = @_;
    my @required_args = qw(query_time1 query_time2);
@@ -5877,11 +6287,11 @@ sub diff_warnings {
    my @diffs;
    foreach my $code ( sort keys %codes ) {
       next if exists $host1_warns->{$code} && exists $host2_warns->{$code};
-      push @diffs, {
+      push @diffs, [
-         code  => $code,
+         $code,
-         host1 => $host1_warns->{$code},
+         $host1_warns->{$code},
-         host2 => $host2_warns->{$code},
+         $host2_warns->{$code},
-      };
+      ];
    }
 
    return \@diffs;
@@ -5896,6 +6306,8 @@ sub diff_rows {
    my $sth1 = $args{sth1};
    my $sth2 = $args{sth2};
 
+   return unless $sth1 && $sth2;
+
    # Optional args
    my $max_diffs = $args{max_diffs} || 3;
 
@@ -5917,11 +6329,11 @@ sub diff_rows {
          # so don't store the reference and then use it after a later fetch."
          my @copy_row1 = $row1 ? @$row1 : ();
          my @copy_row2 = $row2 ? @$row2 : ();
-         push @diffs, {
+         push @diffs, [
-            row_number => $rowno,
+            $rowno,
-            host1      => \@copy_row1,
+            \@copy_row1,
-            host2      => \@copy_row2,
+            \@copy_row2,
-         };
+         ];
       }
    }
 
@@ -5951,6 +6363,19 @@ sub identical_rows {
    return 1;
 } 
 
+sub report_stats {
+   print "
+########################################################################
+# Stats
+########################################################################
+
+";
+   my $fmt = "%-20s  %d\n";
+   foreach my $stat ( sort keys %$stats ) {
+      printf $fmt, $stat, $stats->{$stat} || 0;
+   }
+   return;
+}
 
 # Catches signals so we can exit gracefully.
 sub sig_int {
@@ -6423,6 +6848,12 @@ default: yes
 
 Continue running even if there is an error.
 
+=item --database
+
+short form: -D; type: string
+
+Database to use for queries that do not specify a database.
+
 =item --daemonize
 
 Fork to the background and detach from the shell.

lib/EventExecutor.pm

@@ -0,0 +1,120 @@
+# This program is copyright 2013 Percona Ireland Ltd.
+# Feedback and improvements are welcome.
+#
+# THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
+# WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
+# MERCHANTIBILITY AND FITNESS FOR A PARTICULAR PURPOSE.
+#
+# This program is free software; you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free Software
+# Foundation, version 2; OR the Perl Artistic License.  On UNIX and similar
+# systems, you can issue `man perlgpl' or `man perlartistic' to read these
+# licenses.
+#
+# You should have received a copy of the GNU General Public License along with
+# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
+# Place, Suite 330, Boston, MA  02111-1307  USA.
+# ###########################################################################
+# EventExecutor package
+# ###########################################################################
+{
+package EventExecutor;
+
+use strict;
+use warnings FATAL => 'all';
+use English qw(-no_match_vars);
+use constant PTDEBUG => $ENV{PTDEBUG} || 0;
+
+use Time::HiRes qw(time);
+use Data::Dumper;
+
+use Lmo;
+
+has 'default_database' => (
+   is       => 'rw',
+   isa      => 'Maybe[Str]',
+   required => 0,
+);
+
+has 'current_database' => (
+   is       => 'rw',
+   isa      => 'Maybe[Str]',
+   required => 0,
+);
+
+##
+# Private
+##
+
+has 'stats' => (
+   is       => 'ro',
+   isa      => 'HashRef',
+   required => 0,
+   default  => sub { return {} },
+);
+
+sub exec_event {
+   my ($self, %args) = @_;
+   my @required_args = qw(host event);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my $host  = $args{host};
+   my $event = $args{event};
+
+   my $results = {
+      query_time => undef,
+      sth        => undef,
+      warnings   => undef,
+      error      => undef,
+   };
+
+   eval {
+      my $db = $event->{db} || $event->{Schema} || $self->default_database;
+      if ( !$host->{current_db} || $host->{current_db} ne $db ) {
+         PTDEBUG && _d('New current db:', $db);
+         $host->dbh->do("USE `$db`");
+         $host->{current_db} = $db;
+      }
+      my $sth = $host->dbh->prepare($event->{arg});
+      my $t0 = time;
+      $sth->execute();
+      my $t1 = time - $t0;
+      $results->{query_time} = sprintf('%.6f', $t1);
+      $results->{sth}        = $sth;
+      $results->{warnings}   = $self->get_warnings(host => $host);
+   };
+   if ( $EVAL_ERROR ) {
+      PTDEBUG && _d($EVAL_ERROR);
+      chomp($EVAL_ERROR);
+      $results->{error} = $EVAL_ERROR;
+   }
+   PTDEBUG && _d('Result on', $host->name, Dumper($results));
+   return $results;
+}
+
+sub get_warnings {
+   my ($self, %args) = @_;
+   my @required_args = qw(host);
+   foreach my $arg ( @required_args ) {
+      die "I need a $arg argument" unless $args{$arg};
+   }
+   my $host = $args{host};
+   my $warnings = $host->dbh->selectall_hashref('SHOW WARNINGS', 'code');
+   return $warnings;
+}
+
+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";
+}
+
+no Lmo;
+1;
+}
+# ###########################################################################
+# End EventExecutor package
+# ###########################################################################

lib/QueryIterator.pm

@@ -62,12 +62,6 @@ has 'oktorun' => (
 # Optional
 ##
 
-has 'default_database' => (
-   is       => 'rw',
-   isa      => 'Maybe[Str]',
-   required => 0,
-);
-
 has 'filter' => (
    is       => 'ro',
    isa      => 'CodeRef',
@@ -99,12 +93,6 @@ has 'stats' => (
    default  => sub { return {} },
 );
 
-has 'database' => (
-   is       => 'rw',
-   isa      => 'Maybe[Str]',
-   required => 0,
-);
-
 has '_fh' => (
    is       => 'rw',
    isa      => 'Maybe[FileHandle]',
@@ -236,15 +224,6 @@ sub next {
 
       $event->{fingerprint} = $self->fingerprint->($event->{arg});
 
-      my $current_db = $self->database;
-      my $db = $event->{db} || $event->{Schema} || $self->default_database;
-      if ( $db && (!$current_db || $current_db ne $db) ) {
-         $self->database($db);
-      }
-      else {
-         $self->database(undef);
-      }
-
       return $event;
    }
 

lib/UpgradeResults.pm

@@ -59,24 +59,23 @@ sub save_diffs {
    my $class = $self->class(event => $event);
 
    if ( my $query = $self->_can_save(event => $event, class => $class) ) {
-
       if ( $query_time_diff
            && scalar @{$class->{query_time_diffs}} < $self->max_examples ) {
          push @{$class->{query_time_diffs}}, [
             $query,
-            $query_time_diff,
+            @$query_time_diff,
          ];
       }
 
-      if ( @$warning_diffs
+      if ( $warning_diffs && @$warning_diffs
            && scalar @{$class->{warning_diffs}} < $self->max_examples ) {
-         push @{$class->{warnings_diffs}}, [
+         push @{$class->{warning_diffs}}, [
             $query,
             $warning_diffs,
          ];
       }
 
-      if ( @$row_diffs
+      if ( $row_diffs && @$row_diffs
            && scalar @{$class->{row_diffs}} < $self->max_examples ) {
          push @{$class->{row_diffs}}, [
             $query,
@@ -85,6 +84,8 @@ sub save_diffs {
       }
    }
 
+   $self->report_if_ready(class => $class);
+
    return;
 }
 
@@ -107,6 +108,32 @@ sub save_error {
       }
    }
 
+   $self->report_if_ready(class => $class);
+
+   return;
+}
+
+sub save_failed_query {
+   my ($self, %args) = @_;
+
+   my $event  = $args{event};
+   my $error1 = $args{error1};
+   my $error2 = $args{error2};
+
+   my $class = $self->class(event => $event);
+
+   if ( my $query = $self->_can_save(event => $event, class => $class) ) {
+      if ( scalar @{$class->{failures}} < $self->max_examples ) {
+         push @{$class->{failures}}, [
+            $query,
+            $error1,
+            $error2,
+         ];
+      }
+   }
+
+   $self->report_if_ready(class => $class);
+
    return;
 }
 
@@ -115,6 +142,11 @@ sub _can_save {
    my $event = $args{event};
    my $class = $args{class};
    my $query = $event->{arg};
+   if ( $class->{reported} ) {
+      PTDEBUG && _d('Class already reported');
+      return;
+   }
+   $class->{total_queries}++;
    if ( exists $class->{unique_queries}->{$query}
         || scalar keys %{$class->{unique_queries}} < $self->max_class_size ) {
       $class->{unique_queries}->{$query}++;
@@ -133,7 +165,6 @@ sub class {
    my $classes = $self->classes;
    my $class   = $classes->{$id};
    if ( !$class ) {
-      PTDEBUG && _d('New query class:', $id, $event->{fingerprint});
       $class = $self->_new_class(
          id    => $id,
          event => $event,
@@ -155,6 +186,8 @@ sub _new_class {
       unique_queries   => {
          $event->{arg} => 0,
       },
+      failures         => [],  # error on both hosts
+      errors           => [],  # error on one host
       query_time_diffs => [],
       warning_diffs    => [],
       row_diffs        => [],
@@ -162,6 +195,318 @@ sub _new_class {
    return $class;
 }
 
+sub report_unreported_classes {
+   my ($self) = @_;
+   my $classes = $self->classes;
+   foreach my $id ( sort keys %$classes ) {
+      my $class = $classes->{$id};
+      my $reason;
+      if ( !scalar @{$class->{failures}} ) {
+         $reason = 'it has diffs';
+      }
+      elsif (    scalar @{$class->{errors}}
+              || scalar @{$class->{query_time_diffs}}
+              || scalar @{$class->{warning_diffs}}
+              || scalar @{$class->{row_diffs}} ) {
+         $reason = 'it has SQL errors and diffs';
+      }
+      else {
+         $reason = 'it has SQL errors'
+      }
+      $self->report_class(
+         class   => $class,
+         reasons => ["$reason, but hasn't been reported yet"],
+      );
+      $class = { reported => 1 };
+   }
+   return;
+}
+
+sub report_if_ready {
+   my ($self, %args) = @_;
+   my $class = $args{class};
+
+   my $max_examples = $self->max_class_size;
+   my @report_reasons;
+
+   if ( scalar keys %{$class->{unique_queries}} >= $self->max_class_size ) {
+      push @report_reasons, "it's full (--max-class-size)";
+   }
+
+   if ( scalar @{$class->{query_time_diffs}} >= $max_examples ) {
+      push @report_reasons, "there are $max_examples query diffs";
+   }
+
+   if ( scalar @{$class->{warning_diffs}} >= $max_examples ) {
+      push @report_reasons, "there are $max_examples warning diffs";
+   }
+
+   if ( scalar @{$class->{row_diffs}} >= $self->max_examples ) {
+      push @report_reasons, "there are $max_examples row diffs";
+   }
+
+   if ( scalar @{$class->{errors}} >= $self->max_examples ) {
+      push @report_reasons, "there are $max_examples query errors";
+   }
+
+   if ( scalar @{$class->{failures}} >= $self->max_examples ) {
+      push @report_reasons, "there are $max_examples failed queries";
+   }
+
+   if ( scalar @report_reasons ) {
+      PTDEBUG && _d('Reporting class because', @report_reasons);
+      $self->report_class(
+         class   => $class,
+         reasons => \@report_reasons,
+      );
+      $class = { reported => 1 };
+   }
+
+   return;
+}
+
+sub report_class {
+   my ($self, %args) = @_;
+   my $class   = $args{class};
+   my $reasons = $args{reasons};
+
+   PTDEBUG && _d('Reporting class', $class->{id}, $class->{fingerprint});
+
+   $self->_print_class_header(
+      class   => $class,
+      reasons => $reasons,
+   );
+
+   if ( scalar @{$class->{failures}} ) {
+      $self->_print_failures(
+         failures => $class->{failures},
+      );
+   }
+
+   if ( scalar @{$class->{errors}} ) {
+      $self->_print_diffs(
+         diffs         => $class->{errors},
+         name          => 'Query error',
+         inline        => 0,
+         default_value => 'No error',
+      );
+   }
+
+   if ( scalar @{$class->{query_time_diffs}} ) {
+      $self->_print_diffs(
+         diffs  => $class->{query_time_diffs},
+         name   => 'Query time',
+         inline => 1,
+      );
+   }
+
+   if ( scalar @{$class->{warning_diffs}} ) {
+      $self->_print_multiple_diffs(
+         diffs     => $class->{warning_diffs},
+         name      => 'Warning',
+         formatter => \&_format_warnings,
+      );
+   }
+
+   if ( scalar @{$class->{row_diffs}} ) {
+      $self->_print_multiple_diffs(
+         diffs     => $class->{row_diffs},
+         name      => 'Row',
+         formatter => \&_format_rows,
+      );
+   }
+
+   return;
+}
+
+# This is a terrible hack due to two things: 1) our own util/update-modules
+# things lines starting with multiple # are package headers; 2) the same
+# util strips all comment lines start with #.  So if we use the literal #
+# for this header, util/update-modules will remove them from the code.
+# *facepalm*
+my $class_header_format = <<'EOF';
+%s
+%s
+%s
+
+Reporting class because %s.
+
+Total queries      %s
+Unique queries     %s
+Discarded queries  %s
+
+%s
+EOF
+
+sub _print_class_header {
+   my ($self, %args) = @_;
+   my $class   = $args{class};
+   my @reasons = @{ $args{reasons} };
+
+   my $unique_queries = do {
+      my $i = 0;
+      map { $i += $_ } values %{$class->{unique_queries}};
+      $i;
+   };
+   PTDEBUG && _d('Unique queries:', $unique_queries);
+
+   my $reasons;
+   if ( scalar @reasons > 1 ) {
+      $reasons = join(', ', @reasons[0..($#reasons - 1)])
+               . ', and ' . $reasons[-1];
+   }
+   else {
+      $reasons = $reasons[0];
+   }
+   PTDEBUG && _d('Reasons:', $reasons);
+
+   printf $class_header_format,
+      ('#' x 72),
+      ('# Query class ' . ($class->{id} || '?')),
+      ('#' x 72),
+      ($reasons                || '?'),
+      (defined $class->{total_queries} ? $class->{total_queries} : '?'),
+      (defined $unique_queries         ? $unique_queries         : '?'),
+      (defined $class->{discarded}     ? $class->{discarded}     : '?'),
+      ($class->{fingerprint}   || '?');
+
+   return;
+}
+
+sub _print_diff_header {
+   my ($self, %args) = @_;
+   my $name  = $args{name}  || '?';
+   my $count = $args{count} || '?';
+   print "\n##\n## $name diffs: $count\n##\n";
+   return;
+}
+
+sub _print_diffs {
+   my ($self, %args) = @_;
+   my $diffs         = $args{diffs};
+   my $name          = $args{name};
+   my $inline        = $args{inline};
+   my $default_value = $args{default_value} || '?';
+
+   $self->_print_diff_header(
+      name  => $name,
+      count => scalar @$diffs,
+   );
+
+   my $fmt = $inline ? "\n%s vs. %s\n" : "\n%s\n\nvs.\n\n%s\n";
+
+   my $diffno = 1;
+   foreach my $diff ( @$diffs ) {
+      print "\n-- $diffno.\n";
+      printf $fmt,
+         ($diff->[1] || $default_value), 
+         ($diff->[2] || $default_value);
+      print "\n" . ($diff->[0] || '?') . "\n";
+      $diffno++;
+   }
+
+   return;
+}
+
+sub _print_multiple_diffs {
+   my ($self, %args) = @_;
+   my $diffs     = $args{diffs};
+   my $name      = $args{name};
+   my $formatter = $args{formatter};
+
+   $self->_print_diff_header(
+      name  => $name,
+      count => scalar @$diffs,
+   );
+
+   my $diffno = 1;
+   foreach my $diff ( @$diffs ) {
+      print "\n-- $diffno.\n";
+      my $formatted_diff = $formatter->($diff->[1]);
+      print $formatted_diff || '?';
+      print "\n" . ($diff->[0] || '?') . "\n";
+      $diffno++;
+   }
+
+   return;
+}
+
+sub _print_failures {
+   my ($self, %args) = @_;
+   my $failures = $args{failures};
+
+   my $n_failures = scalar @$failures;
+
+   print "\n##\n## SQL errors: $n_failures\n##\n";
+
+   my $failno = 1;
+   foreach my $failure ( @$failures ) {
+      print "\n-- $failno.\n";
+      if ( ($failure->[1] || '') eq ($failure->[2] || '') ) {
+         printf "\nOn both hosts:\n\n" . ($failure->[1] || '') . "\n";
+      }
+      else {
+         printf "\n%s\n\nvs.\n\n%s\n",
+            ($failure->[1] || ''),
+            ($failure->[2] || '');
+      }
+      print "\n" . ($failure->[0] || '?') . "\n";
+      $failno++;
+   }
+
+   return;
+}
+
+my $warning_format = <<'EOL';
+   Code: %s
+  Level: %s
+Message: %s
+EOL
+
+sub _format_warnings {
+   my ($warnings) = @_;
+   return unless $warnings && @$warnings;
+   my @warnings;
+   foreach my $warn ( @$warnings ) {
+      my $code  = $warn->[0];
+      my $warn1 = $warn->[1];
+      my $warn2 = $warn->[2];
+      my $host1_warn
+         = $warn1 ? sprintf $warning_format, 
+                       ($warn1->{Code}    || $warn1->{code}    || '?'),
+                       ($warn1->{Level}   || $warn1->{level}   || '?'),
+                       ($warn1->{Message} || $warn1->{message} || '?')
+         :          "No warning $code\n";
+      my $host2_warn
+         = $warn2 ? sprintf $warning_format, 
+                       ($warn2->{Code}    || $warn2->{code}    || '?'),
+                       ($warn2->{Level}   || $warn2->{level}   || '?'),
+                       ($warn2->{Message} || $warn2->{message} || '?')
+         :          "No warning $code\n";
+
+      my $warning = sprintf "\n%s\nvs.\n\n%s", $host1_warn, $host2_warn;
+      push @warnings, $warning;
+   }
+   return join("\n\n", @warnings);
+}
+
+sub _format_rows {
+   my ($rows) = @_;
+   return unless $rows && @$rows;
+   my @diffs;
+   foreach my $row ( @$rows ) {
+      my $rowno = $row->[0];
+      my $cols1 = $row->[1];
+      my $cols2 = $row->[2];
+      my $diff
+         = "@ row " . ($rowno || '?') . "\n"
+         . '< ' . join(',', map {defined $_ ? $_ : 'NULL'} @$cols1) . "\n"
+         . '> ' . join(',', map {defined $_ ? $_ : 'NULL'} @$cols2) . "\n";
+      push @diffs, $diff;
+   }
+   return "\n" . join("\n", @diffs);
+}
+
 sub _d {
    my ($package, undef, $line) = caller 0;
    @_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }

t/pt-upgrade/samples/001/insert-warning.log

@@ -0,0 +1,4 @@
+# User@Host: root[root] @ localhost []
+# Query_time: 1  Lock_time: 0  Rows_sent: 0  Rows_examined: 1
+use test;
+INSERT INTO t (id, username, last_login) VALUES (6, 'new_user_name_is_too_long', '2013-01-01 00:00:00');

t/pt-upgrade/samples/001/select-all-rows.log

@@ -0,0 +1,4 @@
+# User@Host: root[root] @ localhost []
+# Query_time: 1  Lock_time: 0  Rows_sent: 7  Rows_examined: 7
+use test;
+select * from test.t order by id;

t/pt-upgrade/samples/001/tables.sql

@@ -4,7 +4,7 @@ USE test;
 DROP TABLE IF EXISTS t;
 CREATE TABLE `t` (
   `id` int(10) NOT NULL,
-  `name` varchar(255) default NULL,
+  `username` varchar(32) default NULL,
   `last_login` datetime default NULL,
   PRIMARY KEY  (`id`)
 );