File Coverage

blib/lib/MySQL/Binlog_RBR_Data.pm

Criterion	Covered	Total	%
statement	68	77	88.3
branch	41	52	78.8
condition	15	30	50.0
subroutine	6	6	100.0
pod	1	1	100.0
total	131	166	78.9

package MySQL::Binlog_RBR_Data;

use 5.008005;

# be as strict and verbose as possible
use warnings;
use strict;

# set version info
our $VERSION = "1.02";

use Carp qw( confess croak );

sub parse {
    my ( $class, $handle, $start_position, @interesting_tables ) = @_;

    my %interesting_tables;
    @interesting_tables{ @interesting_tables } = @_;

    my $pos;
    my @transactions;
    my $trans;

    # mysqlbinlog outputs the binlog data between two lines with:
    # DELIMITER
    # so start working after the first one
    while ( <$handle> ) {
        if ( substr( $_, 0, 5 ) eq '# at ' ) {
            $pos = 0 + substr( $_, 5 );
            next;
        }

        last
            if substr( $_, 0, 9 ) eq 'DELIMITER';
    }

    # it's an error if we didn't get the DELIMITER, probably
    # not a binlog file or a corrupted one
    $_ && substr( $_, 0, 9 ) eq 'DELIMITER'
        or croak "not a binlog file, didn't find DELIMITER";

    # want to start at some position? mysqlbinlog seeks on it, but we
    # must parse the full output until we get at that position
    if ( $start_position ) {
        while ( <$handle> ) {
            if ( substr( $_, 0, 5 ) eq '# at ' ) {
                $pos = 0 + substr( $_, 5 );

                last if $pos >= $start_position;
            }

            # got to EOF?
            last
                if substr( $_, 0, 9 ) eq 'DELIMITER';
        }

        if ( $pos < $start_position ) {
            # no guarantee $. points to 'at'
            croak "position $start_position not found in binlog, last one found was: $pos";
        }
        elsif ( $pos > $start_position ) {
            croak "position $start_position not found in binlog, next one after was: $pos, at $.";
        }
    }

    my ( $new_row, $old_row, $row );
    my $end_pos;

    return sub {
        while ( <$handle> ) {
            # position update
            if ( substr( $_, 0, 5 ) eq '# at ' ) {
                $pos = 0 + substr( $_, 5 );
                next;
            }

            # EOF?
            last
                if substr( $_, 0, 9 ) eq 'DELIMITER';

            # new transaction?
            # transactions have a BEGIN and a COMMIT
            if ( substr( $_, 0, 5 ) eq 'BEGIN' ) {
                my %transaction;

                while ( <$handle> ) {
                    # EOT?
                    last
                        if substr( $_, 0, 6 ) eq 'COMMIT';

                    # end of current record indicated by
                    # # end_log_pos
                    if ( substr( $_, 0, 2 ) eq '#1' # FIXME in 2020
                            && /^#\d.*end_log_pos (\d+)/ ) {
                        $end_pos = $1;
                        next;
                    }

                    # not RBR? not interested
                    substr( $_, 0, 4 ) eq '### '
                        or next;

                    # RBR statements format:

                    # INSERT INTO ...
                    # SET
                    #   values...

                    # UPDATE ...
                    # WHERE
                    #   old values...
                    # SET
                    #   new values...

                    # DELETE FROM ...
                    # WHERE
                    #   old values...

                    # so... SET defines new values
                    #       WHERE defines old values

                    if ( $old_row && substr( $_, 4, 5 ) eq 'WHERE' ) {
                        $row = $old_row;
                    }
                    elsif ( $new_row && substr ( $_, 4, 3 ) eq 'SET' ) {
                        $row = $new_row;
                    }
                    elsif ( /^### INSERT INTO (\S+)/ ) {
                        $old_row = $row = undef;

                        if ( ! @interesting_tables
                            || exists $interesting_tables{ $1 } ) {
                            $new_row = [];
                            push @{ $transaction{ $1 } }, [ $new_row ];
                        }
                        else {
                            $new_row = undef;
                        }
                    }
                    elsif ( /^### DELETE FROM (\S+)/ ) {
                        $new_row = $row = undef;

                        if ( ! @interesting_tables
                            || exists $interesting_tables{ $1 } ) {
                            $old_row = [];
                            push @{ $transaction{ $1 } }, [ undef, $old_row ];
                        }
                        else {
                            $old_row = undef;
                        }
                    }
                    elsif ( /^### UPDATE (\S+)/ ) {
                        $row = undef;

                        if ( ! @interesting_tables
                            || exists $interesting_tables{ $1 } ) {
                            $new_row = [];
                            $old_row = [];
                            push @{ $transaction{ $1 } }, [ $new_row, $old_row ];
                        }
                        else {
                            $new_row = $old_row = undef;
                        }
                    }
                    elsif ( $row && /@(\d+)=(.+)/ ) {
                        $row->[ $1 - 1 ] = $2;
                    }
                }

                # did actually get EOT?
                $_ && substr( $_, 0, 6 ) eq 'COMMIT'
                    or croak "truncated binlog file, at " . $handle->input_line_number() . ": $_";

                # last 'end_log_pos' seen
                $transaction{ end_position } = $end_pos;
                # first 'at' seen
                $transaction{ start_position } = $pos;

                return \%transaction;
            }
        }

        # actually got EOF?
        $_ && substr( $_, 0, 9 ) eq 'DELIMITER'
            or croak "truncated binlog file, at " . $handle->input_line_number() . ": $_";

        return;
    };
}

1;

__END__