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
        $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 "GetServerId($DB,$Host) == $row[0]\n" if ($debug);

    return $row[0];
}

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

print STDERR "## d: $debug v: $verbose q: $quiet\n";

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

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

    # first, we must know for wich tables the slave subscribed
    my $result = $sconn->exec("SELECT tname FROM _RSERV_SLAVE_TABLES_");
    if ($result->resultStatus ne PGRES_TUPLES_OK)
    {
        print STDERR $mconn->errorMessage unless ($quiet);
        return(-1);
    }
    
    my @row;
    while (@row = $result->fetchrow) {
        $Stables{$row[0]} = 1;
    }
    
    $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("set transaction isolation level serializable");
    if ($result->resultStatus ne PGRES_COMMAND_OK)
    {
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    
    # MAP oid --> tabname, keyname, key_type
    $result = $mconn->exec("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");
    if ($result->resultStatus ne PGRES_TUPLES_OK)
    {
        print STDERR $mconn->errorMessage unless ($quiet);
        $mconn->exec("ROLLBACK");
        return(-1);
    }
    
    while (@row = $result->fetchrow)
    {
        #       printf "$row[0], $row[1], $row[2]\n";
                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 "Prepare snapshot for tables with oid: ",join(",",keys %Mtables),"\n" if ($debug);

    # Read last succeeded sync
    my $sql = "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)";
    
    printf "$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 @lastsync = $result->fetchrow;

    # 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 $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]);
    }
    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 $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)
        {
            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 $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)
        {
            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)
    {
        $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);
    }
    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);
}

# 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, $onlytables) = @_; # (@_[0], @_[1]);
    
    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 = "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);

    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("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 $sconn->errorMessage unless ($quiet);
                $sconn->exec("ROLLBACK");
                return(-1);
            }
            my @row = $result->fetchrow;
            if (! defined $row[0])
            {
                $result = $sconn->exec("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 = $sconn->exec("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);
        }
        else
        {
            printf STDERR "Unknown command $cmd\n" unless ($quiet);
            $sconn->exec("ROLLBACK");
            return(-2);
        }
    }
    
    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 "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 "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 "@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 "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 "@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;
    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;
    $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) {
            print STDERR "Failed opening $info\n";
            exit 1;
        }
        return $conn;
}

sub Exec {
        my $conn = shift @_;
        my $sql = shift @_;

        my $result = $conn->exec($sql);
        print STDERR "$sql\n" if ($debug);
        RollbackAndQuit($conn) if ($result->resultStatus ne PGRES_COMMAND_OK);
}

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);
}

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;