#!/usr/bin/perl

# This is mk-slave-prefetch, a program to pipeline relay logs on a MySQL slave.
#
# This program is copyright (c) 2007 Baron Schwartz.  Feedback and improvements
# are welcome.
#
# THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
# WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
# MERCHANTIBILITY AND FITNESS FOR A PARTICULAR PURPOSE.
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2; OR the Perl Artistic License.  On UNIX and similar
# systems, you can issue `man perlgpl' or `man perlartistic' to read these
# licenses.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place, Suite 330, Boston, MA  02111-1307  USA.

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

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

use English qw(-no_match_vars);
$OUTPUT_AUTOFLUSH = 1;

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

package OptionParser;

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

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

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

      }
   }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

   return %vals;
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

1;

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

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

package VersionParser;

use English qw(-no_match_vars);

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

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

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

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

1;

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

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

package DSNParser;

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

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

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

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

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

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

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

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

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

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

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

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

1;

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

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

package LogParser;


use English qw(-no_match_vars);

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

my $general_log_first_line = qr{
   \A
   (?:(\d{6}\s+\d{1,2}:\d\d:\d\d)|\t)? # Timestamp
   \t
   (?:\s*(\d+))                        # Thread ID
   \s
   (.*)                                # Everything else
   \Z
}xs;

my $general_log_any_line = qr{
   \A(
      Connect
      |Field\sList
      |Init\sDB
      |Query
      |Quit
   )
   (?:\s+(.*\Z))?
}xs;

my $slow_log_ts_line = qr/^# Time: (\d{6}\s+\d{1,2}:\d\d:\d\d)/;
my $slow_log_uh_line = qr/# User\@Host: ([^\[]+).*?@ (\S*) \[(.*)\]/;

my $binlog_line_1 = qr{^# at (\d+)};
my $binlog_line_2 = qr/^#(\d{6}\s+\d{1,2}:\d\d:\d\d)\s+server\s+id\s+(\d+)\s+end_log_pos\s+(\d+)\s+(\S+)\s*([^\n]*)$/;
my $binlog_line_2_rest = qr{Query\s+thread_id=(\d+)\s+exec_time=(\d+)\s+error_code=(\d+)};

sub parse_event {
   my ( $self, $fh, $code, $mode ) = @_;
   my $event; # Don't initialize, that'll cause a loop.

   my $done = 0;
   my $type = 0; # 0 = comments, 1 = USE and SET etc, 2 = the actual query
   my $line = defined $self->{last_line} ? $self->{last_line} : <$fh>;
   $mode  ||= '';

   LINE:
   while ( !$done && defined $line ) {
      $ENV{MKDEBUG} && _d('type: ', $type, ' ', $line);

      if ( !$mode && $line =~ m/^# [A-Z]/ ) {
         $mode = 'slow';
      }

      if ( $line =~ m/Version:.+ started with:/ ) {
         $ENV{MKDEBUG} && _d('Chomping out header lines');
         <$fh>; # Tcp port: etc
         <$fh>; # Column headers
         $line = <$fh>;
         $type = 0;
         redo LINE;
      }

      elsif ( $line =~ m/^# No InnoDB statistics available/ ) {
         $ENV{MKDEBUG} && _d('Ignoring line');
         $line = <$fh>;
         $type = 0;
         next LINE;
      }

      elsif ( my ( $ts, $id, $rest ) = $line =~ m/$general_log_first_line/s ) {
         $ENV{MKDEBUG} && _d('Beginning of general log event');
         $mode ||= 'log';
         $self->{last_line} = undef;
         if ( $type == 0 ) {
            my ( $cmd, $arg ) = $rest =~ m/$general_log_any_line/;
            $event = {
               ts  => $ts || '',
               id  => $id,
               cmd => $cmd,
               arg => $arg || '',
            };
            if ( $cmd ne 'Query' ) {
               $ENV{MKDEBUG} && _d('Not a query, done with this event');
               $done = 1;
               chomp $event->{arg} if $event->{arg};
            }
            $type = 2;
         }
         else {
            $ENV{MKDEBUG} && _d('Saving line for next invocation');
            $self->{last_line} = $line;
            $done = 1;
            chomp $event->{arg} if $event->{arg};
         }
      }

      elsif ( my ( $time ) = $line =~ m/$slow_log_ts_line/ ) {
         $ENV{MKDEBUG} && _d('Beginning of slow log event');
         $self->{last_line} = undef;
         if ( $type == 0 ) {
            $event->{ts} = $time;
            if ( my ( $user, $host, $ip ) = $line =~ m/$slow_log_uh_line/ ) {
               @{$event}{qw(user host ip)} = ($user, $host, $ip);
            }
         }
         else {
            $ENV{MKDEBUG} && _d('Saving line for next invocation');
            $self->{last_line} = $line;
            $done = 1;
         }
         $type = 0;
      }

      elsif ( my ( $user, $host, $ip ) = $line =~ m/$slow_log_uh_line/ ) {
         if ( $type == 0 ) {
            @{$event}{qw(user host ip)} = ($user, $host, $ip);
         }
         else {
            $ENV{MKDEBUG} && _d('Saving line for next invocation');
            $self->{last_line} = $line;
            $done = 1;
         }
         $type = 0;
      }

      elsif ( $line =~ m/^# / && (my %hash = $line =~ m/(\w+):\s+(\S+)/g ) ) {
         if ( $type == 0 ) {
            $ENV{MKDEBUG} && _d('Splitting line into fields');
            @{$event}{keys %hash} = values %hash;
         }
         else {
            $ENV{MKDEBUG} && _d('Saving line for next invocation');
            $self->{last_line} = $line;
            $done = 1;
         }
         $type = 0;
      }

      else {
         if ( $mode eq 'slow' && $line =~ m/;\s+\Z/ ) {
            $ENV{MKDEBUG} && _d('Line is the end of a query within event');
            if ( my ( $db ) = $line =~ m/^use (.*);/ ) {
               $ENV{MKDEBUG} && _d('Setting event DB to ', $db);
               $event->{db} = $db;
               $type = 1;
            }
            elsif ( $type < 2 && (my ( $setting ) = $line =~ m/^(SET .*);\s+\Z/ ) ) {
               $ENV{MKDEBUG} && _d('Setting a property for event');
               push @{$event->{settings}}, $setting;
               $type = 1;
            }
            else {
               $ENV{MKDEBUG} && _d('Line is a continuation of prev line');
               $event->{arg} .= $line;
               $type = 2;
            }
         }
         else {
            $ENV{MKDEBUG} && _d('Line is a continuation of prev line');
            $event->{arg} .= $line;
            $type = 2;
         }
         $event->{cmd} = 'Query';
      }

      $event->{NR} = $NR;

      $line = <$fh> unless $done;
   }

   if ( !defined $line ) {
      $self->{last_line} = undef;
   }

   if ( $mode && $mode eq 'slow' ) {
      $event->{arg} =~ s/;\s*\Z// if $event->{arg};
   }

   $code->($event) if $event && $code;
   return $event;
}

sub parse_binlog_event {
   my ( $self, $fh, $code ) = @_;
   my $event;

   my $term  = $self->{term} || ";\n"; # Corresponds to DELIMITER
   my $tpat  = quotemeta $term;
   local $RS = $term;
   my $line  = <$fh>;

   LINE: {
      return unless $line;

      if ( $line =~ m/^DELIMITER/m ) {
         my($del)      = $line =~ m/^DELIMITER ([^\n]+)/m;
         $self->{term} = $del;
         local $RS     = $del;
         $line         = <$fh>; # Throw away DELIMITER line
         $ENV{MKDEBUG} && _d('New record separator: ', $del);
         redo LINE;
      }

      $line =~ s/$tpat\Z//;

      if ( my ( $offset ) = $line =~ m/$binlog_line_1/m ) {
         $self->{last_line} = undef;
         $event = {
            offset => $offset,
         };
         my ( $ts, $sid, $end, $type, $rest ) = $line =~ m/$binlog_line_2/m;
         @{$event}{qw(ts server_id end type)} = ($ts, $sid, $end, $type);
         (my $arg = $line) =~ s/\n*^#.*\n//gm; # Remove comment lines
         $event->{arg} = $arg;
         if ( $type eq 'Xid' ) {
            my ($xid) = $rest =~ m/(\d+)/;
            $event->{xid} = $xid;
         }
         elsif ( $type eq 'Query' ) {
            @{$event}{qw(id time code)} = $rest =~ m/$binlog_line_2_rest/;
         }
         else {
            die "Unknown event type $type"
               unless $type =~ m/Start|Execute_load_query|Append_block|Begin_load_query|Rand|User_var|Intvar/;
         }
      }
      else {
         $event = {
            arg => $line,
         };
      }
   }

   if ( !defined $line ) {
      delete $self->{term};
   }

   $code->($event) if $event && $code;
   return $event;
}

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

1;

# ###########################################################################
# End LogParser package
# ###########################################################################

# ###########################################################################
# QueryRewriter package 1767
# ###########################################################################
use strict;
use warnings FATAL => 'all';

package QueryRewriter;

use English qw(-no_match_vars);

my $quote_re = qr/"(?:(?!(?<!\\)").)*"|'(?:(?!(?<!\\)').)*'/; # Costly!
my $bal;
$bal         = qr/
                  \(
                  (?:
                     (?> [^()]+ )    # Non-parens without backtracking
                     |
                     (??{ $bal })    # Group with matching parens
                  )*
                  \)
                 /x;


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

sub fingerprint {
   my ( $self, $query ) = @_;
   $query = lc $query;
   $query =~ s/^\s*(?:--|#).*//gm;              # One-line comments
   $query =~ s{
              (?<![\w.+-])
              [+-]?
              (?:
                \d+
                (?:[.]\d*)?
                |[.]\d+
              )
              (?:e[+-]?\d+)?
              \b
             }
             {N}gx;                             # Float/real into N
   $query =~ s/\b0(?:x[0-9a-f]+|b[01]+)\b/N/g;  # Hex/bin into N
   $query =~ s/[xb]'N'/N/g;                     # Hex/bin into N
   $query =~ s/\\["']//g;                       # Turn quoted strings into S
   $query =~ s/(["']).*?\1/S/g;                 # Turn quoted strings into S
   $query =~ s/\A\s+//;                         # Chop off leading whitespace
   $query =~ s/\s{2,}/ /g;                      # Collapse all whitespace
   $query =~ s/[\n\r\f]+/ /g;                   # Collapse newlines etc
   $query =~ s{
               \b(in|values)\s*\(\s*([NS])\s*,[^\)]*\)
              }
              {$1($2+)}gx;      # Collapse IN() and VALUES() lists
   $query =~ s/(?<=\w_)\d+(_\d+)?\b/$1 ? "N_N" : "N"/eg;
   return $query;
}

sub convert_to_select {
   my ( $self, $query ) = @_;
   return unless $query;
   $query =~ s{
                 \A.*?
                 update\s+(.*?)
                 \s+set\b(.*?)
                 (?:\s+where\b(.*?))?
                 (limit\s*\d+(?:\s*,\s*\d+)?)?
                 \Z
              }
              {__update_to_select($1, $2, $3, $4)}exsi
      || $query =~ s{
                    \A.*?
                    (?:insert|replace)\s+
                    .*?\binto\b(.*?)\(([^\)]+)\)\s*
                    values\s*(\(.*?\))\s*
                    (?:\blimit\b|on\s*duplicate\s*key.*)?\s*
                    \Z
                 }
                 {__insert_to_select($1, $2, $3)}exsi
      || $query =~ s{
                    \A.*?
                    delete\s+(.*?)
                    \bfrom\b(.*)
                    \Z
                 }
                 {__delete_to_select($1, $2)}exsi;
   $query =~ s/\s*on\s+duplicate\s+key\s+update.*\Z//si;
   $query =~ s/\A.*?(?=\bSELECT\s*\b)//ism;
   return $query;
}

sub convert_select_list {
   my ( $self, $query ) = @_;
   $query =~ s{
               \A\s*select(.*?)\bfrom\b
              }
              {$1 =~ m/\*/ ? "select 1 from" : "select isnull(coalesce($1)) from"}exi;
   return $query;
}

sub __delete_to_select {
   my ( $delete, $join ) = @_;
   if ( $join =~ m/\bjoin\b/ ) {
      return "select 1 from $join";
   }
   return "select * from $join";
}

sub __insert_to_select {
   my ( $tbl, $cols, $vals ) = @_;
   $ENV{MKDEBUG} && _d('Args: ', @_);
   my @cols = split(/,/, $cols);
   $ENV{MKDEBUG} && _d('Cols: ', @cols);
   $vals =~ s/^\(|\)$//g; # Strip leading/trailing parens
   my @vals = $vals =~ m/($quote_re|[^,]*${bal}[^,]*|[^,]+)/g;
   $ENV{MKDEBUG} && _d('Vals: ', @vals);
   if ( @cols == @vals ) {
      return "select * from $tbl where "
         . join(' and ', map { "$cols[$_]=$vals[$_]" } (0..$#cols));
   }
   else {
      return "select * from $tbl limit 1";
   }
}

sub __update_to_select {
   my ( $from, $set, $where, $limit ) = @_;
   return "select $set from $from "
      . ( $where ? "where $where" : '' )
      . ( $limit ? " $limit "      : '' );
}

sub wrap_in_derived {
   my ( $self, $query ) = @_;
   return unless $query;
   return $query =~ m/\A\s*select/i
      ? "select 1 from ($query) as x limit 1"
      : $query;
}

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

1;

# ###########################################################################
# End QueryRewriter package
# ###########################################################################

package main;

use Data::Dumper;
use English qw(-no_match_vars);
use List::Util qw(min max sum);
use Time::HiRes qw(gettimeofday);
use sigtrap qw(handler finish untrapped normal-signals);

$Data::Dumper::Quotekeys = 0;
$Data::Dumper::Indent    = 0;

my @opt_spec = (
   { s => 'askpass',           d => 'Prompt for password for connections' },
   { s => 'charset|A=s',       d => 'Default character set' },
   { s => 'checkint|i=A',      d => 'How often to check the slave: '
                                  . 'init,min,max (default 16,1,1024)' },
   { s => 'daemonize',         d => 'Fork to background and detach '
                                  . '(POSIX only)' },
   { s => 'database|D=s',      d => 'Database to use' },
   { s => 'defaults-file|F=s', d => 'Only read mysql options from the '
                                  . 'given file' },
   { s => 'execute|x!',        d => 'Execute transformed queries (default)' },
   { s => 'host|h=s',          d => 'Connect to host' },
   { s => 'iolag|l=z',         d => 'How many bytes to lag the I/O thread '
                                  . '(default 1k)' },
   { s => 'maxquerytime|q=f',  d => 'Do not run queries longer than this '
                                  . 'many seconds; fractions allowed '
                                  . '(default 1)' },
   { s => 'querysampsize=i',   d => 'Average query exec time over this '
                                  . 'many queries (default 4)' },
   { s => 'offset|o=z',        d => 'How many bytes to lead the SQL thread '
                                  . '(default 128)' },
   { s => 'password|p=s',      d => 'Password to use when connecting' },
   { s => 'print',             d => 'Print transformed queries' },
   { s => 'printnonrewritten', d => 'Print non-rewritable queries' },
   { s => 'permitregexp=s',    d => 'Permit queries matching '
                                  . 'this Perl regexp' },
   { s => 'port|P=i',          d => 'Port number to use for connection' },
   { s => 'rejectregexp=s',    d => 'Reject queries matching this '
                                  . 'Perl regexp' },
   { s => 'sentinel=s',        d => 'Sentinel file; default '
                                  . '/tmp/mk-slave-prefetch-sentinel' },
   {  s => 'setvars=s',        d => 'Set these MySQL variables (default '
                                  . 'wait_timeout=10000)' },
   { s => 'socket|S=s',        d => 'Socket file to use for connection' },
   { s => 'statistics',        d => 'Print statistics after exiting' },
   { s => 'stop',              d => 'Stop running instances by creating '
                                  . 'the sentinel file' },
   { s => 'time|t=m',          d => 'Time to run before exiting' },
   { s => 'tmpdir=s',          d => 'Where to create temp files for LOAD '
                                  . 'DATA INFILE queries (default /dev/null)' },
   { s => 'user|u=s',          d => 'User for login if not current user' },
   { s => 'window|w=z',        d => 'How many bytes ahead is too far '
                                  . 'ahead (default 4k)' },
   '--print and --daemonize are mutually exclusive',
   'at least one of --print or --execute are required',
);

my $dsn_parser = new DSNParser();
my $qn         = new QueryRewriter();
my $lp         = new LogParser();
my $opt_parser = new OptionParser(@opt_spec);
$opt_parser->{strict} = 0;
$opt_parser->{prompt} = '[OPTION...] [FILE]';
$opt_parser->{descr}  = q{pipelines relay logs to pre-warm the slave's caches.};
my %opts = $opt_parser->parse();
$dsn_parser->prop('setvars', $opts{setvars});

if ( $opts{t} ) {
   $opts{t} = max($opts{t}, 1);
}

my ($chk_int, $chk_min, $chk_max) = @{$opts{i}};
if ( grep { !defined $_ || $_ !~ m/^\d+$/ } ($chk_int, $chk_min, $chk_max) ) {
   $opt_parser->error("You must specify three elements for --checkint");
}
elsif ( $chk_int > $chk_max || $chk_int < $chk_min
   || $chk_max < $chk_min || $chk_min < 0 )
{
   $opt_parser->error("You specified an invalid range for --checkint");
}

if ( @ARGV > 1 ) {
   $opt_parser->error("You can specify only one FILE");
}

$opt_parser->usage_or_errors(%opts);

# ############################################################################
# First things first: if --stop was given, create the sentinel file.
# ############################################################################
if ( $opts{stop} ) {
   open my $file, ">", $opts{sentinel}
      or die "Cannot open $opts{sentinel}: $OS_ERROR\n";
   print $file "Remove this file to permit mk-slave-prefetch to run\n"
      or die "Cannot write to $opts{sentinel}: $OS_ERROR\n";
   close $file
      or die "Cannot close $opts{sentinel}: $OS_ERROR\n";
   print STDOUT "Successfully created file $opts{sentinel}\n";
   exit(0);
}

# ############################################################################
# Initialize the query stats from the file on the commandline, if any.
# ############################################################################
my %query_stats;
if ( @ARGV ) {
   open my $fh, "<", $ARGV[0] or die $OS_ERROR;
   my ($type, $rest);
   while ( my $line = <$fh> ) {
      ($type, $rest) = $line =~ m/^# (query|stats): (.*)$/;
      next unless $type;
      if ( $type eq 'query' ) {
         $query_stats{$rest} = { seen => 1, samples => [] };
      }
      else {
         my ( $seen, $exec, $sum, $avg )
            = $rest =~ m/seen=(\S+) exec=(\S+) sum=(\S+) avg=(\S+)/;
         if ( $seen ) {
            $query_stats{$rest}->{samples}
               = [ map { $avg } (1 ..  $opts{querysampsize}) ];
            $query_stats{$rest}->{avg} = $avg;
         }
      }
   }
   close $fh or die $OS_ERROR;
}

# ############################################################################
# Get the database connection and set it up as desired: Lowercase all column
# names for fetchrow_hashref. Don't disconnect on fork.  Disable the query
# cache.
# ############################################################################
if ( $opts{askpass} ) {
   $opts{p} = OptionParser::prompt_noecho("Enter password: ");
}
my $dbh = $dsn_parser->get_dbh(
   $dsn_parser->get_cxn_params(\%opts), { AutoCommit => 1 });
$dbh->{FetchHashKeyName} = 'NAME_lc';
$dbh->{InactiveDestroy}  = 1;
$dbh->do('/*!40001 set @@session.query_cache_type=OFF */');

# ############################################################################
# Daemonize only after (potentially) asking for passwords for --askpass.
# ############################################################################
if ( $opts{daemonize} ) {
   require POSIX;
   chdir '/'                 or die "Can't chdir to /: $OS_ERROR";
   open STDIN, '/dev/null'   or die "Can't read /dev/null: $OS_ERROR";
   open STDOUT, '>/dev/null' or die "Can't write to /dev/null: $OS_ERROR";
   defined( my $pid = fork ) or die "Can't fork: $OS_ERROR";
   exit if $pid;
   POSIX::setsid()           or die "Can't start a new session: $OS_ERROR";
   open STDERR, '>&STDOUT'   or die "Can't dup STDOUT: $OS_ERROR";
   delete $ENV{MKDEBUG};     # Debugging is useless
}


# TODO:
# * filter: Throw away non-replicated DBs/tables
# * Optionally read the binlog from the server's master.  advantages: can run
#   tool on another server than the slave.

# ############################################################################
# Ready to work now.
# ############################################################################
my $vp              = new VersionParser();
my $have_subqueries = $vp->version_ge($dbh, '4.1.0');

my $now       = time();
my $end       = $now + ( $opts{t} || 0 );    # When we should exit
my ($datadir) = ($dbh->selectrow_array('show variables like "datadir"'))[1];
$ENV{MKDEBUG} && _d("Data directory ", $datadir);

my %stats;
my %slave;
my $last_chk = 0;
my $oktorun  = 1;

eval {
   while ( oktorun() ) {
      %slave = get_status($dbh);

      my $pos  = 0; # Current position we're reading in relay log.
      my $next = 0; # Start of next relay log event.
      my $ts   = 0; # Last seen timestamp.

      if ( $slave{running} ) {

         my $cmd = "mysqlbinlog -l $opts{tmpdir} "
                 . "--start-pos=$slave{pos} $datadir/$slave{file}";
         $ENV{MKDEBUG} && _d($cmd);
         open my $fh, "$cmd 2>/dev/null |" or die $OS_ERROR;
         $stats{mysqlbinlog}++;

         my $i = 0;
         EVENT:
         while ( oktorun(1) && (my $event = $lp->parse_binlog_event($fh) )) {
            $stats{events}++;
            $pos  = $event->{offset} || $pos;
            $next = max($next, $pos + ($event->{end} || 0));
            $ENV{MKDEBUG} && _d("i=$i pos=$pos next=$next slave=$slave{pos}");

            # If it's a LOAD DATA INFILE, rm the temp file.
            if (
               $event->{arg}
               && (my ($file) = $event->{arg} =~ m/INFILE ('[^']+')/i)
            ) {
               $stats{load_data_infile}++;
               if ( !unlink($file) ) {
                  $ENV{MKDEBUG} && _d("Could not unlink ", $file);
                  $stats{could_not_unlink}++;
               }
               next EVENT;
            }

            # Stay ahead of the slave.
            next EVENT if not_far_enough_ahead($pos);

            # Time to check the slave's status again?
            if ( $pos > $slave{pos} && (++$i - $last_chk) >= $chk_int ) {
               %slave    = get_status($dbh);
               $last_chk = $i;
               $chk_int  = $pos <= $slave{pos} # The slave caught up to us
                  ? max($chk_min, $chk_int / 2)
                  : min($chk_max, $chk_int * 2);
               next EVENT if not_far_enough_ahead($pos);
            }

            # But don't get too far ahead or too close to the end of the binlog.
            while ( oktorun(1)
               && ( too_far_ahead($pos) || too_close_to_io($pos) )
            ) {

               # Don't increment stats if the slave didn't catch up while we
               # slept. TODO: if the slave is very caught up to the I/O, this
               # will be a problem.
               if ( wait_for_master(\%slave, $pos - $opts{w} + 1) > 0 ) {
                  if (too_far_ahead($pos)) {
                     $ENV{MKDEBUG}
                        && _d("Event $pos too far ahead of $slave{pos}");
                     $stats{too_far_ahead}++;
                  }
                  elsif (too_close_to_io($pos)) {
                     $ENV{MKDEBUG}
                        && _d("Event $pos too close to I/O thread "
                            . "($slave{pos} + $slave{lag})");
                     $stats{too_close_to_io_thread}++;
                  }
               }
               else {
                  $ENV{MKDEBUG} && _d('SQL thread did not advance');
               }

               %slave    = get_status($dbh);
               $last_chk = $i;
            }

            if ( ($event->{arg}||'')
               !~ m/\A\s*(?:set [t@]|use|insert|update|delete|replace)/i
            ) {
               $ENV{MKDEBUG} && _d('Event arg: ',
                  (defined $event->{arg}
                     ?  substr($event->{arg}, 0, 50)
                     : 'undef'));
               $ENV{MKDEBUG} && _d('Skipping this event because not allowable');
               $stats{event_not_allowed}++;
               next EVENT;
            }

            if (
                  ( $opts{rejectregexp}
                     && $event->{arg} =~ m/$opts{rejectregexp}/ )
               || ( $opts{permitregexp}
                     && $event->{arg} !~ m/$opts{permitregexp}/ )
            ) {
               $ENV{MKDEBUG} && _d('Skipping because of permit/reject regexp');
               $stats{event_filtered_out}++;
               next EVENT;
            }

            # If the event is SET TIMESTAMP and we've already set the timestamp
            # to that value, skip it.
            if ( (my ($newts) = $event->{arg} =~ m/SET TIMESTAMP=(\d+)/) ) {
               if ( $newts == $ts ) {
                  $ENV{MKDEBUG} && _d('Already saw timestamp ', $newts);
                  $stats{same_timestamp}++;
                  next EVENT;
               }
               else {
                  $ts = $newts;
               }
            }

            # Convert the event to a SELECT and print/execute it.
            my $select = $qn->convert_to_select($event->{arg});
            if ( $select =~ m/\A\s*(?:set|select|use)/i ) {
               my $fingerprint = $qn->fingerprint($event->{arg});
               if ( (my $avg = get_avg($fingerprint)) < $opts{q} ) {

                  # Safeguard as much as possible against really enormous
                  # result sets.
                  my $sql = $qn->convert_select_list($select);
                  if ( $have_subqueries && !have_seen_query($fingerprint)) {
                     # Wrap in a "derived table," but only if it hasn't been
                     # seen before.  This way, really short queries avoid the
                     # overhead of creating the temp table.
                     $sql = $qn->wrap_in_derived($sql);
                  }

                  # Do it!
                  $ENV{MKDEBUG} && _d($sql);
                  $stats{do_query}++;
                  if ( $opts{x} ) {
                     eval {
                        my $start = gettimeofday();
                        $dbh->do($sql);
                        store_avg($fingerprint, gettimeofday() - $start);
                     };
                     if ( $EVAL_ERROR ) {
                        $stats{query_error}++;
                        if ( $ENV{MKDEBUG} ) {
                           _d($EVAL_ERROR);
                           _d("SQL was: ", $event->{arg});
                        }
                     }
                  }
                  elsif ( $opts{print} ) {
                     print $sql, ";\n";
                  }

               }

               # The query's average execution time is longer than the
               # specified limit, so we skip it and just wait for the master
               # to pass it by.
               else {
                  $ENV{MKDEBUG}
                     && _d("Avg time $avg too long for ", $fingerprint);
                  $stats{query_too_long}++;
                  wait_for_master(\%slave, $pos + 1);
                  %slave    = get_status($dbh);
                  $last_chk = $i;
               }

            }
            else {
               $stats{query_not_rewritten}++;
               if ( $ENV{MKDEBUG} || $opts{printnonrewritten} ) {
                  _d($event->{arg});
               }
            }
         }

         $ENV{MKDEBUG} && _d('Closing filehandle');

         # Unfortunately, mysqlbinlog does NOT like me to close the pipe before
         # reading all data from it.  It hangs and prints angry messages about a
         # closed file.  So I'll find the mysqlbinlog process created by the
         # open() and kill it.
         my $procs = `ps -eaf | grep mysqlbinlog`;
         $ENV{MKDEBUG} && _d($procs);
         if ( my ($line) = $procs =~ m/^(.*?\d\s+$cmd)$/m ) {
            chomp $line;
            $ENV{MKDEBUG} && _d($line);
            if ( my ( $proc ) = $line =~ m/(\d+)/ ) {
               $ENV{MKDEBUG} && _d("Will kill process ", $proc);
               kill(15, $proc);
            }
         }

         if ( !close($fh) ) {
            if ( $OS_ERROR ) {
               warn "Error closing mysqlbinlog pipe: $OS_ERROR\n";
            }
            else {
               $ENV{MKDEBUG} && _d("Exit status $CHILD_ERROR from mysqlbinlog");
            }
         }
      }
      if ( oktorun() ) {
         $stats{sleep}++;
         sleep(1);
      }
   }
};
if ( $EVAL_ERROR ) {
   print $EVAL_ERROR;
}

# Print statistics
if ( $opts{statistics} ) {

   # Print operations in order of descending count, with percentage.
   my $maxlen = max(0, map { length($_) } keys %stats);
   my $total  = sum(0, values %stats);
   printf("# %-${maxlen}s \%10s %10s\n", qw(Action Count Pct));
   my $fmt = "# %-${maxlen}s \%10d %10.2f\n";
   foreach my $key ( reverse sort { $stats{$a} <=> $stats{$b} } keys %stats ) {
      printf($fmt, $key, $stats{$key}, $stats{$key} / $total * 100);
   }

   # Print normalized queries, their average exec times, times seen and times
   # executed.  Sort in order of times seen descending.
   foreach my $query (
      reverse sort {
         $query_stats{$a}->{seen} <=> $query_stats{$b}->{seen}
      } keys %query_stats
   ) {
      my $stats = $query_stats{$query};
      print
         "# query: ", $query, "\n# stats: ",
         join(' ',
            (map { "$_=" . ($stats->{$_} || '0') } qw(seen exec sum avg))),
         "\n";
   }
}

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

# Catches signals so we can exit gracefully.
sub finish {
   my ($signal) = @_;
   print STDERR "Exiting on SIG$signal.\n";
   $oktorun = 0;
}

# It's ok to run if we haven't been told to stop, we haven't exceeded the
# time.  The parameter adds the further restriction that the slave must be
# running.
sub oktorun {
   my ( $only_if_slave_running ) = @_;
   $now = time();
   return (!$only_if_slave_running || $slave{running})
      && !-f $opts{sentinel}
      && ((!$opts{t} || $now < $end) && $oktorun);
}

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

# Whether we are far enough ahead of the slave.
sub not_far_enough_ahead {
   my ( $pos ) = @_;
   if ( $pos < $slave{pos} + $opts{o} ) {
      $ENV{MKDEBUG} && _d("$pos is not $opts{o} ahead of $slave{pos}");
      $stats{not_far_enough_ahead}++;
      return 1;
   }
   return 0;
}

# Whether we are too far ahead of the slave.
sub too_far_ahead {
   my ( $pos ) = @_;
   return ($pos > $slave{pos} + $opts{w});
}

# Whether we are too close to where the I/O thread is writing.
sub too_close_to_io {
   my ( $pos ) = @_;
   return $slave{lag} && $pos >= $slave{pos} + $slave{lag} - $opts{l};
}

sub wait_for_master {
   my ( $slave, $pos ) = @_;
   $stats{master_pos_wait}++;
   my $sql = "SELECT COALESCE(MASTER_POS_WAIT('$slave->{mfile}', "
      . ($slave->{mpos} + ($pos - $slave->{pos})) . ", 1), 0)";
   $ENV{MKDEBUG} && _d($sql);
   my $start = gettimeofday();
   my ($events) = $dbh->selectrow_array($sql);
   $ENV{MKDEBUG} && _d("Waited ", (gettimeofday - $start), " and got ",
      (defined $events ? $events : 'NULL'));
   return $events;
}

# The average is weighted so we don't quit trying a statement when we have
# only a few samples.  So if we want to collect 16 samples and the first one
# is huge, it will be weighted as 1/16th of its size.
sub store_avg {
   my ( $query, $time ) = @_;
   $ENV{MKDEBUG} && _d('Execution time: ', $query, ' ', $time);
   $query_stats{$query}->{samples} ||= [];
   my $samples = $query_stats{$query}->{samples};
   push @$samples, $time;
   if ( @$samples > $opts{querysampsize} ) {
      shift @$samples;
   }
   $query_stats{$query}->{avg} = sum(@$samples) / $opts{querysampsize};
   $query_stats{$query}->{exec}++;
   $query_stats{$query}->{sum} += $time;
   $ENV{MKDEBUG} && _d('Average time: ', $query_stats{$query}->{avg});
}

sub have_seen_query {
   my ( $query ) = @_;
   return $query_stats{$query}->{seen};
}

sub get_avg {
   my ( $query ) = @_;
   $query_stats{$query}->{seen}++;
   return $query_stats{$query}->{avg} || 0;
}

sub get_status {
   my ( $dbh ) = @_;
   $stats{show_slave_status}++;
   my $status = $dbh->selectrow_hashref("SHOW SLAVE STATUS");
   if ( !$status || ! %$status ) {
      die "No output from SHOW SLAVE STATUS.\n";
   }
   # $ENV{MKDEBUG} && _d(Dumper($status));
   my %status = (
      running => ($status->{slave_sql_running} || '') eq 'Yes',
      file    => $status->{relay_log_file},
      pos     => $status->{relay_log_pos},
                 # If the slave SQL thread is executing from the same log the
                 # I/O thread is reading from, in general (except when the
                 # master or slave starts a new binlog or relay log) we can tell
                 # how many bytes the SQL thread lags the I/O thread.
      lag     => $status->{master_log_file} eq $status->{relay_master_log_file}
               ? $status->{read_master_log_pos} - $status->{exec_master_log_pos}
               : 0,
      mfile   => $status->{relay_master_log_file},
      mpos    => $status->{exec_master_log_pos},
   );
   $ENV{MKDEBUG} && _d(Dumper(\%status));
   return %status;
}

# ############################################################################
# Documentation.
# ############################################################################

=pod

=head1 NAME

mk-slave-prefetch - Pipeline relay logs on a MySQL slave to pre-warm caches.

=head1 SYNOPSIS

 mk-slave-prefetch
 mk-slave-prefetch --statistics > /path/to/saved/statistics
 mk-slave-prefetch /path/to/saved/statistics

=head1 DESCRIPTION

mk-slave-prefetch reads the slave's relay log slightly ahead of where the
slave's SQL thread is reading, converts statements into C<SELECT>, and
executes them.  In theory, this should help alleviate the effects of the
slave's single-threaded SQL execution.  It will help take advantage of
multiple CPUs and disks by pre-reading the data from disk, so the data is
already in the cache when the slave SQL thread executes the un-modified
version of the statement.

Statements that can't be converted into C<SELECT> are ignored.  However, there
is always a chance of bugs.  It would be a very good idea to connect as a
read-only user.  Here is an example of how to grant the necessary privileges:

   GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.*
   TO 'prefetch'@'%' IDENTIFIED BY 'sp33dmeup!';

C<mk-slave-prefetch> learns how long it takes statements to execute, and doesn't
try to execute those that take a very long time.  You can ask it to print what
it has learned after it executes.  You can also specify a filename on the
command line.  The file should contain the statistics printed by a previous
run.  These will be used to pre-populate the statistics so it doesn't have to
re-learn.

This program is based on concepts I heard Paul Tuckfield explain at the November
2006 MySQL Camp un-conference.  However, the code is my own work.  I have not
seen any other implementation of Paul's idea.

=head1 DOES IT WORK?

Does it work?  Does it actually speed up the slave?

That depends on your workload, hardware, and other factors.  It might work when
the following are true:

=over

=item *

The slave's data is much larger than memory, and the workload is mostly randomly
scattered small (single-row is ideal) changes.

=item *

There are lots of high-concurrency C<UPDATE> and C<DELETE> statements on the
master.

=item *

The slave SQL thread is I/O-bound, but the slave overall has plenty of spare I/O
capacity (definitely more than one disk spindle).

=item *

The slave uses InnoDB or another storage engine with row-level locking.

=back

It does B<not> speed up replication on my slaves, which mostly have large
queries like C<INSERT .. SELECT .. GROUP BY>.  In my benchmarks it seemed to
make no difference at all, positive or negative.

On the wrong workload or slave configuration, this technique might actually make
the slaves slower.  Your mileage will vary.

User-contributed benchmarks are welcome.

=head1 OPTIONS

Some options are negatable by specifying them in their long form with a --no
prefix.

=over

=item --askpass

Prompts the user for a password when connecting to MySQL.

=item --charset

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

=item --checkint

How many relay log events should pass before checking the output of C<SHOW
SLAVE STATUS>.  The syntax is a three-number range: initial, minimum, and
maximum.  You should be able to leave this at the defaults.

C<mk-slave-prefetch> varies the check interval in powers of two, depending on
whether it decides the check was necessary.

=item --daemonize

Fork to the background and detach from the shell.  This probably doesn't work
on Microsoft Windows.

=item --database

Connect to this database.  C<mk-slave-prefetch> will issue C<USE> statements
as required by the binary log events.

=item --defaults-file

Only read MySQL options from the given file.

=item --execute

Execute the transformed queries.  This is enabled by default, because this is
what actually warms the caches.

=item --help

Displays a help message.

=item --host

Host to connect to.

=item --iolag

How many bytes to lag the slave I/O thread (to avoid C<mysqlbinlog> reading
right off the end of the relay log file).

=item --maxquerytime

The time limit for queries.  The value can be a floating-point number.  If
C<mk-slave-prefetch> predicts the query will take longer to execute, it will
skip the query.  This is based on the theory that pre-warming the cache is
most beneficial for short queries.

C<mk-slave-prefetch> learns how long queries require to execute.  It keeps an
average over the last L<"--querysampsize"> samples of each query.  The averages
are based on an abstracted version of the query, with specific parameters
replaced by placeholders.  The result is a sort of "fingerprint" for the query,
not executable SQL.  You can see the learned statistics with the
L<"--statistics"> option.

You can pre-load query fingerprints, and average execution times, from a file.
This way you don't have to wait for C<mk-slave-prefetch> to learn all over
every time you start it.  Just specify the file on the command line.  The
format should be the same as the output from L<"--statistics">.

You might also want to filter out some statements completely, or let only some
statements through.  See the L<"--rejectregexp"> and L<"--permitregexp">
options.

If C<mk-slave-prefetch> hasn't seen a query's fingerprint before, and thus
doesn't know how long it will take to execute, it wraps it in a subuery, like
this:

   SELECT 1 FROM ( <query> ) AS X LIMIT 1;

This helps avoid fetching a lot of data back to the client when a query is
very large.  It requires a version of MySQL that supports subqueries (version
4.1 and newer).  If yours doesn't, the subquery trick can't be used, so the
query might fetch a lot of data back to the client.

Once a query's fingerprint has been seen, so it's known that the query isn't
enormously slow, C<mk-slave-prefetch> just rewrites the C<SELECT> list for
efficiency.  (Avoiding the subquery reduces the query's overhead for short
queries).  The rewritten query will then look like the following;

   SELECT ISNULL(COALESCE(<columns>)) FROM ...

=item --offset

The number of bytes C<mk-slave-prefetch> will try to stay in front of the slave
SQL thread.  It will not execute log events it doesn't think are at least this
far ahead of the SQL thread.  See also L<"--window">.

=item --password

The password to use when connecting.

=item --permitregexp

A Perl regular expression to filter log events.  The regular expression is
matched against the raw log event, before any transformations are applied.  If
specified, this option will permit only log events matching the regular
expression.

=item --port

Port number to use for connection.

=item --print

Print the transformed relay log events to standard output.

=item --printnonrewritten

Print queries that could not be transformed into C<SELECT>.

=item --querysampsize

How many queries to average together to get the average query execution time
(see L<"--maxquerytime">).  The last C<N> queries with a given fingerprint are
averaged together.

=item --rejectregexp

Similar to L<"--permitregexp">, but has the opposite effect: log events must
B<not> match the regular expression.

=item --setvars

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

=item --socket

Socket file to use for connection.

=item --statistics

Prints statistics after executing.  The statistics are in two sections:
counters, and queries.  The counters simply count the number of times events
occur.  You may see the following counters:

   NAME                    MEANING
   ======================  =======================================
   mysqlbinlog             Executed mysqlbinlog to read log events.
   events                  The total number of relay log events.
   not_far_enough_ahead    An event was not at least --offset
                           bytes ahead of the SQL thread.
   too_far_ahead           An event was more than --offset
                           + --window bytes ahead of the SQL thread.
   too_close_to_io_thread  An event was less than --iolag bytes
                           away from the I/O thread's position.
   event_not_allowed       An event wasn't a SET, USE, INSERT,
                           UPDATE, DELETE or REPLACE query.
   event_filtered_out      An event was filtered out because of
                           --permitregexp or --rejectregexp.
   same_timestamp          A SET TIMESTAMP event was ignored because
                           it had the same timestamp as the last one.
   do_query                A transformed event was executed
                           or printed.
   query_error             An executed query had an error.
   query_too_long          An event was not executed because its
                           average query length exceeded
                           --maxquerytime.
   query_not_rewritten     A query could not be rewritten to a
                           SELECT.
   master_pos_wait         The tool waited for the SQL thread to
                           catch up.
   show_slave_status       The tool queried SHOW SLAVE STATUS.
   load_data_infile        The tool found a LOAD DATA INFILE query
                           and unlinked (deleted) the temp file.
   could_not_unlink        The tool failed to unlink a temp file.
   sleep                   The tool slept for a second because the 
                           slave's SQL thread was not running, or
                           because it read past the end of the log.

After the counters, C<mk-slave-prefetch> prints information about each query
fingerprint it has seen, two lines per fingerprint.  The first line contains
the query's fingerprint.  The second line contains the number of times the
fingerprint was seen, number of times executed, the sum of the execution
times, and the average execution time over the last L<"--querysampsize">
samples.

=item --time

How long C<mk-slave-prefetch> should run before exiting.  Default is to run
forever.

=item --tmpdir

Where to create temporary files for C<LOAD DATA INFILE> queries inlined in the
log file.  The default is C</dev/null>, which will cause C<mysqlbinlog> to
skip the file and the associated C<LOAD DATA INFILE> command entirely.

If C<mk-slave-prefetch> sees a C<LOAD DATA INFILE> command (which it won't, if
this is left at the default), it will try to remove the temporary file, then
skip the event.

=item --user

User for login if not current user.

=item --version

Output version information and exit.

=item --window

Defines the window within which C<mk-slave-prefetch> considers a query OK to
execute.  The window begins at the slave SQL thread's last known position plus
L<"--offset"> bytes, and extends for the specified number of bytes.

If C<mk-slave-prefetch> sees a log event that is too far in the future, it will
increment the C<too_far_ahead> counter and wait for the slave SQL thread to
catch up (which increments the C<master_pos_wait> counter).  If an event isn't
far enough ahead of the SQL thread, it will be discarded and the
C<not_far_enough_ahead> counter increments.

Watching the mentioned statistics can help you understand how to tune the
window.  You want C<mk-slave-prefetch> to run just ahead of the SQL thread, not
throwing out a lot of events for being too far ahead or not far enough ahead.

=back

=head1 SYSTEM REQUIREMENTS

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

=head1 ENVIRONMENT

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

   MKDEBUG=1 mk-....

=head1 BUGS

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

=head1 COPYRIGHT, LICENSE AND WARRANTY

This program is copyright (c) 2007 Baron Schwartz.
Feedback and improvements are welcome.

THIS PROGRAM IS PROVIDED "AS IS" AND WITHOUT ANY EXPRESS OR IMPLIED
WARRANTIES, INCLUDING, WITHOUT LIMITATION, THE IMPLIED WARRANTIES OF
MERCHANTIBILITY AND FITNESS FOR A PARTICULAR PURPOSE.

This program is free software; you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation, version 2; OR the Perl Artistic License.  On UNIX and similar
systems, you can issue `man perlgpl' or `man perlartistic' to read these
licenses.

You should have received a copy of the GNU General Public License along with
this program; if not, write to the Free Software Foundation, Inc., 59 Temple
Place, Suite 330, Boston, MA  02111-1307  USA.

=head1 AUTHOR

Baron Schwartz.

=head1 VERSION

This manual page documents Ver 1.0.1 Distrib 1877 $Revision: 1871 $.

=cut