Fix for 1052722: pt-fifo-split is processing n-1 rows initially

bin/pt-fifo-split

@@ -1284,13 +1284,9 @@ sub main {
    if ( $o->get('force') && -e $file ) {
       unlink($file) or die "Can't unlink $file: $OS_ERROR";
    }
-   mkfifo($file, 0777) or die "Can't make fifo $file: $OS_ERROR";
 
    my $fh;
 
-   $fh = IO::File->new($file, '>') or die "Can't open $file: $OS_ERROR";
-   $fh->autoflush(1);
-
    if ( $o->get('statistics') ) {
       printf("%5s %9s %5s %8s %8s\n", qw(chunks lines time overall current));
    }
@@ -1299,21 +1295,25 @@ sub main {
    my $OFFSET = $o->get('offset');
    my $LINES  = $o->get('lines');
 
-   my $lines  = 0;
+   my $chunks  = 0;
-   my $chunks = 0;
+   my $start   = time();
-   my $start  = time();
+   my $cstart  = time();
-   my $cstart = time();
+   my $printed = 0;
    while ( my $line = <> ) {
-      $lines++;
+      my $lines = $INPUT_LINE_NUMBER;
       next if $OFFSET && $lines < $OFFSET;
-      if ( $lines % $LINES == 0 ) {
+      if ( $printed == 0 ) {
+         mkfifo($file, 0777) or die "Can't make fifo $file: $OS_ERROR";
+         $fh = IO::File->new($file, '>') or die "Can't open $file: $OS_ERROR";
+         $fh->autoflush(1);
+      }
+      print $fh $line or die "Can't print: $OS_ERROR";
+      $printed++;
+      if ( ($lines % $LINES) == 0 ) {
          close $fh or die "Can't close: $OS_ERROR";
 
          unlink($file) or die "Can't unlink $file: $OS_ERROR";
-         mkfifo($file, 0777) or die "Can't make fifo $file: $OS_ERROR";
+         $printed = 0;
-
-         $fh = IO::File->new($file, '>') or die "Can't open $file: $OS_ERROR";
-         $fh->autoflush(1);
 
          $chunks++;
          my $end = time();
@@ -1325,11 +1325,10 @@ sub main {
          }
          $cstart = $end;
       }
-      print $fh $line or die "Can't print: $OS_ERROR";
    }
-   close $fh or die "Can't close: $OS_ERROR";
 
-   unlink($file) or die "Can't unlink $file: $OS_ERROR";
+   close $fh or die "Can't close: $OS_ERROR" if $fh && $fh->opened;
+   unlink($file) or die "Can't unlink $file: $OS_ERROR" if -e $file;
 
    return 0;
 }

t/pt-fifo-split/pt-fifo-split.t

@@ -9,6 +9,7 @@ BEGIN {
 use strict;
 use warnings FATAL => 'all';
 use English qw(-no_match_vars);
+use File::Temp qw(tempfile);
 use Test::More;
 
 if ( !$ENV{SLOW_TESTS} ) {
@@ -26,6 +27,62 @@ my $cmd = "$trunk/bin/pt-fifo-split";
 my $output = `$cmd --help`;
 like($output, qr/Options and values/, 'It lives');
 
+require IO::File;
+my ($fh, $filename) = tempfile("pt-fifo-split-data.XXXXXXXXX", OPEN => 1, TMPDIR => 1, UNLINK => 1);
+$fh->autoflush(1);
+print { $fh } "$_\n" for 1..9;
+
+local $SIG{CHLD} = 'IGNORE';
+my $pid = fork();
+die "Cannot fork: $OS_ERROR" unless defined $pid;
+if ( !$pid ) {
+   exec { $cmd } $cmd, qw(--lines 2), $filename;
+   exit 1;
+}
+
+PerconaTest::wait_for_files($fifo);
+my @fifo;
+while (kill 0, $pid) {
+   push @fifo, slurp_file($fifo) if -e $fifo;
+}
+waitpid($pid, 0);
+
+is_deeply(
+   \@fifo,
+   [
+      "1\n2\n",
+      "3\n4\n",
+      "5\n6\n",
+      "7\n8\n",
+      "9\n",
+   ],
+   "--lines=2 with 9 lines works as expected"
+);
+
+$pid = fork();
+die "Cannot fork: $OS_ERROR" unless defined $pid;
+if ( !$pid ) {
+   exec { $cmd } $cmd, qw(--lines 15), $filename;
+   exit 1;
+}
+PerconaTest::wait_for_files($fifo);
+
+@fifo = ();
+while (kill 0, $pid) {
+   push @fifo, slurp_file($fifo) if -e $fifo;
+}
+waitpid($pid, 0);
+
+is_deeply(
+   \@fifo,
+   [
+      "1\n2\n3\n4\n5\n6\n7\n8\n9\n",
+   ],
+   "--lines=15 with 9 lines works as expected"
+);
+
+close $fh or die "Cannot close $filename: $OS_ERROR";
+
 system("($cmd --lines 10000 $trunk/bin/pt-fifo-split > /dev/null 2>&1 < /dev/null)&");
 PerconaTest::wait_for_files($fifo);
 
@@ -48,6 +105,8 @@ is($contents, <<EOF
 EOF
 , 'Offset works');
 
+#>"
+
 # #########################################################################
 # Issue 391: Add --pid option to all scripts
 # #########################################################################