rserv/share/RServ.pm

# -*- perl -*-
# RServ.pm
# Vadim Mikheev, (c) 2000, PostgreSQL Inc.

package RServ;

require Exporter;
@ISA = qw(Exporter);
@EXPORT = qw(PrepareSnapshot ApplySnapshot GetSyncID SyncSyncID CleanLog GetServerId
        Rollback RollbackAndQuit Connect Exec Exec2 MkInfo ExecDebug
        $debug $quiet $verbose
        );
@EXPORT_OK = qw();
use strict;
use Pg;

my $debug = 0;
my $quiet = 1;
my $verbose = 0;

$debug = 1;
$quiet = 0;
$verbose = 1;

my %Mtables = ();
my %Stables = ();

sub GetServerId
{
    my ($mconn, $DB, $Host) = @_; # (@_[0], @_[1]. @_[2]);

    print STDERR "GetServerId: host $Host, database $DB\n" if ($debug);

    my $result = $mconn->exec("SELECT server FROM _RSERV_SERVERS_ WHERE".
                          " host='$Host' AND dbase='$DB'");

    if ($result->resultStatus ne PGRES_TUPLES_OK)
    {
        print STDERR $mconn->errorMessage unless ($quiet);
        return(-1);
    }
    
    if ($result->cmdTuples && $result->cmdTuples > 1)
    {
        printf STDERR "Duplicate host definitions.\n" unless ($quiet);
        return(-2);
    }

    my @row = $result->fetchrow;

    print STDERR "GetServerId($DB,$Host) == $row[0]\n" if ($debug);

    return $row[0];
}

sub PrepareSnapshot
{
    my ($mconn, $sconn, $outf, $mserver, $sserver, $multimaster, $onlytables) = @_;

    if ($mserver == $sserver) {
        print STDERR "master and slave numbers are same [$mserver] !\n";
        return(-1);
    }

    print STDERR "PrepareSnapshot master: $mserver slave: $sserver\n" if ($debug);

    # dump master server ID into snapshot file (to prevent replication
    # of colums from master back to slave)
    print $outf "-- SERVER $mserver\n";

    # first, we must know for wich tables the slave subscribed
    my $result = Exec($sconn,"SELECT tname FROM _RSERV_SLAVE_TABLES_", -1);
    return (-1) if ($result == -1);

    my @row;
    while (@row = $result->fetchrow) {
        $Stables{$row[0]} = 1;
    }
    
    print STDERR "Prepare snapshot for tables: ",join(",",keys %Stables),"\n" if ($debug);

    Exec($mconn,"BEGIN");
    Exec($mconn,"set transaction isolation level serializable");
    
    # MAP oid --> tabname, keyname, key_type
    my $sql = qq{
        select pgc.oid, pgc.relname, pga.attname, pgt.typname
                from _RSERV_TABLES_ rt, pg_class pgc, pg_attribute pga,
                        pg_type pgt
                where pgc.oid = rt.reloid AND pga.attrelid = rt.reloid
                        AND pga.attnum = rt.key AND pga.atttypid=pgt.oid
    };
    $result = Exec($mconn,$sql);

    while (@row = $result->fetchrow)
    {
                printf "$row[0], $row[1], $row[2]\n" if ($debug);
                if (ref($onlytables) eq 'HASH') {
                        next unless (exists $onlytables->{$row[1]});
                        $onlytables->{$row[1]} = $row[0] unless ($onlytables->{$row[1]});
                }
        push @{$Mtables{$row[0]}}, $row[1], $row[2], $row[3];
    }

    print STDERR "Master database table oids: ",join(",",keys %Mtables),"\n" if ($debug);
    if (! %Mtables) {
        print STDERR "FATAL: can't find oids for tables in master! Did you run SlaveAddTable?\n";
        RollbackAndQuit($mconn);
    }

    # Read last succeeded sync
    $sql = qq{
        select syncid, synctime, minid, maxid, active from _RSERV_SYNC_
        where server = $sserver AND syncid =
                (select max(syncid) from _RSERV_SYNC_
                        where server = $sserver AND status > 0)
    };
    
    $result = Exec($mconn,$sql);
 
    my @lastsync = $result->fetchrow;
    print STDERR "lastsync: ",join(",",@lastsync),"\n" if ($debug);

    # exclude data which originated from master server
    my $sel_server = " and l.server = $mserver ";

    my $sinfo = "";
    if (@lastsync && $lastsync[3] ne '')        # sync info
    {
        $sinfo = "and (l.logid >= $lastsync[3]";
        $sinfo .= " or l.logid in ($lastsync[4])" if $lastsync[4] ne '';
        $sinfo .= ")";
    }

    my @keys;   # keys in this snapshot

    my $havedeal = 0;
    
    # DELETED rows
    $sql = "select l.reloid, l.key from _RSERV_LOG_ l" .
        " where l.delete = 1 $sinfo $sel_server order by l.reloid";
    
    printf "DELETED: $sql\n" if $debug;
    
    $result = $mconn->exec($sql);
    if ($result->resultStatus ne PGRES_TUPLES_OK) {
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    
    my $lastoid = -1;
    while (@row = $result->fetchrow) {
        next unless exists $Mtables{$row[0]};
        next unless exists $Stables{$Mtables{$row[0]}[0]};

        if ($lastoid != $row[0]) {
            if ($lastoid == -1) {
                my $syncid = GetSYNCID($mconn, $outf);
                return($syncid) if $syncid < 0;
                $havedeal = 1;
            } else {
                printf $outf "\\.\n";
            }
            printf $outf "-- DELETE $Mtables{$row[0]}[0]\n";
            $lastoid = $row[0];
        }
        if (! defined $row[1]) {
            print STDERR "NULL key\n" unless ($quiet);
            $mconn->exec("ROLLBACK");
            return(-2);
        }
        printf $outf "%s\n", OutputValue($row[1]);
        push @keys,OutputKey($row[2],$Mtables{$row[0]}[2]);
    }
    printf $outf "\\.\n" if ($lastoid != -1);
    
    # UPDATED rows
    
    my ($taboid, $tabname, $tabkey);
    foreach $taboid (keys %Mtables)
    {
        my ($tabname, $tabkey, $keytype) = @{$Mtables{$taboid}};
        next unless exists $Stables{$tabname};

        my $oidkey = ($tabkey eq 'oid') ? "_$tabname.oid," : '';

        $sql = sprintf "SELECT \"_$tabname\".\"${tabkey}\",$oidkey \"_$tabname\".* FROM \"$tabname\" ".
          "\"_$tabname\", _RSERV_LOG_ l WHERE l.reloid = $taboid AND l.update = 1".
          " $sinfo AND \"_$tabname\".\"${tabkey}\"=l.key::${keytype}".
          $sel_server;
        
        printf "UPDATED: $sql\n" if $debug;
        
        $result = $mconn->exec($sql);
        if ($result->resultStatus ne PGRES_TUPLES_OK)
        {
            printf $outf "-- ERROR\n" if $havedeal;
            print STDERR $mconn->errorMessage unless ($quiet);
            $mconn->exec("ROLLBACK");
            return(-1);
        }
        next if $result->ntuples <= 0;
        if (! $havedeal)
        {
            my $syncid = GetSYNCID($mconn, $outf);
            return($syncid) if $syncid < 0;
            $havedeal = 1;
        }
        printf $outf "-- UPDATE $tabname\n";
        printf "-- UPDATE $tabname\n" if $debug;
        while (@row = $result->fetchrow)
        {
            push @keys,OutputKey(shift @row,$keytype);
            for (my $i = 0; $i <= $#row; $i++)
            {
                printf $outf "  " if $i;
                printf "        " if $i && $debug;
                printf $outf "%s", OutputValue($row[$i]);
                printf "%s", OutputValue($row[$i]) if $debug;;
            }
            printf $outf "\n";
            printf "\n" if $debug;
        }
        printf $outf "\\.\n";
        printf "\\.\n" if $debug;;
    }

    # INSERTED rows

    foreach $taboid (keys %Mtables)
    {
        my ($tabname, $tabkey, $keytype) = @{$Mtables{$taboid}};
        next unless exists $Stables{$tabname};

        my $oidkey = ($tabkey eq 'oid') ? "_$tabname.oid," : '';

        $sql = sprintf "SELECT \"_$tabname\".\"${tabkey}\", $oidkey \"_$tabname\".* FROM \"$tabname\" ".
          "\"_$tabname\", _RSERV_LOG_ l WHERE l.reloid = $taboid AND l.insert = 1".
          " $sinfo AND \"_$tabname\".\"${tabkey}\"=l.key::${keytype}".
          $sel_server;
        
        printf "INSERTED: $sql\n" if $debug;
        
        $result = $mconn->exec($sql);
        if ($result->resultStatus ne PGRES_TUPLES_OK)
        {
            printf $outf "-- ERROR\n" if $havedeal;
            print STDERR $mconn->errorMessage unless ($quiet);
            $mconn->exec("ROLLBACK");
            return(-1);
        }
        next if $result->ntuples <= 0;
        if (! $havedeal)
        {
            my $syncid = GetSYNCID($mconn, $outf);
            return($syncid) if $syncid < 0;
            $havedeal = 1;
        }
        printf $outf "-- INSERT $tabname\n";
        printf "-- INSERT $tabname\n" if $debug;
        while (@row = $result->fetchrow)
        {
            push @keys,OutputKey(shift @row,$keytype);
            for (my $i = 0; $i <= $#row; $i++)
            {
                printf $outf "  " if $i;
                printf "        " if $i && $debug;
                printf $outf "%s", OutputValue($row[$i]);
                printf "%s", OutputValue($row[$i]) if $debug;
            }
            printf $outf "\n";
            printf "\n" if $debug;
        }
        printf $outf "\\.\n";
        printf "\\.\n" if $debug;;
    }
    
    
    unless ($havedeal)
    {
        print STDERR "hon't have deal, rollback...\n" if ($debug);
        $mconn->exec("ROLLBACK");
        return(0);
    }
    
    # Remember this snapshot info
    $result = $mconn->exec("select _rserv_sync_($sserver)");
    if ($result->resultStatus ne PGRES_TUPLES_OK)
    {
        printf $outf "-- ERROR\n";
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    
    $result = $mconn->exec("COMMIT");
    if ($result->resultStatus ne PGRES_COMMAND_OK)
    {
        printf $outf "-- ERROR\n";
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }

    if ($multimaster) {
        # save keys
#       my $key_out = "-- KEYS ".($#keys+1)."\n(\"key\" like ".join(") or (\"key\" like ",@keys).")\n";
        my $key_out = "-- KEYS ".($#keys+1)."\n".join(",",@keys)."\n";
        print $outf $key_out;
        print $key_out if ($debug);
    }

    printf $outf "-- OK\n";
    printf "-- OK\n" if $debug;
    
    return(1);
    
}

sub OutputValue
{
        my ($val) = @_; # @_[0];

        return("\\N") unless defined $val;

        $val =~ s/\\/\\\\/g;
        $val =~ s/      /\\011/g;
        $val =~ s/\n/\\012/g;
        $val =~ s/\'/\\047/g;

        return($val);
}

sub OutputKey {
        my $val = shift;
        my $cast = shift || '';
#       $cast = "::$cast" if ($cast);
        $cast = "::text";
#       $cast = "";

        return "null" if (! defined($val));

        print STDERR "Key: ${val}${cast}\n" if ($debug);

        if ($val =~ m/^\d+$/) {
                return "${val}${cast}";
        } else {
                return "'$val'${cast}";
        }
}

# Get syncid for new snapshot
sub GetSYNCID
{
    my ($conn, $outf) = @_; # (@_[0], @_[1]);
    
    my $result = $conn->exec("select nextval('_rserv_sync_seq_')");
    if ($result->resultStatus ne PGRES_TUPLES_OK)
    {
        print STDERR $conn->errorMessage unless ($quiet);
        $conn->exec("ROLLBACK");
        return(-1);
    }
 
    my @row = $result->fetchrow;
    
    printf $outf "-- SYNCID $row[0]\n";
    printf "-- SYNCID $row[0]\n" if $debug;
    return($row[0]);
}


sub CleanLog
{
    my ($conn, $howold, $onlytables) = @_; # (@_[0], @_[1]);
    
    my $result = $conn->exec("BEGIN");
    if ($result->resultStatus ne PGRES_COMMAND_OK)
    {
        print STDERR $conn->errorMessage unless ($quiet);
        $conn->exec("ROLLBACK");
        return(-1);
    }
    
    my $sql = "select rs.maxid, rs.active from _RSERV_SYNC_ rs" .
        " where rs.syncid = (select max(rs2.syncid) from _RSERV_SYNC_ rs2" .
        " where rs2.server = rs.server AND rs2.status > 0) order by rs.maxid";
    
    printf "$sql\n" if $debug;
    
    $result = $conn->exec($sql);
    if ($result->resultStatus ne PGRES_TUPLES_OK)
    {
        print STDERR $conn->errorMessage unless ($quiet);
        return(-1);
    }
    my $maxid = '';
    my %active = ();
    while (my @row = $result->fetchrow)
    {
        $maxid = $row[0] if $maxid eq '';
        last if $row[0] > $maxid;
        my @ids = split(/[      ]+,[    ]+/, $row[1]);
        foreach my $aid (@ids)
        {
            $active{$aid} = 1 unless exists $active{$aid};
        }
    }
    if ($maxid eq '')
    {
        print STDERR "No Sync IDs\n" unless ($quiet);
        return(0);
    }
    my $alist = join(',', keys %active);
    my $sinfo = "logid < $maxid";
    $sinfo .= " AND logid not in ($alist)" if $alist ne '';
    #if (ref($onlytables) eq 'HASH') {
    #   foreach my $onlytable (keys %{$onlytables}) {
    #           $sinfo
    #   }
    #}
    $sql = "delete from _RSERV_LOG_ where " . 
        "logtime < now() - '$howold second'::interval AND $sinfo";
    
    printf "$sql\n" if $debug;
    
    $result = $conn->exec($sql);
    if ($result->resultStatus ne PGRES_COMMAND_OK)
    {
        print STDERR $conn->errorMessage unless ($quiet);
        $conn->exec("ROLLBACK");
        return(-1);
    }
    $maxid = $result->cmdTuples;
    
    $result = $conn->exec("COMMIT");
    if ($result->resultStatus ne PGRES_COMMAND_OK)
    {
        print STDERR $conn->errorMessage unless ($quiet);
        $conn->exec("ROLLBACK");
        return(-1);
    }
    
    return($maxid);
}

sub ApplySnapshot
{
    my ($sconn, $inpf, $multimaster, $onlytables) = @_; # (@_[0], @_[1]);

    my $serverId;

    my $result = $sconn->exec("BEGIN");
    if ($result->resultStatus ne PGRES_COMMAND_OK) {
        print STDERR $sconn->errorMessage unless ($quiet);
        $sconn->exec("ROLLBACK");
        return(-1);
    }
    
    $result = $sconn->exec("SET CONSTRAINTS ALL DEFERRED");
    if ($result->resultStatus ne PGRES_COMMAND_OK) {
        print STDERR $sconn->errorMessage unless ($quiet);
        $sconn->exec("ROLLBACK");
        return(-1);
    }
    
    # MAP name --> oid, keyname, keynum
    my $sql = qq{
        select pgc.oid, pgc.relname, pga.attname, rt.key
        from _RSERV_SLAVE_TABLES_ rt, pg_class pgc, pg_attribute pga
        where pgc.oid = rt.reloid AND pga.attrelid = rt.reloid
                AND pga.attnum = rt.key
    };
    
    $result = $sconn->exec($sql);
    if ($result->resultStatus ne PGRES_TUPLES_OK) {
        print STDERR $sconn->errorMessage unless ($quiet);
        $sconn->exec("ROLLBACK");
        return(-1);
    }
    %Stables = ();
    while (my @row = $result->fetchrow) {
        #       printf "\t%s\t%s\t%s\t\n", $row[1], $row[0], $row[2], $row[3];
                if (ref($onlytables) eq 'HASH') {
                        next unless (exists $onlytables->{$row[1]});
                        $onlytables->{$row[1]} = $row[0] unless ($onlytables->{$row[1]});
                }
        push @{$Stables{$row[1]}}, $row[0], $row[2], $row[3];
    }

    print STDERR "Snapshot tables oids: ",join(",",keys %Stables),"\n" if ($debug);

    # save keys from snapshot because we want to update _rserv_log_ with
    # correct source server later...
    my @keys;
    my $keys_sql;

    my $ok = 0;
    my $syncid = -1;
    while(<$inpf>) {
        $_ =~ s/\n//;
        my ($cmt, $cmd, $prm) = split (/[       ]+/, $_, 3);
        die "FATAL: snapshot format unknown or snapshot corrupted!" if (! $cmt);
        if ($cmt ne '--') {
            printf STDERR "Invalid format\n" unless ($quiet);
            $sconn->exec("ROLLBACK");
            return(-2);
        }
        if ($cmd eq 'DELETE') {
            if ($syncid == -1) {
                printf STDERR "Sync ID unspecified\n" unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-2);
            }
            $result = DoDelete($sconn, $inpf, $prm); 
            if ($result) {
                $sconn->exec("ROLLBACK");
                return($result);
            }
        } elsif ($cmd eq 'INSERT') {
            if ($syncid == -1) {
                printf STDERR "Sync ID unspecified\n" unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-2);
            }
            $result = DoInsert($sconn, $inpf, $prm);
            if ($result) {
                $sconn->exec("ROLLBACK");
                return($result);
            }
        } elsif ($cmd eq 'UPDATE') {
            if ($syncid == -1) {
                printf STDERR "Sync ID unspecified\n" unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-2);
            }
            $result = DoUpdate($sconn, $inpf, $prm);
            if ($result) {
                $sconn->exec("ROLLBACK");
                return($result);
            }
        } elsif ($cmd eq 'SYNCID') {
            if ($syncid != -1) {
                printf STDERR "Second Sync ID ?!\n" unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-2);
            }
            if ($prm !~ /^\d+$/) {
                printf STDERR "Invalid Sync ID $prm\n" unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-2);
            }
            $syncid = $prm;
            
            printf STDERR "Sync ID $syncid\n" unless ($quiet);
            
            $result = $sconn->exec(qq{
                select syncid, synctime
                from _RSERV_SLAVE_SYNC_
                where syncid = 
                        (select max(syncid) from _RSERV_SLAVE_SYNC_)
            });
            if ($result->resultStatus ne PGRES_TUPLES_OK) {
                print STDERR "can't get current syncid from _rserv_slave_sync_: ",$sconn->errorMessage unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-1);
            }

            my @row = $result->fetchrow;
            print STDERR "Slave Sync ID ",($row[0] || "null"),"\n" if ($debug);
            if (! defined $row[0]) {
                $result = Exec($sconn,qq{
                        insert into _RSERV_SLAVE_SYNC_ (syncid, synctime)
                        values ($syncid, now())
                });
            } elsif ($row[0] >= $prm) {
                printf STDERR "Sync-ed to ID $row[0] ($row[1])\n" unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(0);
            } else {
                $result = Exec($sconn,qq{
                        update _RSERV_SLAVE_SYNC_ set syncid = $syncid, synctime = now()
                });
            }
            if ($result->resultStatus ne PGRES_COMMAND_OK) {
                print STDERR $sconn->errorMessage unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-1);
            }
        } elsif ($cmd eq 'OK') {
            $ok = 1;
            last;
        } elsif ($cmd eq 'ERROR') {
            printf STDERR "ERROR signaled\n" unless ($quiet);
            $sconn->exec("ROLLBACK");
            return(-2);
        } elsif ($cmd eq 'SERVER') {
            if ($prm !~ /^\d+$/) {
                printf STDERR "Invalid Server ID $prm\n" unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-2);
            }
            $serverId = $prm;
            print STDERR "Server ID $serverId\n" unless ($quiet);
        } elsif ($cmd eq 'KEYS') {
            if ($prm !~ /^\d+$/) {
                printf STDERR "Invalid numer of keys $prm\n" unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-2);
             }
             my $keys = <$inpf>;
             chomp($keys);
             if ($multimaster) {
ExecFatch($sconn,"select count(*) from _rserv_log_");
ExecDebug($sconn,"select * from _rserv_log_");
my ($logid) = ExecFatch($sconn,"select distinct logid from _rserv_log_ where key in ($keys)");
                $keys_sql = qq{
                        update _rserv_log_ set server=$serverId
                        where key in ($keys)
                };
                $keys_sql = qq{
                        update _rserv_log_ set server=$serverId
                        where logid = $logid
                };
                                
                print STDERR "$keys_sql\n" if ($debug);
                $result = $sconn->exec($keys_sql);
ExecDebug($sconn,"explain analyze $keys_sql");
                print STDERR "expected $prm updates, got ",$result->ntuples,"\n" if ($result->ntuples != $prm);
#               if ($result->resultStatus ne PGRES_COMMAND_OK || $result->ntuples != $prm) {
                if (0) {
                        print STDERR "FATAL: Cannot update source server in _rserv_log_: ",$sconn->errorMessage,"\n";
                        $sconn->exec("ROLLBACK");
                        return(-1);
                }
            }
        } else {
            printf STDERR "Unknown command $cmd\n" unless ($quiet);
            $sconn->exec("ROLLBACK");
            return(-2);
        }
    }
    
ExecFatch($sconn,"select count(*) from _rserv_log_");
    if (! $ok) {
        printf STDERR "No OK flag in input\n" unless ($quiet);
        $sconn->exec("ROLLBACK");
        return(-2);
    }
    
    $result = $sconn->exec("COMMIT");
    if ($result->resultStatus ne PGRES_COMMAND_OK) {
        print STDERR $sconn->errorMessage unless ($quiet);
        $sconn->exec("ROLLBACK");
        return(-1);
    }
    
    return(1);
}

sub DoDelete
{
    my ($sconn, $inpf, $tabname) = @_; # (@_[0], @_[1], @_[2]);

    # only delete tables that the slave wants
    if (! defined($Stables{$tabname})) {
        print STDERR "Not configured to delete rows from table $tabname\n" unless $quiet;
        while (<$inpf>) {
            my $istring = $_;
            $istring =~ s/\n//;
            last if ($istring eq '\.');
        }
        return(0);
    }

    my $ok = 0;
    while(<$inpf>)
    {
        if ($_ !~ /\n$/)
        {
            printf STDERR "Invalid format\n" unless ($quiet);
            return(-2);
        }
        my $key = $_;
        $key =~ s/\n//;
        if ($key eq '\.')
        {
            $ok = 1;
            last;
        }
        
        my $sql = "delete from \"$tabname\" where ".
            "\"$Stables{$tabname}->[1]\" = '$key'";
        
        printf "$sql\n" if $debug;
        
        my $result = $sconn->exec($sql);
        if ($result->resultStatus ne PGRES_COMMAND_OK)
        {
            print STDERR $sconn->errorMessage unless ($quiet);
            return(-1);
        }
    }
    
    if (! $ok)
    {
        printf STDERR "No end of input in DELETE section\n" unless ($quiet);
        return(-2);
    }
    
    return(0);
}


sub DoUpdate
{
    my ($sconn, $inpf, $tabname) = @_; # (@_[0], @_[1], @_[2]);

    # only update the tables that the slave wants
    if (! defined($Stables{$tabname})) {
        print STDERR "Not configured to update rows from table $tabname\n" unless $quiet;
        while (<$inpf>) {
            my $istring = $_;
            $istring =~ s/\n//;
            last if ($istring eq '\.');
        }
        return(0);
    }

    my $oidkey = ($Stables{$tabname}->[2] < 0) ? 1 : 0;
    
    my @CopyBuf = ();
    my $CBufLen = 0;
    my $CBufMax = 16 * 1024 * 1024;     # max size of buf for copy

    my $sql = "select attnum, attname from pg_attribute" .
        " where attrelid = $Stables{$tabname}->[0] AND attnum > 0";
    
    my $result = $sconn->exec($sql);
    if ($result->resultStatus ne PGRES_TUPLES_OK)
    {
        print STDERR $sconn->errorMessage unless ($quiet);
        return(-1);
    }
    
    my @anames = ();
    while (my @row = $result->fetchrow) {
        $anames[$row[0]] = $row[1];
    }
    
    my $istring;
    my $ok = 0;
    while(<$inpf>) {
        if ($_ !~ /\n$/) {
            printf STDERR "Invalid format\n" unless ($quiet);
            return(-2);
        }
        $istring = $_;
        $istring =~ s/\n//;
        if ($istring eq '\.') {
            $ok = 1;
            last;
        }
        my @vals = split(/      /, $istring);
        if ($oidkey) {
            if ($vals[0] !~ /^\d+$/ || $vals[0] <= 0) {
                printf STDERR "Invalid OID\n" unless ($quiet);
                return(-2);
            }
            $oidkey = $vals[0];
        } else {
            unshift @vals, '';
        }
        
        $sql = "update \"$tabname\" set ";
        my $ocnt = 0;
        for (my $i = 1; $i <= $#anames; $i++) {
            if ($vals[$i] eq '\N') {
                if ($i == $Stables{$tabname}->[2]) {
                    printf STDERR "NULL key\n" unless ($quiet);
                    return(-2);
                }
                $vals[$i] = 'null';
            } else {
                $vals[$i] = "'" . $vals[$i] . "'";
                next if $i == $Stables{$tabname}->[2];
            }
            $ocnt++;
            $sql .= ', ' if $ocnt > 1;
            $sql .= "\"$anames[$i]\" = $vals[$i]";
        } if ($oidkey) {
            $sql .= " where \"$Stables{$tabname}->[1]\" = $oidkey";
        } else {
            $sql .= " where \"$Stables{$tabname}->[1]\" = ".
                $vals[$Stables{$tabname}->[2]];
        }
        
        printf "$sql\n" if $debug;
        
        $result = $sconn->exec($sql);
        
        if ($result->resultStatus ne PGRES_COMMAND_OK) {
            print STDERR $sconn->errorMessage unless ($quiet);
            return(-1);
        }
        next if $result->cmdTuples == 1;        # updated
        
        if ($result->cmdTuples > 1) {
            printf STDERR "Duplicate keys\n" unless ($quiet);
            return(-2);
        } 

        # no key - copy
        push @CopyBuf, "$istring\n";
        $CBufLen += length($istring);
        
        if ($CBufLen >= $CBufMax) {
            $result = DoCopy($sconn, $tabname, $oidkey, \@CopyBuf);
            return($result) if $result;
            @CopyBuf = ();
            $CBufLen = 0;
        }
    }
    
    if (! $ok) {
        printf STDERR "No end of input in UPDATE section\n" unless ($quiet);
        return(-2);
    }
    
    if ($CBufLen) {
        print STDERR "@CopyBuf\n" if $debug;
        $result = DoCopy($sconn, $tabname, $oidkey, \@CopyBuf);
        return($result) if $result;
    }

    return(0);
}

sub DoInsert
{
    my ($sconn, $inpf, $tabname) = @_; # (@_[0], @_[1], @_[2]);

    # only insert rows into tables that the slave wants
    if (! defined($Stables{$tabname})) {
        print STDERR "Not configured to insert rows from table $tabname\n" unless $quiet;
        while (<$inpf>) {
            my $istring = $_;
            $istring =~ s/\n//;
            last if ($istring eq '\.');
        }
        return(0);
    }

    my $oidkey = ($Stables{$tabname}->[2] < 0) ? 1 : 0;
    
    my @CopyBuf = ();
    my $CBufLen = 0;
    my $CBufMax = 16 * 1024 * 1024;     # max size of buf for copy
    
    my $istring;
    my $ok = 0;
    while(<$inpf>) {
        if ($_ !~ /\n$/) {
            printf STDERR "Invalid format\n" unless ($quiet);
            return(-2);
        }
        $istring = $_;
        $istring =~ s/\n//;
        if ($istring eq '\.') {
            $ok = 1;
            last;
        }

        # no key - copy
        push @CopyBuf, "$istring\n";
        $CBufLen += length($istring);
        
        if ($CBufLen >= $CBufMax) {
            my $result = DoCopy($sconn, $tabname, $oidkey, \@CopyBuf);
            return($result) if $result;
            @CopyBuf = ();
            $CBufLen = 0;
        }
    }
    
    if (! $ok) {
        printf STDERR "No end of input in INSERT section\n" unless ($quiet);
        return(-2);
    }
    
    if ($CBufLen) {
        print STDERR "@CopyBuf\n" if $debug;
        my $result = DoCopy($sconn, $tabname, $oidkey, \@CopyBuf);
        return($result) if $result;
    }
    
    return(0);
}


sub DoCopy {
    my ($sconn, $tabname, $withoids, $CBuf) = @_; # (@_[0], @_[1], @_[2], @_[3]);
    
    my $sql = "COPY \"$tabname\" " . (($withoids) ? "WITH OIDS " : '') . 
        "FROM STDIN";
    my $result = $sconn->exec($sql);
    if ($result->resultStatus ne PGRES_COPY_IN) {
        print STDERR $sconn->errorMessage unless ($quiet);
        return(-1);
    }
    
    foreach my $str (@{$CBuf}) {
        $sconn->putline($str);
    }
    
    $sconn->putline("\\.\n");
    
    if ($sconn->endcopy) {
        print STDERR $sconn->errorMessage unless ($quiet);
        return(-1);
    }
    
    return(0);
}


#
# Returns last SyncID applied on Slave
#
sub GetSyncID {
    my ($sconn) = @_; # (@_[0]);
    
    my $result = $sconn->exec("select max(syncid) from _RSERV_SLAVE_SYNC_");
    if ($result->resultStatus ne PGRES_TUPLES_OK) {
        print STDERR $sconn->errorMessage unless ($quiet);
        return(-1);
    }
    my @row = $result->fetchrow;
    print STDERR "GetSyncID: ",($row[0] || 'null'),"\n" if ($debug);
    return(undef) unless defined $row[0];       # null
    return($row[0]);
}

#
# Updates _RSERV_SYNC_ on Master with Slave SyncID
#
sub SyncSyncID {
    my ($mconn, $sserver, $syncid) = @_; # (@_[0], @_[1], @_[2]);
    
    my $result = $mconn->exec("BEGIN");
    if ($result->resultStatus ne PGRES_COMMAND_OK) {
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    
    $result = $mconn->exec("select synctime, status from _RSERV_SYNC_" .
                          " where server = $sserver AND syncid = $syncid" .
                          " for update");
    if ($result->resultStatus ne PGRES_TUPLES_OK) {
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    my @row = $result->fetchrow;
    if (! defined $row[0]) {
        printf STDERR "No SyncID $syncid found for server $sserver\n" unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(0);
    }
    if ($row[1] > 0) {
        printf STDERR "SyncID $syncid for server ".
            "$sserver already updated\n" unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(0);
    }
    $result = $mconn->exec("update _RSERV_SYNC_" .
                          " set synctime = now(), status = 1" .
                          " where server = $sserver AND syncid = $syncid");
    if ($result->resultStatus ne PGRES_COMMAND_OK) {
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    $result = $mconn->exec("delete from _RSERV_SYNC_" .
                          " where server = $sserver AND syncid < $syncid");
    if ($result->resultStatus ne PGRES_COMMAND_OK) {
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    
    $result = $mconn->exec("COMMIT");
    if ($result->resultStatus ne PGRES_COMMAND_OK) {
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    
    return(1);
}

# stuff moved from perl scripts for better re-use

sub Rollback {
    my $conn = shift @_;

    print STDERR $conn->errorMessage unless ($quiet);
    $conn->exec("ROLLBACK");
}

sub RollbackAndQuit {
    my $conn = shift @_;

    Rollback($conn);
    exit (-1);
}

sub Connect {
        my $info = shift @_;

        print("Connecting to $info\n") if ($debug || $verbose);
        my $conn = Pg::connectdb($info);
        if ($conn->status != PGRES_CONNECTION_OK) {
            die "Failed opening $info";
        }
        return $conn;
}

sub Exec {
        my $conn = shift || die "Exec needs connection!";
        my $sql = shift || die "Exec needs SQL statement!";
        # used to return error code if no tuples are retured
        my $return_code = shift;

        if ($debug) {
                # re-format SQL in one line (for nicer output)
                $sql =~ s/[\s\n\r]+/ /gs;
                print STDERR "Exec: $sql\n";
        }
        my $result = $conn->exec($sql);
        if ($result->resultStatus eq PGRES_COMMAND_OK) {
                return $result;
        } elsif ($result->resultStatus eq PGRES_TUPLES_OK) {
                print STDERR "Returned ",$result->ntuples," tuples\n" if ($debug);
                return $result;
        } else {
                if (defined($return_code)) {
                        print STDERR "ERROR: ",$conn->errorMessage,"\n" unless ($quiet);
                        return($return_code);
                } else {
                        RollbackAndQuit($conn)
                }
        }
}

sub Exec2 {
        my $mconn = shift @_;
        my $sconn = shift @_;
        my $sql = shift @_;

        my $result = $mconn->exec($sql);
        RollbackAndQuit($mconn) if ($result->resultStatus ne PGRES_COMMAND_OK);
        $result = $sconn->exec($sql);
        RollbackAndQuit($sconn) if ($result->resultStatus ne PGRES_COMMAND_OK);
        # XXX TODO: return results?!
}

sub ExecFatch {
        my $conn = shift || die "ExecFatch need conn!";
        my $sql = shift || die "ExecFatch need SQL!";

        print STDERR "ExecFatch: $sql\n" if ($debug);

        my $result = $conn->exec($sql);
        RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_TUPLES_OK);

        print STDERR "Returned ",$result->ntuples," tuples\n" if ($debug);

        my @row = $result->fetchrow;
        print STDERR "DATA: ",join(",",@row),"\n" if ($debug);
        return @row;
}

sub ExecDebug {
        my $conn = shift || die "ExecDebug need conn!";
        my $sql = shift || die "ExecDebug need SQL!";

        print STDERR "ExecDebug: $sql\n" if ($debug);

        my $result = $conn->exec($sql);
        RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_TUPLES_OK);

        print STDERR "Returned ",$result->ntuples," tuples\n" if ($debug);

        while (my @row = $result->fetchrow) {
                print STDERR "DATA: ",join(",",@row),"\n" if ($debug);
        }
        return $result->ntuples;
}
sub MkInfo {
        my $db = shift || die "need database name!";
        my $host = shift;
        my $port = shift;
        my $user = shift;
        my $password = shift;

        my $info = "dbname=$db";
        $info = "$info host=$host" if (defined($host));
        $info = "$info port=$port" if (defined($port));
        $info = "$info user=$user" if (defined($user));
        $info = "$info password=$password" if (defined($password));

        return $info;
}

1;