#!/usr/bin/perl

# This is a program to load files into MySQL in parallel.
#
# 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';

# ###########################################################################
# This is a combination of modules and programs in one -- a runnable module.
# http://www.perl.com/pub/a/2006/07/13/lightning-articles.html?page=last
# Or, look it up in the Camel book on pages 642 and 643 in the 3rd edition.
# ###########################################################################

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

package OptionParser;

use Getopt::Long;
use List::Util qw(max);

sub new {
   my ( $class, @opts ) = @_;
   bless { specs => \@opts }, $class;
}

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

   my %opt_seen;
   foreach my $spec ( @specs ) {
      my ( $long, $short ) = $spec->{s} =~ m/^([\w-]+)(?:\|([^!+=]*))?/;
      $spec->{k} = $short || $long;
      $spec->{l} = $long;
      $spec->{t} = $short;
      $spec->{n} = $spec->{s} =~ m/!/;
      $defaults{$spec->{k}} = undef unless defined $defaults{$spec->{k}};
      die "Duplicate option $spec->{k}" if $opt_seen{$spec->{k}}++;
   }

   foreach my $key ( keys %defaults ) {
      die "No such option '$key'\n" unless exists $opt_seen{$key};
   }

   Getopt::Long::Configure('no_ignore_case', 'bundling');
   GetOptions( map { $_->{s} => \$defaults{$_->{k}} } @specs ) or $defaults{help} = 1;
   return %defaults;
}

sub usage {
   my ( $self ) = @_;
   my @specs = @{$self->{specs}};
   my $maxw = max(map { length($_->{l}) + ($_->{n} ? 4 : 0)} @specs);
   my $usage = '';
   foreach my $spec ( sort { $a->{l} cmp $b->{l} } @specs ) {
      my $long  = $spec->{n} ? "[no]$spec->{l}" : $spec->{l};
      my $short = $spec->{t} ? "-$spec->{t}" : '';
      $usage .= sprintf("  --%-${maxw}s %-4s %s\n", $long, $short, $spec->{d});
   }
   return $usage;
}

1;

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

package main;

use DBI;
use English qw(-no_match_vars);
use File::Basename qw(dirname);
use File::Find;
use File::Spec;
use List::Util qw(max sum);
use POSIX;
use Time::HiRes qw(time);

our $VERSION = '0.9.0';
our $DISTRIB = '1053';
our $SVN_REV = sprintf("%d", q$Revision: 1049 $ =~ m/(\d+)/g || 0);

# Globals -- as few as possible.
my %opts;
my @conn_params;

if ( !caller ) {

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

   my @opt_spec = (
      { s => 'basedir=s',         d => 'Base directory for creating files (default cwd)' },
      { s => 'bulkinsbufsize=i',  d => 'Set bulk_insert_buffer_size before LOAD DATA INFILE' },
      { s => 'charset=s',         d => 'Set the character set (default=binary)' },
      { s => 'commit',            d => 'Commit after each LOAD DATA INFILE' },
      { s => 'createdb',          d => 'Create databases that do not exist' },
      { s => 'csv',               d => 'Files are in in CSV format (implies --tab)' },
      { s => 'database|D=s',      d => 'Specify the database for all tables' },
      { s => 'databases|d=s',     d => 'Restore only this comma-separated list of databases' },
      { s => 'dbregex=s',         d => 'Restore only databases whose names match this pattern' },
      { s => 'defaults-file|F=s', d => 'Only read default options from the given file' },
      { s => 'disablekeys!',      d => 'Set DISABLE KEYS before loading files' },
      { s => 'fifo!',             d => 'Stream files into a FIFO for --tab (sets --umask 0)' },
      { s => 'help',              d => 'Show this help message' },
      { s => 'host|h=s',          d => 'Connect to host' },
      { s => 'ignoredb|g=s',      d => 'Ignore this comma-separated list of databases' },
      { s => 'ignore|i',          d => 'Use the IGNORE option to LOAD DATA INFILE' },
      { s => 'ignoretbl|n=s',     d => 'Ignore this comma-separated list of tables' },
      { s => 'local|L',           d => 'Use the LOCAL option to LOAD DATA INFILE' },
      { s => 'locktables!',       d => 'Lock tables before LOAD DATA INFILE' },
      { s => 'noautovalon0!',     d => 'Set NO_AUTO_VALUE_ON_ZERO before LOAD DATA INFILE' },
      { s => 'nobinlog!',         d => 'Set SQL_LOG_BIN=0 before LOAD DATA INFILE' },
      { s => 'noforeignkeys!',    d => 'Set FOREIGN_KEY_CHECKS=0 before LOAD DATA INFILE' },
      { s => 'nouniquechecks!',   d => 'Set UNIQUE_CHECKS=0 before LOAD DATA INFILE' },
      { s => 'numthread|m=i',     d => 'Number of threads (default #CPUs or 2)' },
      { s => 'password|p=s',      d => 'Password to use when connecting' },
      { s => 'port|P=i',          d => 'Port number to use for connection' },
      { s => 'quiet|q',           d => 'Set --verbose to 0' },
      { s => 'replace|r',         d => 'Use the REPLACE option to LOAD DATA INFILE' },
      { s => 'socket|S=s',        d => 'Socket file to use for connection' },
      { s => 'sql!',              d => 'Load SQL files first with mysql (default)' },
      { s => 'tables|t=s',        d => 'Restore only this comma-separated list of tables' },
      { s => 'tab|T',             d => 'Restore tab-separated files with LOAD DATA INFILE' },
      { s => 'tblregex=s',        d => 'Restore only tables whose names match this pattern' },
      { s => 'test',     ,        d => 'Print commands instead of executing them' },
      { s => 'truncate',          d => 'TRUNCATE TABLE before LOAD DATA INFILE' },
      { s => 'umask=s',           d => 'Set umask to this value, in octal' },
      { s => 'user|u=s',          d => 'User for login if not current user' },
      { s => 'verbose|v+',        d => 'Verbosity (default 1, can specify multiple times)' },
      { s => 'version',           d => 'Output version information and exit' },
      { s => 'wait|w=s',          d => 'Wait limit when server is down (default 5m)' },
   );

   # Holds command-line options.
   %opts = (
      w        => '5m',
      v        => 1,
      basedir  => File::Spec->curdir(),
      fifo     => 1,
      sql      => 1,
      charset  => 'BINARY',
   );

   my $opt_parser = OptionParser->new(@opt_spec);
   %opts = $opt_parser->parse(%opts);

   # ############################################################################
   # Process options.
   # ############################################################################
   $opts{basedir} = File::Spec->rel2abs($opts{basedir});

   if ( $opts{q} ) {
      $opts{v} = 0;
   }

   if ( $opts{csv} ) {
      $opts{T} = 1;
   }

   if ( $opts{T} ) {
      $opts{disablekeys}    = 1 unless defined $opts{disablekeys};
      $opts{noautovalon0}   = 1 unless defined $opts{noautovalon0};
      $opts{nobinlog}       = 1 unless defined $opts{nobinlog};
      $opts{nouniquechecks} = 1 unless defined $opts{nouniquechecks};
      $opts{noforeignkeys}  = 1 unless defined $opts{noforeignkeys};
   }

   if ( !$opts{m} ) {
      eval {
         # Try to read --numthread from the number of CPUs in /proc/cpuinfo.  This
         # only works on GNU/Linux.
         open my $file, "<", "/proc/cpuinfo"
            or die $OS_ERROR;
         local $INPUT_RECORD_SEPARATOR = undef;
         my $contents = <$file>;
         close $file;
         $opts{m} = scalar( map { $_ } $contents =~ m/(processor)/g );
      };
      $opts{m} ||= $ENV{NUMBER_OF_PROCESSORS}; # MSWin32
      $opts{m} = max(2, $opts{m} || 0);
   }

   if ( !$opts{help} ) {
      if ( !$opts{m} ) {
         warn "You must specify --numthread\n";
         $opts{help} = 1;
      }

      if ( !$opts{help} ) {
         my ($num, $suf ) = $opts{w} =~ m/(\d+)([smhd])$/;
         if ( !defined $num || $num <= 0 ) {
            warn "Invalid --wait argument\n";
            $opts{help} = 1;
         }
         else {
            $opts{w} = $suf eq 's' ? $num            # Seconds
                     : $suf eq 'm' ? $num * 60       # Minutes
                     : $suf eq 'h' ? $num * 3600     # Hours
                     :               $num * 86400;   # Days
         }
      }

      if ( !$opts{help} && !@ARGV ) {
         warn "You did not specify any files to restore\n";
         $opts{help} = 1;
      }

      foreach my $opt ( qw(locktables truncate disablekeys noautovalon0
                           nouniquechecks noforeignkeys nobinlog
                           bulkinsbufsize commit L i r) ) {
         if ( defined $opts{$opt} && !$opts{T} ) {
            warn 'Option ' . (length($opt) > 1 ? '--' : '-')
               . "$opt is ineffective without --tab or --csv\n";
            $opts{help} = 1;
         }
      }
   }

   if ( $opts{fifo} ) {
      if ( !defined $opts{umask} ) {
         $opts{umask} = 0;
      }
   }

   if ( defined $opts{umask} ) {
      umask oct($opts{umask});
   }

   # Gather connection parameters to pass to mysql.  Order matters; mysql
   # will have a problem if --defaults-file isn't first.
   @conn_params = (
      [qw(--defaults-file F)],
      [qw(--host          h)],
      [qw(--password      p)],
      [qw(--port          P)],
      [qw(--socket        S)],
      [qw(--user          u)],
   );
   @conn_params = map { "$_->[0]='$opts{$_->[1]}'" } grep { defined $opts{$_->[1]} } @conn_params;

   if ( $opts{version} ) {
      print "mysql-parallel-restore  Ver $VERSION Distrib $DISTRIB Changeset $SVN_REV\n";
      exit(0);
   }

   if ( $opts{help} ) {
      print "Usage: mysql-parallel-restore <options> PATH [PATH...]\n\n";
      print $opt_parser->usage();
      (my $usage = <<"      USAGE") =~ s/^         //gm;

         mysql-parallel-restore loads files into MySQL in parallel.  For
         more details, please read the documentation:

            perldoc mysql-parallel-restore

      USAGE
      print $usage;
      exit(0);
   }

   # Make comma-separated lists into hashes.
   if ( $opts{d} ) {
      $opts{d} = { map { $_ => 1 } split(/,\s*/, $opts{d}) };
   }
   $opts{g} = { map { $_ => 1 } split(/,\s*/, $opts{g} || '') };
   if ( $opts{t} ) {
      $opts{t} = { map { $_ => 1 } split(/,\s*/, $opts{t}) };
   }
   $opts{n} = { map { $_ => 1 } split(/,\s*/, $opts{n} || '') };
   if ( $opts{e} ) {
      $opts{e} = { map { lc($_) => 1 } split(/,\s*/, $opts{e}) };
   }

   # ############################################################################
   # Connect.
   # ############################################################################
   my $dbh = get_dbh();
   $dbh->{InactiveDestroy}  = 1;         # Don't die on fork().
   $dbh->{FetchHashKeyName} = 'NAME_lc'; # Lowercases all column names for fetchrow_hashref()

   # This signal handler will do nothing but wake up the sleeping parent process
   # and record the exit status and time of the child that exited (as a side
   # effect of not discarding the signal).
   my %exited_children;
   $SIG{CHLD} = sub {
      my $kid;
      while (($kid = waitpid(-1, POSIX::WNOHANG)) > 0) {
         # Must right-shift to get the actual exit status of the child.
         $exited_children{$kid}->{exit_status} = $CHILD_ERROR >> 8;
         $exited_children{$kid}->{exit_time}   = time();
      }
   };

   # ############################################################################
   # Discover files to be restored.
   # ############################################################################
   my @tables_to_do;
   my %files_for_table;
   my %stats;

   # Find directories and files and save them.
   File::Find::find(
      {  no_chdir => 1,
         wanted   => sub {
            my ( $dir, $filename ) = ($File::Find::dir, $File::Find::name);
            if ( -f $filename && $filename !~ m/00_master_data.sql$/ ) {
               my ($vol, $dirs, $file) = File::Spec->splitpath( $filename );
               if ( $file =~ m/\.(?:sql|txt|csv)(?:\.\d+)?(?:\.gz)?$/ ) {
                  my @dirs  = grep { $_ } File::Spec->splitdir($dir);
                  my $db    = $opts{D} || $dirs[-1];
                  my ($tbl) = $file =~ m/^([^.]+)/;

                  if ( ( !$opts{d} || exists($opts{d}->{$db}) )
                     && ( !$opts{dbregex} || $db =~ m/$opts{dbregex}/ )
                     && ( !exists $opts{g}->{$db} )
                     && ( !$opts{t} || exists($opts{t}->{$tbl}) )
                     && ( !$opts{tblregex} || $tbl =~ m/$opts{tblregex}/ )
                  ) {
                     if ( !is_sql_file($file) && !$opts{T} ) {
                        die "$filename isn't a SQL file and you didn't tell me "
                           . "to load tab-delimited files.  Maybe you should "
                           . "specify the --tab option.\n";
                     }
                     $stats{files}++;
                     push @{$files_for_table{$db}->{$tbl}}, $File::Find::name;
                     push @tables_to_do, {
                        D => $db,
                        N => $tbl,
                     };
                  }
               }
            }
         },
      },
      map { File::Spec->rel2abs($_) } @ARGV
   );

   # ############################################################################
   # Canonicalize table list in the order they were discovered, filtering out
   # tables that should not be done.
   # ############################################################################
   {
      my %seen;
      @tables_to_do = grep { !$seen{$_->{D}}->{$_->{N}}++ } @tables_to_do;
      $stats{tables} = scalar(@tables_to_do);
   }

   # #########################################################################
   # Design the format for printing out.
   # #########################################################################
   my ( $maxdb, $maxtbl);
   $maxdb  = max(8, map { length($_->{D}) } @tables_to_do);
   $maxtbl = max(5, map { length($_->{N}) } @tables_to_do);
   my $format = "%-${maxdb}s %-${maxtbl}s %5s %5s %6s %7s";
   info(2, sprintf($format, qw(DATABASE TABLE FILES TIME STATUS THREADS)));

   # #########################################################################
   # Assign the work to child processes.  Initially just start --numthreads
   # number of children.  Each child that exits will trigger a new one to start
   # after that.
   # #########################################################################
   my $start = time();
   my %kids;
   while ( @tables_to_do || %kids ) {

      # Wait for the MySQL server to become responsive.
      my $tries = 0;
      while ( !$dbh->ping && $tries++ < $opts{w} ) {
         sleep(1);
         eval {
            $dbh = get_dbh();
         };
         if ( $EVAL_ERROR ) {
            info(0, 'Waiting: ' . scalar(localtime) . ' ' . mysql_error_msg($EVAL_ERROR));
         }
      }
      if ( $tries >= $opts{w} ) {
         die "Too many retries, exiting.\n";
      }

      # Start a new child process.
      while ( @tables_to_do && $opts{m} > keys %kids ) {
         my $todo = shift @tables_to_do;
         $todo->{time} = time;
         my $pid = fork();
         die "Can't fork: $OS_ERROR" unless defined $pid;
         if ( $pid ) {              # I'm the parent
            $kids{$pid} = $todo;
         }
         else {                     # I'm the child
            my $exit_status = 0;
            $exit_status = do_table(
               @{$todo}{qw(D N)},
               @{$files_for_table{$todo->{D}}->{$todo->{N}}}
            ) || $exit_status;
            exit($exit_status);
         }
      }

      # Possibly wait for child.
      my $reaped = 0;
      foreach my $kid ( keys %exited_children ) {
         my $status = $exited_children{$kid};
         my $todo   = $kids{$kid};
         my $stat   = $status->{exit_status};
         my $time   = $status->{exit_time} - $todo->{time};
         info(2, sprintf($format, @{$todo}{qw(D N)},
            scalar(@{$files_for_table{$todo->{D}}->{$todo->{N}}}),
            sprintf('%.2f', $time), $stat, scalar(keys %kids)));
         $stats{ $stat ? 'failure' : 'success' }++;
         $stats{time} += $time;
         delete $kids{$kid};
         delete $exited_children{$kid};
         $reaped = 1;
      }

      if ( !$reaped ) {
         # Don't busy-wait.  But don't wait forever either, as a child may exit
         # and signal while we're not sleeping, so if we sleep forever we may
         # not get the signal.
         sleep(1);
      }
   }

   $stats{wallclock} = time() - $start;

   info(1, sprintf(
      '%5d tables, %5d files, %5d successes, %2d failures, '
      . '%6.2f wall-clock time, %6.2f load time',
         map {
            $stats{$_} || 0
         } qw(tables files success failure wallclock time)
      ));

   # Exit status is 1 if there were any failures.
   exit( $stats{failure} ? 1 : 0 );

}

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

sub makefifo {
   my $filename = File::Spec->catfile($opts{basedir}, "mpr_fifo_$PID");
   if ( !-p $filename ) {
      if ( -e $filename ) {
         die "Cannot make fifo: $filename exists";
      }
      if ( $opts{test} ) {
         print "mkfifo $filename\n";
      }
      else {
         POSIX::mkfifo($filename, 0777) or die "Cannot make fifo $filename: $OS_ERROR";
      }
   }
   return $filename;
}

sub mysql_error_msg {
   my ( $text ) = @_;
   $text =~ s/^.*?failed: (.*?) at \S+ line (\d+).*$/$1 at line $2/s;
   return $text;
}

# Prints a message.
sub info {
   my ( $level, $msg ) = @_;
   if ( $level <= $opts{v} ) {
      print $msg, "\n";
   }
}

# Actually restores a table.
sub do_table {
   my ( $db, $tbl, @files ) = @_;
   my $exit_status = 0;

   # Sort sql files first.
   if ( $opts{sql} ) {
      @files = reverse sort {
         is_sql_file($a) <=> is_sql_file($b)
      } @files;
   }

   my $dbh;

   if ( $opts{createdb} ) {
      $dbh = get_dbh();
      $dbh->do("CREATE DATABASE IF NOT EXISTS `$db`");
   }

   my ($fifo, $load_from, $loaded);

   foreach my $file ( @files ) {

      if ( is_sql_file($file) ) {
         my @args;
         if ( $file =~ m/\.gz/ ) {
            @args = (qw(gunzip --stdout), qq{'$file'}, qw(| mysql), $db, @conn_params);
         }
         else {
            @args = (qw(mysql), $db, @conn_params, '<', qq{'$file'});
         }
         $exit_status = system_call( @args ) || $exit_status;
      }
      else {

         if ( $file =~ m/\.gz$/ ) {
            if ( $opts{fifo} ) {
               $fifo ||= makefifo();
               $exit_status
                  = system_call(qq{gunzip --stdout '$file' > '$fifo' &}) || $exit_status;
               $load_from = $fifo;
            }
            else {
               $exit_status = system_call(qq{gunzip '$file'}) || $exit_status;
               ( $load_from = $file ) =~ s/\.gz$//;
            }
         }
         else {
            $load_from = $file;
         }

         my $sql;
         my $LOCAL = $opts{L} ? ' LOCAL' : '';
         my $OPT   = $opts{i} ? 'IGNORE' : $opts{r} ? 'REPLACE' : '';
         if ( $opts{csv} ) {
            $sql  = qq{LOAD DATA$LOCAL INFILE ? }
                  . qq{$OPT INTO TABLE `$db`.`$tbl` }
                  . qq{FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\\"' }
                  . qq{LINES TERMINATED BY '\\n' }
                  . qq{CHARACTER SET $opts{charset}};
         }
         elsif ( $opts{T} ) {
            $sql  = qq{LOAD DATA$LOCAL INFILE ? }
                  . qq{$OPT INTO TABLE `$db`.`$tbl` }
                  . qq{CHARACTER SET $opts{charset}};
         }

         if ( $sql ) {
            if ( $opts{test} ) {
               print $sql, "\n";
            }
            else {

               # First loop through.
               if ( !$loaded++ ) {
                  $dbh ||= get_dbh();
                  if ( $opts{locktables} ) {
                     $dbh->do("LOCK TABLES `$db`.`$tbl` WRITE");
                  }
                  if ( $opts{truncate} ) {
                     $dbh->do("TRUNCATE TABLE `$db`.`$tbl`");
                  }
               }

               # Every loop through, set options.
               $dbh ||= get_dbh();
               $dbh->do("USE `$db`"); # For binary logging.
               if ( $opts{disablekeys} ) {
                  $dbh->do("/*!40000 ALTER TABLE `$db`.`$tbl` DISABLE KEYS */");
               }
               if ( $opts{noautovalon0} ) {
                  $dbh->do('/*!40101 SET SQL_MODE="NO_AUTO_VALUE_ON_ZERO" */');
               }
               if ( $opts{nouniquechecks} ) {
                  $dbh->do('SET UNIQUE_CHECKS=0');
               }
               if ( $opts{noforeignkeys} ) {
                  $dbh->do('SET FOREIGN_KEY_CHECKS=0');
               }
               if ( $opts{nobinlog} ) {
                  $dbh->do('SET SQL_LOG_BIN=0');
               }
               if ( $opts{bulkinsbufsize} ) {
                  $dbh->do("SET SESSION bulk_insert_buffer_size=$opts{bulkinsbufsize}");
               }

               eval {
                  $dbh->do($sql, {}, $load_from);
                  $dbh->commit if $opts{commit};
               };
               if ( $EVAL_ERROR ) {
                  die mysql_error_msg($EVAL_ERROR) . " while restoring $db.$tbl";
               }
            }
         }
         else {
            unlink $fifo if $fifo;
            die "I don't understand how to load file $file\n";
         }
      }
   }

   if ( $dbh && $opts{T} ) {
      if ( $opts{disablekeys} ) {
         $dbh->do("/*!40000 ALTER TABLE `$db`.`$tbl` ENABLE KEYS */");
      }
      if ( $opts{locktables} ) {
         $dbh->do("LOCK TABLES `$db`.`$tbl` WRITE");
      }
   }

   if ( !$opts{test} && $fifo ) {
      unlink $fifo;
   }

   $dbh->disconnect() if $dbh;
   return $exit_status;
}

sub is_sql_file {
   my ( $filename ) = @_;
   return $filename =~ m/\.sql(?:\.gz)?$/ ? 1 : 0;
}

sub get_dbh {
   my %conn = (
      F => 'mysql_read_default_file',
      h => 'host',
      P => 'port',
      S => 'mysql_socket'
   );
   my $dsn = 'DBI:mysql:;'
      . join(';', map  { "$conn{$_}=$opts{$_}" } grep { defined $opts{$_} } qw(F h P S))
      . ';mysql_read_default_group=mysql';
   my $dbh = DBI->connect($dsn, @opts{qw(u p)}, { AutoCommit => 0, RaiseError => 1, PrintError => 0 } );
   if ( $opts{charset} ) {
      $dbh->do("SET NAMES $opts{charset}");
      $dbh->do("/*!40101 SET character_set_database=$opts{charset} */");
   }
   return $dbh;
}

sub system_call {
   my ( @cmd ) = @_;
   my $exit_status = 0;
   if ( $opts{test} ) {
      print join(' ', @cmd), "\n";
   }
   else {
      $exit_status = system(join(' ', @cmd));
      # Must right-shift to get the actual exit status of the command.
      # Otherwise the upstream exit() call that's about to happen will get a
      # larger value than it likes, and will just report zero to waitpid().
      $exit_status = $exit_status >> 8;
   }
   return $exit_status;
}

1; # Because this is a runnable module.

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

=pod

=head1 NAME

mysql-parallel-restore - Load files into MySQL in parallel.

=head1 SYNOPSIS

  mysql-parallel-restore /path/to/files
  mysql-parallel-restore --tab /path/to/files

=head1 DESCRIPTION

MySQL Parallel Restore is a way to load SQL or delimited-file dumps into MySQL
in parallel at high speed.  It is especially designed for restoring files
dumped by MySQL Parallel Dump (see L<mysql-parallel-dump>).  It automatically
detects whether a file contains SQL or delimited data from the filename
extension, and either shells out to C<mysql> or executes C<LOAD DATA INFILE>
with the file.  On UNIX-like systems, it will even make a FIFO to decompress
gzipped files for C<LOAD DATA INFILE>.

By default it discovers all files in the directory you specify on the command
line.  It uses the file's parent directory as the database name and the file's
name (up to the first dot) as the table name.  It can deal with files named
like the following:

  dir/tbl.sql
  dir/tbl.txt
  dir/tbl.csv

It is also happy with files that look like this, where C<EXT> is one of the
extensions just listed.

  dir/tbl.EXT.000
  dir/tbl.EXT.000.gz

By default, it loads C<SQL> files first, if they exist, then loads C<CSV> or
C<TXT> files next, in order of the numbers in the filename extension as just
shown.  This makes it easy for you to reload a table's definition followed by
its data, in case you dumped them into separate files (as happens with
C<mysqldump>'s C<--tab> option).

Exit status is 0 if everything went well, 1 if any files failed, and any
other value indicates an internal error.

=head1 OUTPUT

Output depends on verbosity.  When L<"--test"> is given, output includes
commands that would be executed.

When L<"--verbose"> is 0, there is normally no output unless there's an error.

When L<"--verbose"> is 1, there is one line of output for the entire job,
showing how many tables were processed, how many files were loaded with what
status, how much time elapsed, and how much time the parallel load jobs added
up to.

When L<"--verbose"> is 2, there's one line of output per table, showing extra
data such as how many threads were running when each table finished loading:

  DATABASE TABLE            FILES  TIME STATUS THREADS
  sakila   language             2  0.07      0       2
  sakila   film_actor           2  0.07      0       2
  sakila   actor                2  0.06      0       2
  sakila   payment              2  0.07      0       2
  sakila   transport_backup     2  0.05      0       2
  sakila   country              2  0.08      0       2
  sakila   film                 2  0.05      0       2
  sakila   rental               2  0.07      0       2

=head1 SPEED OF PARALLEL LOADING

User-contributed benchmarks are welcome.

=head1 OPTIONS

Some options can be disabled by prefixing them with C<--no>.

=over

=item --basedir

Directory in which temporary files, such as FIFO files (see L<"--fifo">) will be
created.

=item --bulkinsbufsize

Sets the bulk_insert_buffer_size variable before each C<LOAD DATA INFILE>.
Has no effect without L<"--tab">.

=item --charset

Sets the connection, database, and C<LOAD DATA INFILE> character set.  The
default is C<BINARY>, which is the safest value to use for C<LOAD DATA
INFILE>.  Has no effect without L<"--tab">.

=item --commit

Commit between each file it loads via C<LOAD DATA INFILE>.

=item --createdb

Create databases if they don't exist.

=item --csv

Changes L<"--tab"> options so the following C<LOAD DATA INFILE> statement is used:

   LOAD DATA INFILE <filename> INTO TABLE <table>
   FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"'
   LINES TERMINATED BY '\n';

This option implies L<"--tab">.

=item --database

Specifies a database into which all files should be loaded.  Overrides the
database which is normally specified by the directory in which the files live.
Does I<not> specify a default database for the connection.

=item --databases

Load only this comma-separated list of databases.

=item dbregex

Load only databases whose names match this Perl regular expression.

=item --defaults-file

Only read default options from the given file.  You must give an absolute
pathname.

=item --disablekeys

Use C<ALTER TABLE DISABLE KEYS> before loading files into a table.

=item --fifo

Load compressed tab-separated files by piping them into a FIFO and using the
FIFO with C<LOAD DATA INFILE>, instead of by decompressing the files on disk.
Sets L<"--umask"> to 0.

=item --help

Displays a help message.

=item --host

Connect to host.

=item --ignore

Adds the C<IGNORE> modifier to C<LOAD DATA INFILE>.

=item --ignoredb

Do not load this comma-separated list of databases.

=item --ignoretbl

Do not load this comma-separated list of table (not database.table) names.

=item --local

Uses the C<LOCAL> modifier to C<LOAD DATA INFILE>, so the files are read locally
by the client library, not by the server.

=item --locktables

Lock tables before C<LOAD DATA INFILE> (disabled by default).

=item --noautovalon0

Set the SQL mode to C<NO_AUTO_VALUE_ON_ZERO> before C<LOAD DATA INFILE>.

=item --nobinlog

Set C<SQL_LOG_BIN=0> before C<LOAD DATA INFILE>.  This prevents large loads
from being logged to the server's binary log.

=item --noforeignkeys

Set C<FOREIGN_KEY_CHECKS=0 before C<LOAD DATA INFILE>.

=item --nouniquechecks

Set C<UNIQUE_CHECKS=0 before C<LOAD DATA INFILE>.

=item --numthread

Specifies the number of parallel processes to run.  The default is 2 (this is
MySQL Parallel Restore, after all -- 1 is not parallel).  On GNU/Linux machines,
the default is the number of times 'processor' appears in F</proc/cpuinfo>.  On
Windows, the default is read from the environment.  In any case, the default is
at least 2, even when there's only a single processor.

=item --password

Password to use when connecting.

=item --port

Port number to use for connection.

=item --quiet

Sets L<"--verbose"> to 0.

=item --replace

Adds the C<REPLACE> modifier to C<LOAD DATA INFILE>.

=item --socket

Socket file to use for connection.

=item --sql

Causes files with a .sql extension to be loaded (via C<mysql>) before
loading .txt and .csv files via C<LOAD DATA INFILE>.  Enabled by default.

=item --tab

Load via C<LOAD DATA INFILE>, which is similar to what C<mysqlimport> does, but
more flexibly.

Enables the following options, unless they are specifically disabled:
L<"--commit">, L<"--disablekeys">, L<"--noautovalon0">, L<"--nobinlog">,
L<"--nouniquechecks">, L<"--noforeignkeys">.

=item --tables

Restore this comma-separated list of table (not database.table) names.

=item --tblregex

Restore only tables whose names match this Perl regular expression.

=item --test

Print commands instead of executing them.

=item --truncate

Issues C<TRUNCATE TABLE> to delete all rows from a table before loading the
first tab-delimited file into it with C<LOAD DATA INFILE>.

=item --umask

Set the program's C<umask> to this octal value.  This is useful when you want
created files (such as FIFO files) to be readable or writable by other users
(for example, the MySQL server itself).

=item --user

User for login if not current user.

=item --verbose

Sets the verbosity; repeatedly specifying it increments the verbosity.
Default is 1 if not specified.  See L<"OUTPUT">.

=item --version

Output version information and exit.

=item --wait

If the MySQL server crashes during loading, waits until the server comes back
and then continues with the rest of the files.  The value is a number with a
suffix (s=seconds, m=minutes, h=hours, d=days).  MySQL Parallel Restore will
check the server every second until this time is exhausted, at which point it
will give up and exit.

=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 BUGS

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

=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 SEE ALSO

See also L<mysql-parallel-dump>.

=head1 VERSION

This manual page documents Ver 0.9.0 Distrib 1053 $Revision: 1049 $.

=cut