@@ -3931,6 +3931,7 @@ ($SQL = $SQL{delta}) =~ s/\$1/$g->{oid}/go; (my $safedbname = $targetdb) =~ s/\'/''/go; $SQL =~ s/\$2/$safedbname/o; + $sth{source}{$g}{getdelta} = $sourcedbh->prepare($SQL); my $safesourcedb; @@ -4687,6 +4688,9 @@ ## The target table's OID my $toid = $g->{targetoid}{$targetdb}; + ## Since we have systems with little memory and swap space + my $chunksize = 250; + ## If requested, disable all indexes, then enable and rebuild them after we COPY my $hasindex = 0; if ($g->{rebuild_index} == 2) { @@ -4826,13 +4830,55 @@ }; } - ## Build a list of all PK values - my $pkvals = ''; + ## Postgres does not match on null values when using the 'IN ( 'A', 'B', null, 'D' ) notation + ## Hence, we build hashes of lists for pk columns and values + my ( %pkcol_list, %pkcol_null_array, %pkval_lists, %nrow, %zaehler, %listbuf ); + my $maxbitmap = '1' x $g->{pkcols}; + $nrow{$maxbitmap} = 0; + my $bitmap = $maxbitmap; + $pkcol_list{$bitmap} = $g->{pkeycols}; + $listbuf{$maxbitmap} = ''; for my $row (@$info) { - my $inner = join ',' => map { s/\'/''/go; s{\\}{\\\\}go; qq{'$_'}; } @$row; - $pkvals .= $g->{pkcols} > 1 ? "($inner)," : "$inner,"; + my ( $has_null, $bitmap ) = ( 0, $maxbitmap ); + my $inner; + my @inner = map { if ( defined ) { s/\'/''/go; s{\\}{\\\\}go; qq{'$_'} } else { $has_null = 1; 'null' } } @$row; + if ( $has_null ) { + my @cleanvals = @inner; + my @cleankeys = @{$g->{pkey}}; + my @null_keys; + my $idx = 0; + foreach ( @inner ) { + if ( $_ eq 'null' ) { + substr( $bitmap, $idx, 1 ) = '0'; + splice( @cleanvals, $idx, 1 ); + push @null_keys, splice( @cleankeys, $idx, 1 ); + } + $idx++; + } + $nrow{$bitmap} = 0 unless exists $nrow{$bitmap}; + $zaehler{$bitmap} = 0 unless exists $zaehler{$bitmap}; + $listbuf{$bitmap} = '' unless exists $listbuf{$bitmap}; + $pkcol_list{$bitmap} = '(' . join( ',', @cleankeys ) . ')' unless exists $pkcol_list{$bitmap}; + $pkcol_null_array{$bitmap} = \@null_keys unless exists $pkcol_null_array{$bitmap}; + $inner = join ',', @cleanvals; + } else { + $inner = join ',', @inner; + } + $nrow{$bitmap} += 1; + $zaehler{$bitmap} += 1; + if ( $zaehler{$bitmap} > $chunksize ) { + $zaehler{$bitmap} = 1; + chop $listbuf{$bitmap}; + push @{$pkval_lists{$bitmap}}, $listbuf{$bitmap}; + $listbuf{$bitmap} = $g->{pkcols} > 1 ? "($inner)," : "$inner,"; + } else { + $listbuf{$bitmap} .= $g->{pkcols} > 1 ? "($inner)," : "$inner,"; + } + } + foreach ( keys %listbuf ) { + chop $listbuf{$_}; + push @{$pkval_lists{$_}}, $listbuf{$_}; } - chop $pkvals; ## Example: 1234, 221 ## Example MCPK: ('1234','Don''t Stop','2008-01-01'),('221','foobar','2008-11-01') @@ -4872,99 +4918,82 @@ ## If rows were updated on source, we'll insert later (update = delete + insert) $self->glog(qq{Deleting rows from $S.$T}); - ## If we've got a very large number of values, break the DELETEs into multiples - my @delchunks; - if (length $pkvals > 100_000) { - ## How many items in the IN () clause - my $deletebatch = 10_000; - my $dcount = 0; - my $delcount = 0; - for my $row (@$info) { - my $inner = join ',' => map { s/\'/''/go; s{\\}{\\\\}go; qq{'$_'}; } @$row; - ## Put this group of pks into a temporary array - $delchunks[$delcount] .= $g->{pkcols} > 1 ? "($inner)," : "$inner,"; - ## Once we reach out limit, start appending to the next bit of the array - if ($dcount++ >= $deletebatch) { - $delcount++; - $dcount = 0; + ## Looping over permutations of columns constituting a unique key with null values + ## In most cases, there's only one iteration for the primary key! + + foreach my $bm ( keys %nrow ) { + next unless $nrow{$bm}; + my $dcount = 1; + + ## DELETE of affected rows at target + foreach ( @{$pkval_lists{$bm}} ) { + $SQL = "DELETE /* chunk $dcount */ FROM $S.$T WHERE $pkcol_list{$bm} IN ($_)"; + if ( exists $pkcol_null_array{$bm} ) { + $SQL .= " and $_ is null" foreach @{$pkcol_null_array{$bm}}; } - } - $dcount = 1; - for my $chunk (@delchunks) { - ## Remove the trailing comma - chop $chunk; - $SQL = "DELETE /* chunk $dcount */ FROM $S.$T WHERE $g->{pkeycols} IN ($chunk)"; - $self->glog("Deleting chunk $dcount"); + $self->glog("Deleting chunk $dcount") if $nrow{$bm} > $chunksize; ($count = $targetdbh->do($SQL)) =~ s/0E0/0/o; $dmlcount{alldeletes}{target} += $dmlcount{D}{target}{$S}{$T} = $count; $dcount++; } - } - else { - $SQL = "DELETE FROM $S.$T WHERE $g->{pkeycols} IN ($pkvals)"; - ($count = $targetdbh->do($SQL)) =~ s/0E0/0/o; - $dmlcount{alldeletes}{target} += $dmlcount{D}{target}{$S}{$T} = $count; - } - ## COPY over all affected rows from source to target + $dcount = 1; - ## Old versions of Postgres don't support "COPY (query)" - my ($srccmd,$temptable); - if (! $source_modern_copy) { - $temptable = "bucardo_tempcopy_$$"; - $self->glog("Creating temporary table $temptable for copy on $S.$T, and savepoint bucardo_$$ along with it"); - $sourcedbh->pg_savepoint("bucardo_$$"); - $srccmd = "CREATE TEMP TABLE $temptable AS SELECT * FROM $S.$T WHERE $g->{pkeycols} IN ($pkvals)"; + ## COPY over all affected rows from source to target + foreach my $vals( @{$pkval_lists{$bm}} ) { - $sourcedbh->do($srccmd); - $srccmd = "COPY $temptable TO STDOUT"; - } - elsif (! @delchunks) { - $srccmd = "COPY (SELECT * FROM $S.$T WHERE $g->{pkeycols} IN ($pkvals)) TO STDOUT"; - } + ## Old versions of Postgres don't support "COPY (query)" + my ( $srccmd, $temptable ); + if ( ! $source_modern_copy ) { + $temptable = "bucardo_tempcopy_$$"; + $self->glog("Creating temporary table $temptable for copy on $S.$T, and savepoint bucardo_$$ along with it"); + $sourcedbh->pg_savepoint("bucardo_$$"); + $srccmd = "CREATE TEMP TABLE $temptable AS SELECT * FROM $S.$T WHERE " . $pkcol_list{$bm} . " IN ($vals)"; + if ( exists $pkcol_null_array{$bm} ) { + $srccmd .= " and $_ is null" foreach @{$pkcol_null_array{$bm}}; + } + $sourcedbh->do($srccmd); + $srccmd = "COPY $temptable TO STDOUT"; + } else { + $srccmd = "COPY /* chunk $dcount */ (SELECT * FROM $S.$T WHERE " . $pkcol_list{$bm} . " IN ($vals)"; + if ( exists $pkcol_null_array{$bm} ) { + $srccmd .= " and $_ is null" foreach @{$pkcol_null_array{$bm}}; + } + $srccmd .= ') TO STDOUT'; + } - my $tgtcmd = "COPY $S.$T FROM STDIN"; - $targetdbh->do($tgtcmd); - my $buffer = ''; - $self->glog(qq{Begin COPY to $S.$T}); + my $tgtcmd = "COPY $S.$T FROM STDIN"; + $targetdbh->do( $tgtcmd ); + my $buffer = ''; + $self->glog(qq{Begin COPY to $S.$T}) if $dcount eq 1; - my $rows_copied = 0; - if ($source_modern_copy and @delchunks) { - my $dcount = 1; - for my $chunk (@delchunks) { - $srccmd = "COPY /* chunk $dcount */ (SELECT * FROM $S.$T WHERE $g->{pkeycols} IN ($chunk)) TO STDOUT"; - $sourcedbh->do($srccmd); - $self->glog("Copying chunk $dcount"); - $dcount++; - while ($sourcedbh->pg_getcopydata($buffer) >= 0) { - $targetdbh->pg_putcopydata($buffer); + my $rows_copied = 0; + $sourcedbh->do( $srccmd ); + $self->glog( "Copying chunk $dcount" ) if $nrow{$bm} > $chunksize; + while ( $sourcedbh->pg_getcopydata( $buffer ) >= 0 ) { + $targetdbh->pg_putcopydata( $buffer ); $rows_copied++; } - } - } - else { - $sourcedbh->do($srccmd); - while ($sourcedbh->pg_getcopydata($buffer) >= 0) { - $targetdbh->pg_putcopydata($buffer); - $rows_copied++; - } - } - $targetdbh->pg_putcopyend(); - $dmlcount{allinserts}{target} += $dmlcount{I}{target}{$S}{$T} = $rows_copied; + $targetdbh->pg_putcopyend(); + $dmlcount{allinserts}{target} += $dmlcount{I}{target}{$S}{$T} = $rows_copied; - if (! $source_modern_copy) { - $self->glog("Dropping temporary table $temptable"); - $sourcedbh->do("DROP TABLE $temptable"); - } + if ( ! $source_modern_copy ) { + $self->glog( "Dropping temporary table $temptable" ); + $sourcedbh->do( "DROP TABLE $temptable" ); + } + + $dcount++; + } - ## If we disabled the indexes earlier, flip them on and run a REINDEX - if ($hasindex) { - $self->glog("Re-enabling indexes for table $S.$T on $targetdb"); - $SQL = "UPDATE pg_class SET relhasindex = 't' WHERE oid = $toid"; - $targetdbh->do($SQL); - $self->glog("Reindexing table $S.$T on $targetdb"); - $targetdbh->do("REINDEX TABLE $S.$T"); + ## If we disabled the indexes earlier, flip them on and run a REINDEX + if ($hasindex) { + $self->glog("Re-enabling indexes for table $S.$T on $targetdb"); + $SQL = "UPDATE pg_class SET relhasindex = 't' WHERE oid = $toid"; + $targetdbh->do($SQL); + $self->glog("Reindexing table $S.$T on $targetdb"); + $targetdbh->do("REINDEX TABLE $S.$T"); + } } }; ## end of eval