File Coverage

blib/lib/DBD/PgLite/MirrorPgToSQLite.pm

Criterion	Covered	Total	%
statement	12	218	5.5
branch	0	138	0.0
condition	0	41	0.0
subroutine	4	17	23.5
pod	0	12	0.0
total	16	426	3.7

package DBD::PgLite::MirrorPgToSQLite;
use strict;
use DBI;
use Storable;
use File::Copy;
require Exporter;
our @ISA = qw(Exporter);
our @EXPORT_OK = qw(pg_to_sqlite);
our $VERSION = '0.05';

#### MAIN SUBROUTINE ####

sub pg_to_sqlite {
	my %opt = @_;
	my %defaults = defaults();
	for (keys %defaults) {
		next if exists $opt{$_};
		$opt{$_} = $defaults{$_};
	}
	$opt{tables} = _commasplit($opt{tables});
	$opt{views}  = _commasplit($opt{views});
	if ($opt{where}
		&& $opt{where} !~ /^\s*where\s/i
		&& $opt{where} !~ /^\s*(?:natural\s+)?join\s/i) {
		$opt{where} = 'where '.$opt{where};
	}
	die "Incompatible options: 'append' and 'snapshot'" if $opt{append} && $opt{snapshot};
	die "Need either database handle (pg_dbh) or DSN (pg_dsn)" unless $opt{pg_dbh} || $opt{pg_dsn};
	my $disconnect = 0;
	unless ($opt{pg_dbh}) {
		$disconnect++;
		$opt{pg_dbh} = DBI->connect(@opt{ qw(pg_dsn pg_user pg_pass) },{RaiseError=>1})
		  or die "Could not connect to PostgreSQL: $DBI::errstr";
	}
	my $fn = $opt{sqlite_file};
	die "Need both list of source tables and a SQLite file" unless @{$opt{tables}} && $fn;
	$|++ if $opt{verbose};
	my $lockfile = "$fn.lock";
	lockfile('create',$lockfile);
	if (-f "$fn.tmp") {
		warn "WARNING: Removing temp file $fn.tmp - apparently left over from a previous run\n";
		unlink "$fn.tmp" or die "ERROR: Could not remove file.tmp: $!\n";
	}
	if ($opt{append}) {
		unless (copy $fn, "$fn.tmp") {
			warn "WARNING: Could not copy $fn to $fn.tmp for appending: $! - turning --append off\n";
			$opt{append} = 0;
		}
	}
	$opt{sl_dbh} = DBI->connect("dbi:SQLite:dbname=$fn.tmp",undef,undef,{RaiseError=>1});

	my @tables = tablelist($opt{pg_dbh}, $opt{schema}, @{ $opt{tables} }); # handle regexp
	print "MIRRORING ".scalar(@tables)." table(s):\n" if $opt{verbose};
	my @views = viewlist($opt{pg_dbh}, $opt{schema}, @{ $opt{views} })
	  if grep { /^\/.+\/$/ } @{ $opt{views} };

	eval {
		if ($opt{snapshot}) {
			$opt{pg_dbh}->do("set session characteristics as transaction isolation level serializable");
			$opt{pg_dbh}->begin_work;
			mirror_table($_,%opt) for @tables;
			mirror_functions(%opt) if $opt{functions};
			$opt{pg_dbh}->commit;
		} else {
			mirror_table($_,%opt) for @tables;
			mirror_functions(%opt) if $opt{functions};
		}
		if (@views) {
			print "CREATING ".scalar(@views)." view(s):\n" if $opt{verbose};
			create_view($_,%opt) for @views;
		}
		print "done!\n" if $opt{verbose};
	};

	if ($@) {
		lockfile('clear',$lockfile);
		die $@;
	}

	$opt{pg_dbh}->disconnect if $disconnect;
	$opt{sl_dbh}->disconnect;

	if (-f $fn) {
		copy $fn, "$fn.bak" or warn "WARNING: Could not make backup copy of $fn: $!\n";
	}
	move "$fn.tmp", $fn or die "ERROR: Could not move temporary SQLite file $fn.tmp to $fn";
	lockfile('clear',$lockfile);
}



########## OTHER SUBROUTINES ###########

sub _commasplit {
	my $list = shift;
	return [] unless $list;
	$list = [$list] unless ref $list;
	my @new = split(/\s*,\s*/,join(',',@$list));
	return \@new;
}

sub defaults {
	my %defaults = (
					verbose     => 0,
					pg_dsn      => ($ENV{PGDATABASE} ? "dbi:Pg:dbname=$ENV{PGDATABASE}" : undef),
					pg_user     => ($ENV{PGUSER} || $ENV{USER}),
					pg_pass     => $ENV{PGPASSWORD},
					schema      => 'public',
					tables      => [],
					sqlite_file => '',
					where       => '',
					cachedir    => '/tmp/sqlite_mirror_cache',
					append      => 0,
					snapshot    => 0,
					indexes     => 0,
					views       => [],
					functions   => 0,
					page_limit  => 5000, # each page is 8K
					pg_dbh      => undef,
				   );
	$defaults{pg_dsn} ||= "dbi:Pg:dbname=$defaults{pg_user}" if $defaults{pg_user};
	return %defaults;
}

sub mirror_table {
	my ($tn,%opt) = @_;
	my $sn = $opt{schema};
	print "  - $sn.$tn\n" if $opt{verbose};
	my ($create,$colcnt) = get_schema($sn,$tn,%opt);
	my $drop = '';
	if ($opt{append}) {
		$drop = $opt{sl_dbh}->selectrow_array("select name from sqlite_master where type = 'table' and name = ?",{},$tn);
		$opt{sl_dbh}->do("drop table $tn") if $drop;
	}
	$opt{sl_dbh}->do($create);
	my $pages = $opt{pg_dbh}->selectrow_array("select relpages from pg_class where relnamespace = (select oid from pg_namespace where nspname = ?) and relname = ?",{},$sn,$tn);
	my $ins = $opt{sl_dbh}->prepare("insert into $tn values (". join(',', ("?") x $colcnt) . ")");
	if ($pages > $opt{page_limit}) {
		warn "      pagelimit ($opt{page_limit}) kicks in for $sn.$tn ($pages)\n" if $opt{verbose};
		my @pkey = $opt{pg_dbh}->primary_key(undef,$sn,$tn);
		warn "         (pkey is )".join(":",@pkey)."\n" if $opt{verbose};
		if (@pkey) {
			my $pkey_vals = $opt{pg_dbh}->selectall_arrayref("select ".join(', ', @pkey)." from $sn.$tn");
			my $sql = "select * from $sn.$tn where ".join(" and ", map {"$_ = ?"} @pkey);
			my $selh = $opt{pg_dbh}->prepare($sql);
			$opt{sl_dbh}->begin_work;
			foreach (@$pkey_vals) {
				$selh->execute(@$_);
				my $row = $selh->fetchrow_arrayref;
				$ins->execute(@$row);
			}
			$opt{sl_dbh}->commit;
			$selh->finish;
		}
		else {
			warn "*** CANNOT READ $sn.$tn ROW-BY-ROW - NO PRIMARY KEY\n*** SKIPPING TABLE!\n";
			$ins->finish;
			return;
		}
	}
	else {
		my $res = $opt{pg_dbh}->selectall_arrayref("select * from $sn.$tn $opt{where}");
		if (@$res && scalar(@{$res->[0]}) != $colcnt) {
			$ins->finish;
			die "ERROR: Bad schema for table $tn: number of columns does not match\n";
		}
		$opt{sl_dbh}->begin_work;
		$ins->execute(@$_) for @$res;
		$opt{sl_dbh}->commit;
	}
	$ins->finish;
	create_indexes($tn,%opt) if $opt{indexes};
	$opt{sl_dbh}->do("vacuum") if $drop;
}

sub mirror_functions {
	my %opt = @_;
	unless ($opt{sl_dbh}->selectrow_array("select name from sqlite_master where name = 'pglite_functions' and type = 'table'")) {
		$opt{sl_dbh}->do("CREATE TABLE pglite_functions (name text, argnum int, type text, sql text, primary key (name,argnum))");
	}
	my $langnum = $opt{pg_dbh}->selectrow_array("select oid from pg_language where lanname = 'sql'");
	my $snum = $opt{pg_dbh}->selectrow_array("select oid from pg_namespace where nspname = ?",{},$opt{schema});
	my $fun = $opt{pg_dbh}->selectall_arrayref("select proname, pronargs, prosrc from pg_proc where prolang = ? and pronamespace = ?",{},$langnum,$snum);
	print "FUNCTIONS:\n" if $opt{verbose};
	return unless ref $fun eq 'ARRAY' && @$fun;
	for (@$fun) {
		my ($name,$argnum,$sql) = @$_;
		unless ($opt{sl_dbh}->selectrow_array("select name from pglite_functions where name = ? and argnum = ?",{},$name,$argnum)) {
			print "    - $name ($argnum)\n" if $opt{verbose};
			$opt{sl_dbh}->do("insert into pglite_functions (name,argnum,type,sql) values (?,?,'sql',?)",{},$name,$argnum,$sql);
		}
	}
}

sub create_indexes {
	my ($tn,%opt) = @_;
	my $sn = $opt{schema};
	my $ixn = $opt{pg_dbh}->selectcol_arrayref("select indexdef from pg_indexes where schemaname = ? and tablename = ? ", {}, $sn,$tn);
	for (@$ixn) {
		next if /\(oid\)/;
		next if /_pkey\b/; # No need to recreate primary keys
		s/USING btree //;
		s/ ON \w+\.\"?([^\"]+)\"?/ ON $1/;
		print "      + $_\n" if $opt{verbose};
		eval {  $opt{sl_dbh}->do($_); };
		# Pg supports multiple null values in  unique columns - SQLite doesn't
		if ($@ =~ /unique/i && s/ UNIQUE / /) {
			print "      + retry: $_\n" if $opt{verbose};
			eval { $opt{sl_dbh}->do($_); };
		}
	}
}

sub create_view {
	my ($vn,%opt) = @_;
	my $sn = $opt{schema};
	my $def = $opt{pg_dbh}->selectrow_array("select definition from pg_views where schemaname = ? and viewname = ?",{},$sn,$vn);
	print "  - $sn.$vn\n" if $opt{verbose};
	$def =~ s/::\w+//g; # casting is not supported in SQLite
	if ($opt{sl_dbh}->selectrow_array("select name from sqlite_master where name = ? and type = 'view'",{},$vn)) {
		eval { $opt{sl_dbh}->do("drop view $vn") };
	}
	eval { $opt{sl_dbh}->do("create view $vn as $def"); };
	warn "    *** COULD NOT CREATE VIEW $sn.$vn *** \n" if $@;
}

sub tablelist {
	my ($pg,$sn,@pats) = @_;
	my %tables; # prevent duplicate table names
	for my $pat (@pats) {
		if ($pat =~ s/^\/(.+)\/$/$1/) {
			my $res = $pg->selectcol_arrayref("select tablename from pg_tables where lower(schemaname) = lower('$sn') and tablename ~* '$pat'");
			$tables{$_}++ for @$res;
		} else {
			$tables{$pat}++;
		}
	}
	return keys %tables;
}

sub viewlist {
	my ($pg,$sn,@pats) = @_;
	my %views; # prevent duplicate table names
	for my $pat (@pats) {
		if ($pat =~ s/^\/(.+)\/$/$1/) {
			my $res = $pg->selectcol_arrayref("select viewname from pg_views where lower(schemaname) = lower('$sn') and viewname ~* '$pat'");
			$views{$_}++ for @$res;
		} else {
			$views{$pat}++;
		}
	}
	return keys %views;
}


sub get_schema {
	my ($sn,$tn,%opt) = @_;
	# Constructing a schema definition can be rather slow,
	# so we cache the result for up to a week
	my @cached = cached_schema($sn,$tn,undef,undef,%opt);
	return @cached if @cached;
	my @cdef = col_def($sn,$tn,%opt);
	my $colcnt = scalar @cdef;
	my @pknames = $opt{pg_dbh}->primary_key(undef,$sn,$tn);
	push @cdef, "primary key (" . join(',',@pknames) . ")" if @pknames && $pknames[0] ne '';
	my $create = "create table $tn (\n  ".join(",\n  ",@cdef)."\n)\n";
	cached_schema($sn,$tn,$create,$colcnt,%opt);
	return ($create, $colcnt);
}

sub cached_schema {
	my ($sn,$tn,$creat,$cnt,%opt) = @_;
	my $database = $opt{pg_dbh}->{mbl_dbh}
	  ? $opt{pg_dbh}->{mbl_dbh}->[0]->{Name}
	  : $opt{pg_dbh}->{Name};
	unless (-d $opt{cachedir}) {
		mkdir $opt{cachedir};
		chmod 0777, $opt{cachedir};
	}
	my $uid = (getpwuid($>))[0] || $>;
	mkdir "$opt{cachedir}/$uid" unless -d "$opt{cachedir}/$uid";
	my $fn = "$opt{cachedir}/$uid/$database.$sn.$tn";
	if ($cnt) {
		Storable::store [$creat,$cnt], $fn;
	} elsif (-f $fn && time-(stat $fn)[9]<7*24*60*60) {
		my $ret = Storable::retrieve $fn || [];
		return @$ret;
	} else {
		return ();
	}
}

sub col_def {
	my ($sn,$tn,%opt) = @_;
	my $sth = $opt{pg_dbh}->column_info(undef,$sn,$tn,undef);
	$sth->execute;
	my $res = $sth->fetchall_arrayref;
	my @ret;
	foreach my $ci (@$res) {
		my ($colnam,$typnam,$nullable) = @{$ci}[qw(3 5 10)]; #)];
		my $notnull = $nullable ? "" : " not null";
		push @ret, "$colnam $typnam$notnull";
	}
	$sth->finish;
	return @ret;
}

sub lockfile {
	my ($action,$lockfile) = @_;
	if ($action eq 'create') {
		die "ERROR: Lockfile $lockfile exists - cannot continue" if -f $lockfile;
		open LOCK, ">", "$lockfile" or die "ERROR: Could not open lockfile $lockfile: $!";
		print LOCK $$;
		close LOCK;
	} elsif ($action eq 'clear') {
		if (-f $lockfile) {
			unlink $lockfile or die "ERROR: Could not remove lockfile $lockfile: $!";
		} else {
			warn "WARNING: Lockfile $lockfile does not exist - cannot clear" unless -f $lockfile;
		}
	}
}

1;

__END__