1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
# This program is copyright 2009-2011 Percona Inc.
# 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.
# ###########################################################################
# MockSyncStream package
# ###########################################################################
{
# Package: MockSyncStream
# MockSyncStream simulates a <TableSyncStream> module.
# It's used by mk-upgrade to quickly compare result sets for any differences.
# If any are found, mk-upgrade writes all remaining rows to an outfile.
# This causes RowDiff::compare_sets() to terminate early. So we don't actually
# sync anything. Unlike TableSyncStream, we're not working with a table but an
# arbitrary query executed on two servers.
package MockSyncStream;
use strict;
use warnings FATAL => 'all';
use English qw(-no_match_vars);
use constant MKDEBUG => $ENV{MKDEBUG} || 0;
sub new {
my ( $class, %args ) = @_;
foreach my $arg ( qw(query cols same_row not_in_left not_in_right) ) {
die "I need a $arg argument" unless defined $args{$arg};
}
return bless { %args }, $class;
}
sub get_sql {
my ( $self ) = @_;
return $self->{query};
}
sub same_row {
my ( $self, %args ) = @_;
return $self->{same_row}->($args{lr}, $args{rr});
}
sub not_in_right {
my ( $self, %args ) = @_;
return $self->{not_in_right}->($args{lr});
}
sub not_in_left {
my ( $self, %args ) = @_;
return $self->{not_in_left}->($args{rr});
}
sub done_with_rows {
my ( $self ) = @_;
$self->{done} = 1;
}
sub done {
my ( $self ) = @_;
return $self->{done};
}
sub key_cols {
my ( $self ) = @_;
return $self->{cols};
}
# Do any required setup before executing the SQL (such as setting up user
# variables for checksum queries).
sub prepare {
my ( $self, $dbh ) = @_;
return;
}
# Return 1 if you have changes yet to make and you don't want the MockSyncer to
# commit your transaction or release your locks.
sub pending_changes {
my ( $self ) = @_;
return;
}
# RowDiff::key_cmp() requires $tlb and $key_cols but we're syncing query
# result sets not tables so we can't use TableParser. The following sub
# uses sth attributes to return a pseudo table struct for the query's columns.
sub get_result_set_struct {
my ( $dbh, $sth ) = @_;
my @cols = @{$sth->{NAME}};
my @types = map { $dbh->type_info($_)->{TYPE_NAME} } @{$sth->{TYPE}};
my @nullable = map { $dbh->type_info($_)->{NULLABLE} == 1 ? 1 : 0 } @{$sth->{TYPE}};
my @p = @{$sth->{PRECISION}};
my @s = @{$sth->{SCALE}};
my $struct = {
cols => \@cols,
# collation_for => {}, RowDiff::key_cmp() may need this.
};
for my $i ( 0..$#cols ) {
my $col = $cols[$i];
my $type = $types[$i];
$struct->{is_col}->{$col} = 1;
$struct->{col_posn}->{$col} = $i;
$struct->{type_for}->{$col} = $type;
$struct->{is_nullable}->{$col} = $nullable[$i];
$struct->{is_numeric}->{$col}
= ($type =~ m/(?:(?:tiny|big|medium|small)?int|float|double|decimal|year)/ ? 1 : 0);
$struct->{size}->{$col}
= ($type =~ m/(?:float|double)/) ? "($s[$i],$p[$i])"
: ($type =~ m/(?:decimal)/) ? "($p[$i],$s[$i])"
: ($type =~ m/(?:char|varchar)/ && $p[$i]) ? "($p[$i])"
: undef;
}
return $struct;
}
# Transforms a row fetched with DBI::fetchrow_hashref() into a
# row as if it were fetched with DBI::fetchrow_arrayref(). That is:
# the hash values (i.e. column values) are returned as an arrayref
# in the correct column order (because hashes are randomly ordered).
# This is used in mk-upgrade.
sub as_arrayref {
my ( $sth, $row ) = @_;
my @cols = @{$sth->{NAME}};
my @row = @{$row}{@cols};
return \@row;
}
sub _d {
my ($package, undef, $line) = caller 0;
@_ = map { (my $temp = $_) =~ s/\n/\n# /g; $temp; }
map { defined $_ ? $_ : 'undef' }
@_;
print STDERR "# $package:$line $PID ", join(' ', @_), "\n";
}
1;
}
# ###########################################################################
# End MockSyncStream package
# ###########################################################################
|