Annotation of /bin/MasterInit

#!/usr/bin/perl -w
# MasterInit
# Vadim Mikheev, (c) 2000, PostgreSQL Inc.

eval '(exit $?0)' && eval 'exec perl -S $0 ${1+"$@"}'
    & eval 'exec perl -S $0 $argv:q'
    if 0;

use strict;
use Pg;
use Getopt::Long;

$| = 1;

my ($debug,$verbose) = (0,0);
my ($help,$masterhost,$masterport,$masteruser,$masterpassword);
my $lib;

my $result = GetOptions(
        "debug!" => \$debug, "verbose!" => \$verbose, "help" => \$help,
        "masterhost=s" => \$masterhost, "masterport=i" => \$masterport,
        "masteruser=s" => \$masteruser, "masterpassword=s" => \$masterpassword,
        "lib=s" => \$lib);

if (defined($help) || (scalar(@ARGV) < 1)) {
    print "Usage: $0 [options] masterdb
Options:
        --masterhost=hostname --masterport=port
        --masteruser=username --masterpassword=string
        --lib=libpath
";
    exit ((scalar(@ARGV) < 1)? 1:0);
}

my $master = $ARGV[0] || "master";

my $minfo = "dbname=$master";
$minfo = "$minfo host=$masterhost" if (defined($masterhost));
$minfo = "$minfo port=$masterport" if (defined($masterport));
$minfo = "$minfo user=$masteruser" if (defined($masteruser));
$minfo = "$minfo password=$masterpassword" if (defined($masterpassword));

if (!defined($lib) || !-e $lib) {
        # find my compiled rserv.so module
        $lib = $0; $lib =~ s#/[^/]+$#/../lib#;
        if ($lib =~ m#^\.#) {
                my $pwd = `pwd`;
                chomp($pwd);
                $lib = "$pwd/$lib/";
        }
        while ($lib =~ s#/[^/]+/\.\./#/#g) {};
        while ($lib =~ s#/\./#/#g) {};
        $lib =~ s#//#/#g;

        if (-e "$lib/rserv.so") {
                $lib .= "rserv.so";
        } else {
                print STDERR "Can't find compiled rserv.so in $lib. Go there and type make.\n";
                exit 1;
        }
}
print "Using lib '$lib'\n" if ($verbose);

sub RollbackAndQuit {
    my $conn = shift @_;

    print STDERR "Error in query: ", $conn->errorMessage;
    $conn->exec("ROLLBACK");
    exit (-1);
}

my $conn = Pg::connectdb($minfo);
if ($conn->status != PGRES_CONNECTION_OK) {
    print STDERR "Failed opening $minfo\n";
    exit 1;
}

$result = $conn->exec("BEGIN");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

$result = $conn->exec("set transaction isolation level serializable");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# List of slave servers
$result = $conn->exec("create table _RSERV_SERVERS_" .
                      " (server serial primary key, host text not ".
                      "null, port int4 default 5432, dbase text not ".
                      "null, unique(host,port,dbase))");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# List of replicated tables
$result = $conn->exec("create table _RSERV_TABLES_" .
                      " (tname name not null, cname name not null, ".
                      "reloid oid primary key, key int4 not null)");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

## should always call MasterDelTable
#$result = $conn->exec("CREATE RULE _rserv_deltrig_ AS ON delete to _RSERV_TABLES_ DO (DELETE FROM pg_trigger WHERE tgname='_rserv_trigger_t_' AND tgrelid=(SELECT oid FROM pg_class WHERE relname=old.tname);UPDATE pg_class SET reltriggers=reltriggers-1 WHERE relname=old.tname)");
#RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# Bookkeeping log for row replication   
$result = $conn->exec("create table _RSERV_LOG_" .
                      " (reloid oid REFERENCES _RSERV_TABLES_(reloid) ON ".
                      "DELETE CASCADE ON UPDATE CASCADE, logid int4 not ".
                      "null, logtime timestamp not null, insert smallint, ".
                      "update smallint, delete smallint, key text, ".
                      "CONSTRAINT only_one CHECK (insert+update+delete=1))");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# This is to speedup lookup of deleted tuples
$result = $conn->exec("create index _RSERV_LOG_INDX_DLT_ID_ on _RSERV_LOG_ ".
                      "(delete, logid) WHERE delete = 1");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# This is to speedup lookup of updated tuples
$result = $conn->exec("create index _RSERV_LOG_INDX_UPD_ID_ on _RSERV_LOG_ ".
                      "(update, logid) WHERE update = 1");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# This is to speedup lookup of insert tuples
$result = $conn->exec("create index _RSERV_LOG_INDX_INS_ID_ on _RSERV_LOG_ ".
                      "(insert, logid) WHERE insert = 1");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# This is to speedup cleanup
$result = $conn->exec("create index _RSERV_LOG_INDX_TM_ID_ on _RSERV_LOG_ ".
                      "(logtime, logid)");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# This is to speedup trigger 
$result = $conn->exec("create index _RSERV_LOG_INDX_REL_KEY_ on _RSERV_LOG_ ".
                      "(reloid, key)");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# View to help managing _rserv_log_ table
$result = $conn->exec("CREATE VIEW _RSERV_HUMAN_LOG_ AS SELECT log.logid, ".
                      "tab.tname AS table_name, tab.cname AS column_name, ".
                      "log.key AS column_value, log.insert, log.update, ".
                      "log.delete, log.logtime FROM _RSERV_LOG_ log, ".
                      "_RSERV_TABLES_ tab WHERE tab.reloid ".
                      "= log.reloid ORDER BY log.logtime");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# View to help logging daily transactions
$result = $conn->exec("CREATE VIEW _RSERV_DAILY_LOG_ AS ".
                      "SELECT count(*) AS \"# records\", ".
                      "to_char(_rserv_log_.logtime, 'YYYY-MM-DD') ".
                      "AS day FROM _rserv_log_ GROUP BY day");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# Sync point for each slave server
$result = $conn->exec("create table _RSERV_SYNC_ " .
                      "(server int REFERENCES _RSERV_SERVERS_(server) ".
                      "ON DELETE CASCADE ON UPDATE CASCADE, syncid int4 ".
                      "not null, synctime timestamp, status int4 not null,".
                      " minid int4 not null, maxid int4 not null, active text)");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

$result = $conn->exec("create index _RSERV_SYNC_INDX_SRV_ID_ on _RSERV_SYNC_ ".
                      "(server, syncid)");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

# Sync point reference numbers
$result = $conn->exec("create sequence _rserv_sync_seq_");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

$result = $conn->exec("CREATE FUNCTION _rserv_log_() RETURNS opaque" .
                      " AS '$lib' LANGUAGE 'c'");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

$result = $conn->exec("CREATE FUNCTION _rserv_sync_(int4) RETURNS int4" .
                      " AS '$lib' LANGUAGE 'c'");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

$result = $conn->exec("CREATE FUNCTION _rserv_debug_(int4) RETURNS int4" .
                      " AS '$lib' LANGUAGE 'c'");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

$result = $conn->exec("COMMIT");
RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);

exit (0);
1	dpavlin	1.5	#!/usr/bin/perl -w
2	dpavlin	1.1	# MasterInit
3			# Vadim Mikheev, (c) 2000, PostgreSQL Inc.
4
5			eval '(exit $?0)' && eval 'exec perl -S $0 ${1+"$@"}'
6			& eval 'exec perl -S $0 $argv:q'
7			if 0;
8
9	dpavlin	1.5	use strict;
10	dpavlin	1.1	use Pg;
11			use Getopt::Long;
12
13			$\| = 1;
14
15	dpavlin	1.5	my ($debug,$verbose) = (0,0);
16			my ($help,$masterhost,$masterport,$masteruser,$masterpassword);
17	dpavlin	1.6	my $lib;
18	dpavlin	1.1
19	dpavlin	1.5	my $result = GetOptions(
20			"debug!" => \$debug, "verbose!" => \$verbose, "help" => \$help,
21			"masterhost=s" => \$masterhost, "masterport=i" => \$masterport,
22			"masteruser=s" => \$masteruser, "masterpassword=s" => \$masterpassword,
23			"lib=s" => \$lib);
24
25			if (defined($help) \|\| (scalar(@ARGV) < 1)) {
26			print "Usage: $0 [options] masterdb
27			Options:
28			--masterhost=hostname --masterport=port
29			--masteruser=username --masterpassword=string
30			--lib=libpath
31			";
32	dpavlin	1.1	exit ((scalar(@ARGV) < 1)? 1:0);
33			}
34
35			my $master = $ARGV[0] \|\| "master";
36
37			my $minfo = "dbname=$master";
38	dpavlin	1.5	$minfo = "$minfo host=$masterhost" if (defined($masterhost));
39			$minfo = "$minfo port=$masterport" if (defined($masterport));
40			$minfo = "$minfo user=$masteruser" if (defined($masteruser));
41			$minfo = "$minfo password=$masterpassword" if (defined($masterpassword));
42	dpavlin	1.3
43	dpavlin	1.6	if (!defined($lib) \|\| !-e $lib) {
44			# find my compiled rserv.so module
45			$lib = $0; $lib =~ s#/[^/]+$#/../lib#;
46			if ($lib =~ m#^\.#) {
47			my $pwd = `pwd`;
48			chomp($pwd);
49			$lib = "$pwd/$lib/";
50			}
51			while ($lib =~ s#/[^/]+/\.\./#/#g) {};
52			while ($lib =~ s#/\./#/#g) {};
53			$lib =~ s#//#/#g;
54
55			if (-e "$lib/rserv.so") {
56			$lib .= "rserv.so";
57			} else {
58			print STDERR "Can't find compiled rserv.so in $lib. Go there and type make.\n";
59			exit 1;
60			}
61			}
62			print "Using lib '$lib'\n" if ($verbose);
63
64	dpavlin	1.1	sub RollbackAndQuit {
65	dpavlin	1.5	my $conn = shift @_;
66	dpavlin	1.1
67			print STDERR "Error in query: ", $conn->errorMessage;
68			$conn->exec("ROLLBACK");
69			exit (-1);
70			}
71
72			my $conn = Pg::connectdb($minfo);
73			if ($conn->status != PGRES_CONNECTION_OK) {
74			print STDERR "Failed opening $minfo\n";
75			exit 1;
76			}
77
78	dpavlin	1.5	$result = $conn->exec("BEGIN");
79	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
80
81			$result = $conn->exec("set transaction isolation level serializable");
82			RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
83
84			# List of slave servers
85			$result = $conn->exec("create table _RSERV_SERVERS_" .
86	dpavlin	1.3	" (server serial primary key, host text not ".
87			"null, port int4 default 5432, dbase text not ".
88			"null, unique(host,port,dbase))");
89	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
90
91			# List of replicated tables
92			$result = $conn->exec("create table _RSERV_TABLES_" .
93	dpavlin	1.3	" (tname name not null, cname name not null, ".
94			"reloid oid primary key, key int4 not null)");
95	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
96
97	dpavlin	1.3	## should always call MasterDelTable
98			#$result = $conn->exec("CREATE RULE _rserv_deltrig_ AS ON delete to _RSERV_TABLES_ DO (DELETE FROM pg_trigger WHERE tgname='_rserv_trigger_t_' AND tgrelid=(SELECT oid FROM pg_class WHERE relname=old.tname);UPDATE pg_class SET reltriggers=reltriggers-1 WHERE relname=old.tname)");
99			#RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
100
101			# Bookkeeping log for row replication
102	dpavlin	1.1	$result = $conn->exec("create table _RSERV_LOG_" .
103	dpavlin	1.3	" (reloid oid REFERENCES _RSERV_TABLES_(reloid) ON ".
104			"DELETE CASCADE ON UPDATE CASCADE, logid int4 not ".
105			"null, logtime timestamp not null, insert smallint, ".
106			"update smallint, delete smallint, key text, ".
107			"CONSTRAINT only_one CHECK (insert+update+delete=1))");
108	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
109
110			# This is to speedup lookup of deleted tuples
111	dpavlin	1.3	$result = $conn->exec("create index _RSERV_LOG_INDX_DLT_ID_ on _RSERV_LOG_ ".
112			"(delete, logid) WHERE delete = 1");
113			RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
114
115			# This is to speedup lookup of updated tuples
116			$result = $conn->exec("create index _RSERV_LOG_INDX_UPD_ID_ on _RSERV_LOG_ ".
117			"(update, logid) WHERE update = 1");
118			RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
119
120			# This is to speedup lookup of insert tuples
121			$result = $conn->exec("create index _RSERV_LOG_INDX_INS_ID_ on _RSERV_LOG_ ".
122			"(insert, logid) WHERE insert = 1");
123	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
124
125			# This is to speedup cleanup
126	dpavlin	1.3	$result = $conn->exec("create index _RSERV_LOG_INDX_TM_ID_ on _RSERV_LOG_ ".
127			"(logtime, logid)");
128			RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
129
130			# This is to speedup trigger
131			$result = $conn->exec("create index _RSERV_LOG_INDX_REL_KEY_ on _RSERV_LOG_ ".
132			"(reloid, key)");
133			RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
134
135			# View to help managing _rserv_log_ table
136			$result = $conn->exec("CREATE VIEW _RSERV_HUMAN_LOG_ AS SELECT log.logid, ".
137			"tab.tname AS table_name, tab.cname AS column_name, ".
138			"log.key AS column_value, log.insert, log.update, ".
139			"log.delete, log.logtime FROM _RSERV_LOG_ log, ".
140			"_RSERV_TABLES_ tab WHERE tab.reloid ".
141			"= log.reloid ORDER BY log.logtime");
142	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
143
144	dpavlin	1.3	# View to help logging daily transactions
145			$result = $conn->exec("CREATE VIEW _RSERV_DAILY_LOG_ AS ".
146			"SELECT count(*) AS \"# records\", ".
147			"to_char(_rserv_log_.logtime, 'YYYY-MM-DD') ".
148			"AS day FROM _rserv_log_ GROUP BY day");
149	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
150
151			# Sync point for each slave server
152	dpavlin	1.3	$result = $conn->exec("create table _RSERV_SYNC_ " .
153			"(server int REFERENCES _RSERV_SERVERS_(server) ".
154			"ON DELETE CASCADE ON UPDATE CASCADE, syncid int4 ".
155			"not null, synctime timestamp, status int4 not null,".
156			" minid int4 not null, maxid int4 not null, active text)");
157	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
158
159	dpavlin	1.3	$result = $conn->exec("create index _RSERV_SYNC_INDX_SRV_ID_ on _RSERV_SYNC_ ".
160			"(server, syncid)");
161	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
162
163			# Sync point reference numbers
164			$result = $conn->exec("create sequence _rserv_sync_seq_");
165			RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
166
167	dpavlin	1.3	$result = $conn->exec("CREATE FUNCTION _rserv_log_() RETURNS opaque" .
168			" AS '$lib' LANGUAGE 'c'");
169	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
170
171			$result = $conn->exec("CREATE FUNCTION _rserv_sync_(int4) RETURNS int4" .
172	dpavlin	1.3	" AS '$lib' LANGUAGE 'c'");
173	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
174
175			$result = $conn->exec("CREATE FUNCTION _rserv_debug_(int4) RETURNS int4" .
176	dpavlin	1.3	" AS '$lib' LANGUAGE 'c'");
177	dpavlin	1.1	RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
178
179			$result = $conn->exec("COMMIT");
180			RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
181
182			exit (0);