Open-ILS/src/extras/import/parallel_pg_loader.pl

   1 #!/usr/bin/perl
   2 use strict;
   3 use warnings;
   4
   5 use OpenSRF::System;
   6 use OpenSRF::EX qw/:try/;
   7 use OpenSRF::Utils::SettingsClient;
   8 use OpenILS::Utils::Fieldmapper;
   9 use OpenSRF::Utils::JSON;
  10 use FileHandle;
  11
  12 use Time::HiRes qw/time/;
  13 use Getopt::Long;
  14
  15 my @files;
  16 my ($config, $output, @auto, @order, @wipe) =
  17         ('/openils/conf/opensrf_core.xml', 'pg_loader-output');
  18 my $nocommit = 0;
  19
  20 GetOptions(
  21         'config=s'      => \$config,
  22         'output=s'      => \$output,
  23         'wipe=s'        => \@wipe,
  24         'autoprimary=s' => \@auto,
  25         'order=s'       => \@order,
  26         'nocommit=i'    => \$nocommit,
  27 );
  28
  29 my $pwd = `pwd`;
  30 chop($pwd);
  31
  32 my %lineset;
  33 my %fieldcache;
  34
  35 OpenSRF::System->bootstrap_client( config_file => $config );
  36 Fieldmapper->import(IDL => OpenSRF::Utils::SettingsClient->new->config_value("IDL"));
  37
  38 my $main_out = FileHandle->new(">$output.sql") if ($output);
  39
  40 binmode($main_out,'utf8');
  41
  42 $main_out->print("SET CLIENT_ENCODING TO 'UNICODE';\n\n");
  43 $main_out->print("BEGIN;\n\n");
  44
  45 my %out_files;
  46 for my $h (@order) {
  47         $out_files{$h} = FileHandle->new(">$output.$h.sql");
  48         binmode($out_files{$h},'utf8');
  49 }
  50
  51 my $count = 0;
  52 my $starttime = time;
  53 my $after_commit = '';
  54 while ( my $rec = <> ) {
  55         next unless ($rec);
  56
  57         my $row;
  58         try {
  59                 $row = OpenSRF::Utils::JSON->JSON2perl($rec);
  60         } catch Error with {
  61                 my $e = shift;
  62                 warn "\n\n !!! Error : $e \n\n at or around line $count\n";
  63         };
  64         next unless ($row);
  65
  66         my $class = $row->class_name;
  67         my $hint = $row->json_hint;
  68
  69         next unless ( grep /$hint/, @order );
  70
  71         if (!$fieldcache{$hint}) {
  72                 my @cols = $row->real_fields;
  73                 if (grep { $_ eq $hint} @auto) {
  74                         @cols = grep { $_ ne $class->Identity } @cols;
  75                 }
  76
  77                 $fieldcache{$hint} =
  78                         { table => $class->Table,
  79                           sequence => $class->Sequence,
  80                           pkey => $class->Identity,
  81                           fields => \@cols,
  82                         };
  83
  84         #XXX it burnnnsssessss
  85         $fieldcache{$hint}{table} =~ s/\.full_rec/.real_full_rec/o if ($hint eq 'mfr');
  86
  87                 my $fields = join(',', @{ $fieldcache{$hint}{fields} });
  88                 $main_out->print( "DELETE FROM $fieldcache{$hint}{table};\n" ) if (grep {$_ eq $hint } @wipe);
  89                 # Speed up loading of bib records
  90                 $main_out->print( "COPY $fieldcache{$hint}{table} ($fields) FROM '$pwd/$output.$hint.sql';\n" );
  91
  92         }
  93
  94         my $line = [map { $row->$_ } @{ $fieldcache{$hint}{fields} }];
  95         my @data;
  96         my $x = 0;
  97         for my $d (@$line) {
  98                 if (!defined($d)) {
  99                         $d = '\N';
 100                 } else {
 101                         $d =~ s/\f/\\f/gos;
 102                         $d =~ s/\n/\\n/gos;
 103                         $d =~ s/\r/\\r/gos;
 104                         $d =~ s/\t/\\t/gos;
 105                         $d =~ s/\\/\\\\/gos;
 106                 }
 107                 if ($hint eq 'bre' and $fieldcache{$hint}{fields}[$x] eq 'quality') {
 108                         $d = int($d) if ($d ne '\N');
 109                 }
 110                 push @data, $d;
 111                 $x++;
 112         }
 113         $out_files{$hint}->print( join("\t", @data)."\n" );
 114
 115         if (!($count % 500)) {
 116                 print STDERR "\r$count\t". $count / (time - $starttime);
 117         }
 118
 119         $count++;
 120 }
 121
 122 for my $hint (@order) {
 123     next if (grep { $_ eq $hint} @auto);
 124     next unless ($fieldcache{$hint}{sequence});
 125     $after_commit .= "SELECT setval('$fieldcache{$hint}{sequence}'::TEXT, (SELECT MAX($fieldcache{$hint}{pkey}) FROM $fieldcache{$hint}{table}), TRUE);\n";
 126 }
 127
 128 if (grep /^mfr$/, %out_files) {
 129         $main_out->print("SELECT reporter.enable_materialized_simple_record_trigger();\n");
 130         $main_out->print("SELECT reporter.disable_materialized_simple_record_trigger();\n");
 131 }
 132
 133 $main_out->print("COMMIT;\n\n") unless $nocommit;
 134 $main_out->print($after_commit);
 135 $main_out->close;
 136