view release on metacpan or  search on metacpan

lib/DBD/PgLite/MirrorPgToSQLite.pm  view on Meta::CPAN

		&& $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);